Merge branch 'main' of https://github.com/Killea/AgentChatBus

Author: Killea

src/db/crud.py

@@ -790,6 +790,13 @@ async def template_delete(db: aiosqlite.Connection, template_id: str) -> None:
 _VALID_PRIORITIES = {"normal", "urgent", "system"}
 
 
+async def msg_get(db: aiosqlite.Connection, message_id: str) -> Optional[Message]:
+    """Fetch a single message by ID. Returns None if not found."""
+    async with db.execute("SELECT * FROM messages WHERE id = ?", (message_id,)) as cur:
+        row = await cur.fetchone()
+    return _row_to_message(row) if row else None
+
+
 async def msg_post(
     db: aiosqlite.Connection,
     thread_id: str,
@@ -800,11 +807,25 @@ async def msg_post(
     role: str = "user",
     metadata: Optional[dict] = None,
     priority: str = "normal",
+    reply_to_msg_id: Optional[str] = None,
 ) -> Message:
     # Validate priority (UP-16)
     if priority not in _VALID_PRIORITIES:
         raise ValueError(f"Invalid priority '{priority}'. Must be one of: {', '.join(sorted(_VALID_PRIORITIES))}")
 
+    # Validate reply_to_msg_id (UP-14): must exist and belong to the same thread
+    if reply_to_msg_id is not None:
+        async with db.execute(
+            "SELECT thread_id FROM messages WHERE id = ?", (reply_to_msg_id,)
+        ) as cur:
+            parent_row = await cur.fetchone()
+        if parent_row is None:
+            raise ValueError(f"reply_to_msg_id '{reply_to_msg_id}' does not exist.")
+        if parent_row["thread_id"] != thread_id:
+            raise ValueError(
+                f"reply_to_msg_id '{reply_to_msg_id}' belongs to a different thread."
+            )
+
     # Content filter: block known secret patterns before any DB interaction
     if CONTENT_FILTER_ENABLED:
         blocked, pattern_name = check_content(content)
@@ -890,8 +911,8 @@ async def msg_post(
     seq = await next_seq(db)
     meta_json = json.dumps(metadata) if metadata else None
     await db.execute(
-        "INSERT INTO messages (id, thread_id, author, role, content, seq, created_at, metadata, author_id, author_name, priority) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
-        (mid, thread_id, actual_author, role, content, seq, now, meta_json, author_id, author_name, priority),
+        "INSERT INTO messages (id, thread_id, author, role, content, seq, created_at, metadata, author_id, author_name, priority, reply_to_msg_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (mid, thread_id, actual_author, role, content, seq, now, meta_json, author_id, author_name, priority, reply_to_msg_id),
     )
     await db.execute(
         "UPDATE threads SET updated_at = ? WHERE id = ?", (now, thread_id)
@@ -940,11 +961,19 @@ async def msg_post(
                 "msg_id": mid, "thread_id": thread_id,
                 "agent": author_name, "reason": stop_reason,
             })
-    logger.debug(f"Message posted: seq={seq} author={author_name} thread={thread_id} priority={priority}")
+    # SSE event for reply-to threading (UP-14)
+    if reply_to_msg_id is not None:
+        await _emit_event(db, "msg.reply", thread_id, {
+            "msg_id": mid, "reply_to_msg_id": reply_to_msg_id,
+            "thread_id": thread_id, "author": author_name, "seq": seq,
+        })
+
+    logger.debug(f"Message posted: seq={seq} author={author_name} thread={thread_id} priority={priority} reply_to={reply_to_msg_id}")
     return Message(
         id=mid, thread_id=thread_id, author=actual_author, role=role,
         content=content, seq=seq, created_at=_parse_dt(now), metadata=meta_json,
-        author_id=author_id, author_name=author_name, priority=priority
+        author_id=author_id, author_name=author_name, priority=priority,
+        reply_to_msg_id=reply_to_msg_id,
     )
 
 
@@ -1072,6 +1101,7 @@ def _row_to_message(row: aiosqlite.Row) -> Message:
     if not author_name:
         author_name = row["author"]
     priority = row["priority"] if "priority" in keys else "normal"
+    reply_to_msg_id = row["reply_to_msg_id"] if "reply_to_msg_id" in keys else None
 
     return Message(
         id=row["id"],
@@ -1085,6 +1115,7 @@ def _row_to_message(row: aiosqlite.Row) -> Message:
         author_id=author_id,
         author_name=author_name,
         priority=priority,
+        reply_to_msg_id=reply_to_msg_id,
     )
 
 
@@ -1636,9 +1667,9 @@ async def thread_export_markdown(db: aiosqlite.Connection, thread_id: str) -> Op
     return "\n".join(lines)
 
 
-# ─────────────────────────────────────────────
-# Metrics (UP-22)
-# ─────────────────────────────────────────────
+# ──────────────────────────────────────────────────────────────────────────────
+# UP-22: Bus-level observability metrics
+# ──────────────────────────────────────────────────────────────────────────────
 
 async def get_bus_metrics(db: aiosqlite.Connection) -> dict:
     """Return a snapshot of bus-level observability metrics.
@@ -1664,21 +1695,20 @@ async def get_bus_metrics(db: aiosqlite.Connection) -> dict:
                              AGENT_HEARTBEAT_TIMEOUT window
     """
     now = datetime.now(timezone.utc)
-    now_iso = now.isoformat()
 
-    # ── Thread counts ────────────────────────────────────────────────────────
+    # ── Thread counts ──────────────────────────────────────────────────────────
     threads_by_status: dict[str, int] = {}
     async with db.execute("SELECT status, COUNT(*) AS cnt FROM threads GROUP BY status") as cur:
         async for row in cur:
             threads_by_status[row["status"]] = row["cnt"]
     threads_total = sum(threads_by_status.values())
 
-    # ── Message total ────────────────────────────────────────────────────────
+    # ── Message total ──────────────────────────────────────────────────────────
     async with db.execute("SELECT COUNT(*) AS cnt FROM messages") as cur:
         row = await cur.fetchone()
     messages_total = row["cnt"] if row else 0
 
-    # ── Message rates (1m / 5m / 15m) ───────────────────────────────────────
+    # ── Message rates (1m / 5m / 15m) ─────────────────────────────────────────
     cutoffs = {
         "last_1m":  (now - timedelta(minutes=1)).isoformat(),
         "last_5m":  (now - timedelta(minutes=5)).isoformat(),
@@ -1692,11 +1722,9 @@ async def get_bus_metrics(db: aiosqlite.Connection) -> dict:
             row = await cur.fetchone()
         message_rate[key] = row["cnt"] if row else 0
 
-    # ── Inter-message latency (avg ms, threads active in last 15 min) ────────
+    # ── Inter-message latency (avg ms, threads active in last 15 min) ─────────
     # Uses LAG() window function (SQLite >= 3.25.0) to compute time gaps
     # between consecutive messages within each thread, then averages them.
-    # Only considers threads that had activity in the last 15 minutes to keep
-    # the metric relevant to current bus load.
     cutoff_15m = cutoffs["last_15m"]
     avg_latency_ms: Optional[float] = None
     try:
@@ -1718,10 +1746,9 @@ async def get_bus_metrics(db: aiosqlite.Connection) -> dict:
         if row and row["avg_gap"] is not None:
             avg_latency_ms = round(row["avg_gap"], 1)
     except Exception:
-        # Defensive: if the window query fails for any reason, degrade gracefully
         avg_latency_ms = None
 
-    # ── stop_reason distribution (UP-17) ─────────────────────────────────────
+    # ── stop_reason distribution (UP-17) ──────────────────────────────────────
     canonical_reasons = ("convergence", "timeout", "complete", "error", "impasse")
     stop_reasons: dict[str, int] = {r: 0 for r in canonical_reasons}
     async with db.execute(
@@ -1736,7 +1763,7 @@ async def get_bus_metrics(db: aiosqlite.Connection) -> dict:
             reason = row["reason"]
             stop_reasons[reason] = stop_reasons.get(reason, 0) + row["cnt"]
 
-    # ── Agent counts ─────────────────────────────────────────────────────────
+    # ── Agent counts ───────────────────────────────────────────────────────────
     agents_total = 0
     agents_online = 0
     heartbeat_cutoff = (

src/db/database.py

@@ -486,6 +486,17 @@ async def init_schema(db: aiosqlite.Connection) -> None:
     # Migration: Add priority column to messages (UP-16)
     await _add_column_if_missing(db, "messages", "priority", "TEXT NOT NULL DEFAULT 'normal'")
 
+    # Migration: Add reply_to_msg_id column to messages (UP-14)
+    await _add_column_if_missing(db, "messages", "reply_to_msg_id", "TEXT")
+    try:
+        await db.execute(
+            "CREATE INDEX IF NOT EXISTS idx_messages_reply ON messages(reply_to_msg_id)"
+        )
+        await db.commit()
+        logger.info("Migration: ensured reply_to_msg_id column + index exist (UP-14)")
+    except Exception as e:
+        logger.error(f"Migration failed for reply_to_msg_id index: {e}")
+
     # Migration: Create reactions table if it does not exist (UP-13)
     # Safe for existing DBs — CREATE TABLE IF NOT EXISTS + CREATE UNIQUE INDEX IF NOT EXISTS
     try:

src/db/models.py

@@ -45,6 +45,7 @@ class Message:
     metadata: Optional[str]  # JSON string
     author_id: Optional[str] = None
     author_name: Optional[str] = None
+    reply_to_msg_id: Optional[str] = None
     priority: str = "normal"  # normal | urgent | system (UP-16)
 
 

src/main.py

@@ -106,7 +106,7 @@ def _agent_label(agent: object | None, fallback_id: str | None = None) -> str:
 # Support environment variable override via AGENTCHATBUS_DB_TIMEOUT
 DB_TIMEOUT = int(os.getenv("AGENTCHATBUS_DB_TIMEOUT", "5"))
 
-# Server start time — set in lifespan(), used by /api/metrics
+# Server start time — set in lifespan(), used by /api/metrics (UP-22)
 _start_time: datetime | None = None
 
 
@@ -599,6 +599,7 @@ async def api_messages(
             "created_at": m.created_at.isoformat(),
             "metadata": m.metadata,
             "priority": m.priority,
+            "reply_to_msg_id": m.reply_to_msg_id,
             "reactions": reactions_map.get(m.id, []),
         }
         for m in msgs
@@ -850,6 +851,7 @@ class MessageCreate(BaseModel):
     metadata: dict | None = None
     images: list[dict] | None = None  # [{url: str, name: str}, ...]
     priority: Literal["normal", "urgent", "system"] = "normal"  # UP-16
+    reply_to_msg_id: str | None = None  # UP-14: optional parent message ID
 
 class SyncContextRequest(BaseModel):
     agent_id: str | None = None
@@ -1079,7 +1081,8 @@ async def api_post_message(thread_id: str, body: MessageCreate, x_agent_token: s
                          reply_token=reply_token,
                          role=body.role,
                          metadata=msg_metadata if msg_metadata else None,
-                         priority=body.priority),
+                         priority=body.priority,
+                         reply_to_msg_id=body.reply_to_msg_id),
             timeout=DB_TIMEOUT
         )
     except MissingSyncFieldsError as e:
@@ -1132,7 +1135,7 @@ async def api_post_message(thread_id: str, body: MessageCreate, x_agent_token: s
     # Return the full message with metadata
     result = {"id": m.id, "seq": m.seq, "author": m.author,
             "role": m.role, "content": m.content, "created_at": m.created_at.isoformat(),
-            "priority": m.priority}
+            "priority": m.priority, "reply_to_msg_id": m.reply_to_msg_id}
 
     # Add metadata (includes mentions and images)
     if m.metadata:
@@ -1720,12 +1723,9 @@ async def api_thread_admin_decision(thread_id: str, body: AdminDecisionRequest):
 
 
 # ─────────────────────────────────────────────
-# Health check
-# ΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇ
-
-# ─────────────────────────────────────────────
+# ──────────────────────────────────────────────────────────────────────────────
 # Metrics (UP-22)
-# ─────────────────────────────────────────────
+# ──────────────────────────────────────────────────────────────────────────────
 
 @app.get("/api/metrics")
 async def get_metrics():
@@ -1755,6 +1755,9 @@ async def get_metrics():
     }
 
 
+# Health check
+# ──────────────────────────────────────────────────────────────────────────────
+
 @app.get("/health")
 async def health():
     return {"status": "ok", "service": "AgentChatBus"}

src/mcp_server.py

@@ -231,6 +231,10 @@ async def list_tools() -> list[types.Tool]:
                             },
                         },
                     },
+                    "reply_to_msg_id": {
+                        "type": "string",
+                        "description": "Optional ID of the message being replied to. Must belong to the same thread. Triggers a msg.reply SSE event.",
+                    },
                 },
                 "required": ["thread_id", "author", "content", "expected_last_seq", "reply_token"],
             },

src/static/css/main.css

@@ -869,6 +869,28 @@ body[data-theme="light"] .handoff-badge {
   color: #2563eb;
 }
 
+/* Reply-to quote block (UP-14) */
+.msg-reply-quote {
+  display: inline-block;
+  font-size: 0.78em;
+  padding: 2px 8px;
+  margin-bottom: 4px;
+  border-left: 3px solid rgba(150,150,255,0.6);
+  background: rgba(150,150,255,0.08);
+  border-radius: 0 4px 4px 0;
+  color: rgba(200,200,220,0.8);
+  font-style: normal;
+  max-width: 100%;
+  white-space: nowrap;
+  overflow: hidden;
+  text-overflow: ellipsis;
+}
+body[data-theme="light"] .msg-reply-quote {
+  border-left-color: rgba(80,80,200,0.4);
+  background: rgba(80,80,200,0.06);
+  color: #555;
+}
+
 body[data-theme="light"] .stop-tag {
   background: rgba(100,100,100,0.08);
   color: #555;

src/static/index.html

@@ -800,6 +800,11 @@
         priorityBadgeHtml = '<span class="msg-priority-badge msg-priority-system" title="System message">SYSTEM</span>';
       }
 
+      // Reply-to quote (UP-14)
+      const replyQuoteHtml = m.reply_to_msg_id
+        ? `<div class="msg-reply-quote" title="In reply to message ${esc(m.reply_to_msg_id)}">&#8617; In reply to: <em>${esc(m.reply_to_msg_id.slice(0, 8))}…</em></div>`
+        : '';
+
       row.innerHTML = `
       <div class="msg-avatar" style="background:${bgAlpha};color:${color};border:1px solid ${color}44">${avatarEmoji}</div>
       <div class="msg-col">
@@ -809,6 +814,7 @@
           ${priorityBadgeHtml}
           ${handoffBadgeHtml}
         </div>
+        ${replyQuoteHtml}
         <div class="bubble-v2" style="border-color:${color}55;${isHuman ? 'border-right-color' : 'border-left-color'}:${color}"></div>
         ${isAgentMessage ? '<acb-message-tail-meta></acb-message-tail-meta>' : ''}
         <div class="msg-reactions"></div>

src/tools/dispatch.py

@@ -223,7 +223,7 @@ async def handle_thread_create(db, arguments: dict[str, Any]) -> list[types.Text
             agent_info = await crud.agent_get(db, agent_id)
             if agent_info:
                 await crud.thread_settings_set_creator_admin(db, result.id, agent_id, agent_info.name)
-    
+
     token_payload = await crud.issue_reply_token(db, thread_id=result.id, agent_id=agent_id)
 
     return [types.TextContent(type="text", text=json.dumps({
@@ -346,6 +346,7 @@ async def handle_msg_post(db, arguments: dict[str, Any]) -> list[types.TextConte
             role=arguments.get("role", "user"),
             metadata=arguments.get("metadata"),
             priority=arguments.get("priority", "normal"),
+            reply_to_msg_id=arguments.get("reply_to_msg_id"),
         )
 
         # Agent posted a message - exit wait state for this thread
@@ -395,9 +396,19 @@ async def handle_msg_post(db, arguments: dict[str, Any]) -> list[types.TextConte
             "error": "Content blocked by filter",
             "pattern": e.pattern_name,
         }))]
+    except ValueError as e:
+        return [types.TextContent(type="text", text=json.dumps({
+            "error": "INVALID_ARGUMENT",
+            "detail": str(e),
+        }))]
 
     meta = _safe_json_loads(msg.metadata)
-    result: dict[str, Any] = {"msg_id": msg.id, "seq": msg.seq, "priority": msg.priority}
+    result: dict[str, Any] = {
+        "msg_id": msg.id,
+        "seq": msg.seq,
+        "priority": msg.priority,
+        "reply_to_msg_id": msg.reply_to_msg_id,
+    }
     if isinstance(meta, dict):
         if meta.get("handoff_target"):
             result["handoff_target"] = meta["handoff_target"]
@@ -438,6 +449,7 @@ async def handle_msg_list(db, arguments: dict[str, Any]) -> list[types.Content]:
             "created_at": m.created_at.isoformat(),
             "metadata": m.metadata,
             "priority": m.priority,
+            "reply_to_msg_id": m.reply_to_msg_id,
             "reactions": reactions_map.get(m.id, []),
         }
         for m in msgs