diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 124a693bb20aa..6d1de4fc1661e 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -371,6 +371,39 @@ def _call(): if _codex_idle_timeout <= 0: _codex_idle_enabled = False + # Progress-stall watchdog. The stream-idle detector above is satisfied by + # ANY SSE frame — including content-free keepalive / ``response.in_progress`` + # frames. The chatgpt.com/backend-api/codex backend has a failure mode where + # it keeps the socket "alive" with periodic keepalives but never emits a real + # delta or completes, so _last_event_ts stays fresh forever and the idle + # watchdog never trips: the call burns the full blunt stale timeout (observed: + # 90s × 6 retries ≈ 9 min before fallback). ``_codex_stream_last_progress_ts`` + # (set by run_codex_stream only on real deltas / function calls / output items + # / terminal events) lets us catch "events flowing but zero forward progress" + # and reconnect at the fast cutoff instead. Reuses the fast-reconnect target + # (default 40s, clamped below the stale timer) so it fires before the blunt + # kill. Set HERMES_CODEX_PROGRESS_STALE_TIMEOUT_SECONDS=0 to disable, or tune + # it directly; otherwise it defaults to the same fast cutoff as the no-byte + # watchdog. + _codex_progress_enabled = _codex_watchdog_enabled + _codex_progress_default = _ttfb_timeout if _ttfb_enabled else _codex_idle_timeout + _codex_progress_timeout = _env_float( + "HERMES_CODEX_PROGRESS_STALE_TIMEOUT_SECONDS", + _codex_progress_default, + ) + if _codex_progress_timeout <= 0: + _codex_progress_enabled = False + elif _stale_timeout != float("inf"): + # Never let the progress-stall cutoff meet or exceed the blunt stale + # timer (it would then never win); keep the same margin as the no-byte + # fast-reconnect clamp. + _progress_margin = _env_float( + "HERMES_CODEX_TTFB_BELOW_STALE_MARGIN_SECONDS", 10.0 + ) + _codex_progress_timeout = min( + _codex_progress_timeout, max(_stale_timeout - _progress_margin, 5.0) + ) + if _codex_watchdog_enabled: # Reset before the worker starts so a marker left over from a previous # call on this agent can't be misread as first-byte for this one. @@ -494,6 +527,54 @@ def _call(): ) break + # Progress-stall detector: the Codex stream is emitting frames (keepalive + # / in_progress refresh _codex_stream_last_event_ts, so the idle detector + # above stays quiet) but has produced ZERO real forward progress — no text + # / reasoning delta, no function call, no output item, no terminal event — + # for longer than the fast-reconnect cutoff. This is the keepalive-only + # hang on chatgpt.com/backend-api/codex: without this branch the call + # rides the socket all the way to the blunt wall-clock stale timeout + # (observed: 90s × 6 retries ≈ 9 min) instead of reconnecting in ~1s. + # We only arm once at least one event has arrived (_last_event_ts set) so + # this never races the no-byte TTFB detector for the first-byte case. + _last_progress_ts = getattr(agent, "_codex_stream_last_progress_ts", None) + _last_event_ts_for_progress = getattr(agent, "_codex_stream_last_event_ts", None) + if ( + _codex_progress_enabled + and _last_event_ts_for_progress is not None + and _last_progress_ts is None + and _elapsed > _codex_progress_timeout + ): + logger.warning( + "Codex stream emitted events but no forward progress for %.0fs " + "(threshold %.0fs, model=%s, context=~%s tokens). Backend is " + "sending keepalives without producing output. Killing connection " + "so the retry loop can reconnect.", + _elapsed, + _codex_progress_timeout, + api_kwargs.get("model", "unknown"), + f"{_est_tokens_for_codex_watchdog:,}", + ) + agent._buffer_status( + f"⚠️ Codex stream stalled with no progress for {int(_elapsed)}s " + f"(keepalives only, model: {api_kwargs.get('model', 'unknown')}). " + f"Reconnecting." + ) + try: + _close_request_client_once("codex_progress_stall_kill") + except Exception: + pass + agent._touch_activity( + f"codex stream killed after {int(_elapsed)}s with no forward progress" + ) + t.join(timeout=2.0) + if result["error"] is None and result["response"] is None: + result["error"] = TimeoutError( + f"Codex stream emitted events but made no forward progress for " + f"{int(_elapsed)}s (threshold: {int(_codex_progress_timeout)}s)" + ) + break + # Stale-call detector: kill the connection if no response # arrives within the configured timeout. if _elapsed > _stale_timeout: diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 398deed3c166f..b4c5643744674 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -444,7 +444,31 @@ def _on_reasoning_delta(text: str) -> None: def _on_event(event: Any) -> None: # TTFB watchdog and activity touch — runs once per SSE event. - agent._codex_stream_last_event_ts = time.time() + # ``_codex_stream_last_event_ts`` is refreshed by ANY event, including + # content-free keepalive / ``response.in_progress`` frames — that's the + # signal the no-byte and stream-idle watchdogs key on. + now = time.time() + agent._codex_stream_last_event_ts = now + # ``_codex_stream_last_progress_ts`` tracks REAL forward progress only: + # text/reasoning deltas, function-call frames, output-item completions, + # or a terminal event. A backend that emits only periodic keepalives + # (observed on chatgpt.com/backend-api/codex: the socket stays "alive" + # with in_progress frames but never produces a delta or completes) keeps + # _last_event_ts fresh forever, so the idle watchdog never trips and the + # call burns the full blunt stale timeout. Stamping progress separately + # lets a progress-stall watchdog reconnect at the fast cutoff instead. + _etype = _event_field(event, "type", "") + if not isinstance(_etype, str): + _etype = "" + _is_progress = ( + "delta" in _etype + or "function_call" in _etype + or _etype == "response.output_item.done" + or _etype == "error" + or _etype in _TERMINAL_EVENT_TYPES + ) + if _is_progress: + agent._codex_stream_last_progress_ts = now agent._touch_activity("receiving stream response") def _interrupt_check() -> bool: diff --git a/tests/agent/test_codex_ttfb_watchdog.py b/tests/agent/test_codex_ttfb_watchdog.py index 34bb64e27ebe2..a45ed30b576cc 100644 --- a/tests/agent/test_codex_ttfb_watchdog.py +++ b/tests/agent/test_codex_ttfb_watchdog.py @@ -522,3 +522,104 @@ def fake_stream(api_kwargs, client=None, on_first_delta=None): assert resp is sentinel assert "codex_ttfb_kill" not in closes + +def test_progress_stall_kills_keepalive_only_stream(tmp_path, monkeypatch): + """Regression for the keepalive-only hang on chatgpt.com/backend-api/codex. + + The backend emits periodic content-free keepalive / in_progress frames + (which refresh ``_codex_stream_last_event_ts``) but never produces a real + delta or completes. Before the progress-stall watchdog, the no-byte TTFB + detector was satisfied (events ARE arriving), the stream-idle detector was + satisfied (the gap between keepalives stays under its threshold), and only + the blunt wall-clock stale timer eventually killed it — burning the full + stale timeout (90s × 6 retries ≈ 9 min) before fallback. The progress-stall + watchdog must kill such a stream at the fast-reconnect cutoff instead. + """ + from agent import chat_completion_helpers as h + + agent = _make_codex_agent(tmp_path, monkeypatch) + # Real prod shape: blunt 90s stale timer, fast-reconnect cutoff 4s for a + # quick test. Idle threshold high so only the progress watchdog can fire. + monkeypatch.setattr(agent, "_compute_non_stream_stale_timeout", lambda *a, **k: 90.0) + monkeypatch.setenv("HERMES_CODEX_TTFB_FAST_RECONNECT_SECONDS", "4") + monkeypatch.setenv("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", "60") + + closes: list = [] + dummy_client = SimpleNamespace() + monkeypatch.setattr(agent, "_create_request_openai_client", lambda **k: dummy_client) + monkeypatch.setattr( + agent, "_abort_request_openai_client", lambda c, reason=None: closes.append(reason) + ) + monkeypatch.setattr( + agent, "_close_request_openai_client", lambda c, reason=None: closes.append(reason) + ) + + stop = {"flag": False} + + def fake_keepalive(api_kwargs, client=None, on_first_delta=None): + # Emit a keepalive every 1s: refreshes _last_event_ts but never sets + # _last_progress_ts (no delta / function_call / output_item / terminal). + deadline = time.time() + 60 + while time.time() < deadline and not stop["flag"] and not agent._interrupt_requested: + agent._codex_stream_last_event_ts = time.time() + time.sleep(1.0) + raise RuntimeError("connection closed") + + monkeypatch.setattr(agent, "_run_codex_stream", fake_keepalive) + + t0 = time.time() + try: + with pytest.raises(TimeoutError) as excinfo: + h.interruptible_api_call(agent, {"model": "gpt-5.5", "input": "hi"}) + elapsed = time.time() - t0 + assert "no forward progress" in str(excinfo.value) + # Killed via the progress-stall fast-reconnect path, not the 90s stale path. + assert "codex_progress_stall_kill" in closes + assert "stale_call_kill" not in closes + # Must reconnect fast (~4s cutoff + 2s join), far under the 90s stale. + assert elapsed < 20, f"progress-stall did not fire before stale timer: {elapsed:.1f}s" + finally: + stop["flag"] = True + + +def test_progress_stall_does_not_kill_streaming_progress(tmp_path, monkeypatch): + """A stream that keeps making real forward progress (text deltas) must NOT + be killed by the progress-stall watchdog even when it runs longer than the + fast-reconnect cutoff — progress events refresh ``_codex_stream_last_progress_ts``. + """ + from agent import chat_completion_helpers as h + + agent = _make_codex_agent(tmp_path, monkeypatch) + monkeypatch.setattr(agent, "_compute_non_stream_stale_timeout", lambda *a, **k: 90.0) + monkeypatch.setenv("HERMES_CODEX_TTFB_FAST_RECONNECT_SECONDS", "4") + monkeypatch.setenv("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", "60") + + closes: list = [] + dummy_client = SimpleNamespace() + monkeypatch.setattr(agent, "_create_request_openai_client", lambda **k: dummy_client) + monkeypatch.setattr( + agent, "_abort_request_openai_client", lambda c, reason=None: closes.append(reason) + ) + monkeypatch.setattr( + agent, "_close_request_openai_client", lambda c, reason=None: closes.append(reason) + ) + + sentinel = SimpleNamespace(ok=True) + + def fake_progress(api_kwargs, client=None, on_first_delta=None): + # Stream real progress (deltas) for ~8s — longer than the 4s fast cutoff — + # then complete. Each delta stamps both event_ts and progress_ts. + for _ in range(8): + now = time.time() + agent._codex_stream_last_event_ts = now + agent._codex_stream_last_progress_ts = now + time.sleep(1.0) + return sentinel + + monkeypatch.setattr(agent, "_run_codex_stream", fake_progress) + + resp = h.interruptible_api_call(agent, {"model": "gpt-5.5", "input": "hi"}) + assert resp is sentinel + assert "codex_progress_stall_kill" not in closes + assert "stale_call_kill" not in closes +