fix(agent): defer codex idle watchdog until progress

Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>

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)."""

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)

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:

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
 
 

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
-
-
-
-