diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 57db47a0f4..8901747332 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -537,6 +537,14 @@ _CODEX_PROGRESS_DELTA_TYPES = frozenset( ) +# Progress-aware auxiliary stream deadlines (Aug 2026, masoria report): +# a dead stream fails fast at the no-progress window (first token AND +# between tokens), a live stream re-arms per substantive event and is +# bounded only by _aux_stream_total_ceiling() (shared with the streamed +# chat.completions path). +_AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS = 60.0 + + def _codex_event_has_content(event: Any) -> bool: """Whether a Codex Responses event carries a non-empty payload.""" event_type = _event_field(event, "type") @@ -1777,9 +1785,38 @@ class _CodexCompletionsAdapter: tool_calls_raw: List[Any] = [] usage = None total_timeout = timeout if isinstance(timeout, (int, float)) and timeout > 0 else None - deadline = time.monotonic() + float(total_timeout) if total_timeout else None + # Progress-aware stream deadlines (supersedes the old single absolute + # kill at ``total_timeout``). Three regimes: + # 1. First token: the stream must produce its first substantive + # payload within ``no_progress_timeout`` (60s default) or we + # fail fast and let the caller's normal retry/fallback chain + # run — a dead (or keepalive-only zombie) Codex stream no + # longer holds the full 300s compression budget before falling + # back (masoria report, Aug 2026: 3 stacked 300s waits -> + # 20+ min stuck on "Summarizing"). + # 2. Streaming: every substantive event re-arms the deadline by + # ``no_progress_timeout`` — a live stream is never killed by an + # absolute total, so a long reasoning summary that is actually + # producing tokens completes instead of timing out at 300s and + # falling back (#54915's original complaint, fixed properly). + # Keepalive/lifecycle frames do NOT re-arm, mirroring the + # commit-fence progress gating (#96707). + # 3. Hard ceiling: an absolute backstop from + # ``_aux_stream_total_ceiling`` (max(600s, 4x configured + # timeout) — the same bound the streamed chat.completions path + # uses) so a pathological one-token-per-59s drip still + # terminates. + _start_monotonic = time.monotonic() + no_progress_timeout = _AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS + if total_timeout is not None: + no_progress_timeout = min(no_progress_timeout, float(total_timeout)) + hard_deadline = _start_monotonic + _aux_stream_total_ceiling(total_timeout) + deadline_lock = threading.Lock() + progress_deadline = [_start_monotonic + no_progress_timeout] + saw_content = threading.Event() timed_out = threading.Event() - timeout_timer: Optional[threading.Timer] = None + stream_finished = threading.Event() + timeout_timer: List[Optional[threading.Timer]] = [None] # A protected provider call may outlive its owning compression attempt: # the owner returns promptly on hard cancellation while this adapter is # still blocked in the SDK stream on its isolated worker. Timer threads @@ -1791,8 +1828,34 @@ class _CodexCompletionsAdapter: attempt_stream_lock = threading.Lock() attempt_stream: List[Any] = [] + def _effective_deadline() -> float: + with deadline_lock: + return min(hard_deadline, progress_deadline[0]) + + def _record_stream_progress() -> None: + # A substantive payload re-arms the no-progress window. The hard + # ceiling is never extended. + with deadline_lock: + progress_deadline[0] = time.monotonic() + no_progress_timeout + def _timeout_message() -> str: - return f"Codex auxiliary Responses stream exceeded {float(total_timeout):.1f}s total timeout" + elapsed = time.monotonic() - _start_monotonic + if time.monotonic() >= hard_deadline: + return ( + "Codex auxiliary Responses stream exceeded " + f"{hard_deadline - _start_monotonic:.1f}s hard ceiling" + ) + if not saw_content.is_set(): + return ( + "Codex auxiliary Responses stream produced no output " + f"within {float(no_progress_timeout):.1f}s " + f"(no-progress timeout, {elapsed:.1f}s elapsed)" + ) + return ( + "Codex auxiliary Responses stream stalled: no new output " + f"for {float(no_progress_timeout):.1f}s " + f"({elapsed:.1f}s elapsed)" + ) def _close_client_on_timeout() -> None: begin_timeout_cleanup = getattr( @@ -1845,7 +1908,7 @@ class _CodexCompletionsAdapter: logger.debug("Codex auxiliary: cache eviction on timeout failed", exc_info=True) def _check_cancelled() -> None: - if deadline is not None and time.monotonic() >= deadline: + if total_timeout is not None and time.monotonic() >= _effective_deadline(): if not timed_out.is_set(): _close_client_on_timeout() raise TimeoutError(_timeout_message()) @@ -1867,11 +1930,30 @@ class _CodexCompletionsAdapter: # new failure mode for auxiliary calls. pass + def _watchdog_fire() -> None: + # Re-armable watchdog: if progress moved the deadline forward + # since this timer was scheduled, reschedule instead of killing + # a live stream. Only kill when the effective deadline (progress + # window or hard ceiling, whichever is sooner) has truly passed. + remaining = _effective_deadline() - time.monotonic() + if remaining > 0: + if timed_out.is_set() or stream_finished.is_set(): + return + t = threading.Timer(remaining, _watchdog_fire) + t.daemon = True + timeout_timer[0] = t + t.start() + return + _close_client_on_timeout() + try: if total_timeout: - timeout_timer = threading.Timer(float(total_timeout), _close_client_on_timeout) - timeout_timer.daemon = True - timeout_timer.start() + timeout_timer[0] = threading.Timer( + max(_effective_deadline() - time.monotonic(), 0.0), + _watchdog_fire, + ) + timeout_timer[0].daemon = True + timeout_timer[0].start() _check_cancelled() # Event-driven Responses streaming via the low-level @@ -1903,7 +1985,13 @@ class _CodexCompletionsAdapter: # compression commit fence) counts only substantive # payloads — lifecycle and keepalive events must not reset # the compression idle clock. + # The transport no-progress window likewise re-arms only on + # substantive payloads: a zombie stream that drips SSE + # keepalives but never produces output dies at the same 60s + # window as a fully dead connection. if _codex_event_has_content(_event): + _record_stream_progress() + saw_content.set() _notify_aux_provider_response() else: _notify_aux_timing_response() @@ -1998,8 +2086,10 @@ class _CodexCompletionsAdapter: logger.debug("Codex auxiliary Responses API call failed: %s", exc) raise finally: - if timeout_timer is not None: - timeout_timer.cancel() + stream_finished.set() + _t = timeout_timer[0] + if _t is not None: + _t.cancel() content = "".join(text_parts).strip() or None @@ -10029,12 +10119,20 @@ def _call_llm_impl( # fall straight through to provider/model fallback; fast blips (a # streaming-close or a 5xx) still retry, since those are cheap. if task == "compression" and _is_timeout_error(transient_err): - logger.info( - "Auxiliary compression: timeout on the critical path; " - "skipping same-provider retry and falling back: %s", - transient_err, - ) - raise + # A fast first-token fail (dead stream detected within the + # 60s no-progress window, zero output seen) is cheap — take + # the normal same-provider retry chain first; the provider + # is often fine and only that one stream was stillborn. A + # mid-stream stall or hard-ceiling timeout skips straight to + # fallback, because re-running a multi-minute summary on the + # same provider doubles the user-visible stall (#54465). + if "no-progress timeout" not in str(transient_err): + logger.info( + "Auxiliary compression: timeout on the critical path; " + "skipping same-provider retry and falling back: %s", + transient_err, + ) + raise _max_transient_retries = _transient_retry_count() _last_transient = transient_err for _attempt in range(1, _max_transient_retries + 1): diff --git a/tests/agent/test_codex_aux_no_progress_timeout.py b/tests/agent/test_codex_aux_no_progress_timeout.py new file mode 100644 index 0000000000..93401c8d82 --- /dev/null +++ b/tests/agent/test_codex_aux_no_progress_timeout.py @@ -0,0 +1,277 @@ +"""Progress-aware deadlines for the Codex auxiliary Responses stream. + +Regression tests for the masoria report (Aug 2026): each compression +attempt sat the FULL 300s absolute timeout on a dead Codex stream before +falling back, and repeated attempts stacked into a 20+ minute +"Summarizing thread" stall. + +New contract for ``_CodexCompletionsAdapter.create``: + +1. No first token within the 60s no-progress window -> fail fast + (``no-progress timeout`` in the message) so the fallback chain runs + after ~60s, not 300s. +2. A live stream re-arms the window on every substantive event: a slow + summary that keeps producing tokens is never killed by the old + absolute ``total_timeout``. +3. A mid-stream stall (tokens seen, then silence) dies one no-progress + window after the last token (``stalled`` in the message). +4. The compression critical-path retry gate distinguishes the two: a + cheap first-token failure still gets the same-provider retry; a + full-budget stall skips straight to fallback (#54465 semantics). +""" + +import threading +import time +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from agent.auxiliary_client import _CodexCompletionsAdapter, call_llm + + +def _content_event(text="tok"): + return SimpleNamespace(type="response.output_text.delta", delta=text) + + +def _keepalive_event(): + return SimpleNamespace(type="response.in_progress") + + +def _make_adapter(event_iter): + real_client = SimpleNamespace( + base_url="https://chatgpt.com/backend-api/codex", + responses=SimpleNamespace(create=lambda **_kwargs: event_iter), + close=lambda: None, + ) + return _CodexCompletionsAdapter(real_client, "gpt-5.6-sol") + + +def _consume(stream, *, model, on_event): + del model + for event in stream: + on_event(event) + return SimpleNamespace( + output=[SimpleNamespace( + type="message", + content=[SimpleNamespace(type="output_text", text="summary")], + )], + usage=None, + ) + + +class TestNoProgressFailFast: + def test_dead_stream_fails_at_no_progress_window_not_total_timeout(self): + """Keepalive-only stream dies at the (patched) no-progress window, + long before the 300s-style total timeout.""" + + def _zombie(): + deadline = time.monotonic() + 30.0 + while time.monotonic() < deadline: + time.sleep(0.02) + yield _keepalive_event() + + adapter = _make_adapter(_zombie()) + start = time.monotonic() + with ( + patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.3), + patch("agent.codex_runtime._consume_codex_event_stream", _consume), + pytest.raises(TimeoutError, match="no-progress timeout"), + ): + adapter.create( + messages=[{"role": "user", "content": "summarize"}], + timeout=300, + ) + elapsed = time.monotonic() - start + assert elapsed < 5.0, f"fail-fast took {elapsed:.1f}s" + + def test_live_stream_outlives_the_old_absolute_total_timeout(self): + """Tokens arriving inside the window keep the stream alive past + ``timeout`` — the old code killed this call at total_timeout.""" + + def _slow_but_alive(): + # 8 tokens, 0.1s apart: total ~0.8s, well past timeout=0.4. + for _ in range(8): + time.sleep(0.1) + yield _content_event() + + adapter = _make_adapter(_slow_but_alive()) + with ( + patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.4), + patch("agent.codex_runtime._consume_codex_event_stream", _consume), + ): + response = adapter.create( + messages=[{"role": "user", "content": "summarize"}], + timeout=0.4, + ) + assert response.choices[0].message.content == "summary" + + def test_mid_stream_stall_raises_stalled_timeout(self): + def _stalls_after_two_tokens(): + yield _content_event() + yield _content_event() + deadline = time.monotonic() + 30.0 + while time.monotonic() < deadline: + time.sleep(0.02) + yield _keepalive_event() + + adapter = _make_adapter(_stalls_after_two_tokens()) + start = time.monotonic() + with ( + patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.3), + patch("agent.codex_runtime._consume_codex_event_stream", _consume), + pytest.raises(TimeoutError, match="stalled: no new output"), + ): + adapter.create( + messages=[{"role": "user", "content": "summarize"}], + timeout=300, + ) + assert time.monotonic() - start < 5.0 + + def test_hard_ceiling_bounds_a_token_drip(self): + """A degenerate one-token-per-window drip still terminates at the + _aux_stream_total_ceiling backstop.""" + + def _dripper(): + while True: + time.sleep(0.05) + yield _content_event() + + adapter = _make_adapter(_dripper()) + with ( + patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 5.0), + patch("agent.auxiliary_client._aux_stream_total_ceiling", + return_value=0.3), + patch("agent.codex_runtime._consume_codex_event_stream", _consume), + pytest.raises(TimeoutError, match="hard ceiling"), + ): + adapter.create( + messages=[{"role": "user", "content": "summarize"}], + timeout=300, + ) + + def test_watchdog_timer_fires_while_blocked_before_first_event(self): + """responses.create() itself can block with zero bytes; the re-armable + watchdog must close the client and surface the no-progress timeout + without any event ever reaching _check_cancelled.""" + release = threading.Event() + + def _blocked_create(**_kwargs): + release.wait(timeout=30.0) + return iter([]) + + closed = threading.Event() + real_client = SimpleNamespace( + base_url="https://chatgpt.com/backend-api/codex", + responses=SimpleNamespace(create=_blocked_create), + close=closed.set, + ) + adapter = _CodexCompletionsAdapter(real_client, "gpt-5.6-sol") + try: + with ( + patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.3), + patch("agent.auxiliary_client._evict_cached_client_instance"), + ): + # The watchdog closes the shared client at the window; the + # blocked create keeps waiting (SimpleNamespace has no real + # transport), so unblock it and verify the timeout surfaced. + waiter: dict = {} + + def _run(): + try: + adapter.create( + messages=[{"role": "user", "content": "x"}], + timeout=300, + ) + except Exception as exc: # noqa: BLE001 + waiter["exc"] = exc + + t = threading.Thread(target=_run, daemon=True) + t.start() + assert closed.wait(timeout=5.0), "watchdog never closed client" + release.set() + t.join(timeout=5.0) + assert isinstance(waiter.get("exc"), TimeoutError) + assert "no-progress timeout" in str(waiter["exc"]) + finally: + release.set() + + +class TestCompressionRetryGate: + """First-token failures retry same-provider; stalls skip to fallback.""" + + def _run_call_llm(self, primary_error, second_response=None): + primary_client = MagicMock() + primary_client.base_url = "https://chatgpt.com/backend-api/codex" + if second_response is not None: + primary_client.chat.completions.create.side_effect = [ + primary_error, second_response, + ] + else: + primary_client.chat.completions.create.side_effect = primary_error + + fallback_client = MagicMock() + fallback_client.chat.completions.create.return_value = SimpleNamespace( + choices=[SimpleNamespace( + index=0, + message=SimpleNamespace(role="assistant", content="fallback"), + finish_reason="stop", + )], + model="fb", usage=None, + ) + + with ( + patch("agent.auxiliary_client._get_cached_client", + return_value=(primary_client, "gpt-5.6-sol")), + patch("agent.auxiliary_client._resolve_task_provider_model", + return_value=("auto", "gpt-5.6-sol", None, None, None)), + patch("agent.auxiliary_client._try_configured_fallback_chain", + return_value=(None, None, "")), + patch("agent.auxiliary_client._try_main_fallback_chain", + return_value=(None, None, "")), + patch("agent.auxiliary_client._try_payment_fallback", + return_value=(fallback_client, "fb", "openrouter")) as mock_fb, + patch("agent.auxiliary_client._TRANSIENT_RETRY_BACKOFF_BASE", 0.0), + ): + result = call_llm( + task="compression", + messages=[{"role": "user", "content": "summarize"}], + ) + return result, primary_client, mock_fb + + def test_no_progress_timeout_retries_same_provider(self): + err = TimeoutError( + "Codex auxiliary Responses stream produced no output within " + "60.0s (no-progress timeout, 60.2s elapsed)" + ) + good = SimpleNamespace( + choices=[SimpleNamespace( + index=0, + message=SimpleNamespace(role="assistant", content="retried"), + finish_reason="stop", + )], + model="gpt-5.6-sol", usage=None, + ) + result, primary, mock_fb = self._run_call_llm(err, second_response=good) + assert result.choices[0].message.content == "retried" + assert primary.chat.completions.create.call_count == 2 + assert not mock_fb.called + + def test_stalled_timeout_skips_same_provider_retry(self): + err = TimeoutError( + "Codex auxiliary Responses stream stalled: no new output for " + "60.0s (247.3s elapsed)" + ) + result, primary, mock_fb = self._run_call_llm(err) + assert result.choices[0].message.content == "fallback" + assert primary.chat.completions.create.call_count == 1 + assert mock_fb.called + + def test_hard_ceiling_timeout_skips_same_provider_retry(self): + err = TimeoutError( + "Codex auxiliary Responses stream exceeded 600.0s hard ceiling" + ) + result, primary, mock_fb = self._run_call_llm(err) + assert result.choices[0].message.content == "fallback" + assert primary.chat.completions.create.call_count == 1 + assert mock_fb.called