From 3d30232eba801a2b2028d3372bc4a4d885146e5b Mon Sep 17 00:00:00 2001 From: atakan g Date: Tue, 28 Jul 2026 12:00:54 -0700 Subject: [PATCH] fix(delegate): isolate async batches from parent interrupts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Detached background delegation batches (_batch_runner) no longer honor the foreground parent's interrupt flag — a busy-submit interrupt in the TUI/desktop previously fabricated 'interrupted' results for background children that should outlive the turn. Explicit cancellation still works via _batch_interrupt. Rebased onto current main from PR #65040; both interrupt-suppression regression tests aligned with the current _session test helper. Original work by @AtakanGs in #65040. --- tests/test_tui_gateway_queue_on_busy.py | 214 ++++++++++++++++++++++++ tools/delegate_tool.py | 11 +- 2 files changed, 222 insertions(+), 3 deletions(-) diff --git a/tests/test_tui_gateway_queue_on_busy.py b/tests/test_tui_gateway_queue_on_busy.py index 44f2363e5c9a..e8e721e6d92b 100644 --- a/tests/test_tui_gateway_queue_on_busy.py +++ b/tests/test_tui_gateway_queue_on_busy.py @@ -13,6 +13,7 @@ import threading import time import types +import tools.async_delegation as ad from tui_gateway import server @@ -130,6 +131,62 @@ def test_queued_flag_overrides_mode_never_touches_live_turn(monkeypatch): assert session["queued_prompt"]["text"] == "next turn text" +def test_busy_interrupt_mode_ignores_completed_background_delegation(monkeypatch): + """A terminal delegation must not suppress normal busy-turn interruption.""" + monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt") + calls = {"interrupt": 0} + agent = types.SimpleNamespace( + interrupt=lambda *a, **k: calls.__setitem__("interrupt", calls["interrupt"] + 1) + ) + session = _session(agent=agent, running=True) + + with ad._records_lock: + ad._records["deleg_completed"] = { + "delegation_id": "deleg_completed", + "status": "completed", + "session_key": "session-key", + "origin_ui_session_id": "sid", + } + + try: + resp = server._handle_busy_submit("r1", "sid", session, "continue", "ws-1") + finally: + with ad._records_lock: + ad._records.clear() + + assert resp["result"]["status"] == "queued" + assert calls["interrupt"] == 1 + assert session["queued_prompt"]["text"] == "continue" + + +def test_busy_interrupt_mode_ignores_foreign_background_delegation(monkeypatch): + """Another tab's background work must not suppress this tab's interrupt.""" + monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt") + calls = {"interrupt": 0} + agent = types.SimpleNamespace( + interrupt=lambda *a, **k: calls.__setitem__("interrupt", calls["interrupt"] + 1) + ) + session = _session(agent=agent, running=True) + + with ad._records_lock: + ad._records["deleg_foreign"] = { + "delegation_id": "deleg_foreign", + "status": "running", + "session_key": "foreign-key", + "origin_ui_session_id": "foreign-sid", + } + + try: + resp = server._handle_busy_submit("r1", "sid", session, "interrupt me", "ws-1") + finally: + with ad._records_lock: + ad._records.clear() + + assert resp["result"]["status"] == "queued" + assert calls["interrupt"] == 1 + assert session["queued_prompt"]["text"] == "interrupt me" + + def test_busy_steer_mode_injects_when_accepted(monkeypatch): monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "steer") agent = types.SimpleNamespace(steer=lambda text: True, interrupt=lambda *a, **k: None) @@ -289,3 +346,160 @@ def test_drain_releases_running_on_dispatch_failure(monkeypatch): assert server._drain_queued_prompt("r1", "sid", session) is True # Failure must not leave the session wedged as running. assert session["running"] is False + + +def test_busy_interrupt_mode_preserves_real_background_batch_completion( + monkeypatch, tmp_path +): + """Foreground interruption must not cancel its detached async batch.""" + import json + import queue + import time + + import tools.delegate_tool as dt + from gateway.session_context import clear_session_vars, set_session_vars + from tools.process_registry import process_registry + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt") + + isolated_queue = queue.Queue() + monkeypatch.setattr(process_registry, "completion_queue", isolated_queue) + ad._reset_for_tests() + + calls = {"interrupt": 0} + + class _Parent: + def __init__(self): + self._delegate_depth = 0 + self.session_id = "session-key" + self._interrupt_requested = False + self._active_children = [] + self._active_children_lock = None + + def interrupt(self, *_args, **_kwargs): + calls["interrupt"] += 1 + self._interrupt_requested = True + + parent = _Parent() + session = _session(agent=parent, running=True) + + release_children = threading.Event() + all_children_started = threading.Event() + started_lock = threading.Lock() + started = {"count": 0} + child_ids = iter(("child-1", "child-2", "child-3")) + + def _build_child(**_kwargs): + return types.SimpleNamespace( + _delegate_role="leaf", + _subagent_id=next(child_ids), + ) + + def _blocking_child(task_index, goal, child=None, parent_agent=None, **_kwargs): + with started_lock: + started["count"] += 1 + if started["count"] == 3: + all_children_started.set() + + release_children.wait(timeout=10) + return { + "task_index": task_index, + "status": "completed", + "summary": f"done: {goal}", + "api_calls": 1, + "duration_seconds": 0.1, + "model": "test-model", + "exit_reason": "completed", + } + + credentials = { + "model": "test-model", + "provider": None, + "base_url": None, + "api_key": None, + "api_mode": None, + "command": None, + "args": None, + } + + monkeypatch.setattr(dt, "_build_child_agent", _build_child) + monkeypatch.setattr(dt, "_run_single_child", _blocking_child) + monkeypatch.setattr( + dt, + "_resolve_delegation_credentials", + lambda *_args, **_kwargs: credentials, + ) + + context_tokens = set_session_vars( + source="tui", + session_key="session-key", + ui_session_id="sid", + ) + + response = None + event = None + try: + dispatched = json.loads( + dt.delegate_task( + tasks=[ + {"goal": "first"}, + {"goal": "second"}, + {"goal": "third"}, + ], + background=True, + parent_agent=parent, + ) + ) + assert dispatched["status"] == "dispatched" + assert all_children_started.wait(timeout=5) + + response = server._handle_busy_submit( + "r1", + "sid", + session, + "follow-up", + "ws-1", + ) + + # The old detached-batch loop polls this parent flag every 0.5 seconds. + time.sleep(0.7) + release_children.set() + event = isolated_queue.get(timeout=5) + finally: + release_children.set() + clear_session_vars(context_tokens) + ad._reset_for_tests() + + assert response["result"]["status"] == "queued" + assert session["queued_prompt"]["text"] == "follow-up" + assert calls["interrupt"] == 1 + + assert event["type"] == "async_delegation" + assert event["origin_ui_session_id"] == "sid" + assert event["session_key"] == "session-key" + assert [result["status"] for result in event["results"]] == [ + "completed", + "completed", + "completed", + ] + assert sorted(result["summary"] for result in event["results"]) == [ + "done: first", + "done: second", + "done: third", + ] + + # Exercise the same positive-proof ownership gate used by the TUI's + # post-turn delivery path, not just event production. + isolated_queue.put(event) + drained = process_registry.drain_notifications( + session_key=session.get("session_key", ""), + owns_event=lambda candidate: server._session_owns_notification_event( + "sid", session, candidate + ), + ) + assert len(drained) == 1 + delivered_event, synthetic_prompt = drained[0] + assert delivered_event is event + assert synthetic_prompt + assert isolated_queue.empty() diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index b1d32e33d146..2513fa804086 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2979,7 +2979,7 @@ def delegate_task( child._live_transcript_path = str(_writer.path) children.append((i, t, child)) - def _execute_and_aggregate() -> dict: + def _execute_and_aggregate(*, honor_parent_interrupt: bool = True) -> dict: """Run all built children (1 or N), join on them, aggregate results, fire subagent_stop hooks + cost rollup, and return the combined result dict. Used by BOTH the synchronous path and the background runner. In @@ -3028,7 +3028,10 @@ def delegate_task( pending = set(futures.keys()) while pending: - if getattr(parent_agent, "_interrupt_requested", False) is True: + if ( + honor_parent_interrupt + and getattr(parent_agent, "_interrupt_requested", False) is True + ): # Parent interrupted — collect whatever finished and # abandon the rest. Children already received the # interrupt signal; we just can't wait forever. @@ -3270,7 +3273,9 @@ def delegate_task( pass def _batch_runner(): - return _execute_and_aggregate() + # This batch is detached from the foreground turn. Its lifecycle is + # owned by the async registry and cancelled only via _batch_interrupt. + return _execute_and_aggregate(honor_parent_interrupt=False) def _batch_interrupt(): for _c in _child_agents: