diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 345517d660..a94a11921a 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -1167,37 +1167,23 @@ def _inline_nonstream_hard_timeout(stale_timeout: float): def direct_api_call(agent, api_kwargs: dict): """Run a non-streaming LLM call inline on the conversation thread. - Used when ``should_use_direct_api_call`` is True (cron turns and - delegated children). Skips the interrupt worker (whose only job is - interactive-interrupt responsiveness, which these contexts do not have) - so the nested-pool deadlock (#62151, #60203) cannot occur. - - While the inline request blocks, a lightweight activity heartbeat keeps - ``last_activity_ts`` advancing. Subagents use this path (non-streaming), - and without mid-call ticks the async stall monitor / sync heartbeat treat - a slow-but-healthy local model wait as "no progress" and interrupt around - 450s — surfacing as ``Operation interrupted: waiting for model response``. - - A stale-call watchdog bounds the request the same way the interrupt - worker's poll loop does (#80759). The keepalive httpx client uses - ``read=None`` (SSE), so the socket itself is not a usable bound: a - provider that accepts the request and then goes silent never trips a - read timeout, and a stranger-thread abort cannot ``close()`` the FD - (#29507). The watchdog aborts in-flight sockets through the already- - registered abort hook; a per-call ``timeout`` matching the stale budget - is the hard backstop when that abort finds nothing to shut down - (#85252). Either path surfaces a retryable ``TimeoutError`` so the - outer retry loop reconnects with backoff / credential rotation / - provider fallback. + Used when ``should_use_direct_api_call`` is True (cron turns, delegated + children): no interrupt worker, so the nested-pool deadlock (#62151, + #60203) cannot occur. An activity heartbeat keeps ``last_activity_ts`` + advancing or the stall monitor interrupts a slow-but-healthy wait at + ~450s. A stale-call watchdog bounds the request (#80759): the keepalive + client uses ``read=None``, so a silent provider never trips a read + timeout — the timer aborts in-flight sockets via the registered hook, and + a per-call ``timeout`` equal to the stale budget is the backstop when the + abort finds nothing to shut down (#85252). Both surface a retryable + ``TimeoutError`` for the outer retry loop. """ _check_stale_giveup(agent) agent._touch_activity("waiting for non-streaming API response") - # Request-lifecycle state, every transition under ``request_client_lock`` - # (ported from the #75301 design): ``done`` stops a late timer from - # bumping the stale streak after the call already unwound; ``cancelled`` - # marks a user/monitor interrupt as the owner of the outcome so a racing - # stale timer cannot misclassify the kill as provider staleness; - # ``stale`` is the one-shot stale transition itself. + # Lifecycle state, every transition under the lock (#75301): ``done`` + # stops a late timer bumping the stale streak after unwind; ``cancelled`` + # makes a user/monitor interrupt own the outcome so a racing timer can't + # misclassify the kill as staleness; ``stale`` is the one-shot transition. request_state = {"client": None, "done": False, "stale": False, "cancelled": False} request_client_lock = threading.Lock() activity_hb_stop = threading.Event() @@ -1209,30 +1195,23 @@ def direct_api_call(agent, api_kwargs: dict): timer callback only reports/bumps once, and never after an interrupt or a completed request). """ - # Abort while still holding the holder lock: the instant it is - # released, the inline finally may pop + cache the client for reuse - # and the NEXT call check it out — a late abort would then poison - # the slot and shut down an innocent in-flight request's sockets - # (same atomicity contract as _close_request_client_once in the - # interruptible variants; the abort itself never blocks). + # Abort under the lock (same contract as _RequestClientRegistry): + # once released the finally may cache the client and the NEXT call + # check it out, so a late abort would poison an innocent request. with request_client_lock: if request_state["done"]: return False if reason == "stale_call_kill" and request_state["cancelled"]: return False if reason != "stale_call_kill": - # A user interrupt/redirect that wins this lock owns the - # request outcome. Do not let a later timer misclassify the - # cancelled call as provider staleness and advance the - # cross-turn circuit breaker. + # Interrupt wins the lock -> owns the outcome; a later timer + # must not count it as staleness. request_state["cancelled"] = True newly_stale = reason == "stale_call_kill" and not request_state["stale"] if newly_stale: request_state["stale"] = True - # Advance the breaker before releasing the lock: the inline - # owner can unwind the moment the socket abort lands, and a - # fast retry's success/reset must not be overtaken by this - # older timer restoring the streak afterwards. + # Bump BEFORE releasing: a fast retry's reset must not be + # overtaken by this older timer restoring the streak. _bump_stale_streak(agent) request_client = request_state["client"] if request_client is not None: @@ -1245,23 +1224,17 @@ def direct_api_call(agent, api_kwargs: dict): return newly_stale def _make_client(reason: str, kind: str = "openai"): - # direct_api_call only runs for OpenAI-wire chat_completions cron - # requests (see should_use_direct_api_call), so the anthropic branch of - # the dispatch — the only caller that passes kind — is never reached - # here; the ``kind`` parameter exists purely for signature parity. + # Only OpenAI-wire requests reach direct_api_call; ``kind`` exists + # for signature parity with the dispatch helper. client = agent._create_request_openai_client(reason=reason, api_kwargs=api_kwargs) stale_before_dispatch = False with request_client_lock: request_state["client"] = client if request_state["stale"]: - # The timer expired while client construction was in flight - # (registration race): the abort found no sockets to kill, so - # handing this client to dispatch would open a brand-new - # socket after the only watchdog already fired. Abort it here - # and fail the call instead. (The residual window — timer - # firing between this registration and httpx opening its - # socket — is accepted: it is ms-scale against a >=90s budget - # and self-bounds at the OS connect-retry cap.) + # Timer fired during client construction: the abort found no + # socket, so dispatching now would open one AFTER the only + # watchdog fired. Fail here instead. (Residual ms-scale window + # before httpx opens its socket is accepted.) stale_before_dispatch = True try: agent._abort_request_openai_client( @@ -1295,16 +1268,13 @@ def direct_api_call(agent, api_kwargs: dict): name="direct-api-activity-hb", daemon=True, ) - # Resolve the budget BEFORE starting the heartbeat: the resolver may - # raise (fail-closed contract), and a raise after start() would leak the - # heartbeat thread — ticking last_activity_ts forever and masking real - # stalls from the stall monitor. + # Resolve the budget BEFORE start(): the resolver may raise (fail-closed), + # and a leaked heartbeat thread would mask real stalls forever. call_start = time.time() stale_timeout = _resolve_direct_stale_timeout(agent, api_kwargs) - # Do not override an explicit per-call timeout (provider config / - # transport already set one). Otherwise pin read=stale_timeout so a - # no-op stranger-thread abort cannot leave the keepalive client's - # read=None socket hanging until TCP dies (#85252). + # Never override an explicit per-call timeout; otherwise pin + # read=stale_timeout so a no-op abort can't leave the read=None socket + # hanging until TCP dies (#85252). hard_timeout = _inline_nonstream_hard_timeout(stale_timeout) if hard_timeout is not None and "timeout" not in api_kwargs: api_kwargs = dict(api_kwargs) @@ -1312,12 +1282,9 @@ def direct_api_call(agent, api_kwargs: dict): activity_hb.start() def _on_stale() -> None: - # Runs on the timer thread. It only aborts the in-flight sockets — - # it never issues a request — so the inline / no-worker property that - # fixes #62151 and #60203 is preserved. The abort helper owns the - # stale transition under the request lock: it returns False (and this - # callback stays silent) when the request already finished or a user - # interrupt owns the outcome. + # Timer thread: aborts sockets only, never issues a request (keeps + # the no-worker property). False = request finished or an interrupt + # owns the outcome; stay silent. if not _abort_active_request("stale_call_kill"): return elapsed = time.time() - call_start @@ -1333,9 +1300,8 @@ def direct_api_call(agent, api_kwargs: dict): stale_watchdog.daemon = True stale_watchdog.start() - # Only a clean return may report the reuse reason (request_complete): - # after an error or interrupt the wire client is really closed so the - # retry builds a fresh pool (see _REQUEST_CLIENT_REUSE_REASONS). + # Only a clean return reports the reuse reason; errors/interrupts really + # close the client so the retry builds a fresh pool. succeeded = False try: response = _dispatch_nonstreaming_api_request( @@ -1347,10 +1313,8 @@ def direct_api_call(agent, api_kwargs: dict): with request_client_lock: was_stale = request_state["stale"] if was_stale: - # The transport error is the expected consequence of our own - # abort. Raise a retryable TimeoutError (never InterruptedError, - # which the outer loop treats as "the user wants to stop") so the - # retry loop reconnects on a fresh pool. + # Our own abort caused the transport error: raise a retryable + # TimeoutError, never InterruptedError ("the user wants to stop"). raise TimeoutError( f"Non-streaming API call timed out after " f"{int(time.time() - call_start)}s with no response " @@ -1360,13 +1324,10 @@ def direct_api_call(agent, api_kwargs: dict): else: if getattr(agent, "_interrupt_requested", False): raise InterruptedError("Agent interrupted during API call") - # Close the race window against a timer firing between response - # arrival and this unwind: marking ``done`` under the lock makes any - # later timer callback a no-op, so the reset below cannot be - # overwritten by a stray bump after a successful call. A timer that - # already won the lock left ``stale`` set — the request still - # completed, so return the response (the streak reset undoes the - # bump; the poisoned client is discarded by the finally). + # Mark ``done`` under the lock so a timer firing between response + # arrival and unwind is a no-op and cannot overwrite the reset below. + # If a timer already won, the request still completed: return it (the + # reset undoes the bump; the finally discards the poisoned client). with request_client_lock: request_state["done"] = True _reset_stale_streak(agent) @@ -1643,24 +1604,17 @@ def interruptible_api_call(agent, api_kwargs: dict): _check_stale_giveup(agent) _clients = _RequestClientRegistry(agent) - # Request-local cancellation flag. Distinct from agent._interrupt_requested - # because that flag is cleared at run_conversation() turn boundaries, but - # this daemon worker thread can outlive the turn (the gateway caches - # AIAgent instances per session). Tracks whether THIS specific request was - # cancelled by the main thread's interrupt handler, so the transport error - # that is the expected consequence of our own force-close isn't misread as - # a network bug and surfaced to the caller. (PR #6600 — cascading interrupt - # hang.) + # Request-local cancel flag: agent._interrupt_requested is cleared at turn + # boundaries but this daemon worker can outlive the turn, so the worker + # needs to know THIS request was force-closed and not surface the + # resulting transport error as a network bug (#6600). _request_cancelled = {"value": False} - # Codex Responses retirement token (codex_responses only). The worker - # thread reads it through ``agent._active_codex_stream_request_token`` to - # tell whether it still owns the turn. When a watchdog below force-closes - # the connection it clears the agent-level token, so a worker still - # draining SSE frames raises instead of returning its partial output as a - # "completed" response (see run_codex_stream's _request_is_current). - # ``_codex_request_retired`` is the request-local mirror, used to swallow - # the transport error our own force-close causes — same split as - # ``_request_cancelled`` above. + # Codex retirement token: the worker checks + # ``agent._active_codex_stream_request_token`` to know it still owns the + # turn; a watchdog kill clears it so a worker still draining SSE raises + # instead of returning partial output as "completed" + # (run_codex_stream._request_is_current). ``_codex_request_retired`` is + # the request-local mirror used to swallow our own force-close error. _codex_request_token = object() if agent.api_mode == "codex_responses" else None _codex_request_retired = {"value": False} @@ -1685,11 +1639,9 @@ def interruptible_api_call(agent, api_kwargs: dict): def _call(): try: _install_codex_request_token() - # _clients.set_client registers each per-request client with the - # stranger-thread abort machinery above; the shared dispatch helper - # builds it via this callback (openai- or anthropic-kind) so the - # interrupt / stale-call detectors can force-close the worker's - # connection without touching the shared client (#67142). + # Per-request clients are registered with the abort machinery so + # the watchdogs force-close the worker's connection, never the + # shared client (#67142). result["response"] = _dispatch_nonstreaming_api_request( agent, api_kwargs, @@ -1703,16 +1655,11 @@ def interruptible_api_call(agent, api_kwargs: dict): ), ) except Exception as e: - # If the request was cancelled by the main thread's interrupt - # handler, the transport error is the expected consequence of our - # own force-close, NOT a network bug. Swallow it instead of - # surfacing — the main thread raises InterruptedError. (#6600) + # Our own force-close caused this error: swallow it, the main + # thread raises InterruptedError (#6600). Retirement logs at info + # (a watchdog discarded output the provider already sent — what an + # operator debugging a truncated reply needs); cancellation at debug. if _request_cancelled["value"] or _codex_request_retired["value"]: - # Retirement is logged at info: it means a watchdog discarded - # output the provider had already sent, which is exactly the - # event an operator debugging a truncated reply needs to see. - # Cancellation stays at debug — a user interrupt is a normal, - # high-frequency outcome and the caller already surfaces it. if _codex_request_retired["value"]: logger.info( "Codex worker caught %s after request retirement — " @@ -1731,14 +1678,11 @@ def interruptible_api_call(agent, api_kwargs: dict): return result["error"] = e finally: - # Retire first: _clients.close_once can raise (every other - # call site wraps it in try/except), and a leaked token would let a - # later worker mistake itself for the owning attempt. + # Retire first: close_once can raise, and a leaked token would let + # a later worker mistake itself for the owning attempt. _retire_codex_request_token() - # Reuse reason only on a clean response; any other outcome — - # error, or the cancel-swallow return above (which leaves both - # result slots None) — really closes so the next attempt builds - # a fresh pool (see _REQUEST_CLIENT_REUSE_REASONS). + # Reuse reason only on a clean response; error or cancel-swallow + # really closes so the next attempt builds a fresh pool. _clients.close_once( "request_complete" if result["response"] is not None @@ -1785,11 +1729,8 @@ def interruptible_api_call(agent, api_kwargs: dict): t.join(timeout=0.3) _poll_count += 1 - # Every ~30s: touch activity for the gateway inactivity monitor AND - # rewrite the live spinner/status line so CLI/TUI/Desktop users see - # what the agent is waiting on instead of an unexplained generic - # spinner (the "infinite thinking" complaint — the wait itself is - # usually a slow/overloaded provider, but the UI never said so). + # Every ~30s: gateway inactivity heartbeat + rewrite the status line + # so users see WHAT the wait is (the "infinite thinking" complaint). if _poll_count % 100 == 0: # 100 × 0.3s = 30s _elapsed = time.time() - _call_start try: @@ -1815,11 +1756,8 @@ def interruptible_api_call(agent, api_kwargs: dict): _elapsed = time.time() - _call_start - # TTFB detector: the Codex stream has produced no event at all and - # we're past the first-byte cutoff → the backend opened the - # connection but isn't responding. Kill it so the retry loop can - # reconnect (a fresh connection typically succeeds in seconds), - # instead of waiting out the much longer wall-clock stale timeout. + # TTFB detector: no Codex event past the first-byte cutoff — kill so + # the retry loop reconnects instead of waiting out the stale timeout. if ( _ttfb_enabled and _elapsed > _ttfb_timeout @@ -1860,9 +1798,8 @@ def interruptible_api_call(agent, api_kwargs: dict): ) break - # Stream-idle detector: the Codex backend emitted at least one SSE - # frame, then stopped emitting events. Valid keepalive / in_progress - # frames refresh _codex_stream_last_event_ts and should not be killed. + # Stream-idle detector: first byte arrived, then events stopped + # (keepalive/in_progress frames refresh the timestamp and don't count). _last_codex_event_ts = getattr(agent, "_codex_stream_last_event_ts", None) if ( _codex_idle_enabled @@ -1922,26 +1859,17 @@ def interruptible_api_call(agent, api_kwargs: dict): and getattr(agent, "_codex_stream_last_event_ts", None) is not None ), ) - # Mark THIS request cancelled before force-closing so the worker's - # exception handler recognizes the forced transport error as a - # cancel and exits cleanly instead of surfacing a network error or - # (in the streaming path) burning full retry cycles. (#6600) + # Mark cancelled BEFORE force-closing so the worker treats the + # transport error as a cancel (#6600). _request_cancelled["value"] = True logger.debug( "Force-closing httpx client due to interrupt (not a network error)." ) - # Force-close the in-flight worker-local HTTP connection to stop - # token generation without poisoning the shared client used to - # seed future retries. #67142: for anthropic this aborts the - # request-local client's sockets from this poll (stranger) thread - # rather than closing the shared _anthropic_client, which could - # release a TLS FD mid-SSL-BIO and corrupt an unrelated SQLite DB. + # Force-close the worker-local connection (never the shared client: + # releasing a TLS FD mid-SSL-BIO corrupted an unrelated SQLite DB, + # #67142), then let the worker unwind Relay scopes before raising + # (#81521). _abort_request("interrupt_abort") - # #81521 (sibling of the streaming-path fix): wait for the worker - # to unwind Relay-managed scopes before surfacing - # InterruptedError, so turn teardown cannot race a still-open - # physical scope and corrupt the LIFO stack. No-op when Relay - # managed execution is not live. _join_worker_for_relay_teardown(t, label="Non-streaming") raise InterruptedError("Agent interrupted during API call") if result["error"] is not None: