Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 81 additions & 0 deletions agent/chat_completion_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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:
Expand Down
26 changes: 25 additions & 1 deletion agent/codex_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
101 changes: 101 additions & 0 deletions tests/agent/test_codex_ttfb_watchdog.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Loading