From d39c62409b5f00da6f7bd31dd3fb437d8c2daad6 Mon Sep 17 00:00:00 2001 From: nankingjing <1079826437@qq.com> Date: Fri, 3 Jul 2026 13:00:10 +0800 Subject: [PATCH] fix(delegate): pin async completion to spawning parent session (#57498) Background delegate_task completions only carried session_key. When multiple active sessions shared a routing peer, get_or_create_session could recover the latest ended_at IS NULL row and inject the subagent result into the wrong session. Capture parent_agent.session_id at dispatch time, include it on async-delegation completion events, and pin gateway routing via switch_session when the synthetic completion message is handled. Fixes #57498 --- gateway/run.py | 20 ++++++++++++++++++++ tests/tools/test_async_delegation.py | 2 ++ tools/async_delegation.py | 11 +++++++++++ tools/delegate_tool.py | 2 ++ 4 files changed, 35 insertions(+) diff --git a/gateway/run.py b/gateway/run.py index 7d5293ef411..252836ada9f 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -10633,6 +10633,21 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew session_entry = self.session_store.get_or_create_session(source) session_key = session_entry.session_key + pinned_session_id = str( + (getattr(event, "metadata", None) or {}).get("gateway_session_id") or "" + ).strip() + if pinned_session_id and pinned_session_id != session_entry.session_id: + prior_session_id = session_entry.session_id + switched = self.session_store.switch_session(session_key, pinned_session_id) + if switched is not None: + session_entry = switched + logger.info( + "Pinned async-delegation completion to spawning session %s " + "(was %s) for routing key %s (#57498)", + pinned_session_id, + prior_session_id, + session_key, + ) self._cache_session_source(session_key, source) if await asyncio.to_thread(self._is_telegram_topic_lane, source): try: @@ -15202,12 +15217,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if not adapter: return try: + metadata = {} + parent_session_id = str(evt.get("parent_session_id") or "").strip() + if parent_session_id: + metadata["gateway_session_id"] = parent_session_id synth_event = MessageEvent( text=synth_text, message_type=MessageType.TEXT, source=source, internal=True, message_id=str(evt.get("message_id") or "").strip() or None, + metadata=metadata, ) logger.info( "Watch pattern notification — injecting for %s chat=%s thread=%s", diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 9a473394b51..7714d3c8c08 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -99,6 +99,7 @@ def test_completion_event_lands_on_shared_queue_with_session_key(): res = ad.dispatch_async_delegation( goal="compute X", context="some context", toolsets=["web", "file"], role="leaf", model="test-model", session_key="agent:main:cli:dm:local", + parent_session_id="20260703_parent_sid", runner=runner, max_async_children=3, ) assert res["status"] == "dispatched" @@ -108,6 +109,7 @@ def test_completion_event_lands_on_shared_queue_with_session_key(): assert evt["type"] == "async_delegation" assert evt["summary"] == "the result" assert evt["session_key"] == "agent:main:cli:dm:local" + assert evt["parent_session_id"] == "20260703_parent_sid" assert evt["delegation_id"] == res["delegation_id"] diff --git a/tools/async_delegation.py b/tools/async_delegation.py index cc22a7f6e9a..ce4b094beb4 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -129,6 +129,7 @@ def dispatch_async_delegation( role: str, model: Optional[str], session_key: str, + parent_session_id: Optional[str] = None, runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, @@ -146,6 +147,11 @@ def dispatch_async_delegation( captured on the parent thread BEFORE dispatch, because the daemon worker thread won't carry the contextvar. Used to route the completion back to the originating session. + parent_session_id + The durable ``state.db`` session id of the parent agent that spawned + the delegation. Carried on the completion event so the gateway can + pin routing to the spawning session instead of recovering the latest + ``ended_at IS NULL`` row for the peer tuple (#57498). runner Zero-arg callable that builds + runs the child and returns the same result dict ``_run_single_child`` produces. Runs on the worker thread. @@ -174,6 +180,7 @@ def dispatch_async_delegation( "model": model, "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, + "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, "completed_at": None, @@ -285,6 +292,7 @@ def _push_completion_event( # session; empty string => CLI (single-session) path. "session_key": record.get("session_key", ""), "origin_ui_session_id": record.get("origin_ui_session_id", ""), + "parent_session_id": record.get("parent_session_id"), "goal": record.get("goal", ""), "context": record.get("context"), "toolsets": record.get("toolsets"), @@ -319,6 +327,7 @@ def dispatch_async_delegation_batch( role: str, model: Optional[str], session_key: str, + parent_session_id: Optional[str] = None, runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, @@ -361,6 +370,7 @@ def dispatch_async_delegation_batch( "model": model, "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, + "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, "completed_at": None, @@ -459,6 +469,7 @@ def _finalize_batch( "delegation_id": delegation_id, "session_key": event_record.get("session_key", ""), "origin_ui_session_id": event_record.get("origin_ui_session_id", ""), + "parent_session_id": event_record.get("parent_session_id"), "goal": event_record.get("goal", ""), "goals": event_record.get("goals"), "context": event_record.get("context"), diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 4179926091a..911e25204ab 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2815,6 +2815,7 @@ def delegate_task( _session_key = _agent_session_id except Exception: _origin_ui_session_id = "" + _parent_session_id = getattr(parent_agent, "session_id", None) _child_agents = [c for (_, _, c) in children] # Detach every child from the parent's interrupt-propagation list — the @@ -2856,6 +2857,7 @@ def delegate_task( model=creds["model"], session_key=_session_key, origin_ui_session_id=_origin_ui_session_id, + parent_session_id=_parent_session_id, runner=_batch_runner, interrupt_fn=_batch_interrupt, max_async_children=_get_max_async_children(),