281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""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 mark the timeout (without releasing FDs from its own
|
|
stranger thread — #29507) and the owning thread must surface the
|
|
no-progress TimeoutError and perform the close on unwind."""
|
|
release = threading.Event()
|
|
|
|
def _blocked_create(**_kwargs):
|
|
release.wait(timeout=30.0)
|
|
return iter([])
|
|
|
|
closed_by = []
|
|
real_client = SimpleNamespace(
|
|
base_url="https://chatgpt.com/backend-api/codex",
|
|
responses=SimpleNamespace(create=_blocked_create),
|
|
close=lambda: closed_by.append(threading.get_ident()),
|
|
)
|
|
adapter = _CodexCompletionsAdapter(real_client, "gpt-5.6-sol")
|
|
owner_result: dict = {}
|
|
try:
|
|
with (
|
|
patch("agent.auxiliary_client._AUX_STREAM_NO_PROGRESS_TIMEOUT_SECONDS", 0.3),
|
|
patch("agent.auxiliary_client._evict_cached_client_instance"),
|
|
):
|
|
def _run():
|
|
owner_result["tid"] = threading.get_ident()
|
|
try:
|
|
adapter.create(
|
|
messages=[{"role": "user", "content": "x"}],
|
|
timeout=300,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
owner_result["exc"] = exc
|
|
|
|
t = threading.Thread(target=_run, daemon=True)
|
|
t.start()
|
|
# Watchdog fires within the patched window; it must NOT
|
|
# close() from its own thread (FD ownership, #29507).
|
|
time.sleep(1.0)
|
|
assert not closed_by, f"stranger-thread close: {closed_by}"
|
|
release.set()
|
|
t.join(timeout=5.0)
|
|
assert isinstance(owner_result.get("exc"), TimeoutError)
|
|
assert "no-progress timeout" in str(owner_result["exc"])
|
|
# The OWNER released the FDs on unwind.
|
|
assert closed_by == [owner_result["tid"]], closed_by
|
|
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
|