diff --git a/tests/tools/test_subagent_steer.py b/tests/tools/test_subagent_steer.py index f72c6b9a32..d5418e9337 100644 --- a/tests/tools/test_subagent_steer.py +++ b/tests/tools/test_subagent_steer.py @@ -720,7 +720,9 @@ class TestSubagentSteerRPC: finally: _unregister_subagent("sid-rpc-param-spoof") - def test_session_transport_rebinding_does_not_transfer_ownership(self): + def test_session_transport_rebinding_moves_ownership_to_the_live_slot(self): + """Authority is the owning session's CURRENT transport slot (7befa11bf25 reversed the + original never-transfer rule): the reattached peer steers, the displaced one cannot.""" original_transport = self._Transport() rebound_transport = self._Transport() owner_record = { @@ -738,7 +740,7 @@ class TestSubagentSteerRPC: ) owner_record["transport"] = rebound_transport try: - for transport in (original_transport, rebound_transport): + for transport, expected in ((original_transport, "rejected"), (rebound_transport, "queued")): envelope = self._call( { "session_id": "owner-session", @@ -748,8 +750,8 @@ class TestSubagentSteerRPC: transport=transport, session_record=owner_record, ) - assert envelope["result"]["status"] == "rejected" - assert agent.steered == [] + assert envelope["result"]["status"] == expected + assert agent.steered == ["rebound authority"] finally: _unregister_subagent("sid-rpc-rebound") diff --git a/tests/tui_gateway/test_subagent_snapshot.py b/tests/tui_gateway/test_subagent_snapshot.py index 642db33fb4..774e36303c 100644 --- a/tests/tui_gateway/test_subagent_snapshot.py +++ b/tests/tui_gateway/test_subagent_snapshot.py @@ -224,3 +224,30 @@ def test_reattach_does_not_adopt_foreign_or_retired_generations(runtime): assert call("subagent.steer", via=new, subagent_id=sid, text="deny")["result"]["status"] == "rejected" assert not call("subagent.interrupt", via=new, subagent_id=sid)["result"]["found"] assert effects == [] + + +def test_any_attach_path_carries_subagent_authority_without_registry_sync(runtime): + """Authority follows the live session slot, not a per-record transport copy. Every reattach site + (prompt.submit, queued-prompt drain, resume, activate, future ones) goes through + _attach_session_transport; none of them may need to remember a registry sync step.""" + from tools.delegate_tool_child_run import _register_child + + server, owner, old, call = runtime + new = type("Transport", (), {"write": lambda self, frame: True})() + steered, stopped = [], [] + child = SimpleNamespace(_subagent_id="child", _delegate_depth=1, model="test", + steer=lambda text: steered.append(text) or True, + hard_interrupt=lambda text: stopped.append(text)) + _register_child(child, None, "owned", owner_session_id="ui-owner", + owner_transport=old, owner_session_record=owner) + owner["transport"] = server._detached_ws_transport + assert server._attach_session_transport(owner, new) + assert [r["subagent_id"] for r in call("subagent.list", via=new)["result"]["subagents"]] == ["child"] + assert call("subagent.steer", via=new, subagent_id="child", text="go")["result"]["status"] == "queued" + assert call("subagent.interrupt", via=new, subagent_id="child")["result"]["found"] + assert steered == ["go"] and len(stopped) == 1 + # The detached pre-reconnect transport lost membership and with it every control. + for method in ("list", "steer", "interrupt"): + denied = call("subagent." + method, via=old, subagent_id="child", text="stale") + assert "error" in denied or denied["result"].get("status") == "rejected" + assert steered == ["go"] and len(stopped) == 1 diff --git a/tools/delegate_tool_registry.py b/tools/delegate_tool_registry.py index f5beb16d2e..d4aed02363 100644 --- a/tools/delegate_tool_registry.py +++ b/tools/delegate_tool_registry.py @@ -53,11 +53,6 @@ def _register_subagent(record: Dict[str, Any]) -> None: return record.setdefault("accepting_steer", True) with _active_subagents_lock: - owner = record.get("owner_session_record") - if owner is not None and record.get("owner_transport") is not None: - # Child construction can finish after its captured dispatch transport - # was replaced. The exact session object retains generation authority. - record["owner_transport"] = owner.get("transport") _active_subagents[sid] = record def _unregister_subagent(subagent_id: str, *, agent: Any = None) -> None: @@ -110,9 +105,19 @@ def interrupt_subagent(subagent_id: str) -> bool: return False def _subagent_transport_matches(record, transport) -> bool: + """Authority follows the owning session's LIVE transport slot, read at check time. + + ``owner_transport`` on the record is only the capture-time marker that a gateway session + commissioned the child (``None`` = no RPC authority ever). The slot is authoritative because + every reattach path (prompt.submit, queued drain, resume, activate, viewer failover) already + mutates it; a per-record copy needed a matching registry sync at each of those sites and two + were missed (#106663). Records whose owner is not a session dict keep the exact-object rule.""" from tui_gateway.transport import FanoutTransport - bound = record.get("owner_transport") + if record.get("owner_transport") is None: + return False + owner = record.get("owner_session_record") + bound = owner.get("transport") if isinstance(owner, dict) else record.get("owner_transport") return bound is transport or (isinstance(bound, FanoutTransport) and bound.contains(transport)) diff --git a/tui_gateway/AGENTS.md b/tui_gateway/AGENTS.md index d4181645bd..d8b0ec5456 100644 --- a/tui_gateway/AGENTS.md +++ b/tui_gateway/AGENTS.md @@ -43,9 +43,10 @@ existing topical sibling, registered in the table — no `if method == ...` chai `subagent.list({session_id})` returns `{subagents, delegations}` for the calling transport's live session. Live child records are pinned to the exact session -record and transport. Authenticated live reattachment transfers that exact generation's -child authority to the new transport (also for late child registration and surviving -viewers); foreign or retired generations remain inaccessible. `last_tool` is the last started tool, not an in-flight +record and transport. Child authority is resolved at RPC time against the owning session's +LIVE transport slot, so every authenticated reattach path (prompt.submit, queued drain, +resume, activate, viewer failover) carries it with no registry bookkeeping — never add a +per-record transport sync at an attach site; foreign or retired generations remain inaccessible. `last_tool` is the last started tool, not an in-flight indicator. Async completion units are not agents and lack exact generation authority; `delegations` remains an empty array for wire compatibility. No dispatch context, results, callbacks, or routing keys are sent. Clients hydrate from this snapshot diff --git a/tui_gateway/session_lifecycle.py b/tui_gateway/session_lifecycle.py index 30fbc43166..251636b3c2 100644 --- a/tui_gateway/session_lifecycle.py +++ b/tui_gateway/session_lifecycle.py @@ -465,18 +465,10 @@ def _reattach_refusal(rid, sid: str, session: dict) -> dict | None: def _rebind_live_transport(sid: str, session: dict, transport: Transport) -> None: - """Attach a live peer without displacing existing subscribers (caller holds ``history_lock``).""" - from tools.delegate_tool_registry import _active_subagents, _active_subagents_lock - - # Transfer only this exact live generation's capabilities at the authenticated - # attachment seam, including records spawned through an older dispatch context. - with _active_subagents_lock: - _attach_session_transport(session, transport) - for record in _active_subagents.values(): - if (record.get("owner_session_id") == sid - and record.get("owner_session_record") is session - and record.get("owner_transport") is not None): - record["owner_transport"] = session["transport"] + """Attach a live peer without displacing existing subscribers (caller holds ``history_lock``). + Subagent control authority needs no bookkeeping here: it resolves against ``session["transport"]`` + at RPC time (``tools.delegate_tool_registry._subagent_transport_matches``).""" + _attach_session_transport(session, transport) # Every transport that showed this session (pop-outs resume the same sid); on disconnect the last # viewer becomes the transport instead of the drop sentinel. session.setdefault("viewers", {})[transport] = time.time()