feat: Implement acb-compose-shell custom element for chat composition, enabling mentions and image uploads, and update chat message handling.

Author: Killea

src/config.py

@@ -40,6 +40,11 @@
 MSG_WAIT_TIMEOUT = int(os.getenv("AGENTCHATBUS_WAIT_TIMEOUT", config_data.get("MSG_WAIT_TIMEOUT", "300")))
 BUS_VERSION = "0.1.0"
 
+# Strict message sync mode (mandatory)
+REPLY_TOKEN_LEASE_SECONDS = int(os.getenv("AGENTCHATBUS_REPLY_TOKEN_LEASE_SECONDS", "10"))
+SEQ_TOLERANCE = int(os.getenv("AGENTCHATBUS_SEQ_TOLERANCE", "5"))
+SEQ_MISMATCH_MAX_MESSAGES = int(os.getenv("AGENTCHATBUS_SEQ_MISMATCH_MAX_MESSAGES", "20"))
+
 # Rate limiting: max messages per minute per author identity (0 = disabled)
 RATE_LIMIT_MSG_PER_MINUTE = int(os.getenv("AGENTCHATBUS_RATE_LIMIT", "30"))
 RATE_LIMIT_ENABLED = RATE_LIMIT_MSG_PER_MINUTE > 0
@@ -62,6 +67,9 @@ def get_config_dict():
         "PORT": PORT,
         "AGENT_HEARTBEAT_TIMEOUT": AGENT_HEARTBEAT_TIMEOUT,
         "MSG_WAIT_TIMEOUT": MSG_WAIT_TIMEOUT,
+        "REPLY_TOKEN_LEASE_SECONDS": REPLY_TOKEN_LEASE_SECONDS,
+        "SEQ_TOLERANCE": SEQ_TOLERANCE,
+        "SEQ_MISMATCH_MAX_MESSAGES": SEQ_MISMATCH_MAX_MESSAGES,
         "EXPOSE_THREAD_RESOURCES": EXPOSE_THREAD_RESOURCES,
     }
 

src/db/crud.py

@@ -13,8 +13,15 @@
 import aiosqlite
 
 from src.db.models import Thread, Message, AgentInfo, Event
-from src.config import AGENT_HEARTBEAT_TIMEOUT, RATE_LIMIT_MSG_PER_MINUTE, RATE_LIMIT_ENABLED
-from src.config import AGENT_HEARTBEAT_TIMEOUT, CONTENT_FILTER_ENABLED
+from src.config import (
+    AGENT_HEARTBEAT_TIMEOUT,
+    RATE_LIMIT_MSG_PER_MINUTE,
+    RATE_LIMIT_ENABLED,
+    CONTENT_FILTER_ENABLED,
+    REPLY_TOKEN_LEASE_SECONDS,
+    SEQ_TOLERANCE,
+    SEQ_MISMATCH_MAX_MESSAGES,
+)
 from src.content_filter import check_content, ContentFilterError
 
 logger = logging.getLogger(__name__)
@@ -30,6 +37,46 @@ def __init__(self, limit: int, window: int, retry_after: int, scope: str) -> Non
         super().__init__(f"Rate limit exceeded: {limit} messages/{window}s")
 
 
+class MissingSyncFieldsError(Exception):
+    """Raised when strict sync fields are absent from msg_post."""
+
+    def __init__(self, missing_fields: list[str]) -> None:
+        self.missing_fields = missing_fields
+        super().__init__(f"Missing required sync fields: {', '.join(missing_fields)}")
+
+
+class SeqMismatchError(Exception):
+    """Raised when too many unseen messages exist since expected seq."""
+
+    def __init__(self, expected_last_seq: int, current_seq: int, new_messages: list[dict]) -> None:
+        self.expected_last_seq = expected_last_seq
+        self.current_seq = current_seq
+        self.new_messages = new_messages
+        super().__init__(
+            f"SEQ_MISMATCH: expected_last_seq={expected_last_seq}, current_seq={current_seq}"
+        )
+
+
+class ReplyTokenInvalidError(Exception):
+    def __init__(self, token: str) -> None:
+        self.token = token
+        super().__init__("TOKEN_INVALID")
+
+
+class ReplyTokenExpiredError(Exception):
+    def __init__(self, token: str, expires_at: str) -> None:
+        self.token = token
+        self.expires_at = expires_at
+        super().__init__("TOKEN_EXPIRED")
+
+
+class ReplyTokenReplayError(Exception):
+    def __init__(self, token: str, consumed_at: Optional[str]) -> None:
+        self.token = token
+        self.consumed_at = consumed_at
+        super().__init__("TOKEN_REPLAY")
+
+
 GLOBAL_SYSTEM_PROMPT = """**SYSTEM DIRECTIVE: ACTIVE AGENT COLLABORATION WORKSPACE**
 
 Welcome to this Thread. You are participating in a multi-agent workspace sharing the same underlying codebase and execution environment. You MUST collaborate proactively and keep progress moving.
@@ -233,6 +280,69 @@ async def thread_latest_seq(db: aiosqlite.Connection, thread_id: str) -> int:
     return row["max_seq"] or 0
 
 
+async def _expire_old_reply_tokens(db: aiosqlite.Connection) -> None:
+    now = _now()
+    await db.execute(
+        "UPDATE reply_tokens SET status = 'expired' "
+        "WHERE status = 'issued' AND expires_at <= ?",
+        (now,),
+    )
+    await db.commit()
+
+
+async def issue_reply_token(
+    db: aiosqlite.Connection,
+    thread_id: str,
+    agent_id: Optional[str] = None,
+) -> dict:
+    """Issue a short-lived reply token bound to a thread (and optionally an agent)."""
+    await _expire_old_reply_tokens(db)
+    token = secrets.token_urlsafe(24)
+    issued_at = _now()
+    expires_at = (datetime.now(timezone.utc) + timedelta(seconds=REPLY_TOKEN_LEASE_SECONDS)).isoformat()
+    await db.execute(
+        "INSERT INTO reply_tokens (token, thread_id, agent_id, issued_at, expires_at, consumed_at, status) "
+        "VALUES (?, ?, ?, ?, ?, NULL, 'issued')",
+        (token, thread_id, agent_id, issued_at, expires_at),
+    )
+    await db.commit()
+    current_seq = await thread_latest_seq(db, thread_id)
+    return {
+        "reply_token": token,
+        "current_seq": current_seq,
+        "reply_window": {
+            "expires_at": expires_at,
+            "max_new_messages": SEQ_TOLERANCE,
+        },
+    }
+
+
+async def _get_new_messages_since(
+    db: aiosqlite.Connection,
+    thread_id: str,
+    expected_last_seq: int,
+    limit: int = SEQ_MISMATCH_MAX_MESSAGES,
+) -> list[dict]:
+    msgs = await msg_list(
+        db,
+        thread_id=thread_id,
+        after_seq=expected_last_seq,
+        limit=limit,
+        include_system_prompt=False,
+    )
+    return [
+        {
+            "msg_id": m.id,
+            "seq": m.seq,
+            "author": m.author,
+            "role": m.role,
+            "content": m.content,
+            "created_at": m.created_at.isoformat(),
+        }
+        for m in msgs
+    ]
+
+
 def _row_to_thread(row: aiosqlite.Row) -> Thread:
     system_prompt = row["system_prompt"] if "system_prompt" in row.keys() else None
     updated_at = _parse_dt(row["updated_at"]) if "updated_at" in row.keys() and row["updated_at"] else None
@@ -258,6 +368,8 @@ async def msg_post(
     thread_id: str,
     author: str,
     content: str,
+    expected_last_seq: int,
+    reply_token: str,
     role: str = "user",
     metadata: Optional[dict] = None,
 ) -> Message:
@@ -310,6 +422,41 @@ async def msg_post(
                 scope=scope,
             )
 
+    missing_fields: list[str] = []
+    if expected_last_seq is None:
+        missing_fields.append("expected_last_seq")
+    if not reply_token:
+        missing_fields.append("reply_token")
+    if missing_fields:
+        raise MissingSyncFieldsError(missing_fields)
+
+    await _expire_old_reply_tokens(db)
+    async with db.execute(
+        "SELECT token, thread_id, agent_id, expires_at, consumed_at, status "
+        "FROM reply_tokens WHERE token = ?",
+        (reply_token,),
+    ) as cur:
+        token_row = await cur.fetchone()
+
+    if token_row is None:
+        raise ReplyTokenInvalidError(reply_token)
+    if token_row["thread_id"] != thread_id:
+        raise ReplyTokenInvalidError(reply_token)
+    if token_row["status"] == "consumed":
+        raise ReplyTokenReplayError(reply_token, token_row["consumed_at"])
+    if token_row["status"] == "expired":
+        raise ReplyTokenExpiredError(reply_token, token_row["expires_at"])
+
+    token_agent_id = token_row["agent_id"]
+    if token_agent_id and author_id and token_agent_id != author_id:
+        raise ReplyTokenInvalidError(reply_token)
+
+    current_seq = await thread_latest_seq(db, thread_id)
+    new_messages_count = current_seq - expected_last_seq
+    if new_messages_count > SEQ_TOLERANCE:
+        new_messages = await _get_new_messages_since(db, thread_id, expected_last_seq)
+        raise SeqMismatchError(expected_last_seq, current_seq, new_messages)
+
     mid = str(uuid.uuid4())
     now = _now()
     seq = await next_seq(db)
@@ -323,6 +470,22 @@ async def msg_post(
         "UPDATE threads SET updated_at = ? WHERE id = ?",
         (now, thread_id),
     )
+    async with db.execute(
+        "UPDATE reply_tokens SET status = 'consumed', consumed_at = ? "
+        "WHERE token = ? AND status = 'issued'",
+        (now, reply_token),
+    ) as cur:
+        consumed = cur.rowcount
+    if consumed == 0:
+        await db.rollback()
+        async with db.execute(
+            "SELECT consumed_at FROM reply_tokens WHERE token = ?",
+            (reply_token,),
+        ) as cur:
+            row = await cur.fetchone()
+        consumed_at = row["consumed_at"] if row else None
+        raise ReplyTokenReplayError(reply_token, consumed_at)
+
     await db.commit()
     if author_id:
         await agent_msg_post(db, author_id)

src/db/database.py

@@ -150,6 +150,24 @@ async def init_schema(db: aiosqlite.Connection) -> None:
         CREATE INDEX IF NOT EXISTS idx_messages_thread_seq
             ON messages(thread_id, seq);
 
+        -- ----------------------------------------------------------------
+        -- Reply token lease: mandatory sync token for msg_post in strict mode
+        -- ----------------------------------------------------------------
+        CREATE TABLE IF NOT EXISTS reply_tokens (
+            token       TEXT PRIMARY KEY,
+            thread_id   TEXT NOT NULL REFERENCES threads(id),
+            agent_id    TEXT,
+            issued_at   TEXT NOT NULL,
+            expires_at  TEXT NOT NULL,
+            consumed_at TEXT,
+            status      TEXT NOT NULL CHECK (status IN ('issued', 'consumed', 'expired'))
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_reply_tokens_thread_status
+            ON reply_tokens(thread_id, status);
+        CREATE INDEX IF NOT EXISTS idx_reply_tokens_expires_at
+            ON reply_tokens(expires_at);
+
         -- ----------------------------------------------------------------
         -- Sequence counter: single-row table for thread-safe seq increment
         -- ----------------------------------------------------------------

src/main.py

@@ -28,7 +28,14 @@
 from src.config import HOST, PORT, get_config_dict, save_config_dict
 from src.db.database import get_db, close_db
 from src.db import crud
-from src.db.crud import RateLimitExceeded
+from src.db.crud import (
+    RateLimitExceeded,
+    MissingSyncFieldsError,
+    SeqMismatchError,
+    ReplyTokenInvalidError,
+    ReplyTokenExpiredError,
+    ReplyTokenReplayError,
+)
 from src.config import THREAD_TIMEOUT_ENABLED, THREAD_TIMEOUT_MINUTES, THREAD_TIMEOUT_SWEEP_INTERVAL, RELOAD_ENABLED
 from src.mcp_server import server as mcp_server, _session_language
 from src.content_filter import ContentFilterError
@@ -385,10 +392,34 @@ class MessageCreate(BaseModel):
     author: str = "human"
     role: Literal["user", "assistant", "system"] = "user"
     content: str
+    expected_last_seq: int
+    reply_token: str
     mentions: list[str] | None = None
     metadata: dict | None = None
     images: list[dict] | None = None  # [{url: str, name: str}, ...]
 
+
+class SyncContextRequest(BaseModel):
+    agent_id: str | None = None
+
+
+@app.post("/api/threads/{thread_id}/sync-context")
+async def api_sync_context(thread_id: str, body: SyncContextRequest | None = None):
+    try:
+        db = await asyncio.wait_for(get_db(), timeout=DB_TIMEOUT)
+        t = await asyncio.wait_for(crud.thread_get(db, thread_id), timeout=DB_TIMEOUT)
+    except asyncio.TimeoutError:
+        raise HTTPException(status_code=503, detail="Database operation timeout")
+    if t is None:
+        raise HTTPException(status_code=404, detail="Thread not found")
+
+    agent_id = body.agent_id if body else None
+    sync = await asyncio.wait_for(
+        crud.issue_reply_token(db, thread_id=thread_id, agent_id=agent_id),
+        timeout=DB_TIMEOUT,
+    )
+    return sync
+
 @app.post("/api/threads", status_code=201)
 async def api_create_thread(body: ThreadCreate):
     try:
@@ -421,10 +452,44 @@ async def api_post_message(thread_id: str, body: MessageCreate):
     try:
         m = await asyncio.wait_for(
             crud.msg_post(db, thread_id=thread_id, author=body.author,
-                         content=body.content, role=body.role,
+                         content=body.content,
+                         expected_last_seq=body.expected_last_seq,
+                         reply_token=body.reply_token,
+                         role=body.role,
                          metadata=msg_metadata if msg_metadata else None),
             timeout=DB_TIMEOUT
         )
+    except MissingSyncFieldsError as e:
+        raise HTTPException(status_code=400, detail={
+            "error": "MISSING_SYNC_FIELDS",
+            "missing_fields": e.missing_fields,
+            "action": "CALL_SYNC_CONTEXT_THEN_RETRY",
+        })
+    except SeqMismatchError as e:
+        raise HTTPException(status_code=409, detail={
+            "error": "SEQ_MISMATCH",
+            "expected_last_seq": e.expected_last_seq,
+            "current_seq": e.current_seq,
+            "new_messages": e.new_messages,
+            "action": "RE_READ_AND_RETRY",
+        })
+    except ReplyTokenInvalidError:
+        raise HTTPException(status_code=400, detail={
+            "error": "TOKEN_INVALID",
+            "action": "CALL_SYNC_CONTEXT_THEN_RETRY",
+        })
+    except ReplyTokenExpiredError as e:
+        raise HTTPException(status_code=400, detail={
+            "error": "TOKEN_EXPIRED",
+            "expires_at": e.expires_at,
+            "action": "CALL_SYNC_CONTEXT_THEN_RETRY",
+        })
+    except ReplyTokenReplayError as e:
+        raise HTTPException(status_code=400, detail={
+            "error": "TOKEN_REPLAY",
+            "consumed_at": e.consumed_at,
+            "action": "CALL_SYNC_CONTEXT_THEN_RETRY",
+        })
     except ContentFilterError as e:
         raise HTTPException(status_code=400, detail={"error": "Content blocked by filter", "pattern": e.pattern_name})
     except asyncio.TimeoutError:

src/mcp_server.py

@@ -158,6 +158,14 @@ async def list_tools() -> list[types.Tool]:
                     "thread_id": {"type": "string"},
                     "author":    {"type": "string", "description": "Agent ID, 'system', or 'human'."},
                     "content":   {"type": "string"},
+                    "expected_last_seq": {
+                        "type": "integer",
+                        "description": "Strict sync field. Thread seq the sender used as context baseline.",
+                    },
+                    "reply_token": {
+                        "type": "string",
+                        "description": "Strict sync field. Unconsumed reply token from msg_wait.",
+                    },
                     "role":      {"type": "string", "enum": ["user", "assistant", "system"], "default": "user"},
                     "mentions":  {
                         "type": "array",
@@ -166,7 +174,7 @@ async def list_tools() -> list[types.Tool]:
                     },
                     "metadata":  {"type": "object"},
                 },
-                "required": ["thread_id", "author", "content"],
+                "required": ["thread_id", "author", "content", "expected_last_seq", "reply_token"],
             },
         ),
         types.Tool(
@@ -202,6 +210,8 @@ async def list_tools() -> list[types.Tool]:
             description=(
                 "Block until at least one new message arrives in the thread after `after_seq`. "
                 "Returns immediately if messages are already available. "
+                "Always includes sync context (`current_seq`, `reply_token`, `reply_window`) "
+                "for the next strict `msg_post` call. "
                 "If this tool returns an empty list (timeout), avoid spammy waiting messages, "
                 "but after repeated timeouts you SHOULD send a concise, meaningful progress update "
                 "(status/blocker/next action) and optionally @mention a relevant online agent."