fix: resolve subagent control authority from the live session slot

Subagent list/tail/steer/interrupt authorized against a per-record copy of
the owning session's transport (`owner_transport`). That copy had to be
re-synced at every reattach site; `_rebind_live_transport` did it for
session.resume/activate but prompt.submit and the queued-prompt drain still
attached bare, so a client that reconnected through a prompt (the common
path on a remote gateway / Bot Mode switch) streamed fine while
`subagent.list` returned [] and controls rejected.

Read `owner_session_record["transport"]` at check time instead: the slot is
already mutated by every attach/detach/viewer-failover path, so no site can
forget the sync. `owner_transport` stays as the capture-time "commissioned
by a gateway session" marker (None = no RPC authority ever); non-dict owners
keep the exact-object rule. Drops the registration-time re-read and the
attach-time registry loop.

Diagnosis credit: nftpoetrist (#106663) — their prompt.submit / drain
regression tests pass against this change with no call-site edits.
This commit is contained in:
Teknium
2026-09-10 09:46:10 -07:00
parent 37eb6e1b05
commit ac07e20407
5 changed files with 52 additions and 25 deletions
+6 -4
View File
@@ -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")
@@ -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
+11 -6
View File
@@ -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))
+4 -3
View File
@@ -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
+4 -12
View File
@@ -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()