fix-forward #3942 (tsk-bnohyn): agent reply route returns 202 when the trace_id message lookup raises (CodeRabbit Major, pre-existing)

changelog.d/3942-agent-chat-reply-trace-lookup-503.md

@@ -0,0 +1,2 @@
+### Fixed
+- Agent chat reply trace_id lookup failures now return 503 instead of falling back to self-DM (tsk-4wadcj)

changelog.d/tsk-bnohyn-agent-chat-reply-fail-closed.md

@@ -0,0 +1,2 @@
+### Fixed
+- fail closed on registry errors and unknown channels in the agent reply route

changelog.d/tsk-wfkhb2-agent-chat-session-fix.md

@@ -0,0 +1,4 @@
+### Fixed
+
+- Agent chat session reply endpoint now enforces channel membership: a registry agent can only reply in channels where its slug is listed as a member. Requests targeting a foreign channel (via `channel_id` or `trace_id`) return 403 `{"error": "not a member of this channel"}`.
+- Registry agent slug collision prevention: when two active registry agents have canonical IDs that strip to the same slug, only the first-registered agent may use the session; subsequent agents receive 403 `{"error": "handle collides with another registry agent"}`.

docs/agent-coordination.md

@@ -2434,6 +2434,14 @@ never drain or answer a local agent's session. A request with no Bearer is refus
 middleware before the route runs (401); a JWT without the grant gets 403; an
 unavailable bridge registry gets 503.
 
+A reply targeting a channel (via explicit `channel_id` or via `trace_id`
+resolving to a message's channel) is accepted only when the agent's slug is
+a member of that channel; otherwise 403 `not a member of this channel` is
+returned and the reply is not recorded. Two registry agents whose canonical
+IDs strip to the same slug cannot share a session: only the first-registered
+agent may use it; subsequent agents receive 403 `handle collides with another
+registry agent`.
+
 ## In-process LLM gateway (`/api/llm/v1`, scoped gateway keys, session or host local token)
 
 `tinyagentos/llm_gateway/` is the in-controller replacement for the LiteLLM

tests/test_agent_chat_session_routes.py

@@ -14,7 +14,9 @@
 
 from tinyagentos.agent_registry_store import mint_registry_token
 from tinyagentos.bridge_session import BridgeSessionRegistry
+from tinyagentos.chat.channel_store import ChatChannelStore
 from tinyagentos.chat.guest_outbox import GuestOutbox
+from tinyagentos.chat.message_store import ChatMessageStore
 from taos_test_csrf import csrf_event_hooks
 
 EVENTS = "/api/agents/self/chat/events"
@@ -42,6 +44,39 @@ async def chat_app(app, tmp_data_dir):
             await store.close()
 
 
+@pytest_asyncio.fixture
+async def chat_app_with_stores(app, tmp_data_dir):
+    """Chat app with real chat_channels and chat_messages stores for membership tests."""
+    for attr in ("agent_registry", "agent_grants", "metrics"):
+        store = getattr(app.state, attr)
+        if store._db is None:
+            await store.init()
+    # Initialise real chat stores
+    chat_messages = app.state.chat_messages
+    chat_channels = app.state.chat_channels
+    if chat_messages._db is None:
+        await chat_messages.init()
+    if chat_channels._db is None:
+        await chat_channels.init()
+    app.state.bridge_sessions = BridgeSessionRegistry(
+        trace_registry=None,
+        chat_messages=chat_messages,
+        chat_channels=chat_channels,
+        chat_hub=None,
+        archive=None,
+    )
+    app.state._startup_complete = True
+    yield app
+    for attr in ("agent_registry", "agent_grants", "metrics"):
+        store = getattr(app.state, attr)
+        if store._db is not None:
+            await store.close()
+    if chat_messages._db is not None:
+        await chat_messages.close()
+    if chat_channels._db is not None:
+        await chat_channels.close()
+
+
 async def _agent(app, handle, *, scopes=("chat_session",), display_name=None):
     registry = app.state.agent_registry
     priv, _pub = app.state.agent_registry_keypair
@@ -210,6 +245,126 @@ async def test_body_identity_is_ignored(self, chat_app):
         assert "guest-b" not in slugs
 
 
+@pytest.mark.asyncio
+class TestChatSessionReplyMembership:
+    """Tests for channel membership enforcement on reply (Fix 1)."""
+
+    async def test_reply_foreign_channel_id_gets_403(self, chat_app_with_stores):
+        """Reply with a channel_id the agent is not a member of -> 403."""
+        _cid_a, token_a = await _agent(chat_app_with_stores, "@guest-a")
+        _cid_b, token_b = await _agent(chat_app_with_stores, "@guest-b")
+
+        # Create a channel that only guest-b is a member of
+        channels = chat_app_with_stores.state.chat_channels
+        foreign_channel = await channels.create_channel(
+            name="private",
+            type="group",
+            created_by="user",
+            members=["user", "guest-b"],
+        )
+        foreign_channel_id = foreign_channel["id"]
+
+        mock = AsyncMock()
+        chat_app_with_stores.state.bridge_sessions.record_reply = mock
+
+        async with _bare(chat_app_with_stores) as c:
+            r = await c.post(
+                REPLY,
+                headers=_hdr(token_a),
+                json={"kind": "final", "content": "hi", "channel_id": foreign_channel_id},
+            )
+        assert r.status_code == 403
+        assert r.json() == {"error": "not a member of this channel"}
+        mock.assert_not_called()
+
+    async def test_reply_trace_id_in_foreign_channel_gets_403(self, chat_app_with_stores):
+        """Reply with trace_id of a message in a channel the agent is not a member of -> 403."""
+        _cid_a, token_a = await _agent(chat_app_with_stores, "@guest-a")
+        _cid_b, token_b = await _agent(chat_app_with_stores, "@guest-b")
+
+        # Create a channel that only guest-b is a member of
+        channels = chat_app_with_stores.state.chat_channels
+        messages = chat_app_with_stores.state.chat_messages
+        foreign_channel = await channels.create_channel(
+            name="private",
+            type="group",
+            created_by="user",
+            members=["user", "guest-b"],
+        )
+        foreign_channel_id = foreign_channel["id"]
+
+        # Create a message in that channel from guest-b
+        msg = await messages.send_message(
+            channel_id=foreign_channel_id,
+            author_id="guest-b",
+            author_type="agent",
+            content="hello from guest-b",
+            state="complete",
+        )
+        trace_id = msg["id"]
+
+        mock = AsyncMock()
+        chat_app_with_stores.state.bridge_sessions.record_reply = mock
+
+        async with _bare(chat_app_with_stores) as c:
+            r = await c.post(
+                REPLY,
+                headers=_hdr(token_a),
+                json={"kind": "final", "content": "hi", "trace_id": trace_id},
+            )
+        assert r.status_code == 403
+        assert r.json() == {"error": "not a member of this channel"}
+        mock.assert_not_called()
+
+    async def test_reply_into_own_dm_is_202(self, chat_app_with_stores):
+        """Reply into the agent's own DM channel -> 202 (positive control)."""
+        _cid_a, token_a = await _agent(chat_app_with_stores, "@guest-a")
+
+        # Ensure the agent has a DM channel (create one if needed)
+        channels = chat_app_with_stores.state.chat_channels
+        await channels.create_channel(
+            name="DM with guest-a",
+            type="dm",
+            created_by="user",
+            members=["user", "guest-a"],
+        )
+
+        mock = AsyncMock()
+        chat_app_with_stores.state.bridge_sessions.record_reply = mock
+
+        async with _bare(chat_app_with_stores) as c:
+            r = await c.post(
+                REPLY,
+                headers=_hdr(token_a),
+                json={"kind": "final", "content": "hi"},
+            )
+        assert r.status_code == 202
+        mock.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+class TestChatSessionSlugCollision:
+    """Tests for registry agent slug collision enforcement (Fix 2)."""
+
+    async def test_second_registry_agent_with_colliding_slug_gets_403(self, chat_app):
+        """Two registry agents whose canonical_ids strip to the same slug -> 403."""
+        _cid1, token1 = await _agent(chat_app, "@agent-one", display_name="Shared Name")
+        _cid2, token2 = await _agent(chat_app, "@agent-two", display_name="Shared Name")
+
+        # First agent should work
+        mock = AsyncMock()
+        chat_app.state.bridge_sessions.record_reply = mock
+        async with _bare(chat_app) as c:
+            r1 = await c.post(REPLY, headers=_hdr(token1), json={"kind": "final", "content": "hi"})
+        assert r1.status_code == 202
+
+        # Second agent with same slug should be rejected on reply
+        async with _bare(chat_app) as c:
+            r2 = await c.post(REPLY, headers=_hdr(token2), json={"kind": "final", "content": "hi"})
+        assert r2.status_code == 403
+        assert r2.json() == {"error": "handle collides with another registry agent"}
+
+
 def test_chat_session_scope_registered_in_both_sets():
     from tinyagentos.routes.agent_auth_requests import VALID_SCOPES
     from tinyagentos.routes.agent_registry import _ALLOWED_SCOPES
@@ -320,3 +475,66 @@ async def empty_subscribe(slug):
             assert called_reader_id == "guest-replay"
         finally:
             await outbox.close()
+
+
+@pytest.mark.asyncio
+class TestChatSessionSlugRegistryError:
+    async def test_registry_list_all_raises_returns_503_and_no_reply(self, chat_app):
+        _cid, token = await _agent(chat_app, "@guest-reg-err")
+        registry = chat_app.state.agent_registry
+        registry.list_all = AsyncMock(side_effect=RuntimeError("db"))
+        chat_app.state.bridge_sessions.record_reply = AsyncMock()
+        async with _bare(chat_app) as c:
+            r = await c.post(REPLY, headers=_hdr(token), json={"kind": "final", "content": "hi"})
+        assert r.status_code == 503
+        assert r.json() == {"error": "agent registry unavailable"}
+        chat_app.state.bridge_sessions.record_reply.assert_not_called()
+
+
+@pytest.mark.asyncio
+class TestChatSessionReplyTraceLookupError:
+    async def test_trace_id_lookup_raises_returns_503_and_no_reply(self, chat_app_with_stores):
+        _cid, token = await _agent(chat_app_with_stores, "@guest-trace-err")
+        messages = chat_app_with_stores.state.chat_messages
+        messages.get_message = AsyncMock(side_effect=RuntimeError("db"))
+        chat_app_with_stores.state.bridge_sessions.record_reply = AsyncMock()
+        async with _bare(chat_app_with_stores) as c:
+            r = await c.post(
+                REPLY,
+                headers=_hdr(token),
+                json={"kind": "final", "content": "hi", "trace_id": "msg-x"},
+            )
+        assert r.status_code == 503
+        assert r.json() == {"error": "chat store unavailable"}
+        chat_app_with_stores.state.bridge_sessions.record_reply.assert_not_called()
+
+    async def test_trace_id_not_found_still_returns_202(self, chat_app_with_stores):
+        _cid, token = await _agent(chat_app_with_stores, "@guest-trace-miss")
+        messages = chat_app_with_stores.state.chat_messages
+        messages.get_message = AsyncMock(return_value=None)
+        chat_app_with_stores.state.bridge_sessions.record_reply = AsyncMock()
+        async with _bare(chat_app_with_stores) as c:
+            r = await c.post(
+                REPLY,
+                headers=_hdr(token),
+                json={"kind": "final", "content": "hi", "trace_id": "msg-missing"},
+            )
+        assert r.status_code == 202
+        chat_app_with_stores.state.bridge_sessions.record_reply.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+class TestChatSessionUnknownChannel:
+    async def test_reply_unknown_channel_id_gets_403(self, chat_app_with_stores):
+        _cid_a, token_a = await _agent(chat_app_with_stores, "@guest-a")
+        mock = AsyncMock()
+        chat_app_with_stores.state.bridge_sessions.record_reply = mock
+        async with _bare(chat_app_with_stores) as c:
+            r = await c.post(
+                REPLY,
+                headers=_hdr(token_a),
+                json={"kind": "final", "content": "hi", "channel_id": "chan-does-not-exist"},
+            )
+        assert r.status_code == 403
+        assert r.json() == {"error": "not a member of this channel"}
+        mock.assert_not_called()

tinyagentos/routes/agent_chat_session.py

@@ -40,6 +40,31 @@ async def _resolve_slug(request: Request) -> str | JSONResponse:
             return JSONResponse(
                 {"error": "handle collides with a local agent"}, status_code=403
             )
+    # Two registry agents whose canonical_ids strip to the same slug must not
+    # share a session. Check for any OTHER active registry entry with the same
+    # derived slug. Allow the oldest (first registered) agent to proceed; reject
+    # any newer agents that collide.
+    registry = getattr(request.app.state, "agent_registry", None)
+    if registry is not None:
+        try:
+            active_agents = await registry.list_all(status="active")
+        except Exception:
+            logger.warning("agent registry list_all failed while resolving slug", exc_info=True)
+            return JSONResponse({"error": "agent registry unavailable"}, status_code=503)
+        # list_all returns agents ordered by id (oldest first) via
+        # SELECT * FROM agent_registry WHERE status = ? ORDER BY id. Find the first
+        # agent with this slug; if it's not the current agent, reject.
+        for other in active_agents:
+            other_cid = other.get("canonical_id")
+            if other_cid:
+                other_slug = _CANONICAL_AGENT_ID_RE.sub("", other_cid)
+                if other_slug == slug:
+                    if other_cid != cid:
+                        return JSONResponse(
+                            {"error": "handle collides with another registry agent"}, status_code=403
+                        )
+                    # This is the first agent with this slug; allow it.
+                    break
     return slug
 
 
@@ -126,6 +151,31 @@ async def event_generator():
     )
 
 
+async def _resolve_target_channel(request: Request, body: dict) -> str | None | JSONResponse:
+    """Determine the target channel for a reply, mirroring _handle_reply logic.
+
+    Returns the channel_id if one can be determined, else None (falls back to
+    the agent's own DM in the bridge). Returns a JSONResponse on store errors.
+    """
+    # Explicit channel_id in body takes precedence.
+    channel_id = body.get("channel_id")
+    if channel_id:
+        return channel_id
+    # Fall back to the channel of the message named by trace_id.
+    trace_id = body.get("trace_id")
+    if trace_id:
+        chat_messages = getattr(request.app.state, "chat_messages", None)
+        if chat_messages is not None:
+            try:
+                orig = await chat_messages.get_message(trace_id)
+            except Exception:
+                logger.warning("chat message lookup failed while resolving reply target", exc_info=True)
+                return JSONResponse({"error": "chat store unavailable"}, status_code=503)
+            if orig and orig.get("channel_id"):
+                return orig["channel_id"]
+    return None
+
+
 @router.post("/api/agents/self/chat/reply")
 async def self_chat_reply(request: Request):
     """Record a reply from the calling agent under its own slug.
@@ -144,5 +194,25 @@ async def self_chat_reply(request: Request):
         return JSONResponse({"error": "invalid JSON"}, status_code=400)
     if not isinstance(body, dict):
         return JSONResponse({"error": "invalid JSON"}, status_code=400)
+
+    # Fix 1: enforce channel membership before accepting the reply.
+    target_channel_id = await _resolve_target_channel(request, body)
+    if isinstance(target_channel_id, JSONResponse):
+        return target_channel_id
+    if target_channel_id:
+        chat_channels = getattr(request.app.state, "chat_channels", None)
+        if chat_channels is not None:
+            channel = await chat_channels.get_channel(target_channel_id)
+            if channel:
+                members = channel.get("members") or []
+                if slug not in members:
+                    return JSONResponse(
+                        {"error": "not a member of this channel"}, status_code=403
+                    )
+            else:
+                return JSONResponse(
+                    {"error": "not a member of this channel"}, status_code=403
+                )
+
     await bridge.record_reply(slug, body)
     return JSONResponse({"accepted": True}, status_code=202)