From 463d4bcd158edc46dea0832a8bde47c97db697b3 Mon Sep 17 00:00:00 2001 From: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Date: Sat, 5 Sep 2026 22:38:59 +0800 Subject: [PATCH] fix(agent): defer codex idle watchdog until progress Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> --- agent/auxiliary_client.py | 18 +- agent/chat_completion_helpers.py | 139 +++++++++++---- agent/codex_runtime.py | 60 ++++++- tests/agent/test_aux_progress_streaming.py | 2 +- tests/agent/test_codex_ttfb_watchdog.py | 190 +++++++++++++++++++-- 5 files changed, 339 insertions(+), 70 deletions(-) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 83f8586f76..8d28c9feb9 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -30,6 +30,7 @@ from agent.codex_headers import ( codex_cloudflare_headers as _codex_cloudflare_headers, is_official_codex_base_url as _is_official_codex_base_url, ) +from agent.codex_runtime import _codex_event_has_content # `openai.OpenAI` is imported lazily (~240 ms cold); `OpenAI` below is a proxy # so in-module calls, `auxiliary_client.OpenAI` reads and @@ -377,28 +378,11 @@ def _anthropic_aux_stream_event_hook() -> Callable[[Any], None]: return _on_event -_CODEX_PROGRESS_DELTA_TYPES = frozenset({ - "response.output_text.delta", "response.reasoning_summary_text.delta", "response.text.delta", - "response.audio.delta", "response.function_call_arguments.delta", "response.reasoning_text.delta", -}) - # A dead stream fails at the no-progress window (first token AND between tokens); a live # stream re-arms per event, bounded by _aux_stream_total_ceiling(). _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 = _field(event, "type") - if event_type in _CODEX_PROGRESS_DELTA_TYPES: - return bool(_field(event, "delta")) - if event_type == "response.output_item.added": - item = _field(event, "item") - return "function_call" in str(_field(item, "type") or "") and any( - bool(_field(item, f)) for f in ("id", "call_id", "name", "arguments")) - return False - - @contextlib.contextmanager def _aux_thread_local_hook(local: threading.local, hook): """Install one thread-local hook, restoring the prior on exit (non-callable = passthrough).""" diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 7b0a5022c3..6c8a8d8f13 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -455,16 +455,20 @@ def _estimate_chunk_bytes(chunk: Any) -> int: def _codex_wait_notice_recovery(*, stale_timeout: float, ttfb_enabled: bool, ttfb_timeout: float, - last_event_ts: Optional[float], call_start: float, idle_enabled: bool, idle_timeout: float, - elapsed: float) -> str: + last_event_ts: Optional[float], last_progress_ts: Optional[float], + retry_started_ts: Optional[float], call_start: float, idle_enabled: bool, + idle_timeout: float, idle_requires_progress: bool, elapsed: float) -> str: """Describe the earliest enabled Codex watchdog on the call timeline.""" deadlines: list[float] = [] if math.isfinite(stale_timeout): deadlines.append(stale_timeout) - if last_event_ts is None: + if retry_started_ts is not None: + if ttfb_enabled and math.isfinite(ttfb_timeout): + deadlines.append(max(0.0, retry_started_ts - call_start) + ttfb_timeout) + elif last_event_ts is None: if ttfb_enabled and math.isfinite(ttfb_timeout): deadlines.append(ttfb_timeout) - elif idle_enabled and math.isfinite(idle_timeout): + elif (not idle_requires_progress or last_progress_ts is not None) and idle_enabled and math.isfinite(idle_timeout): deadlines.append(max(0.0, last_event_ts - call_start) + idle_timeout) if not deadlines or min(deadlines) <= elapsed: return "" @@ -1021,6 +1025,7 @@ class _NonStreamWatchdogs: ttfb_timeout: float idle_enabled: bool idle_timeout: float + idle_requires_progress: bool def _resolve_nonstream_watchdogs(agent, api_kwargs: dict) -> _NonStreamWatchdogs: @@ -1028,9 +1033,12 @@ def _resolve_nonstream_watchdogs(agent, api_kwargs: dict) -> _NonStreamWatchdogs The stale detector kills a hung provider early so the retry loop can rotate credentials / fall back. Codex adds two failure modes: accepting the connection - but never emitting an event (no-byte TTFB cutoff; a reconnect succeeds in ~2s) - and stalling after the opening SSE frame (event-idle gap; any SSE event is - activity). Tunables: HERMES_CODEX_TTFB_TIMEOUT_SECONDS, + but never emitting an event (no-event TTFB cutoff; a reconnect succeeds in ~2s) + and stalling after substantive model progress begins (event-idle gap; any parsed SSE + event remains transport activity). Only the implicit official OpenAI Codex policy + for large contexts defers arming until progress; small requests, compatible backends, + and explicit overrides retain the legacy first-event semantics. Tunables: + HERMES_CODEX_TTFB_TIMEOUT_SECONDS, HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS (0 disables each), HERMES_CODEX_TTFB_DISABLE_ABOVE_TOKENS / HERMES_CODEX_TTFB_STRICT, HERMES_CODEX_TTFB_MAX_SECONDS, HERMES_CODEX_HARD_TIMEOUT_SECONDS. @@ -1039,13 +1047,14 @@ def _resolve_nonstream_watchdogs(agent, api_kwargs: dict) -> _NonStreamWatchdogs codex = agent.api_mode == "codex_responses" openai_codex_backend = _is_openai_codex_backend(agent) est_tokens = estimate_request_context_tokens(api_kwargs) + codex_floor = 0.0 if codex and openai_codex_backend: # Raise the stale floor for large payloads so healthy gateway-scale # requests aren't aborted mid-prefill. codex_floor = openai_codex_stale_timeout_floor(est_tokens) if codex_floor: stale_timeout = max(stale_timeout, codex_floor) - # Flat hard ceiling (#64507) for a request that emits SOME bytes then wedges. + # Flat hard ceiling (#64507) for a request that emits SOME events then wedges. # Default sits ABOVE the max floor (1200s) — a backstop, never tighter. 0 disables. hard_timeout = env_float("HERMES_CODEX_HARD_TIMEOUT_SECONDS", 1500.0) if hard_timeout > 0: @@ -1055,7 +1064,7 @@ def _resolve_nonstream_watchdogs(agent, api_kwargs: dict) -> _NonStreamWatchdogs (default for threshold, default in ((100_000, 180.0), (50_000, 120.0), (10_000, 60.0)) if est_tokens > threshold), 12.0) - # No-byte TTFB cutoff. Default 120s: the SDK's own read timeout is 600s, + # No-event TTFB cutoff. Default 120s: the SDK's own read timeout is 600s, # and a tight 12s killed subscription-backed requests mid-prefill. ttfb_enabled = codex ttfb_timeout = env_float("HERMES_CODEX_TTFB_TIMEOUT_SECONDS", 120.0) @@ -1067,22 +1076,32 @@ def _resolve_nonstream_watchdogs(agent, api_kwargs: dict) -> _NonStreamWatchdogs disable_above = env_float("HERMES_CODEX_TTFB_DISABLE_ABOVE_TOKENS", 10_000.0) strict = os.environ.get("HERMES_CODEX_TTFB_STRICT", "").strip().lower() in {"1", "true", "yes", "on"} if not strict and disable_above > 0 and est_tokens >= disable_above and ttfb_timeout < idle_default: - logger.info("Scaling openai-codex no-byte TTFB watchdog from %.0fs to %.0fs " + logger.info("Scaling openai-codex no-event TTFB watchdog from %.0fs to %.0fs " "for large request (context=~%s tokens >= %.0f). " "Set HERMES_CODEX_TTFB_STRICT=1 to keep the smaller cutoff.", ttfb_timeout, idle_default, f"{est_tokens:,}", disable_above) ttfb_timeout = idle_default ttfb_cap = env_float("HERMES_CODEX_TTFB_MAX_SECONDS", 120.0) if ttfb_cap > 0 and ttfb_timeout > ttfb_cap: - logger.info("Capping openai-codex no-byte TTFB timeout from %.0fs to %.0fs " + logger.info("Capping openai-codex no-event TTFB timeout from %.0fs to %.0fs " "(context=~%s tokens). Set HERMES_CODEX_TTFB_MAX_SECONDS to tune.", ttfb_timeout, ttfb_cap, f"{est_tokens:,}") ttfb_timeout = ttfb_cap + idle_raw = os.getenv("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", "").strip() + try: + float(idle_raw) + except ValueError: + idle_explicit = False + else: + idle_explicit = True idle_timeout = env_float("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", idle_default) return _NonStreamWatchdogs(stale_timeout=stale_timeout, codex=codex, est_tokens=est_tokens, ttfb_enabled=ttfb_enabled, ttfb_timeout=ttfb_timeout, idle_enabled=codex and idle_timeout > 0, - idle_timeout=idle_timeout) + idle_timeout=idle_timeout, + idle_requires_progress=( + codex and openai_codex_backend and codex_floor > 0 and not idle_explicit + )) def _codex_silent_hang_hint(agent, api_kwargs: dict) -> Optional[str]: @@ -1116,6 +1135,18 @@ class _NonStreamRequest: self.codex_token = object() if agent.api_mode == "codex_responses" else None self.codex_retired = False self.wd = _resolve_nonstream_watchdogs(agent, api_kwargs) + self.codex_watchdog_state = ( + SimpleNamespace( + token=self.codex_token, + lock=threading.Lock(), + last_event_ts=None, + last_progress_ts=None, + retry_started_ts=None, + phase_aware=self.wd.idle_requires_progress, + ) + if self.codex_token is not None + else None + ) self.call_start = time.time() self.thread = None @@ -1140,8 +1171,14 @@ class _NonStreamRequest: return self.clients.set_client(client, kind=kind) def _call(self): + watchdog_state_var = watchdog_context_token = None try: self._install_codex_request_token() + if self.codex_watchdog_state is not None: + from agent.codex_runtime import _codex_watchdog_state_var + + watchdog_state_var = _codex_watchdog_state_var + watchdog_context_token = watchdog_state_var.set(self.codex_watchdog_state) self.result["response"] = _dispatch_nonstreaming_api_request( self.agent, self.api_kwargs, make_client=self._make_client) except Exception as e: @@ -1160,6 +1197,8 @@ class _NonStreamRequest: return self.result["error"] = e finally: + if watchdog_state_var is not None: + watchdog_state_var.reset(watchdog_context_token) # Retire first: close_once can raise, and a leaked token would let # a later worker mistake itself for the owning attempt. self._retire_codex_request_token() @@ -1172,9 +1211,11 @@ class _NonStreamRequest: """Watchdog/interrupt kill: abort the request client (kind-aware, #67142) and retire the codex token; the worker sees its own forced close via the cancel flags.""" + # Retire before closing the socket: close can wake the worker immediately, + # and a retired worker must not open a new physical retry. + self._retire_codex_request_token() with contextlib.suppress(Exception): self.clients.close_once(reason) - self._retire_codex_request_token() def _await_worker_after_kill(self, timeout_message: str) -> None: # Wait briefly for the worker to notice the closed connection. @@ -1185,13 +1226,27 @@ class _NonStreamRequest: def _model(self) -> str: return self.api_kwargs.get("model", "unknown") + def _codex_watchdog_snapshot(self): + state = self.codex_watchdog_state + if state is None: + return ( + getattr(self.agent, "_codex_stream_last_event_ts", None), + getattr(self.agent, "_codex_stream_last_progress_ts", None), + None, + ) + with state.lock: + return state.last_event_ts, state.last_progress_ts, state.retry_started_ts + def _emit_wait_notice(self, elapsed: float) -> None: wd = self.wd try: + last_event_ts, last_progress_ts, retry_started_ts = self._codex_watchdog_snapshot() recovery = _codex_wait_notice_recovery(stale_timeout=wd.stale_timeout, ttfb_enabled=wd.ttfb_enabled, ttfb_timeout=wd.ttfb_timeout, - last_event_ts=getattr(self.agent, "_codex_stream_last_event_ts", None), + last_event_ts=last_event_ts, last_progress_ts=last_progress_ts, + retry_started_ts=retry_started_ts, call_start=self.call_start, idle_enabled=wd.idle_enabled, idle_timeout=wd.idle_timeout, + idle_requires_progress=wd.idle_requires_progress, elapsed=elapsed) self.agent._emit_wait_notice( f"⏳ waiting on {self.api_kwargs.get('model', 'the provider')} — " @@ -1200,40 +1255,46 @@ class _NonStreamRequest: logger.debug("wait-notice construction failed", exc_info=True) def _ttfb_kill(self, elapsed: float) -> None: - """No Codex event past the first-byte cutoff — kill so the retry loop + """No parsed Codex event past the first-event cutoff — kill so the retry loop reconnects instead of waiting out the stale timeout.""" agent, wd = self.agent, self.wd silent_hint = _codex_silent_hang_hint(agent, self.api_kwargs) - logger.warning("Codex stream produced no bytes within TTFB cutoff " + logger.warning("Codex stream produced no parsed stream event within TTFB cutoff " "(%.0fs > %.0fs, model=%s). Backend accepted the connection " "but sent no stream events. Killing connection so the retry loop can reconnect.", elapsed, wd.ttfb_timeout, self._model()) agent._buffer_status( - f"⚠️ No first byte from provider in {int(elapsed)}s (codex stream, model: {self._model()}). " + f"⚠️ No first stream event from provider in {int(elapsed)}s (codex stream, model: {self._model()}). " f"Reconnecting." + (f" {silent_hint}" if silent_hint else "")) self._abort_request("codex_ttfb_kill") agent._emit_wait_notice(f"⚠ no response from provider in {int(elapsed)}s — reconnecting...") - agent._touch_activity(f"codex stream killed after {int(elapsed)}s with no first byte") + agent._touch_activity(f"codex stream killed after {int(elapsed)}s with no first stream event") self._await_worker_after_kill( - f"Codex stream produced no bytes within {int(elapsed)}s (TTFB threshold: {int(wd.ttfb_timeout)}s)" + f"Codex stream produced no parsed stream event within {int(elapsed)}s " + f"(TTFB threshold: {int(wd.ttfb_timeout)}s)" + (f". {silent_hint}" if silent_hint else "")) def _idle_kill(self, event_stale_elapsed: float) -> None: - """First byte arrived, then SSE events stopped (keepalive/in_progress - frames refresh the timestamp and don't count).""" + """SSE events stopped after the phase-specific idle arm point. + + Only the implicit official OpenAI Codex policy arms on substantive model + progress; compatible providers and explicit operator timeouts arm on first + parsed event. Once armed, any parsed SSE event refreshes transport activity. + """ agent, wd = self.agent, self.wd - logger.warning("Codex stream produced no SSE events for %.0fs after first byte " + arm_point = "model progress began" if wd.idle_requires_progress else "the first parsed event" + logger.warning("Codex stream produced no SSE events for %.0fs after %s " "(threshold %.0fs, model=%s, context=~%s tokens). Killing " - "connection so the retry loop can reconnect.", event_stale_elapsed, wd.idle_timeout, + "connection so the retry loop can reconnect.", event_stale_elapsed, arm_point, wd.idle_timeout, self._model(), f"{wd.est_tokens:,}") agent._buffer_status( - f"⚠️ Codex stream sent no events for {int(event_stale_elapsed)}s after first byte " + f"⚠️ Codex stream sent no events for {int(event_stale_elapsed)}s after {arm_point} " f"(model: {self._model()}). Reconnecting.") self._abort_request("codex_stream_idle_kill") agent._touch_activity(f"codex stream killed after {int(event_stale_elapsed)}s with no SSE events") self._await_worker_after_kill( f"Codex stream produced no SSE events for {int(event_stale_elapsed)}s " - f"after first byte (threshold: {int(wd.idle_timeout)}s)") + f"after {arm_point} (threshold: {int(wd.idle_timeout)}s)") def _stale_kill(self, elapsed: float) -> None: """No response within the stale timeout: kill and count toward the @@ -1250,8 +1311,9 @@ class _NonStreamRequest: def _interrupt(self, elapsed: float) -> None: agent = self.agent + last_event_ts, _, _ = self._codex_watchdog_snapshot() _record_interrupted_provider_wait(agent, elapsed, - response_started=self.wd.codex and getattr(agent, "_codex_stream_last_event_ts", None) is not None + response_started=self.wd.codex and last_event_ts is not None ) # Mark cancelled BEFORE force-closing so the worker treats the transport # error as a cancel (#6600). Never close the shared client (releasing a @@ -1267,9 +1329,13 @@ class _NonStreamRequest: agent, wd = self.agent, self.wd if wd.codex: # 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. + # call on this agent can't be misread as the first event for this one. agent._codex_stream_last_event_ts = None agent._codex_stream_last_progress_ts = None + with self.codex_watchdog_state.lock: + self.codex_watchdog_state.last_event_ts = None + self.codex_watchdog_state.last_progress_ts = None + self.codex_watchdog_state.retry_started_ts = None agent._touch_activity("waiting for non-streaming API response") self.thread = t = threading.Thread(target=_context_thread_target(self._call), daemon=True) @@ -1280,15 +1346,24 @@ class _NonStreamRequest: poll_count += 1 # Every ~30s: gateway inactivity heartbeat + rewrite the status line # so users see WHAT the wait is (the "infinite thinking" complaint). - elapsed = time.time() - self.call_start + now = time.time() + elapsed = now - self.call_start if poll_count % 100 == 0: # 100 × 0.3s = 30s self._emit_wait_notice(elapsed) - last_event_ts = getattr(agent, "_codex_stream_last_event_ts", None) - if wd.ttfb_enabled and elapsed > wd.ttfb_timeout and last_event_ts is None: + last_event_ts, last_progress_ts, retry_started_ts = self._codex_watchdog_snapshot() + retry_ttfb_elapsed = now - retry_started_ts if retry_started_ts is not None else None + if wd.ttfb_enabled and retry_ttfb_elapsed is not None and retry_ttfb_elapsed > wd.ttfb_timeout: + self._ttfb_kill(retry_ttfb_elapsed) + break + if (retry_started_ts is None and wd.ttfb_enabled + and elapsed > wd.ttfb_timeout and last_event_ts is None): self._ttfb_kill(elapsed) break - if wd.idle_enabled and last_event_ts is not None and (time.time() - last_event_ts) > wd.idle_timeout: - self._idle_kill(time.time() - last_event_ts) + idle_elapsed = now - last_event_ts if last_event_ts is not None else None + if (retry_started_ts is None and wd.idle_enabled and idle_elapsed is not None + and (not wd.idle_requires_progress or last_progress_ts is not None) + and idle_elapsed > wd.idle_timeout): + self._idle_kill(idle_elapsed) break if elapsed > wd.stale_timeout: self._stale_kill(elapsed) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index e5b2820458..d7138807b1 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -4,6 +4,7 @@ AIAgent first: ``run_codex_app_server_turn`` drives one ``codex app-server`` sub from __future__ import annotations +import contextvars import json import logging import os @@ -15,6 +16,9 @@ from typing import Any, Callable, Dict, List from agent.stream_single_writer import claim_stream_writer, stream_writer_is_current logger = logging.getLogger(__name__) +_codex_watchdog_state_var: contextvars.ContextVar[Any | None] = contextvars.ContextVar( + "codex_watchdog_state", default=None +) def _call_guarded(fn: Callable | None, fail_msg: str, *fail_args: Any, args: tuple = (), kwargs: dict | None = None): @@ -507,6 +511,28 @@ def _event_field(event: Any, name: str, default: Any = None) -> Any: return value if value is not None else default +_CODEX_PROGRESS_DELTA_TYPES = frozenset({ + "response.output_text.delta", "response.reasoning_summary_text.delta", "response.text.delta", + "response.audio.delta", "response.function_call_arguments.delta", "response.reasoning_text.delta", +}) + + +def _codex_event_has_content(event: Any) -> bool: + """Whether a Codex Responses event carries substantive forward progress. + + Lifecycle/keepalive frames and empty structural deltas prove transport + liveness, but do not mean the model has begun producing its response. + """ + event_type = _event_field(event, "type") + if event_type in _CODEX_PROGRESS_DELTA_TYPES: + return bool(_event_field(event, "delta")) + if event_type == "response.output_item.added": + item = _event_field(event, "item") + return "function_call" in str(_event_field(item, "type") or "") and any( + bool(_event_field(item, field)) for field in ("id", "call_id", "name", "arguments")) + return False + + def _raise_stream_error(event: Any) -> None: """Raise ``_StreamErrorEvent`` from a ``type=error`` SSE frame. The spec puts code/message/param at the top level, but the SDK and several proxies nest them under ``error``; read top-level first, then the envelope.""" @@ -832,7 +858,12 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta # Retirement token for THIS request (installed by ``interruptible_api_call``). A watchdog that kills the # connection clears the agent-level token, so a worker still draining frames can tell it was retired. # ``None`` = no watchdog; every check passes. - request_token = getattr(agent, "_active_codex_stream_request_token", None) + watchdog_state = _codex_watchdog_state_var.get() + request_token = ( + watchdog_state.token + if watchdog_state is not None + else getattr(agent, "_active_codex_stream_request_token", None) + ) # Delta-sink claim for the CURRENT physical attempt (None until the stream opens). writer_token = {"value": None} @@ -847,8 +878,24 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta agent._codex_streamed_text_parts.append(text) agent._fire_stream_delta(text) - def _on_event(event: Any) -> None: # TTFB watchdog and activity touch — once per SSE event. - agent._codex_stream_last_event_ts = time.time() + def _on_event(event: Any) -> None: # TTFB/activity touch — once per SSE event. + now = time.time() + has_progress = _codex_event_has_content(event) + reset_progress = False + if watchdog_state is not None: + with watchdog_state.lock: + reset_progress = watchdog_state.retry_started_ts is not None + if reset_progress: + watchdog_state.retry_started_ts = None + watchdog_state.last_progress_ts = None + watchdog_state.last_event_ts = now + if has_progress: + watchdog_state.last_progress_ts = now + agent._codex_stream_last_event_ts = now + if reset_progress: + agent._codex_stream_last_progress_ts = None + if has_progress: + agent._codex_stream_last_progress_ts = now agent._touch_activity("receiving stream response") def _interrupt_or_superseded() -> bool: @@ -911,8 +958,15 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta call_role = ("delegated" if getattr(agent, "is_subagent", False) else "fallback" if int(getattr(agent, "_fallback_index", 0) or 0) > 0 else "primary") for attempt in range(max_stream_retries + 1): + if not _request_is_current(): + raise TimeoutError("Codex Responses stream request retired before retry") if agent._interrupt_requested: raise InterruptedError("Agent interrupted before Codex stream retry") + if attempt > 0 and watchdog_state is not None and watchdog_state.phase_aware: + # A physical reconnect has its own no-event TTFB phase. Its first parsed + # event clears this marker and starts a fresh model-progress phase. + with watchdog_state.lock: + watchdog_state.retry_started_ts = time.time() intercepted_events: list = [] writer_token["value"] = event_stream = None try: diff --git a/tests/agent/test_aux_progress_streaming.py b/tests/agent/test_aux_progress_streaming.py index 6d30f17f0b..9979d0fc3f 100644 --- a/tests/agent/test_aux_progress_streaming.py +++ b/tests/agent/test_aux_progress_streaming.py @@ -23,12 +23,12 @@ from agent.auxiliary_client import ( _aggregate_chat_stream_async, _anthropic_event_has_content, _aux_stream_total_ceiling, - _codex_event_has_content, _create_with_progress, _notify_aux_progress, _provider_requires_stream, aux_progress_hook, ) +from agent.codex_runtime import _codex_event_has_content from agent.conversation_compression import CompressionCommitFence diff --git a/tests/agent/test_codex_ttfb_watchdog.py b/tests/agent/test_codex_ttfb_watchdog.py index bdf53061f3..a6e80a869f 100644 --- a/tests/agent/test_codex_ttfb_watchdog.py +++ b/tests/agent/test_codex_ttfb_watchdog.py @@ -1,4 +1,4 @@ -"""Regression tests for the Codex time-to-first-byte (TTFB) watchdog. +"""Regression tests for the Codex TTFB (first parsed stream event) watchdog. The chatgpt.com/backend-api/codex endpoint has an intermittent failure mode where it accepts the connection but never emits a single stream event. The @@ -8,10 +8,12 @@ retry loop can reconnect promptly. Once any stream event arrives, the TTFB watchdog is satisfied and a separate idle watchdog handles streams that stop emitting SSE events. -The "bytes flowing" signal is ``agent._codex_stream_last_event_ts``, set on -*any* event by ``codex_runtime.run_codex_stream`` — so reasoning-only or -tool-call-only turns (which emit no output-text deltas) are not mistaken for a -stall. +Parsed-event activity is recorded in ``agent._codex_stream_last_event_ts``; +substantive model progress is recorded separately. For the implicit official +OpenAI Codex policy on large contexts, lifecycle frames satisfy TTFB without +arming the short post-progress idle budget. Small requests, explicit overrides, +and compatible backends retain their first-parsed-event semantics. Raw SSE +comments are outside this layer. """ from __future__ import annotations @@ -29,7 +31,13 @@ sys.modules.setdefault("firecrawl", types.SimpleNamespace(Firecrawl=object)) sys.modules.setdefault("fal_client", types.SimpleNamespace()) -def _make_codex_agent(tmp_path, monkeypatch): +def _make_codex_agent( + tmp_path, + monkeypatch, + *, + provider="openai-codex", + base_url="https://chatgpt.com/backend-api/codex", +): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) (tmp_path / ".env").write_text("", encoding="utf-8") (tmp_path / "config.yaml").write_text("{}\n", encoding="utf-8") @@ -37,9 +45,9 @@ def _make_codex_agent(tmp_path, monkeypatch): agent = AIAgent( model="gpt-5.5", - provider="openai-codex", + provider=provider, api_key="sk-dummy", - base_url="https://chatgpt.com/backend-api/codex", + base_url=base_url, quiet_mode=True, skip_context_files=True, skip_memory=True, @@ -57,12 +65,40 @@ def _make_codex_agent(tmp_path, monkeypatch): return agent +def _shorten_implicit_idle_watchdog(monkeypatch, helpers, timeout=2.0): + """Keep the resolver on its implicit branch while scaling time for tests.""" + monkeypatch.delenv("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", raising=False) + original = helpers._resolve_nonstream_watchdogs + + def resolve(agent, api_kwargs): + watchdogs = original(agent, api_kwargs) + watchdogs.idle_timeout = timeout + return watchdogs + + monkeypatch.setattr(helpers, "_resolve_nonstream_watchdogs", resolve) +def _install_codex_event_stream(agent, monkeypatch, event_factory, closes): + client = SimpleNamespace( + responses=SimpleNamespace(create=lambda **_kwargs: event_factory()) + ) + monkeypatch.setattr( + agent, "_create_request_openai_client", lambda **_kwargs: client + ) + monkeypatch.setattr( + agent, + "_abort_request_openai_client", + lambda _client, reason=None: closes.append(reason), + ) + monkeypatch.setattr( + agent, + "_close_request_openai_client", + lambda _client, reason=None: closes.append(reason), + ) def test_ttfb_includes_silent_hang_hint_for_gpt_5_5(tmp_path, monkeypatch): - """The no-first-byte watchdog should surface the same actionable hint as the + """The no-first-event watchdog should surface the same actionable hint as the stale-call timeout path when the model matches the silent-hang heuristic.""" from agent import chat_completion_helpers as h @@ -216,9 +252,15 @@ def test_ttfb_does_not_kill_when_events_flow(tmp_path, monkeypatch): sentinel = SimpleNamespace(ok=True) def fake_stream(api_kwargs, client=None, on_first_delta=None): - # Bytes flowing: mark stream activity right away, then keep generating + # A parsed event marks stream activity right away; then keep generating # past the 0.4s TTFB cutoff before returning a real response. - agent._codex_stream_last_event_ts = time.time() + from agent.codex_runtime import _codex_watchdog_state_var + + now = time.time() + state = _codex_watchdog_state_var.get() + with state.lock: + state.last_event_ts = now + agent._codex_stream_last_event_ts = now if on_first_delta: on_first_delta() time.sleep(0.9) @@ -231,10 +273,125 @@ def test_ttfb_does_not_kill_when_events_flow(tmp_path, monkeypatch): assert "codex_ttfb_kill" not in closes +@pytest.mark.parametrize( + ("provider", "base_url", "input_chars", "idle_env", "idle_enabled", "requires_progress"), + [ + ("openai-codex", "https://chatgpt.com/backend-api/codex", 40_004, None, True, True), + ("openai-codex", "https://chatgpt.com/backend-api/codex", 40_004, "2", True, False), + ("openai-codex", "https://chatgpt.com/backend-api/codex", 40_000, None, True, False), + ("xai-oauth", "https://api.x.ai/v1", 40_004, None, True, False), + ("openai-codex", "https://chatgpt.com/backend-api/codex", 40_004, "", True, True), + ("openai-codex", "https://chatgpt.com/backend-api/codex", 40_004, "invalid", True, True), + ("openai-codex", "https://chatgpt.com/backend-api/codex", 40_004, "0", False, False), + ], +) +def test_idle_phase_policy_is_narrow_and_preserves_operator_overrides( + tmp_path, + monkeypatch, + provider, + base_url, + input_chars, + idle_env, + idle_enabled, + requires_progress, +): + """Only an implicit, large, official request uses progress-phase arming.""" + from agent import chat_completion_helpers as h + + agent = _make_codex_agent( + tmp_path, monkeypatch, provider=provider, base_url=base_url + ) + if idle_env is None: + monkeypatch.delenv("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", raising=False) + else: + monkeypatch.setenv("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", idle_env) + + watchdogs = h._resolve_nonstream_watchdogs( + agent, {"model": "gpt-5.6-sol", "input": "x" * input_chars} + ) + + assert watchdogs.est_tokens == input_chars // 4 + assert watchdogs.idle_enabled is idle_enabled + assert watchdogs.idle_requires_progress is requires_progress +@pytest.mark.parametrize( + "mode", ["initial_gap", "stall", "retry_gap", "retry_no_event"] +) +def test_event_stale_phase_is_scoped_to_physical_stream_attempt( + tmp_path, monkeypatch, mode +): + """Retry lifecycle resets phase without hiding a zero-event reconnect hang.""" + from agent import chat_completion_helpers as h + agent = _make_codex_agent(tmp_path, monkeypatch) + _shorten_implicit_idle_watchdog(monkeypatch, h) + monkeypatch.setenv("HERMES_CODEX_TTFB_TIMEOUT_SECONDS", "2") + monkeypatch.setenv("HERMES_CODEX_TTFB_STRICT", "1") + monkeypatch.setenv("HERMES_CODEX_HARD_TIMEOUT_SECONDS", "5") + closes: list = [] + attempts = {"count": 0} + + def stream_attempt(): + attempts["count"] += 1 + if mode == "initial_gap": + yield SimpleNamespace(type="response.created") + yield SimpleNamespace(type="response.in_progress") + time.sleep(3.0) + yield SimpleNamespace(type="response.reasoning_text.delta", delta="working") + yield SimpleNamespace(type="response.output_text.delta", delta="done") + yield SimpleNamespace( + type="response.completed", + response=SimpleNamespace(status="completed", id="resp-1", usage=None), + ) + return + if mode == "retry_no_event" and attempts["count"] == 2: + while getattr(agent, "_active_codex_stream_request_token", None) is not None: + time.sleep(0.02) + raise RuntimeError("retired zero-event retry") + if mode == "retry_gap" and attempts["count"] == 2: + yield SimpleNamespace(type="response.created") + yield SimpleNamespace(type="response.in_progress") + time.sleep(3.0) + yield SimpleNamespace(type="response.reasoning_text.delta", delta="retry step") + yield SimpleNamespace(type="response.output_text.delta", delta="done") + yield SimpleNamespace( + type="response.completed", + response=SimpleNamespace(status="completed", id="resp-2", usage=None), + ) + return + + yield SimpleNamespace(type="response.created") + if mode != "retry_no_event": + yield SimpleNamespace(type="response.reasoning_text.delta", delta="first step") + if mode in {"retry_gap", "retry_no_event"}: + raise ConnectionError("retry physical stream") + while getattr(agent, "_active_codex_stream_request_token", None) is not None: + time.sleep(0.02) + raise ConnectionError("retired stalled stream") + + _install_codex_event_stream(agent, monkeypatch, stream_attempt, closes) + + if mode in {"initial_gap", "retry_gap"}: + response = h.interruptible_api_call( + agent, {"model": "gpt-5.6-sol", "input": "x" * 40_004} + ) + assert response.output_text == "done" + assert attempts["count"] == (1 if mode == "initial_gap" else 2) + else: + error_match = "no parsed stream event" if mode == "retry_no_event" else "no SSE events" + with pytest.raises(TimeoutError, match=error_match): + h.interruptible_api_call( + agent, {"model": "gpt-5.6-sol", "input": "x" * 40_004} + ) + + assert ("codex_stream_idle_kill" in closes) is (mode == "stall") + assert ("codex_ttfb_kill" in closes) is (mode == "retry_no_event") + if mode == "stall": + assert attempts["count"] == 1 + if mode == "retry_no_event": + assert attempts["count"] == 2 @pytest.mark.parametrize( @@ -252,9 +409,12 @@ def test_wait_notice_omits_reconnect_when_all_deadlines_are_non_finite( ttfb_enabled=False, ttfb_timeout=float("nan"), last_event_ts=None, + last_progress_ts=None, + retry_started_ts=None, call_start=100.0, idle_enabled=False, idle_timeout=float("nan"), + idle_requires_progress=False, elapsed=30.0, ) @@ -378,7 +538,7 @@ def test_wait_notice_formatting_error_does_not_abort_request(monkeypatch): def test_large_codex_request_hard_ceiling_reclaims_silent_stall(tmp_path, monkeypatch): """#64507 regression: a large Codex request (TTFB watchdog disabled by the - size gate, stale floor *raised*) that never emits a single byte must still + size gate, stale floor *raised*) that never emits a parsed event must still be reclaimed at a finite hard ceiling — not hang for 13+ minutes while the worker stays idle and the session shows as active. @@ -389,7 +549,7 @@ def test_large_codex_request_hard_ceiling_reclaims_silent_stall(tmp_path, monkey agent = _make_codex_agent(tmp_path, monkeypatch) # Real default TTFB threshold (no HERMES_CODEX_TTFB_* override) → for a - # >10k-token request the no-byte TTFB watchdog is auto-disabled. + # >10k-token request the no-event TTFB watchdog is auto-disabled. monkeypatch.setenv("HERMES_CODEX_HARD_TIMEOUT_SECONDS", "3") closes: list = [] @@ -428,7 +588,3 @@ def test_large_codex_request_hard_ceiling_reclaims_silent_stall(tmp_path, monkey assert "with no response" in str(excinfo.value) finally: stop["flag"] = True - - - -