diff --git a/gateway/run.py b/gateway/run.py index e85cd3b0e062..9287a6d85b41 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -985,7 +985,16 @@ def _clarify_send_then_wait(fut, *, clarify_id: str, session_key: str, clarify_m timeout = clarify_mod.get_clarify_timeout() response = clarify_mod.wait_for_response(clarify_id, timeout=float(timeout)) if response is None or response == "": - # Timeout or session-boundary cancellation + # wait_for_response also returns on the thread-scoped interrupt + # (#83889). Report that outcome before treating the empty response as + # a timeout or session-boundary cancellation. + try: + from tools.interrupt import is_interrupted + + if is_interrupted(): + return "[interrupted by user]" + except Exception: + pass return f"[user did not respond within {int(timeout / 60)}m]" return response diff --git a/tests/gateway/test_clarify_send_timeout_ambiguity.py b/tests/gateway/test_clarify_send_timeout_ambiguity.py index a226a8977400..6b652d13da24 100644 --- a/tests/gateway/test_clarify_send_timeout_ambiguity.py +++ b/tests/gateway/test_clarify_send_timeout_ambiguity.py @@ -154,6 +154,22 @@ def test_no_response_returns_timeout_sentinel(): ) +def test_interrupted_wait_returns_interrupt_sentinel(monkeypatch): + fut = MagicMock() + fut.result.return_value = _Result(True) + clarify_mod = MagicMock() + clarify_mod.get_clarify_timeout.return_value = 600 + clarify_mod.wait_for_response.return_value = None + monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: True) + + assert ( + _clarify_send_then_wait( + fut, clarify_id="cid123", session_key="sk", clarify_mod=clarify_mod + ) + == "[interrupted by user]" + ) + + # --- Definitive failures keep their diagnostic detail in the log ---------- diff --git a/tests/tools/test_clarify_interrupt_wait.py b/tests/tools/test_clarify_interrupt_wait.py new file mode 100644 index 000000000000..beac0ebbd354 --- /dev/null +++ b/tests/tools/test_clarify_interrupt_wait.py @@ -0,0 +1,117 @@ +"""wait_for_response must observe the thread-scoped interrupt (#83889 RC1). + +Salvage of #25506 (@liuhao1024): the wait loop checks +``tools.interrupt.is_interrupted()`` once per slice. Without it, an +interrupted agent thread holds until the full clarify timeout (600s default), +or forever in unlimited mode — the session-boundary ``clear_session`` cleanup +only runs after the turn ends, which cannot happen while the thread is +blocked here. Family E anchor in the stall triage (#84047). +""" + +from __future__ import annotations + +import threading +import time + +import tools.clarify_gateway as cg +from tools.interrupt import set_interrupt + + +_ID_COUNTER = iter(range(10_000)) + + +def _register(session_key: str = "agent:main:test:dm:1"): + clarify_id = f"clarify-test-{next(_ID_COUNTER)}" + return cg.register( + clarify_id=clarify_id, + session_key=session_key, + question="q", + choices=None, + ) + + +def _wait_in_thread(clarify_id: str, timeout: float): + """Run the wait in a worker thread; expose its ident for set_interrupt.""" + result = {} + ready = threading.Event() + + def run(): + result["tid"] = threading.get_ident() + ready.set() + result["response"] = cg.wait_for_response(clarify_id, timeout) + + t = threading.Thread(target=run, daemon=True) + t.start() + assert ready.wait(timeout=5.0) + return t, result + + +def test_interrupt_unblocks_bounded_wait(): + entry = _register() + t, result = _wait_in_thread(entry.clarify_id, timeout=600.0) + try: + time.sleep(0.1) + assert t.is_alive(), "wait returned before any interrupt/resolve" + set_interrupt(True, thread_id=result["tid"]) + t.join(timeout=5.0) + assert not t.is_alive(), "interrupt did not unblock the bounded wait" + assert result["response"] is None + finally: + set_interrupt(False, thread_id=result["tid"]) + + +def test_interrupt_unblocks_unlimited_wait(): + """timeout<=0 is the forever-wedge: no deadline ever fires.""" + entry = _register() + t, result = _wait_in_thread(entry.clarify_id, timeout=0.0) + try: + time.sleep(0.1) + assert t.is_alive() + set_interrupt(True, thread_id=result["tid"]) + t.join(timeout=5.0) + assert not t.is_alive(), "interrupt did not unblock the unlimited wait" + assert result["response"] is None + finally: + set_interrupt(False, thread_id=result["tid"]) + + +def test_resolve_still_wins_and_returns_response(): + entry = _register() + t, result = _wait_in_thread(entry.clarify_id, timeout=600.0) + time.sleep(0.05) + assert cg.resolve_gateway_clarify(entry.clarify_id, "picked A") is True + t.join(timeout=5.0) + assert not t.is_alive() + assert result["response"] == "picked A" + + +def test_response_resolved_before_interrupt_wins_deterministically(): + entry = _register() + assert cg.resolve_gateway_clarify(entry.clarify_id, "picked A") is True + set_interrupt(True) + try: + assert cg.wait_for_response(entry.clarify_id, timeout=600.0) == "picked A" + finally: + set_interrupt(False) + + +def test_timeout_still_works_without_interrupt(): + entry = _register() + t, result = _wait_in_thread(entry.clarify_id, timeout=0.2) + t.join(timeout=5.0) + assert not t.is_alive() + assert result["response"] is None + + +def test_entry_cleaned_up_after_interrupt(): + """An interrupted wait must not leak its entry/session-index rows.""" + key = "agent:main:test:dm:cleanup" + entry = _register(session_key=key) + set_interrupt(True) # current thread; wait aborts on first slice check + try: + response = cg.wait_for_response(entry.clarify_id, 600.0) + finally: + set_interrupt(False) + assert response is None + # A second resolve should find nothing — the entry is gone. + assert cg.resolve_gateway_clarify(entry.clarify_id, "late") is False diff --git a/tools/clarify_gateway.py b/tools/clarify_gateway.py index e9bec3e95cae..074a0a914844 100644 --- a/tools/clarify_gateway.py +++ b/tools/clarify_gateway.py @@ -105,7 +105,7 @@ def register( def wait_for_response(clarify_id: str, timeout: float) -> Optional[str]: - """Block on the entry's event until resolved or timeout fires. + """Block on the entry's event until resolved, timeout, or interrupt. Polls in 1-second slices so the agent's inactivity heartbeat keeps firing — without this, ``Event.wait(timeout=600)`` blocks the thread @@ -116,7 +116,17 @@ def wait_for_response(clarify_id: str, timeout: float) -> Optional[str]: heartbeat still fires each slice so inactivity watchdogs don't kill a live prompt. - Returns the resolved response string, or ``None`` on timeout. + The thread-scoped interrupt flag (``tools.interrupt.is_interrupted``) is + checked once per slice. This is how ``/stop`` / an interrupt-mode message + unblocks a waiting agent thread: ``AIAgent.interrupt()`` propagates the + thread-scoped signal to active tool workers, but the session-boundary + ``clear_session`` cleanup only runs after the turn ends — which never + happens while this wait is blocked (#83889). Without the check, an + interrupted agent sits here until the full clarify timeout, or forever + in unlimited mode. + + Returns the resolved response string, or ``None`` on timeout/interrupt. + A response that raced the interrupt and resolved first still wins. """ with _lock: entry = _entries.get(clarify_id) @@ -128,11 +138,27 @@ def wait_for_response(clarify_id: str, timeout: float) -> Optional[str]: except Exception: # pragma: no cover - optional touch_activity_if_due = None + try: + from tools.interrupt import is_interrupted as _is_interrupted + except Exception: # pragma: no cover - defensive: never re-wedge the wait + _is_interrupted = None + # 0 / negative → unlimited: no deadline, poll forever in 1s slices. unlimited = timeout is None or float(timeout) <= 0.0 deadline = None if unlimited else time.monotonic() + float(timeout) activity_state = {"last_touch": time.monotonic(), "start": time.monotonic()} while True: + # A response that completed before the interrupt wins. Besides being + # the least surprising user-visible result, checking the event first + # makes the interrupt-vs-resolve contract deterministic. + if entry.event.is_set(): + break + if _is_interrupted is not None: + try: + if _is_interrupted(): + break + except Exception: # pragma: no cover - defensive + pass if deadline is None: slice_s = 1.0 else: