From c8c268e37d84e83ee242165bf38d0140642eb414 Mon Sep 17 00:00:00 2001 From: yebof Date: Thu, 16 Jul 2026 02:40:16 +0800 Subject: [PATCH 1/2] feat(agents): Harden the LLM call path against relay failure modes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five changes to src/agents/base.py, all motivated by observed production incidents on the OPENAI_BASE_URL relay: - Stream OpenAI calls (stream=True + include_usage). The relay sits behind Cloudflare's ~120s proxy read timeout (HTTP 524), so any non-streamed generation over 120s could never succeed (2026-06-08/09: every long call 524'd and mornings died). Streaming keeps bytes flowing; missing usage falls back to a loud chars/4 estimate instead of 0/0. - Wall-clock retry deadline (480s, QUANT_AGENT_RETRY_DEADLINE_S). The attempt budget alone doesn't bound time: 7 attempts at 120-380s each collided with the wrapper's 1200s kill, so the Anthropic failover below the loop never fired in exactly the sustained-outage scenario it was built for. Past the deadline the primary is abandoned and failover runs while the session window still has room (480 + 300 failover < 1200). - Honor server retry-after hints on 429/524 (header, body field, or message text), capped at 120s so a hostile hint can't stall a session. Pure exponential jitter retried in 2-15s against a server that said 'come back in 120s', burning attempts for nothing. - Per-provider in-flight semaphores around the HTTP call (OpenAI 3, QUANT_AGENT_MAX_CONCURRENT_LLM; Anthropic 4, independent so failover never queues behind a wedged relay slot). The morning fan-out self-inflicted 'Concurrency limit exceeded' 429 storms against the relay's per-user cap (175 occurrences in the 06-16..06-29 logs). - max_retries=0 on every SDK client: both SDKs default to 2 internal retries, silently tripling each agent-loop attempt and invalidating the retry-budget/deadline math. The agent loop is the single retry owner. Degenerate responses now fail loudly instead of passing as clean no-signals: an empty 200 body raises LLMEmptyResponseError and a stream that ends without finish_reason raises LLMStreamInterruptedError (partial text discarded — a half-emitted PM decision parses like 'no trades'). Both are retryable and reach the failover. Truncation-family finish reasons (max_tokens / length / insufficient_system_resource) are exempt: an empty body there is a legit ceiling hit surfaced via truncated=True, never a retry (shared _TRUNCATION_FINISH_REASONS constant). Tests: streaming assembly + usage fallback, interrupt/empty guards on all three providers, deadline-triggered failover, hint floor + cap, SDK internal-retry disable on all constructors, semaphore leak check. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01PQkESoSTYx2bCy7WYnTPXR --- src/agents/base.py | 325 ++++++++++++++++++++++++++++++++++----- tests/test_base_agent.py | 319 +++++++++++++++++++++++++++++++++++++- 2 files changed, 599 insertions(+), 45 deletions(-) diff --git a/src/agents/base.py b/src/agents/base.py index 019779f5..e088f9d6 100644 --- a/src/agents/base.py +++ b/src/agents/base.py @@ -3,6 +3,8 @@ import os import random import re +import threading +import time from abc import ABC, abstractmethod from dataclasses import dataclass @@ -99,6 +101,139 @@ def _retry_backoff_seconds(attempt: int) -> float: # src/execution/broker.py. _LLM_HTTP_TIMEOUT = 300.0 +# Wall-clock deadline for the PRIMARY retry loop in _execute(), seconds. +# +# Why a deadline at all: the attempt budget alone doesn't bound time. Under +# the relay's Cloudflare 524 mode each attempt burned 120-380s, so exhausting +# 7 attempts needed 40+ minutes — the wrapper SIGKILLed the session at 1200s +# mid-loop and the Anthropic failover (which fires only AFTER the loop) never +# ran in exactly the sustained-outage scenario it was built for (2026-06-08/09: +# two days of mornings died with a funded failover key sitting idle). +# +# Why 480: it must leave room for one full failover call inside the wrapper's +# 1200s kill. Worst-case failover = one Anthropic call bounded by +# _LLM_HTTP_TIMEOUT (300s), so 480 + 300 = 780s per agent, ~420s of headroom +# for the rest of the session. And 480s still allows 2-4 real primary attempts +# even in the slow-failure mode (~120-380s each), so a transient blip is +# ridden out before the failover engages. +# +# Overridable via QUANT_AGENT_RETRY_DEADLINE_S (read at call time, like +# _max_retries, so tests can monkeypatch per case). +_DEFAULT_RETRY_DEADLINE_S = 480.0 + + +def _retry_deadline_s() -> float: + raw = os.environ.get("QUANT_AGENT_RETRY_DEADLINE_S") + if raw is None: + return _DEFAULT_RETRY_DEADLINE_S + try: + v = float(raw) + except ValueError: + return _DEFAULT_RETRY_DEADLINE_S + return max(1.0, v) + + +# Server-provided retry hints. The relay's 429 ("Concurrency limit exceeded") +# and 524 payloads carry retry-after semantics (a Retry-After header and/or a +# "retry_after": N field in the JSON body) that pure exponential jitter +# ignored — agents retried in 2-15s against a server that said "come back in +# 120s", burning attempts for nothing. We sleep max(backoff, hint), capped so +# a hostile/buggy hint can't park an agent past the session window. +_RETRY_AFTER_CAP_S = 120.0 + + +def _retry_after_hint_seconds(exc: Exception) -> float | None: + """Best-effort extraction of a server retry-after hint from an SDK error. + + Looks in (a) the Retry-After header of the attached httpx response + (numeric-seconds form only — the HTTP-date form isn't worth parsing for + a hint), (b) a retry_after field in the error body dict, (c) the message + text (relay 524 bodies embed '"retry_after": 120'). Returns None when no + usable hint exists; never raises. + """ + response = getattr(exc, "response", None) + headers = getattr(response, "headers", None) + if headers is not None: + try: + raw = headers.get("retry-after") + except Exception: # noqa: BLE001 — a weird headers object must not mask the real error + raw = None + if raw is not None: + try: + return max(0.0, float(raw)) + except (TypeError, ValueError): + pass + body = getattr(exc, "body", None) + if isinstance(body, dict): + val = body.get("retry_after", body.get("retry-after")) + if isinstance(val, (int, float)) and not isinstance(val, bool): + return max(0.0, float(val)) + m = re.search(r'retry[_-]after["\']?\s*[:=]\s*"?(\d+(?:\.\d+)?)', str(exc), re.IGNORECASE) + if m: + return float(m.group(1)) + return None + + +def _int_env(name: str, default: int) -> int: + raw = os.environ.get(name) + if raw is None: + return default + try: + return max(1, int(raw)) + except ValueError: + return default + + +# Per-provider in-flight caps around the LLM HTTP call itself (NOT around +# run() — building the user message / parsing must never hold a slot). +# +# Why: the relay enforces a per-user concurrency cap, and morning fans out +# macro + news + tech (multi-chunk) + earnings through a +# ThreadPoolExecutor(max_workers=4) on one relay account — the fan-out +# self-inflicted "Concurrency limit exceeded" 429 storms (175 occurrences in +# the 06-16..06-29 logs), and each kill-looped morning re-spawned the full +# team into the already-limited relay. A module-level semaphore serializes +# the excess instead of bouncing it off the server. Anthropic is direct +# (no relay) so it gets a looser, independent cap — failover calls must not +# queue behind a wedged relay slot. +_OPENAI_MAX_CONCURRENT = _int_env("QUANT_AGENT_MAX_CONCURRENT_LLM", 3) +_OPENAI_LLM_SEMAPHORE = threading.Semaphore(_OPENAI_MAX_CONCURRENT) +_ANTHROPIC_MAX_CONCURRENT = 4 +_ANTHROPIC_LLM_SEMAPHORE = threading.Semaphore(_ANTHROPIC_MAX_CONCURRENT) + + +# finish/stop reasons that mean "output hit a ceiling mid-generation". +# Shared by the truncation flag in _execute() and the empty-content guards: +# an empty body WITH one of these reasons is a legitimate truncation (e.g. a +# reasoner burning the whole budget on CoT) that must surface as +# truncated=True, NOT trigger retry/failover (truncation never fails over — +# see CLAUDE.md). insufficient_system_resource is DeepSeek-specific: the +# inference system ran out of resources and returned a cut-off body on a 200. +_TRUNCATION_FINISH_REASONS = ("max_tokens", "length", "insufficient_system_resource") + + +class LLMEmptyResponseError(RuntimeError): + """HTTP 200 whose body carries no usable content (choices empty / + content None or ""). Previously returned as a *successful* '' — which + parses to None downstream and masquerades as a deliberate no-signal, + consuming the agent's one shot for the session while bypassing both the + retry budget and the Anthropic failover. Raised instead, and classified + retryable (a degenerate 200 from a relay is transient territory).""" + + +class LLMStreamInterruptedError(RuntimeError): + """A streamed response ended without a finish_reason — the connection + was cut mid-generation (relay/proxy drop, no error frame). Partial text + is NOT a success: a half-emitted PM decision parses like 'no trades'. + Retryable.""" + + +def _estimate_tokens(text: str) -> int: + """~4 chars/token heuristic — usage-chunk fallback only, so a relay that + doesn't honor stream_options include_usage still yields a nonzero cost + estimate instead of a 0/0 that run() flags as cost-unknown.""" + return max(1, len(text) // 4) if text else 0 + def _max_retries() -> int: """Read at call time so tests can monkeypatch the env var per case @@ -138,6 +273,10 @@ def _is_deepseek_model(model: str) -> bool: "APIConnectionError", "APITimeoutError", "APIConnectionTimeoutError", "InternalServerError", "RateLimitError", "APIError", "Timeout", "ConnectionError", "ConnectTimeout", "ReadTimeout", + # Our own degenerate-response classes (see definitions above): explicit + # here so they stay retryable even if the unknown-exception fallback in + # _is_retryable is ever tightened. + "LLMEmptyResponseError", "LLMStreamInterruptedError", }) @@ -293,11 +432,17 @@ def __init__(self, api_key: str, model: str, max_tokens: int = 4096, # back to Claude" can't crash construction. Empty => failover disabled. self._fallback_api_key = (fallback_api_key or "").strip() + # max_retries=0 on EVERY SDK client construction: both SDKs default to + # 2 internal retries on 429/5xx (incl. the relay's CF 524) with their + # own backoff, silently turning each _execute() attempt into ~3 HTTP + # calls (~380s under sustained 524s) and invalidating the retry-budget + # math documented at _DEFAULT_MAX_RETRIES. The agent-level loop in + # _execute() is the SINGLE owner of retry policy. if self._use_deepseek: # OpenAI-compatible endpoint at a custom base_url with the DeepSeek key. from openai import OpenAI self.client = OpenAI(api_key=api_key, base_url=_DEEPSEEK_BASE_URL, - timeout=_LLM_HTTP_TIMEOUT) + timeout=_LLM_HTTP_TIMEOUT, max_retries=0) elif self._use_openai: from openai import OpenAI # OPENAI_BASE_URL lets OpenAI traffic go through an OpenAI-compatible @@ -320,12 +465,14 @@ def __init__(self, api_key: str, model: str, max_tokens: int = 4096, self.client = OpenAI( api_key=api_key, base_url=base_url, http_client=httpx.Client(verify=ca_bundle, timeout=_LLM_HTTP_TIMEOUT), + max_retries=0, ) else: - self.client = OpenAI(api_key=api_key, base_url=base_url, timeout=_LLM_HTTP_TIMEOUT) + self.client = OpenAI(api_key=api_key, base_url=base_url, + timeout=_LLM_HTTP_TIMEOUT, max_retries=0) else: from anthropic import Anthropic - self.client = Anthropic(api_key=api_key, timeout=_LLM_HTTP_TIMEOUT) + self.client = Anthropic(api_key=api_key, timeout=_LLM_HTTP_TIMEOUT, max_retries=0) @property @abstractmethod @@ -355,6 +502,8 @@ def _execute(self, user_message: str) -> AgentResult: logger.info("Agent %s input:\n%s", self.name, user_message) max_retries = _max_retries() + deadline_s = _retry_deadline_s() + loop_start = time.monotonic() finish_reason: str | None = None primary_error: Exception | None = None for attempt in range(max_retries): @@ -384,10 +533,30 @@ def _execute(self, user_message: str) -> AgentResult: logger.warning("Agent %s attempt %d failed: %s. Primary exhausted.", self.name, attempt + 1, e) break + # Wall-clock deadline: the attempt budget alone doesn't bound + # time (each attempt can burn 120-380s in the relay-524 mode), + # so exhausting it can collide with the wrapper's 1200s kill — + # which is where the failover below became unreachable + # (2026-06-08/09). Past the deadline, abandon the primary NOW + # so failover fires while the session window still has room. + elapsed = time.monotonic() - loop_start + if elapsed >= deadline_s: + logger.warning( + "Agent %s attempt %d failed: %s. Retry deadline %.0fs " + "exceeded (elapsed %.0fs) — abandoning primary, " + "proceeding to failover if configured.", + self.name, attempt + 1, e, deadline_s, elapsed, + ) + break wait = _retry_backoff_seconds(attempt) + # Honor a server retry-after hint (429/5xx): sleeping shorter + # than the server asked just burns attempts against a closed + # door. Capped so a hostile hint can't stall the session. + hint = _retry_after_hint_seconds(e) + if hint is not None: + wait = min(max(wait, hint), _RETRY_AFTER_CAP_S) logger.warning("Agent %s attempt %d failed: %s. Retrying in %.1fs...", self.name, attempt + 1, e, wait) - import time time.sleep(wait) # Model that actually produced the output — primary unless failover wins. @@ -411,13 +580,11 @@ def _execute(self, user_message: str) -> AgentResult: # Truncation detection: a max_tokens / length cutoff means the output # is incomplete, NOT a deliberate "no action". Flag + log loudly so a # truncated decision isn't silently collapsed into "no trades". - truncated = isinstance(finish_reason, str) and finish_reason.lower() in ( - # max_tokens (Anthropic) / length (OpenAI+DeepSeek) = hit the ceiling. - # insufficient_system_resource is DeepSeek-specific: the inference - # system ran out of resources and returned a cut-off body on a 200 — - # incomplete output, so flag it the same as a token-limit truncation. - "max_tokens", "length", "insufficient_system_resource", - ) + # max_tokens (Anthropic) / length (OpenAI+DeepSeek) = hit the ceiling; + # insufficient_system_resource = DeepSeek cut-off-on-200. Shared + # constant with the empty-content guards in the _call_* paths. + truncated = (isinstance(finish_reason, str) + and finish_reason.lower() in _TRUNCATION_FINISH_REASONS) if truncated: logger.warning( "Agent %s response was TRUNCATED (finish_reason=%s) — output is " @@ -471,23 +638,30 @@ def _anthropic_call(self, client, model: str, user_message: str) -> tuple[str, i cache breakpoint (static per agent → cheaper + lower latency; no-op below the cache minimum, safe unconditionally). """ - response = client.messages.create( - model=model, - max_tokens=self.max_tokens, - system=[{ - "type": "text", - "text": self.system_prompt, - "cache_control": {"type": "ephemeral"}, - }], - messages=[{"role": "user", "content": user_message}], - ) + with _ANTHROPIC_LLM_SEMAPHORE: + response = client.messages.create( + model=model, + max_tokens=self.max_tokens, + system=[{ + "type": "text", + "text": self.system_prompt, + "cache_control": {"type": "ephemeral"}, + }], + messages=[{"role": "user", "content": user_message}], + ) in_tok, out_tok = _extract_anthropic_usage(response, self.name) finish_reason = getattr(response, "stop_reason", None) if not isinstance(finish_reason, str): finish_reason = None if not response.content or not hasattr(response.content[0], "text"): - logger.warning("Anthropic returned empty content (stop_reason=%s)", finish_reason) - return ("", in_tok, out_tok, finish_reason) + if finish_reason in _TRUNCATION_FINISH_REASONS: + # Legit truncation (whole budget burned before any text) — + # surface as truncated '', don't retry/fail over. + logger.warning("Anthropic returned empty content (stop_reason=%s)", finish_reason) + return ("", in_tok, out_tok, finish_reason) + raise LLMEmptyResponseError( + f"Anthropic returned empty content (stop_reason={finish_reason})" + ) return (response.content[0].text, in_tok, out_tok, finish_reason) def _call_anthropic(self, user_message: str) -> tuple[str, int, int, str | None]: @@ -508,7 +682,8 @@ def _try_failover(self, user_message: str, primary_error: Exception): ) try: from anthropic import Anthropic - client = Anthropic(api_key=self._fallback_api_key, timeout=_LLM_HTTP_TIMEOUT) + client = Anthropic(api_key=self._fallback_api_key, + timeout=_LLM_HTTP_TIMEOUT, max_retries=0) result = self._anthropic_call(client, _FALLBACK_MODEL, user_message) logger.warning( "Agent %s: FAILOVER to %s SUCCEEDED (in=%d out=%d) — session " @@ -523,23 +698,82 @@ def _try_failover(self, user_message: str, primary_error: Exception): return None def _call_openai(self, user_message: str) -> tuple[str, int, int, str | None]: - response = self.client.chat.completions.create( - model=self.model, - max_completion_tokens=self.max_tokens, - messages=[ - {"role": "system", "content": self.system_prompt}, - {"role": "user", "content": user_message}, - ], - ) - choice = response.choices[0] - content = choice.message.content or "" - if not content: - refusal = getattr(choice.message, "refusal", None) - logger.warning("OpenAI returned empty content (refusal=%s)", refusal) - in_tok, out_tok = _extract_openai_usage(response, self.name) - finish_reason = getattr(choice, "finish_reason", None) - if not isinstance(finish_reason, str): - finish_reason = None + """OpenAI path is STREAMED on purpose. The OPENAI_BASE_URL relay sits + behind Cloudflare, whose ~120s Proxy Read Timeout (HTTP 524) kills any + call that sends zero bytes until the model finishes — and PM / tech / + evening generations legitimately run 120s+, so non-streaming could + never succeed through the relay (the 2026-06-08/09 outage: every long + call 524'd, book froze sell-only). Streaming keeps bytes flowing so + the proxy window never trips; _LLM_HTTP_TIMEOUT becomes a per-chunk + read timeout, so a long *healthy* generation isn't axed either. + + The semaphore covers create + iteration: for a streamed response the + request is in flight (and counts against the relay's per-user + concurrency cap) until the last chunk is read. + """ + with _OPENAI_LLM_SEMAPHORE: + stream = self.client.chat.completions.create( + model=self.model, + max_completion_tokens=self.max_tokens, + messages=[ + {"role": "system", "content": self.system_prompt}, + {"role": "user", "content": user_message}, + ], + stream=True, + stream_options={"include_usage": True}, + ) + parts: list[str] = [] + finish_reason: str | None = None + usage = None + for chunk in stream: + # include_usage delivers usage on a final extra chunk whose + # choices list is empty. + chunk_usage = getattr(chunk, "usage", None) + if chunk_usage is not None: + usage = chunk_usage + choices = getattr(chunk, "choices", None) or [] + if not choices: + continue + choice = choices[0] + delta = getattr(choice, "delta", None) + piece = getattr(delta, "content", None) if delta is not None else None + if piece: + parts.append(piece) + fr = getattr(choice, "finish_reason", None) + if isinstance(fr, str): + finish_reason = fr + content = "".join(parts) + if finish_reason is None: + # Stream ended without a finish_reason = connection cut + # mid-generation (relay drop, no error frame). Partial text must + # NOT be returned as success — a half-emitted PM decision parses + # like 'no trades'. Raise retryable instead. + raise LLMStreamInterruptedError( + f"OpenAI stream ended without finish_reason after {len(content)} " + "chars — connection cut mid-generation; partial output discarded" + ) + if not content and finish_reason not in _TRUNCATION_FINISH_REASONS: + # Degenerate 200 (empty body / refusal / stripped relay response): + # entering the retry → failover machinery beats masquerading as a + # clean no-signal. Truncation-family reasons are exempt — an empty + # body there is a legit ceiling hit, flagged via truncated=True. + raise LLMEmptyResponseError( + f"OpenAI returned empty content (finish_reason={finish_reason})" + ) + if usage is not None: + in_tok = _coerce_token_count(getattr(usage, "prompt_tokens", 0)) + out_tok = _coerce_token_count(getattr(usage, "completion_tokens", 0)) + else: + # Relay didn't honor include_usage — estimate rather than report + # 0/0 (which run() flags as cost-unknown and can't sum into daily + # totals). Loud so the operator knows these numbers are soft. + in_tok = _estimate_tokens(self.system_prompt + user_message) + out_tok = _estimate_tokens(content) + logger.warning( + "OpenAI stream for %s carried no usage chunk — token counts " + "are chars/4 estimates (in≈%d out≈%d).", + self.name, in_tok, out_tok, + ) return (content, in_tok, out_tok, finish_reason) def _deepseek_max_output(self) -> int: @@ -576,6 +810,15 @@ def _call_deepseek(self, user_message: str) -> tuple[str, int, int, str | None]: finish_reason = None if not content: reasoning = getattr(choice.message, "reasoning_content", None) + if finish_reason not in _TRUNCATION_FINISH_REASONS: + # Same guard as _call_openai: a degenerate 200 must enter the + # retry/failover machinery, not pass as a clean no-signal. + raise LLMEmptyResponseError( + f"DeepSeek returned empty content (finish_reason={finish_reason}, " + f"reasoning_content present={bool(reasoning)})" + ) + # Truncation-family: reasoner burned the whole budget on CoT — + # legit empty, surfaced via truncated=True downstream. logger.warning( "DeepSeek returned empty content (finish_reason=%s, reasoning_content present=%s)", finish_reason, bool(reasoning), diff --git a/tests/test_base_agent.py b/tests/test_base_agent.py index e4e558ce..42d5fea7 100644 --- a/tests/test_base_agent.py +++ b/tests/test_base_agent.py @@ -186,7 +186,9 @@ def test_anthropic_client_gets_explicit_http_timeout(): with patch("anthropic.Anthropic") as mock_cls: ConcreteAgent(api_key="k", model="claude-sonnet-4-6-20250514", max_tokens=1024) - mock_cls.assert_called_once_with(api_key="k", timeout=_LLM_HTTP_TIMEOUT) + # max_retries=0: the SDK's 2 internal retries would silently triple + # every _execute() attempt — the agent loop is the single retry owner. + mock_cls.assert_called_once_with(api_key="k", timeout=_LLM_HTTP_TIMEOUT, max_retries=0) def test_openai_client_gets_explicit_http_timeout(): @@ -200,7 +202,8 @@ def test_openai_client_gets_explicit_http_timeout(): ConcreteAgent(api_key="k", model="gpt-5.4", max_tokens=1024) # base_url defaults to None (api.openai.com) unless OPENAI_BASE_URL is set # for relay routing — see test_openai_base_url_routes_through_relay. - mock_cls.assert_called_once_with(api_key="k", base_url=None, timeout=_LLM_HTTP_TIMEOUT) + mock_cls.assert_called_once_with(api_key="k", base_url=None, + timeout=_LLM_HTTP_TIMEOUT, max_retries=0) def test_parse_json_prefers_agent_shape_over_larger_fragment(): @@ -582,9 +585,10 @@ def test_deepseek_sends_max_tokens_not_max_completion_tokens(): assert "max_tokens" in call_kwargs assert "max_completion_tokens" not in call_kwargs - # mirror: the OpenAI path still uses max_completion_tokens + # mirror: the OpenAI path still uses max_completion_tokens (and streams — + # see the relay-hardening section at the end of this file) with patch("openai.OpenAI") as oai_cls: - client = _deepseek_oai_mock() + client = _openai_stream_mock() oai_cls.return_value = client agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=4096) agent.run(data="x") @@ -766,3 +770,310 @@ def test_openai_no_ca_bundle_uses_default_trust(monkeypatch): ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64) assert "http_client" not in oai_cls.call_args.kwargs assert oai_cls.call_args.kwargs.get("timeout") is not None + + +# === streamed OpenAI path + relay hardening (CF-524 / 429 storms / degenerate 200s) === + +def _stream_chunk(piece=None, finish_reason=None, usage=None): + """One streamed chunk. A usage-only chunk (include_usage's final extra + chunk) carries an empty choices list.""" + chunk = MagicMock() + chunk.usage = usage # explicit None — MagicMock auto-attrs would count as usage + if piece is None and finish_reason is None and usage is not None: + chunk.choices = [] + return chunk + delta = MagicMock() + delta.content = piece + choice = MagicMock() + choice.delta = delta + choice.finish_reason = finish_reason + chunk.choices = [choice] + return chunk + + +def _stream_usage(prompt=100, completion=50): + u = MagicMock() + u.prompt_tokens = prompt + u.completion_tokens = completion + return u + + +def _openai_stream_mock(pieces=('{"result"', ': "ok"}'), finish_reason="stop", usage="default"): + """Mock OpenAI client whose chat.completions.create returns a streamed + (iterable-of-chunks) response — the shape _call_openai consumes. + finish_reason=None simulates a connection cut mid-generation (no final + chunk ever carries a finish_reason). usage=None simulates a relay that + ignores stream_options include_usage.""" + oai = MagicMock() + chunks = [_stream_chunk(piece=p) for p in pieces] + if finish_reason is not None: + chunks.append(_stream_chunk(finish_reason=finish_reason)) + if usage == "default": + usage = _stream_usage() + if usage is not None: + chunks.append(_stream_chunk(usage=usage)) + oai.chat.completions.create.return_value = chunks + return oai + + +def test_openai_path_streams_and_assembles_content(): + """_call_openai streams (stream=True) on purpose: the relay sits behind + Cloudflare's ~120s proxy read timeout (HTTP 524), so a non-streamed 120s+ + generation can never succeed through it (2026-06-08/09 outage). Content is + assembled from the deltas; usage arrives on the include_usage final chunk.""" + with patch("openai.OpenAI") as oai_cls: + client = _openai_stream_mock() + oai_cls.return_value = client + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=4096) + result = agent.run(data="x") + kw = client.chat.completions.create.call_args.kwargs + assert kw.get("stream") is True + assert kw.get("stream_options") == {"include_usage": True} + assert result.raw_text == '{"result": "ok"}' + assert result.parse_json() == {"result": "ok"} + assert result.input_tokens == 100 and result.output_tokens == 50 + assert result.truncated is False + + +def test_openai_stream_without_usage_estimates_tokens(): + """Relay ignored include_usage → chars/4 estimates, never 0/0 (0/0 would + flag the call cost-unknown and drop it from daily cost totals).""" + with patch("openai.OpenAI") as oai_cls: + oai_cls.return_value = _openai_stream_mock(usage=None) + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=4096) + result = agent.run(data="x") + assert result.raw_text == '{"result": "ok"}' + assert result.input_tokens > 0 and result.output_tokens > 0 + + +def test_openai_stream_interrupted_discards_partial_and_retries(monkeypatch): + """finish_reason never arrives = connection cut mid-generation. Partial + text must be DISCARDED (a half-emitted PM decision parses like 'no + trades') and the error must be retryable.""" + from src.agents.base import LLMStreamInterruptedError, _is_retryable + monkeypatch.setattr("time.sleep", lambda s: None) + monkeypatch.setenv("QUANT_AGENT_MAX_RETRIES", "2") + with patch("openai.OpenAI") as oai_cls: + client = _openai_stream_mock(pieces=('{"half":',), finish_reason=None, usage=None) + oai_cls.return_value = client + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64) + with pytest.raises(LLMStreamInterruptedError): + agent.run(data="x") + assert client.chat.completions.create.call_count == 2 # retried, then exhausted + assert _is_retryable(LLMStreamInterruptedError("x")) is True + + +def test_openai_empty_content_raises_retryable_not_silent_success(monkeypatch): + """A degenerate 200 (finish_reason=stop, empty body) must enter the + retry/failover machinery — previously it returned '' as a SUCCESS, + masquerading as a deliberate no-signal and burning the agent's one shot + for the session.""" + from src.agents.base import LLMEmptyResponseError, _is_retryable + monkeypatch.setattr("time.sleep", lambda s: None) + monkeypatch.setenv("QUANT_AGENT_MAX_RETRIES", "2") + with patch("openai.OpenAI") as oai_cls: + oai_cls.return_value = _openai_stream_mock(pieces=(), finish_reason="stop") + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64) + with pytest.raises(LLMEmptyResponseError): + agent.run(data="x") + assert _is_retryable(LLMEmptyResponseError("x")) is True + + +def test_openai_empty_content_on_length_is_truncation_not_error(): + """Empty body + truncation-family finish_reason = the whole budget burned + before any visible text. That's a legit truncation (truncated=True), NOT + a degenerate response — must not raise / retry / fail over.""" + with patch("openai.OpenAI") as oai_cls: + oai_cls.return_value = _openai_stream_mock(pieces=(), finish_reason="length") + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64) + result = agent.run(data="x") + assert result.raw_text == "" + assert result.truncated is True + + +def test_openai_empty_content_fails_over_to_anthropic(monkeypatch): + """The point of raising on empty: the Anthropic failover can rescue the + session instead of the agent silently contributing a blank.""" + monkeypatch.setattr("time.sleep", lambda s: None) + monkeypatch.setenv("QUANT_AGENT_MAX_RETRIES", "2") + anth = MagicMock() + anth.messages.create.return_value = _good_anthropic_response() + with patch("openai.OpenAI", return_value=_openai_stream_mock(pieces=(), finish_reason="stop")), \ + patch("anthropic.Anthropic", return_value=anth): + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64, fallback_api_key="fk") + result = agent.run(data="x") + assert result.raw_text == '{"result": "ok"}' + assert result.model == "claude-opus-4-7" + + +def test_anthropic_empty_content_raises_unless_truncation(monkeypatch): + """Same guard on the Anthropic path: end_turn + empty content raises + (degenerate 200); max_tokens + empty content is a legit whole-budget + truncation and returns '' flagged truncated.""" + from src.agents.base import LLMEmptyResponseError + monkeypatch.setattr("time.sleep", lambda s: None) + monkeypatch.setenv("QUANT_AGENT_MAX_RETRIES", "1") + with patch("anthropic.Anthropic") as anth_cls: + r = _good_anthropic_response() + r.content = [] + anth_cls.return_value.messages.create.return_value = r + agent = ConcreteAgent(api_key="k", model="claude-opus-4-7", max_tokens=64) + with pytest.raises(LLMEmptyResponseError): + agent.run(data="x") + with patch("anthropic.Anthropic") as anth_cls: + r = _good_anthropic_response() + r.content = [] + r.stop_reason = "max_tokens" + anth_cls.return_value.messages.create.return_value = r + agent = ConcreteAgent(api_key="k", model="claude-opus-4-7", max_tokens=64) + result = agent.run(data="x") + assert result.raw_text == "" and result.truncated is True + + +def test_deepseek_empty_content_nontruncation_raises(monkeypatch): + """DeepSeek mirror of the degenerate-200 guard (the truncation-family + empty case is covered by test_deepseek_empty_content_with_reasoning...).""" + from src.agents.base import LLMEmptyResponseError + monkeypatch.setattr("time.sleep", lambda s: None) + monkeypatch.setenv("QUANT_AGENT_MAX_RETRIES", "1") + with patch("openai.OpenAI") as oai_cls: + oai_cls.return_value = _deepseek_oai_mock(content="", finish_reason="stop") + agent = ConcreteAgent(api_key="dk", model="deepseek-v4-flash", max_tokens=64) + with pytest.raises(LLMEmptyResponseError): + agent.run(data="x") + + +def test_retry_deadline_abandons_primary_for_failover(monkeypatch): + """The attempt budget alone doesn't bound wall-clock: under the relay's + CF-524 mode each attempt burned 120-380s, so exhausting 7 attempts + collided with the wrapper's 1200s kill and the failover below the loop + never fired (2026-06-08/09 mornings). Past the deadline the primary is + abandoned with attempts still left, and failover fires.""" + monkeypatch.setattr("time.sleep", lambda s: None) + # monotonic: loop_start=0; attempt 1 fails at 200s (<480 → retry); + # attempt 2 fails at 600s (>=480 → abandon primary, fail over). + ticks = [0.0, 200.0, 600.0] + monkeypatch.setattr("src.agents.base.time.monotonic", + lambda: ticks.pop(0) if ticks else 600.0) + oai = MagicMock() + oai.chat.completions.create.side_effect = ConnectionError("relay 524 storm") + anth = MagicMock() + anth.messages.create.return_value = _good_anthropic_response() + with patch("openai.OpenAI", return_value=oai), patch("anthropic.Anthropic", return_value=anth): + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64, fallback_api_key="fk") + result = agent.run(data="x") + assert oai.chat.completions.create.call_count == 2 # NOT the full 7-attempt budget + assert result.model == "claude-opus-4-7" + + +def test_retry_deadline_env_override(monkeypatch): + from src.agents.base import _retry_deadline_s, _DEFAULT_RETRY_DEADLINE_S + monkeypatch.delenv("QUANT_AGENT_RETRY_DEADLINE_S", raising=False) + assert _retry_deadline_s() == _DEFAULT_RETRY_DEADLINE_S + monkeypatch.setenv("QUANT_AGENT_RETRY_DEADLINE_S", "60") + assert _retry_deadline_s() == 60.0 + monkeypatch.setenv("QUANT_AGENT_RETRY_DEADLINE_S", "garbage") + assert _retry_deadline_s() == _DEFAULT_RETRY_DEADLINE_S + monkeypatch.setenv("QUANT_AGENT_RETRY_DEADLINE_S", "0") + assert _retry_deadline_s() == 1.0 # floor — 0/negative would break the loop + + +def test_retry_after_hint_extraction(): + from src.agents.base import _retry_after_hint_seconds + + class HeaderErr(Exception): + pass + e = HeaderErr("429") + resp = MagicMock() + resp.headers = {"retry-after": "37"} + e.response = resp + assert _retry_after_hint_seconds(e) == 37.0 + + class BodyErr(Exception): + pass + b = BodyErr("429") + b.body = {"retry_after": 15} + assert _retry_after_hint_seconds(b) == 15.0 + + # relay 524 bodies embed the hint in the message text + m = Exception('Concurrency limit exceeded, "retry_after": 120') + assert _retry_after_hint_seconds(m) == 120.0 + + # bool must not pass the numeric check; plain errors carry no hint + t = BodyErr("429") + t.body = {"retry_after": True} + assert _retry_after_hint_seconds(t) is None + assert _retry_after_hint_seconds(Exception("plain connection error")) is None + + +def test_retry_sleeps_at_least_the_server_hint(monkeypatch): + """Backoff honors a server retry-after hint — sleeping 2s against a + 'come back in 90s' 429 just burns attempts against a closed door. A + hostile/buggy hint is capped so it can't park the agent past the + session window.""" + sleeps = [] + monkeypatch.setattr("time.sleep", lambda s: sleeps.append(s)) + monkeypatch.setenv("QUANT_AGENT_MAX_RETRIES", "2") + + oai = MagicMock() + oai.chat.completions.create.side_effect = ConnectionError( + 'Concurrency limit exceeded {"retry_after": 90}') + with patch("openai.OpenAI", return_value=oai): + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64) + with pytest.raises(ConnectionError): + agent.run(data="x") + assert sleeps == [90.0] + + sleeps.clear() + oai2 = MagicMock() + oai2.chat.completions.create.side_effect = ConnectionError( + 'slow down {"retry_after": 6000}') + with patch("openai.OpenAI", return_value=oai2): + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64) + with pytest.raises(ConnectionError): + agent.run(data="x") + assert sleeps == [120.0] # _RETRY_AFTER_CAP_S + + +def test_deepseek_client_disables_sdk_internal_retries(): + """max_retries=0 on EVERY SDK client: both SDKs default to 2 internal + retries on 429/5xx, silently tripling each _execute() attempt and + invalidating the retry-budget/deadline math. The agent loop is the + single retry owner. (OpenAI/Anthropic primary constructors are asserted + in the http-timeout tests above; this covers the DeepSeek constructor.)""" + with patch("openai.OpenAI") as oai_cls: + oai_cls.return_value = _deepseek_oai_mock() + ConcreteAgent(api_key="dk", model="deepseek-v4-flash", max_tokens=64) + assert oai_cls.call_args.kwargs.get("max_retries") == 0 + + +def test_failover_client_disables_sdk_internal_retries(monkeypatch): + """The failover Anthropic client is single-shot BY DESIGN — SDK-internal + retries would quietly turn it into 3 shots.""" + monkeypatch.setattr("time.sleep", lambda s: None) + monkeypatch.setenv("QUANT_AGENT_MAX_RETRIES", "1") + oai = MagicMock() + oai.chat.completions.create.side_effect = ConnectionError("down") + with patch("openai.OpenAI", return_value=oai), patch("anthropic.Anthropic") as anth_cls: + anth_cls.return_value.messages.create.return_value = _good_anthropic_response() + agent = ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64, fallback_api_key="fk") + agent.run(data="x") + assert anth_cls.call_args.kwargs.get("max_retries") == 0 + + +def test_llm_semaphore_released_after_success_and_failure(monkeypatch): + """The per-provider in-flight caps must never leak a slot — a leaked slot + would permanently shrink OpenAI concurrency for the whole process.""" + from src.agents import base as base_mod + monkeypatch.setattr("time.sleep", lambda s: None) + monkeypatch.setenv("QUANT_AGENT_MAX_RETRIES", "1") + start = base_mod._OPENAI_LLM_SEMAPHORE._value + with patch("openai.OpenAI") as oai_cls: + oai_cls.return_value = _openai_stream_mock() + ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64).run(data="x") + assert base_mod._OPENAI_LLM_SEMAPHORE._value == start + with patch("openai.OpenAI") as oai_cls: + oai_cls.return_value = _openai_stream_mock(pieces=('{"half":',), finish_reason=None, usage=None) + with pytest.raises(Exception): + ConcreteAgent(api_key="k", model="gpt-5.5", max_tokens=64).run(data="x") + assert base_mod._OPENAI_LLM_SEMAPHORE._value == start From d2c781a40da79e16a344539d32134b050b4122e7 Mon Sep 17 00:00:00 2001 From: yebof Date: Thu, 16 Jul 2026 02:46:37 +0800 Subject: [PATCH 2/2] test(scripts): Add relay streaming smoke test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One real gpt-5.5 call through the OPENAI_BASE_URL relay with stream=True + stream_options include_usage — the load-bearing assumption of the streamed _call_openai path. Run this against the live relay BEFORE merging the streaming change (the relay was down during development, so live verification is still pending): a 400 on stream_options would mean every production call fails over to Anthropic. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01PQkESoSTYx2bCy7WYnTPXR --- scripts/relay_stream_smoke.py | 63 +++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 scripts/relay_stream_smoke.py diff --git a/scripts/relay_stream_smoke.py b/scripts/relay_stream_smoke.py new file mode 100644 index 00000000..bdfb1894 --- /dev/null +++ b/scripts/relay_stream_smoke.py @@ -0,0 +1,63 @@ +"""Smoke: does the relay accept stream=True + stream_options include_usage? + +This is THE load-bearing assumption of the audit-hardening change: if the +relay 400s on stream_options (some OpenAI-compatible gateways do), every +production call would fail over to Anthropic. One tiny real call settles it. +Cost: a few hundred tokens on gpt-5.5. +""" +import os +import sys + +# Load .env the same way the app does (no external deps needed). +for line in open("/home/yebo/quant-agent/.env"): + line = line.strip() + if line.startswith("export "): + line = line[len("export "):] + if line and not line.startswith("#") and "=" in line: + k, _, v = line.partition("=") + os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) + +from openai import OpenAI + +base_url = os.environ.get("OPENAI_BASE_URL") or None +print(f"base_url: {base_url or 'api.openai.com (default)'}") +client = OpenAI(api_key=os.environ["OPENAI_API_KEY"], base_url=base_url, + timeout=60.0, max_retries=0) + +stream = client.chat.completions.create( + model="gpt-5.5", + max_completion_tokens=512, + messages=[ + {"role": "system", "content": "You are a smoke test."}, + {"role": "user", "content": "Reply with exactly: OK"}, + ], + stream=True, + stream_options={"include_usage": True}, +) + +parts, finish_reason, usage, n_chunks = [], None, None, 0 +for chunk in stream: + n_chunks += 1 + if getattr(chunk, "usage", None) is not None: + usage = chunk.usage + choices = getattr(chunk, "choices", None) or [] + if not choices: + continue + delta = getattr(choices[0], "delta", None) + piece = getattr(delta, "content", None) if delta is not None else None + if piece: + parts.append(piece) + fr = getattr(choices[0], "finish_reason", None) + if isinstance(fr, str): + finish_reason = fr + +content = "".join(parts) +print(f"chunks: {n_chunks}") +print(f"content: {content!r}") +print(f"finish_reason: {finish_reason!r}") +print(f"usage: {usage!r}") + +ok = bool(content) and finish_reason == "stop" and usage is not None +print("RESULT:", "PASS — relay supports streaming + include_usage" if ok else + "PARTIAL — see fields above (usage None means the estimate fallback will engage)") +sys.exit(0 if content and finish_reason else 1)