diff --git a/cli.py b/cli.py index e4e85720fb2..4f7f0e83114 100644 --- a/cli.py +++ b/cli.py @@ -15082,7 +15082,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): # and watch pattern matches) while agent is idle. try: from tools.process_registry import process_registry - for _evt, _synth in process_registry.drain_notifications(): + from tools.approval import get_current_session_key + _drain_sk = get_current_session_key(default="") + for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk): self._pending_input.put(_synth) except Exception: pass diff --git a/gateway/run.py b/gateway/run.py index 527da4ff513..7d5293ef411 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -15143,6 +15143,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew chat_type = str(evt.get("chat_type") or derived_chat_type or "").strip().lower() chat_id = str(evt.get("chat_id") or derived_chat_id or "").strip() if not platform_name or not chat_type or not chat_id: + logger.warning( + "Synthetic event source unresolvable: " + "session_key=%r platform=%r chat_type=%r chat_id=%r " + "evt_type=%s", + session_key, platform_name, chat_type, chat_id, + evt.get("type", "?"), + ) return None try: diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index 62d491bf82c..5dac30f2a1e 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -1346,6 +1346,109 @@ def test_drain_notifications_empty_queue(): assert results == [] +def test_drain_notifications_filters_async_delegation_by_session_key(): + """Async-delegation events should only be consumed by the matching session's drain. + + Regression test for issue #58684: background delegation results delivered + to the wrong session when the user switches sessions while a subagent runs. + """ + from tools.process_registry import process_registry + + # Clear the queue first + while not process_registry.completion_queue.empty(): + process_registry.completion_queue.get_nowait() + + try: + # Put events for different sessions + process_registry.completion_queue.put({ + "type": "async_delegation", + "delegation_id": "deleg_session_a", + "session_key": "telegram:dm:111:user_a", + "goal": "task A", + "status": "completed", + "summary": "done A", + "api_calls": 1, + "duration_seconds": 0.5, + }) + process_registry.completion_queue.put({ + "type": "async_delegation", + "delegation_id": "deleg_session_b", + "session_key": "telegram:dm:222:user_b", + "goal": "task B", + "status": "completed", + "summary": "done B", + "api_calls": 1, + "duration_seconds": 0.3, + }) + + # Drain for session A — should only get deleg_session_a + results_a = process_registry.drain_notifications(session_key="telegram:dm:111:user_a") + assert len(results_a) == 1, ( + f"Expected 1 event for session A, got {len(results_a)}" + ) + assert results_a[0][0]["delegation_id"] == "deleg_session_a" + assert "done A" in results_a[0][1] + + # Session B's event should have been re-queued — drain for session B + results_b = process_registry.drain_notifications(session_key="telegram:dm:222:user_b") + assert len(results_b) == 1, ( + f"Expected 1 event for session B, got {len(results_b)}" + ) + assert results_b[0][0]["delegation_id"] == "deleg_session_b" + assert "done B" in results_b[0][1] + + # No more events should remain + assert process_registry.completion_queue.empty() + finally: + while not process_registry.completion_queue.empty(): + process_registry.completion_queue.get_nowait() + + +def test_drain_notifications_no_filter_passes_all_async_delegation(): + """Without a session_key filter, all async-delegation events are consumed. + + This ensures backward compatibility — the default (session_key="") permits + all events, matching pre-fix behavior. + """ + from tools.process_registry import process_registry + + while not process_registry.completion_queue.empty(): + process_registry.completion_queue.get_nowait() + + try: + process_registry.completion_queue.put({ + "type": "async_delegation", + "delegation_id": "deleg_1", + "session_key": "telegram:dm:111:user_a", + "goal": "task 1", + "status": "completed", + "summary": "done 1", + "api_calls": 1, + "duration_seconds": 0.5, + }) + process_registry.completion_queue.put({ + "type": "async_delegation", + "delegation_id": "deleg_2", + "session_key": "telegram:dm:222:user_b", + "goal": "task 2", + "status": "completed", + "summary": "done 2", + "api_calls": 1, + "duration_seconds": 0.3, + }) + + # No filter — both should be consumed + results = process_registry.drain_notifications() + assert len(results) == 2, ( + f"Expected 2 events without filter, got {len(results)}" + ) + ids = {r[0]["delegation_id"] for r in results} + assert ids == {"deleg_1", "deleg_2"} + finally: + while not process_registry.completion_queue.empty(): + process_registry.completion_queue.get_nowait() + + # --------------------------------------------------------------------------- # _terminate_host_pid — cross-platform process-tree termination # --------------------------------------------------------------------------- diff --git a/tools/process_registry.py b/tools/process_registry.py index 5dfa5de7da4..ae21d94e5c0 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -1148,14 +1148,24 @@ class ProcessRegistry: """ return session_id in self._completion_consumed or session_id in self._poll_observed - def drain_notifications(self) -> "list[tuple[dict, str]]": + def drain_notifications( + self, session_key: str = "", + ) -> "list[tuple[dict, str]]": """Pop all pending notification events and return formatted pairs. Returns a list of (raw_event, formatted_text) tuples. Skips completion events the agent already consumed via wait/log or observed inline via poll() (see ``_drain_should_skip``). + + When ``session_key`` is non-empty, async-delegation events whose + ``session_key`` does NOT match are re-queued rather than consumed, + so they remain available for the correct session. This prevents + background delegation results from being delivered to the wrong + session when the user switches between gateway sessions while a + subagent is running (issue #58684). """ - results = [] + results: "list[tuple[dict, str]]" = [] + requeue: "list[dict]" = [] while not self.completion_queue.empty(): try: evt = self.completion_queue.get_nowait() @@ -1164,9 +1174,18 @@ class ProcessRegistry: _evt_sid = evt.get("session_id", "") if evt.get("type") == "completion" and self._drain_should_skip(_evt_sid): continue + # Filter async-delegation events by the caller's session_key so + # they are not delivered to the wrong session/thread (#58684). + if session_key and evt.get("type") == "async_delegation": + evt_session_key = evt.get("session_key", "") or "" + if evt_session_key != session_key: + requeue.append(evt) + continue text = format_process_notification(evt) if text: results.append((evt, text)) + for evt in requeue: + self.completion_queue.put(evt) return results def get(self, session_id: str) -> Optional[ProcessSession]: diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 80fa9cea95e..c640a4d5a6a 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -9337,7 +9337,9 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None: try: from tools.process_registry import process_registry - for _evt, synth in process_registry.drain_notifications(): + for _evt, synth in process_registry.drain_notifications( + session_key=session.get("session_key", ""), + ): with session["history_lock"]: if session.get("running"): process_registry.completion_queue.put(_evt)