"""Request-local worker lifecycle, watchdog polling, and wait status.""" from agent import chat_completion_helpers as h from agent import chat_completion_wait_notice as wn class _NonStreamRequest: """One non-streaming request on a worker thread, polled by the caller. State shared between the worker (``_call``) and the poll loop lives on the instance; ``_abort_request`` may run from the poll (stranger) thread. """ def __init__(self, agent, api_kwargs: dict): self.agent = agent self.api_kwargs = api_kwargs self.result = {"response": None, "error": None} self.clients = h._RequestClientRegistry(agent) # Request-local cancel flag: agent._interrupt_requested is cleared at turn # boundaries but this daemon worker can outlive the turn, so it must know THIS # request was force-closed and not surface the transport error as a bug (#6600). self.cancelled = False # Codex retirement token: the worker checks ``agent._active_codex_stream_request_token`` # to know it still owns the turn; a watchdog kill clears it so a worker still # draining SSE raises instead of returning partial output as "completed" # (run_codex_stream._request_is_current). ``codex_retired`` mirrors it locally. self.codex_token = object() if agent.api_mode == "codex_responses" else None self.codex_retired = False self.wd = h._resolve_nonstream_watchdogs(agent, api_kwargs) self.codex_watchdog_state = ( h.SimpleNamespace( token=self.codex_token, lock=h.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 = h.time.time() self.wait_notice_started_ts = None self.wait_notice = wn.WaitNoticeState() self.thread = None def _install_codex_request_token(self) -> None: if self.codex_token is not None and not self.codex_retired: # retired before start: don't re-publish self.agent._active_codex_stream_request_token = self.codex_token def _retire_codex_request_token(self) -> None: if self.codex_token is None: return self.codex_retired = True if getattr(self.agent, "_active_codex_stream_request_token", None) is self.codex_token: self.agent._active_codex_stream_request_token = None def _make_client(self, reason: str, kind: str = "openai"): # Per-request clients are registered with the abort machinery so the watchdogs # force-close the worker's connection, never the shared client (#67142). if kind == "anthropic_messages": client = self.agent._create_request_anthropic_client(reason=reason) else: client = self.agent._create_request_openai_client(reason=reason, api_kwargs=self.api_kwargs) 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"] = h._dispatch_nonstreaming_api_request( self.agent, self.api_kwargs, make_client=self._make_client) except Exception as e: # Our own force-close caused this error: swallow it, the main # thread raises InterruptedError (#6600). Retirement logs at info # (a watchdog discarded output the provider already sent — what an # operator debugging a truncated reply needs); cancellation at debug. if self.codex_retired: h.logger.info("Codex worker caught %s after request retirement — " "discarding the stale partial instead of surfacing it as a completed response. %s", type(e).__name__, self.agent._client_log_context()) return if self.cancelled: h.logger.debug("Non-streaming worker caught %s after request " "cancellation — exiting without surfacing a network error.", type(e).__name__) 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() # Reuse reason only on a clean response; error or cancel-swallow # really closes so the next attempt builds a fresh pool. self.clients.close_once( "request_complete" if self.result["response"] is not None else "request_error_cleanup") def _abort_request(self, reason: str) -> None: """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.""" with h.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. self.thread.join(timeout=2.0) if self.result["error"] is None and self.result["response"] is None: self.result["error"] = TimeoutError(timeout_message) def _model(self) -> str: return self.api_kwargs.get("model", "unknown") def _codex_watchdog_snapshot(self): """``(last_event_ts, last_progress_ts, retry_started_ts, attempt_started_ts)`` under one lock. ``attempt_started_ts`` is the physical attempt's origin (the reconnect marker, else ``call_start``); the notice and the kill loop must anchor to the same value.""" state = self.codex_watchdog_state if state is None: # non-codex request: no watchdog reads these return (None, None, None, self.call_start) with state.lock: retry_started_ts = state.retry_started_ts return (state.last_event_ts, state.last_progress_ts, retry_started_ts, retry_started_ts if retry_started_ts is not None else self.call_start) def _pre_progress(self, last_event_ts, last_progress_ts) -> bool: """Stream open on this attempt, but no substantive model progress yet.""" return self.wd.progress_timeout > 0 and last_event_ts is not None and last_progress_ts is None def _emit_wait_notice(self, elapsed: float, *, heartbeat: bool = True) -> None: wd = self.wd try: last_event_ts, last_progress_ts, retry_started_ts, attempt_started_ts = self._codex_watchdog_snapshot() pre_progress = self._pre_progress(last_event_ts, last_progress_ts) activity_ts = attempt_started_ts if pre_progress else ( retry_started_ts if retry_started_ts is not None else last_event_ts) # Lifecycle chatter is not progress: once pre-progress begins it must not # clear/reset this notice. Real progress changes phase and clears it. if (self.wait_notice_started_ts is not None and activity_ts is not None and activity_ts > self.wait_notice_started_ts): self.agent._emit_wait_notice("") self.wait_notice_started_ts = None self.wait_notice.reset() if not heartbeat: return silence = self.call_start + elapsed - ( activity_ts if activity_ts is not None else self.call_start) if silence < 60.0: self.agent._touch_activity( "waiting for first stream event after reconnect" if retry_started_ts is not None else "waiting for provider response") return phase = "first_event" if pre_progress: phase = "pre_progress" elif retry_started_ts is not None: phase = "reconnect" elif last_event_ts is not None: phase = "post_event" watchdog = wn.codex_watchdog_deadline(stale_timeout=wd.stale_timeout, ttfb_enabled=wd.ttfb_enabled, ttfb_timeout=wd.ttfb_timeout, 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, progress_timeout=wd.progress_timeout, elapsed=elapsed) # One neutral notice per silence; repeating it every heartbeat made # healthy long calls read as provider trouble (#92550). if not self.wait_notice.should_emit(phase, watchdog): self.agent._touch_activity(f"waiting for provider response ({int(silence)}s, {phase})") return self.agent._emit_wait_notice(wn.wait_notice_text( self.api_kwargs.get('model', 'the provider'), silence, phase, watchdog)) self.wait_notice_started_ts = self.call_start + elapsed except Exception: h.logger.debug("wait-notice construction failed", exc_info=True) def _ttfb_kill(self, elapsed: float) -> None: """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 = h._codex_silent_hang_hint(agent, self.api_kwargs) h.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_diagnostic_status( 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_diagnostic_wait(f"⚠ no response from provider in {int(elapsed)}s — reconnecting...") 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 parsed stream event within {int(elapsed)}s " f"(TTFB threshold: {int(wd.ttfb_timeout)}s)" + (f". {silent_hint}" if silent_hint else "")) def _progress_kill(self, elapsed: float) -> None: """The stream opened, but this physical attempt never made model progress.""" agent, wd = self.agent, self.wd h.logger.warning("Codex stream produced lifecycle events but no substantive model progress " "for %.0fs (threshold %.0fs, model=%s, context=~%s tokens). Reconnecting.", elapsed, wd.progress_timeout, self._model(), f"{wd.est_tokens:,}") agent._buffer_diagnostic_status( f"⚠️ Codex stream opened but made no model progress for {int(elapsed)}s " f"(model: {self._model()}). Reconnecting.") self._abort_request("codex_progress_kill") agent._touch_activity(f"codex stream killed after {int(elapsed)}s without model progress") self._await_worker_after_kill( f"Codex stream produced no substantive model progress for {int(elapsed)}s " f"(progress threshold: {int(wd.progress_timeout)}s)") def _idle_kill(self, event_stale_elapsed: float) -> None: """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 arm_point = "model progress began" if wd.idle_requires_progress else "the first parsed event" h.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, arm_point, wd.idle_timeout, self._model(), f"{wd.est_tokens:,}") agent._buffer_diagnostic_status( 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 {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 circuit breaker (#58962, see ``_stale_streak``).""" agent, wd = self.agent, self.wd silent_hint = h._codex_silent_hang_hint(agent, self.api_kwargs) h._report_stale_nonstream_kill(agent, self.api_kwargs, elapsed, wd.stale_timeout, hint=silent_hint) self._abort_request("stale_call_kill") h._bump_stale_streak(agent) h._touch_stale_kill_activity(agent, elapsed) self._await_worker_after_kill( f"Non-streaming API call timed out after {int(elapsed)}s with no response (threshold: {int(wd.stale_timeout)}s)" + (f". {silent_hint}" if silent_hint else "")) def _interrupt(self, elapsed: float) -> None: agent = self.agent last_event_ts, _, _, _ = self._codex_watchdog_snapshot() h._record_interrupted_provider_wait(agent, elapsed, 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 # TLS FD mid-SSL-BIO corrupted an unrelated SQLite DB, #67142). Then let # the worker unwind Relay scopes before raising (#81521). self.cancelled = True h.logger.debug("Force-closing httpx client due to interrupt (not a network error).") self._abort_request("interrupt_abort") h._join_worker_for_relay_teardown(self.thread, label="Non-streaming") raise InterruptedError("Agent interrupted during API call") def run(self): 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 the first event for this one. 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 = h.threading.Thread(target=h._context_thread_target(self._call), daemon=True) t.start() poll_count = 0 while t.is_alive(): t.join(timeout=0.3) poll_count += 1 # Keep the quiet gateway heartbeat; only silence warrants a notice. # Resumed events clear our notice on the next poll, not 30s later. now = h.time.time() elapsed = now - self.call_start self._emit_wait_notice(elapsed, heartbeat=poll_count % 100 == 0) last_event_ts, last_progress_ts, retry_started_ts, attempt_started_ts = self._codex_watchdog_snapshot() attempt_elapsed = now - attempt_started_ts if (wd.ttfb_enabled and last_event_ts is None and attempt_elapsed > wd.ttfb_timeout): self._ttfb_kill(attempt_elapsed) break if (self._pre_progress(last_event_ts, last_progress_ts) and attempt_elapsed > wd.progress_timeout): self._progress_kill(attempt_elapsed) break 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) break if agent._interrupt_requested: self._interrupt(elapsed) if self.result["error"] is not None: raise self.result["error"] # Success — the provider proved responsive: clear the breaker (#58962). if self.result["response"] is not None: h._reset_stale_streak(agent) return self.result["response"]