ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

fix(compression): dead Codex summary streams fail over in 60s instead of stacking 5-minute waits

Browse Source 
The Codex auxiliary Responses adapter enforced a single absolute
deadline (300s floor for compression). A dead stream held the entire
budget before fallback ran, and repeated compression attempts stacked
those waits into 20+ minute 'Summarizing thread' stalls (masoria debug
bundle, Aug 31 2026). Meanwhile a healthy-but-slow reasoning summary
was killed at the same absolute deadline even while producing tokens.

Replace the absolute kill with progress-aware deadlines:
- 60s no-progress window for the first substantive payload AND between
  payloads; keepalive/lifecycle frames do not re-arm (mirrors the
  commit-fence gating, #96707)
- a live stream re-arms per token and is bounded only by
  _aux_stream_total_ceiling() (max(600s, 4x configured timeout)), the
  same backstop the streamed chat.completions path already uses
- the compression critical-path retry gate now distinguishes failure
  cost: a cheap first-token no-progress failure retries the same
  provider once; mid-stream stalls and ceiling hits still skip straight
  to provider fallback (#54465 semantics preserved)

Live A/B (real OpenAI SDK against a local SSE server, real adapter):
dead keepalive-only stream: main waits the full budget; fixed fails
over at the window. Slow-but-alive stream (tokens past the configured
timeout): main kills it mid-generation; fixed completes.
This commit is contained in:
Teknium
2026-08-31 11:37:22 -07:00
parent f1e01f4daf
commit f50b5bb0fa
2 changed files with 390 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/auxiliary_client.py
+113 -15
View File
@@ -537,6 +537,14 @@ _CODEX_PROGRESS_DELTA_TYPES = frozenset(
)
# Progress-aware auxiliary stream deadlines (Aug 2026, masoria report):
# a dead stream fails fast at the no-progress window (first token AND
# between tokens), a live stream re-arms per substantive event and is
# bounded only by _aux_stream_total_ceiling() (shared with the streamed
# chat.completions path).
_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 = _event_field(event, "type")
@@ -1777,9 +1785,38 @@ class _CodexCompletionsAdapter:
tool_calls_raw: List[Any] = []
usage = None
total_timeout = timeout if isinstance(timeout, (int, float)) and timeout > 0 else None
deadline = time.monotonic() + float(total_timeout) if total_timeout else None
# Progress-aware stream deadlines (supersedes the old single absolute
# kill at ``total_timeout``). Three regimes:
# 1. First token: the stream must produce its first substantive
# payload within ``no_progress_timeout`` (60s default) or we
# fail fast and let the caller's normal retry/fallback chain
# run — a dead (or keepalive-only zombie) Codex stream no
# longer holds the full 300s compression budget before falling
# back (masoria report, Aug 2026: 3 stacked 300s waits ->
# 20+ min stuck on "Summarizing").
# 2. Streaming: every substantive event re-arms the deadline by
# ``no_progress_timeout`` — a live stream is never killed by an
# absolute total, so a long reasoning summary that is actually
# producing tokens completes instead of timing out at 300s and
# falling back (#54915's original complaint, fixed properly).
# Keepalive/lifecycle frames do NOT re-arm, mirroring the
# commit-fence progress gating (#96707).
# 3. Hard ceiling: an absolute backstop from
# ``_aux_stream_total_ceiling`` (max(600s, 4x configured
# timeout) — the same bound the streamed chat.completions path
# uses) so a pathological one-token-per-59s drip still
# terminates.
_start_monotonic = time.monotonic()
no_progress_timeout = _AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS
if total_timeout is not None:
no_progress_timeout = min(no_progress_timeout, float(total_timeout))
hard_deadline = _start_monotonic + _aux_stream_total_ceiling(total_timeout)
deadline_lock = threading.Lock()
progress_deadline = [_start_monotonic + no_progress_timeout]
saw_content = threading.Event()
timed_out = threading.Event()
timeout_timer: Optional[threading.Timer] = None
stream_finished = threading.Event()
timeout_timer: List[Optional[threading.Timer]] = [None]
# A protected provider call may outlive its owning compression attempt:
# the owner returns promptly on hard cancellation while this adapter is
# still blocked in the SDK stream on its isolated worker. Timer threads
@@ -1791,8 +1828,34 @@ class _CodexCompletionsAdapter:
attempt_stream_lock = threading.Lock()
attempt_stream: List[Any] = []
def _effective_deadline() -> float:
with deadline_lock:
return min(hard_deadline, progress_deadline[0])
def _record_stream_progress() -> None:
# A substantive payload re-arms the no-progress window. The hard
# ceiling is never extended.
with deadline_lock:
progress_deadline[0] = time.monotonic() + no_progress_timeout
def _timeout_message() -> str:
return f"Codex auxiliary Responses stream exceeded {float(total_timeout):.1f}s total timeout"
elapsed = time.monotonic() - _start_monotonic
if time.monotonic() >= hard_deadline:
return (
"Codex auxiliary Responses stream exceeded "
f"{hard_deadline - _start_monotonic:.1f}s hard ceiling"
)
if not saw_content.is_set():
return (
"Codex auxiliary Responses stream produced no output "
f"within {float(no_progress_timeout):.1f}s "
f"(no-progress timeout, {elapsed:.1f}s elapsed)"
)
return (
"Codex auxiliary Responses stream stalled: no new output "
f"for {float(no_progress_timeout):.1f}s "
f"({elapsed:.1f}s elapsed)"
)
def _close_client_on_timeout() -> None:
begin_timeout_cleanup = getattr(
@@ -1845,7 +1908,7 @@ class _CodexCompletionsAdapter:
logger.debug("Codex auxiliary: cache eviction on timeout failed", exc_info=True)
def _check_cancelled() -> None:
if deadline is not None and time.monotonic() >= deadline:
if total_timeout is not None and time.monotonic() >= _effective_deadline():
if not timed_out.is_set():
_close_client_on_timeout()
raise TimeoutError(_timeout_message())
@@ -1867,11 +1930,30 @@ class _CodexCompletionsAdapter:
# new failure mode for auxiliary calls.
pass
def _watchdog_fire() -> None:
# Re-armable watchdog: if progress moved the deadline forward
# since this timer was scheduled, reschedule instead of killing
# a live stream. Only kill when the effective deadline (progress
# window or hard ceiling, whichever is sooner) has truly passed.
remaining = _effective_deadline() - time.monotonic()
if remaining > 0:
if timed_out.is_set() or stream_finished.is_set():
return
t = threading.Timer(remaining, _watchdog_fire)
t.daemon = True
timeout_timer[0] = t
t.start()
return
_close_client_on_timeout()
try:
if total_timeout:
timeout_timer = threading.Timer(float(total_timeout), _close_client_on_timeout)
timeout_timer.daemon = True
timeout_timer.start()
timeout_timer[0] = threading.Timer(
max(_effective_deadline() - time.monotonic(), 0.0),
_watchdog_fire,
)
timeout_timer[0].daemon = True
timeout_timer[0].start()
_check_cancelled()
# Event-driven Responses streaming via the low-level
@@ -1903,7 +1985,13 @@ class _CodexCompletionsAdapter:
# compression commit fence) counts only substantive
# payloads — lifecycle and keepalive events must not reset
# the compression idle clock.
# The transport no-progress window likewise re-arms only on
# substantive payloads: a zombie stream that drips SSE
# keepalives but never produces output dies at the same 60s
# window as a fully dead connection.
if _codex_event_has_content(_event):
_record_stream_progress()
saw_content.set()
_notify_aux_provider_response()
else:
_notify_aux_timing_response()
@@ -1998,8 +2086,10 @@ class _CodexCompletionsAdapter:
logger.debug("Codex auxiliary Responses API call failed: %s", exc)
raise
finally:
if timeout_timer is not None:
timeout_timer.cancel()
stream_finished.set()
_t = timeout_timer[0]
if _t is not None:
_t.cancel()
content = "".join(text_parts).strip() or None
@@ -10029,12 +10119,20 @@ def _call_llm_impl(
# fall straight through to provider/model fallback; fast blips (a
# streaming-close or a 5xx) still retry, since those are cheap.
if task == "compression" and _is_timeout_error(transient_err):
logger.info(
"Auxiliary compression: timeout on the critical path; "
"skipping same-provider retry and falling back: %s",
transient_err,
)
raise
# A fast first-token fail (dead stream detected within the
# 60s no-progress window, zero output seen) is cheap — take
# the normal same-provider retry chain first; the provider
# is often fine and only that one stream was stillborn. A
# mid-stream stall or hard-ceiling timeout skips straight to
# fallback, because re-running a multi-minute summary on the
# same provider doubles the user-visible stall (#54465).
if "no-progress timeout" not in str(transient_err):
logger.info(
"Auxiliary compression: timeout on the critical path; "
"skipping same-provider retry and falling back: %s",
transient_err,
)
raise
_max_transient_retries = _transient_retry_count()
_last_transient = transient_err
for _attempt in range(1, _max_transient_retries + 1):
tests/agent/test_codex_aux_no_progress_timeout.py
+277
View File
@@ -0,0 +1,277 @@
"""Progress-aware deadlines for the Codex auxiliary Responses stream.
Regression tests for the masoria report (Aug 2026): each compression
attempt sat the FULL 300s absolute timeout on a dead Codex stream before
falling back, and repeated attempts stacked into a 20+ minute
"Summarizing thread" stall.
New contract for ``_CodexCompletionsAdapter.create``:
1. No first token within the 60s no-progress window -> fail fast
(``no-progress timeout`` in the message) so the fallback chain runs
after ~60s, not 300s.
2. A live stream re-arms the window on every substantive event: a slow
summary that keeps producing tokens is never killed by the old
absolute ``total_timeout``.
3. A mid-stream stall (tokens seen, then silence) dies one no-progress
window after the last token (``stalled`` in the message).
4. The compression critical-path retry gate distinguishes the two: a
cheap first-token failure still gets the same-provider retry; a
full-budget stall skips straight to fallback (#54465 semantics).
"""
import threading
import time
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from agent.auxiliary_client import _CodexCompletionsAdapter, call_llm
def _content_event(text="tok"):
return SimpleNamespace(type="response.output_text.delta", delta=text)
def _keepalive_event():
return SimpleNamespace(type="response.in_progress")
def _make_adapter(event_iter):
real_client = SimpleNamespace(
base_url="https://chatgpt.com/backend-api/codex",
responses=SimpleNamespace(create=lambda **_kwargs: event_iter),
close=lambda: None,
)
return _CodexCompletionsAdapter(real_client, "gpt-5.6-sol")
def _consume(stream, *, model, on_event):
del model
for event in stream:
on_event(event)
return SimpleNamespace(
output=[SimpleNamespace(
type="message",
content=[SimpleNamespace(type="output_text", text="summary")],
)],
usage=None,
)
class TestNoProgressFailFast:
def test_dead_stream_fails_at_no_progress_window_not_total_timeout(self):
"""Keepalive-only stream dies at the (patched) no-progress window,
long before the 300s-style total timeout."""
def _zombie():
deadline = time.monotonic() + 30.0
while time.monotonic() < deadline:
time.sleep(0.02)
yield _keepalive_event()
adapter = _make_adapter(_zombie())
start = time.monotonic()
with (
patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.3),
patch("agent.codex_runtime._consume_codex_event_stream", _consume),
pytest.raises(TimeoutError, match="no-progress timeout"),
):
adapter.create(
messages=[{"role": "user", "content": "summarize"}],
timeout=300,
)
elapsed = time.monotonic() - start
assert elapsed < 5.0, f"fail-fast took {elapsed:.1f}s"
def test_live_stream_outlives_the_old_absolute_total_timeout(self):
"""Tokens arriving inside the window keep the stream alive past
``timeout`` — the old code killed this call at total_timeout."""
def _slow_but_alive():
# 8 tokens, 0.1s apart: total ~0.8s, well past timeout=0.4.
for _ in range(8):
time.sleep(0.1)
yield _content_event()
adapter = _make_adapter(_slow_but_alive())
with (
patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.4),
patch("agent.codex_runtime._consume_codex_event_stream", _consume),
):
response = adapter.create(
messages=[{"role": "user", "content": "summarize"}],
timeout=0.4,
)
assert response.choices[0].message.content == "summary"
def test_mid_stream_stall_raises_stalled_timeout(self):
def _stalls_after_two_tokens():
yield _content_event()
yield _content_event()
deadline = time.monotonic() + 30.0
while time.monotonic() < deadline:
time.sleep(0.02)
yield _keepalive_event()
adapter = _make_adapter(_stalls_after_two_tokens())
start = time.monotonic()
with (
patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.3),
patch("agent.codex_runtime._consume_codex_event_stream", _consume),
pytest.raises(TimeoutError, match="stalled: no new output"),
):
adapter.create(
messages=[{"role": "user", "content": "summarize"}],
timeout=300,
)
assert time.monotonic() - start < 5.0
def test_hard_ceiling_bounds_a_token_drip(self):
"""A degenerate one-token-per-window drip still terminates at the
_aux_stream_total_ceiling backstop."""
def _dripper():
while True:
time.sleep(0.05)
yield _content_event()
adapter = _make_adapter(_dripper())
with (
patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 5.0),
patch("agent.auxiliary_client._aux_stream_total_ceiling",
return_value=0.3),
patch("agent.codex_runtime._consume_codex_event_stream", _consume),
pytest.raises(TimeoutError, match="hard ceiling"),
):
adapter.create(
messages=[{"role": "user", "content": "summarize"}],
timeout=300,
)
def test_watchdog_timer_fires_while_blocked_before_first_event(self):
"""responses.create() itself can block with zero bytes; the re-armable
watchdog must close the client and surface the no-progress timeout
without any event ever reaching _check_cancelled."""
release = threading.Event()
def _blocked_create(**_kwargs):
release.wait(timeout=30.0)
return iter([])
closed = threading.Event()
real_client = SimpleNamespace(
base_url="https://chatgpt.com/backend-api/codex",
responses=SimpleNamespace(create=_blocked_create),
close=closed.set,
)
adapter = _CodexCompletionsAdapter(real_client, "gpt-5.6-sol")
try:
with (
patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.3),
patch("agent.auxiliary_client._evict_cached_client_instance"),
):
# The watchdog closes the shared client at the window; the
# blocked create keeps waiting (SimpleNamespace has no real
# transport), so unblock it and verify the timeout surfaced.
waiter: dict = {}
def _run():
try:
adapter.create(
messages=[{"role": "user", "content": "x"}],
timeout=300,
)
except Exception as exc: # noqa: BLE001
waiter["exc"] = exc
t = threading.Thread(target=_run, daemon=True)
t.start()
assert closed.wait(timeout=5.0), "watchdog never closed client"
release.set()
t.join(timeout=5.0)
assert isinstance(waiter.get("exc"), TimeoutError)
assert "no-progress timeout" in str(waiter["exc"])
finally:
release.set()
class TestCompressionRetryGate:
"""First-token failures retry same-provider; stalls skip to fallback."""
def _run_call_llm(self, primary_error, second_response=None):
primary_client = MagicMock()
primary_client.base_url = "https://chatgpt.com/backend-api/codex"
if second_response is not None:
primary_client.chat.completions.create.side_effect = [
primary_error, second_response,
]
else:
primary_client.chat.completions.create.side_effect = primary_error
fallback_client = MagicMock()
fallback_client.chat.completions.create.return_value = SimpleNamespace(
choices=[SimpleNamespace(
index=0,
message=SimpleNamespace(role="assistant", content="fallback"),
finish_reason="stop",
)],
model="fb", usage=None,
)
with (
patch("agent.auxiliary_client._get_cached_client",
return_value=(primary_client, "gpt-5.6-sol")),
patch("agent.auxiliary_client._resolve_task_provider_model",
return_value=("auto", "gpt-5.6-sol", None, None, None)),
patch("agent.auxiliary_client._try_configured_fallback_chain",
return_value=(None, None, "")),
patch("agent.auxiliary_client._try_main_fallback_chain",
return_value=(None, None, "")),
patch("agent.auxiliary_client._try_payment_fallback",
return_value=(fallback_client, "fb", "openrouter")) as mock_fb,
patch("agent.auxiliary_client._TRANSIENT_RETRY_BACKOFF_BASE", 0.0),
):
result = call_llm(
task="compression",
messages=[{"role": "user", "content": "summarize"}],
)
return result, primary_client, mock_fb
def test_no_progress_timeout_retries_same_provider(self):
err = TimeoutError(
"Codex auxiliary Responses stream produced no output within "
"60.0s (no-progress timeout, 60.2s elapsed)"
)
good = SimpleNamespace(
choices=[SimpleNamespace(
index=0,
message=SimpleNamespace(role="assistant", content="retried"),
finish_reason="stop",
)],
model="gpt-5.6-sol", usage=None,
)
result, primary, mock_fb = self._run_call_llm(err, second_response=good)
assert result.choices[0].message.content == "retried"
assert primary.chat.completions.create.call_count == 2
assert not mock_fb.called
def test_stalled_timeout_skips_same_provider_retry(self):
err = TimeoutError(
"Codex auxiliary Responses stream stalled: no new output for "
"60.0s (247.3s elapsed)"
)
result, primary, mock_fb = self._run_call_llm(err)
assert result.choices[0].message.content == "fallback"
assert primary.chat.completions.create.call_count == 1
assert mock_fb.called
def test_hard_ceiling_timeout_skips_same_provider_retry(self):
err = TimeoutError(
"Codex auxiliary Responses stream exceeded 600.0s hard ceiling"
)
result, primary, mock_fb = self._run_call_llm(err)
assert result.choices[0].message.content == "fallback"
assert primary.chat.completions.create.call_count == 1
assert mock_fb.called