"""API-call helpers extracted from :class:`AIAgent`: non-streaming and streaming request drivers, request kwargs builder, assistant-message materializer, provider-fallback activator, max-iterations handler, per-turn resource cleanup. Each function takes the parent ``AIAgent`` as ``agent``; AIAgent keeps thin forwarders. Symbols tests patch on ``run_agent`` (``cleanup_vm`` / ``cleanup_browser``) are resolved through :func:`_ra` at call time. """ from __future__ import annotations import contextlib import contextvars import json import logging import math import os import re import sys import threading import time import uuid from dataclasses import dataclass from types import SimpleNamespace from typing import Any, Dict, Optional from hermes_cli.timeouts import get_provider_request_timeout, get_provider_stale_timeout from hermes_constants import PARTIAL_STREAM_STUB_ID, FINISH_REASON_LENGTH from agent.error_classifier import ( FailoverReason, PROVIDER_STREAM_EMPTY_FRAME_ERROR_CODE, PROVIDER_STREAM_NON_JSON_ERROR_CODE, _extract_status_code) from agent.sdk_transform_bypass import bypass_chat_sdk_request_transform from agent.errors import EmptyStreamError from agent.chat_completion_stream_monitor import StreamingWaitMonitor from agent.transports.chat_completions import is_router_timeout_shim, router_timeout_shim_may_follow from agent.fast_mode import effective_request_overrides from agent.turn_context import substitute_api_content from agent.gemini_native_adapter import is_native_gemini_base_url # Remote endpoints must never be fingerprinted: the probe waterfall is only valid for local/LM-Studio/Ollama # boxes. Non-Ollama remotes (sglang, vLLM, OpenAI-compat) expose Ollama-compat endpoints that can # misidentify and, without an api_key, return 401 on every leg (issue #89863). from agent.model_metadata import is_local_endpoint from agent.message_content import flatten_message_text from agent.message_metadata import PERSISTENCE_ONLY_MESSAGE_FIELDS, append_message, stamp_message_timestamp from agent.message_sanitization import ( _sanitize_messages_surrogates, _sanitize_surrogates, _repair_tool_call_arguments, normalize_finish_reason as _normalize_finish_reason, sanitize_outbound_kwargs, strip_images_for_rejecting_model, ) from agent.reasoning_summaries import ( append_streamed_reasoning_detail, separate_glued_reasoning_blocks, streamed_reasoning_detail_text, ) from agent.repetition_guard import RunawayStreamWatch, is_repetition_dominated from agent.stream_single_writer import claim_stream_writer, stream_writer_is_current from tools.terminal_tool_lifecycle import is_persistent_env from utils import base_url_host_matches, base_url_hostname, env_float, env_int logger = logging.getLogger(__name__) _OPENROUTER_PROVIDER_SORT_VALUES = {"throughput", "latency", "price"} _PROVIDER_STREAM_ERROR_FINISH_REASONS = {"error", "error_finish"} _PROVIDER_STREAM_SSE_FIELDS = {"event", "data", "id", "retry"} _PROVIDER_STREAM_ERROR_TEXT_LIMIT = 4096 # Fallback chain exhausted on a non-rate-limit failure (#24996): arm a short # cooldown so the NEXT turn's restore_primary_runtime stays gated instead of # resetting _fallback_index=0 and re-marshaling the whole context across every # provider again (memory/swap exhaustion on constrained hosts). Rate-limit / # billing reasons keep their own longer cooldown. _FALLBACK_EXHAUSTED_COOLDOWN_S = 5.0 # Streaming 5xx unmask probe: one non-streaming re-issue per this window. Covers the # outer retry loop (up to ~3 attempts x backoff, well under 60s) so an outage doesn't # double traffic every attempt, while later turns re-arm automatically. _STREAM_5XX_PROBE_WINDOW_S = 60.0 def _context_thread_target(callback): """Bind a no-argument thread target to the caller's ContextVars.""" context = contextvars.copy_context() return lambda: context.run(callback) def _join_worker_for_relay_teardown(worker, *, label: str) -> None: """Bounded worker join before raising InterruptedError (#81521). Raising immediately lets turn teardown race a still-open Relay LLM scope and corrupt the LIFO stack (CLI EIO / redraw storm). Only joins when Relay managed execution is live — otherwise the join would just delay interrupt detection. """ try: from agent import relay_runtime runtime = relay_runtime.get_runtime(create=False) if runtime is None or not runtime.managed_execution_enabled(): return except Exception: return worker.join(timeout=2.0) if worker.is_alive(): logger.warning("%s worker still alive after interrupt abort (2.0s join " "timeout); Relay teardown will best-effort drain orphaned scopes (#81521).", label) def _ra(): """Lazy ``run_agent`` reference so ``patch("run_agent.cleanup_vm")`` etc. intercept.""" import run_agent return run_agent class ProviderStreamError(Exception): """Provider encoded an API error as streaming content instead of an SDK error.""" def __init__(self, *, status_code: Optional[int], body: dict, raw_text: str, headers: Any = None): self.status_code = status_code self.body = body self.raw_text = raw_text self.response = SimpleNamespace(headers=headers or {}) super().__init__(self._format_message()) def _format_message(self) -> str: error_obj = self.body.get("error", {}) if isinstance(self.body, dict) else {} if not isinstance(error_obj, dict): error_obj = {} parts = ["Provider stream returned an error event"] if self.status_code: parts.append(f"HTTP {self.status_code}") if error_obj.get("code"): parts.append(str(error_obj["code"])) text = " - ".join(parts) if error_obj.get("message"): text += f": {error_obj['message']}" return text def _status_code_from_value(value: Any) -> Optional[int]: if isinstance(value, int) and 100 <= value < 600: return value if not isinstance(value, str): return None match = re.search(r"(?:HTTP_STATUS/)?\b([1-5]\d\d)\b", value, re.IGNORECASE) return int(match.group(1)) if match else None def _status_code_from_payload(payload: Any) -> Optional[int]: if not isinstance(payload, dict): return None candidates = [payload.get(k) for k in ("status_code", "status", "http_status")] error_obj = payload.get("error") if isinstance(error_obj, dict): candidates.extend(error_obj.get(k) for k in ("status_code", "status", "http_status", "code")) candidates.append(payload.get("code")) for candidate in candidates: status_code = _status_code_from_value(candidate) if status_code is not None: return status_code return None def _json_object_from_text(text: str) -> Optional[dict]: stripped = (text or "").strip() with contextlib.suppress(json.JSONDecodeError, TypeError): if stripped.startswith("{"): decoded = json.loads(stripped) return decoded if isinstance(decoded, dict) else None return None def _parse_provider_sse_events(text: str) -> list[dict]: """Parse provider text that looks like Server-Sent Events.""" events: list[dict] = [] current = {"event": None, "data": [], "comments": [], "fields": {}} def _flush_current(): nonlocal current if any(current.values()): status_candidates = list(current["comments"]) + [ current["fields"][key] for key in ("status", "status_code", "http_status") if key in current["fields"] ] events.append({ "event": current["event"], "data": "\n".join(current["data"]), "comments": list(current["comments"]), "fields": dict(current["fields"]), "status_code": next( (s for s in map(_status_code_from_value, status_candidates) if s is not None), None), }) current = {"event": None, "data": [], "comments": [], "fields": {}} for raw_line in (text or "").splitlines(): line = raw_line.rstrip("\r") if line == "": _flush_current() continue if line.startswith(":"): current["comments"].append(line[1:].strip()) continue field, sep, value = line.partition(":") if not sep: current["fields"][field.strip().lower()] = "" continue field = field.strip().lower() if value.startswith(" "): value = value[1:] if field == "event": current["event"] = value.strip() elif field == "data": current["data"].append(value) else: current["fields"][field] = value _flush_current() return events def _provider_error_body(payload: dict, status_code: Optional[int]) -> dict: """Normalize common provider error payloads to OpenAI-style body.error.""" if not isinstance(payload, dict): payload = {} elif isinstance(payload.get("error"), dict): return payload code = (payload.get("code") or payload.get("error_code") or payload.get("type") or (f"HTTP_{status_code}" if status_code else "provider_stream_error")) message = (payload.get("message") or payload.get("error_description") or payload.get("error") or "Provider stream returned an error event.") normalized_error = {"message": str(message)} if code: normalized_error["code"] = str(code) for key in ("request_id", "param", "type"): if payload.get(key): normalized_error[key] = payload[key] return {"error": normalized_error} def _provider_stream_error_from_json_decode_error(error: json.JSONDecodeError, *, response: Any = None) -> ProviderStreamError: """Preserve plain-text SSE data rejected inside the OpenAI SDK: on a non-JSON ``event: error`` the SDK raises from ``sse.json()`` before yielding a chunk, but ``JSONDecodeError.doc`` still carries the provider's original message. An EMPTY ``doc`` is the other case: the frame carried no payload at all (``data:`` / ``event: ping`` / ``id:`` alone — legal SSE keepalives and no-ops), which the SDK's ``json.loads`` rejects the same way. A gateway that is degrading answers EVERY streaming request with such frames, so this is not the provider's malformed payload and must not be reported as one: it gets its own code and the stream helper recovers by retrying without streaming.""" from agent.redact import redact_sensitive_text raw_text = str(getattr(error, "doc", "") or "").strip() headers = getattr(response, "headers", None) if response is not None else None if not raw_text: return ProviderStreamError( status_code=None, body=_provider_error_body( {"code": PROVIDER_STREAM_EMPTY_FRAME_ERROR_CODE, "message": "Provider stream returned an empty SSE data frame (keepalive with no payload)."}, None, ), raw_text="", headers=headers, ) safe_text = redact_sensitive_text(_sanitize_surrogates(raw_text), force=True) safe_text = safe_text[:_PROVIDER_STREAM_ERROR_TEXT_LIMIT] return ProviderStreamError( status_code=None, body=_provider_error_body( {"code": PROVIDER_STREAM_NON_JSON_ERROR_CODE, "message": safe_text or "Provider stream returned non-JSON SSE data."}, None, ), raw_text=safe_text, headers=headers, ) def _is_provider_stream_empty_frame_error(exc: BaseException) -> bool: """True for the translated contentless-SSE-frame error. Re-streaming cannot help (a degraded gateway answers every stream that way), so the caller must change channel.""" body = getattr(exc, "body", None) error_obj = body.get("error") if isinstance(body, dict) else None return isinstance(error_obj, dict) and error_obj.get("code") == PROVIDER_STREAM_EMPTY_FRAME_ERROR_CODE def _iter_provider_stream_chunks(stream, *, response: Any = None): """Yield SDK chunks while translating SDK-level SSE decode failures.""" try: yield from stream except json.JSONDecodeError as error: stream_response = response() if callable(response) else response if stream_response is None: stream_response = getattr(stream, "response", None) raise _provider_stream_error_from_json_decode_error(error, response=stream_response) from error def _payload_has_error_shape(payload: Any) -> bool: if not isinstance(payload, dict): return False if isinstance(payload.get("error"), (dict, str)): return True return bool(payload.get("message")) and bool( payload.get("code") or payload.get("error_code") or _status_code_from_payload(payload) is not None) def _provider_stream_text_may_be_sse(text: str) -> bool: """Return True while pending text still looks like an SSE control block.""" stripped = (text or "").lstrip() if not stripped: return False lines = stripped.splitlines() trailing_newline = stripped.endswith(("\n", "\r")) saw_sse_field = False for index, raw_line in enumerate(lines): line = raw_line.rstrip("\r") if line == "": continue if line.startswith(":"): saw_sse_field = True continue field, sep, _value = line.partition(":") field_name = field.strip().lower() if sep and field_name in _PROVIDER_STREAM_SSE_FIELDS: saw_sse_field = True continue is_last_incomplete = index == len(lines) - 1 and not trailing_newline if is_last_incomplete and any( sse_field.startswith(field_name) for sse_field in _PROVIDER_STREAM_SSE_FIELDS): return True return False return saw_sse_field def _provider_stream_error_from_text(text: str, finish_reason: Optional[str], *, response: Any = None) -> Optional[ProviderStreamError]: """Convert provider-streamed error text into an exception for retry logic.""" if not text: return None if str(finish_reason or "").lower() not in _PROVIDER_STREAM_ERROR_FINISH_REASONS: return None headers = getattr(response, "headers", None) if response is not None else None def _error(payload: dict, status_code: Optional[int]) -> ProviderStreamError: return ProviderStreamError(status_code=status_code, body=_provider_error_body(payload, status_code), raw_text=text, headers=headers) for event in _parse_provider_sse_events(text): is_error_event = str(event.get("event") or "").strip().lower() == "error" payload = _json_object_from_text(event.get("data") or "") or {} status_code = event.get("status_code") or _status_code_from_payload(payload) # The finish_reason is an error here, so an error event always qualifies; # a non-error event needs an error-shaped payload or an HTTP error code. if (status_code is not None and status_code >= 400) or is_error_event or _payload_has_error_shape(payload): return _error(payload, status_code) payload = _json_object_from_text(text) if payload is not None: return _error(payload, _status_code_from_payload(payload)) if text.strip(): return _error({}, None) return None _IMAGE_PART_TYPES = frozenset({"image_url", "input_image", "image"}) def _image_part_chars(part: Dict[str, Any], image_cost: int) -> int: """Char-equivalent of one image content part: the per-image cost learned from provider usage (x4 chars/token), never the base64 payload length. A single native screenshot priced as text read as ~100K+ tokens and selected the giant-conversation watchdog tiers (#63871, #76411).""" text = part.get("text") return image_cost * 4 + (len(text) if isinstance(text, str) else 0) def _payload_chars(value: Any, image_cost: int) -> int: """``len(str(value))`` with image content parts priced at ``image_cost`` tokens each.""" if value is None: return 0 if isinstance(value, dict): part_type = value.get("type") # JSON-Schema nodes may hold a sub-schema (``properties.type``) or a multi-type list # under the "type" key; only scalar content-part types can ever match (#104793). if isinstance(part_type, str) and part_type in _IMAGE_PART_TYPES and any(k in value for k in ("image_url", "image", "source", "file_id")): return _image_part_chars(value, image_cost) return sum(len(str(k)) + 6 + _payload_chars(v, image_cost) for k, v in value.items()) if isinstance(value, list): return sum(_payload_chars(item, image_cost) for item in value) + 2 * len(value) return len(str(value)) def estimate_request_context_tokens(api_payload: Any) -> int: """Cheap char/4 context estimate for the stale-call detectors. Handles both wire shapes so Codex turns don't report ~0 tokens: list -> Chat ``messages``; dict with ``messages`` (+``tools``); dict with ``input`` (Responses API, +``instructions``/``tools``); any other dict -> sum of its values. Image parts cost the learned per-image price, not their base64 length.""" from agent.image_token_cost import current_image_token_cost image_cost = current_image_token_cost() def _chars(value: Any) -> int: return _payload_chars(value, image_cost) if isinstance(api_payload, list): return sum(_chars(item) for item in api_payload) // 4 if not isinstance(api_payload, dict): return _chars(api_payload) // 4 messages = api_payload.get("messages") if isinstance(messages, list): total_chars = sum(_chars(item) for item in messages) if "tools" in api_payload: total_chars += _chars(api_payload.get("tools")) return total_chars // 4 if "input" in api_payload: return sum(_chars(api_payload.get(k)) for k in ("input", "instructions", "tools")) // 4 return sum(_chars(value) for value in api_payload.values()) // 4 def _is_openai_codex_backend(agent) -> bool: from agent.codex_responses_adapter import classify_responses_route return classify_responses_route(agent).is_codex_backend def openai_codex_stale_timeout_floor(est_tokens: int) -> float: """Minimum wall-clock stale timeout for openai-codex by estimated context: subscription-backed Codex can spend minutes in admission/prefill on gateway-scale payloads, so the generic default would abort healthy calls. The floor engages above 10k estimated tokens.""" for threshold, floor in ((100_000, 1200.0), (50_000, 900.0), (10_000, 600.0)): if est_tokens > threshold: return floor return 0.0 def _bound_openai_codex_stale_timeout(stale_timeout: float, est_tokens: int) -> float: """Apply the openai-codex stale bounds: raise to ``openai_codex_stale_timeout_floor`` so healthy gateway-scale requests aren't aborted mid-prefill, then clamp to the flat HERMES_CODEX_HARD_TIMEOUT_SECONDS ceiling (#64507, default 1500s — above the max floor, a backstop for a request that emits SOME events then wedges; 0 disables). Shared by the worker watchdogs and the inline cron path (#69734).""" floor = openai_codex_stale_timeout_floor(est_tokens) if floor: stale_timeout = max(stale_timeout, floor) hard_timeout = env_float("HERMES_CODEX_HARD_TIMEOUT_SECONDS", 1500.0) return min(stale_timeout, hard_timeout) if hard_timeout > 0 else stale_timeout def _validated_openrouter_provider_sort(raw_sort: Any) -> Optional[str]: """Return a normalized OpenRouter provider.sort value or None.""" if not isinstance(raw_sort, str): return None sort_value = raw_sort.strip().lower() if not sort_value: return None if sort_value in _OPENROUTER_PROVIDER_SORT_VALUES: return sort_value logger.warning("Ignoring invalid OpenRouter provider.sort value %r (allowed: %s)", raw_sort, ", ".join(sorted(_OPENROUTER_PROVIDER_SORT_VALUES))) return None def _provider_preferences_for_agent(agent) -> Dict[str, Any]: """Build the validated provider-routing object shared by request paths. ``provider_routing.models.`` overlays the flat constructor values for the CURRENT ``agent.model`` (so ``/model`` switches, fallbacks, and delegated children on another model each get their own pins without any surface re-plumbing the kwargs).""" flat = {"only": agent.providers_allowed, "ignore": agent.providers_ignored, "order": agent.providers_order, "sort": agent.provider_sort, "require_parameters": agent.provider_require_parameters, "data_collection": agent.provider_data_collection} per_model = {} with contextlib.suppress(Exception): from hermes_cli.config import load_config_readonly from hermes_constants import resolve_per_model_provider_routing _pr = load_config_readonly().get("provider_routing") per_model = resolve_per_model_provider_routing(agent.model, (_pr or {}).get("models") if isinstance(_pr, dict) else None) merged = {**flat, **{k: v for k, v in per_model.items() if k in flat}} merged["sort"] = _validated_openrouter_provider_sort(merged["sort"]) merged["require_parameters"] = True if merged["require_parameters"] else None return {key: value for key, value in merged.items() if value} def _prompt_cache_scope_for_agent(agent) -> "str | None": """Rotation-stable logical cache scope for *agent*, or None (transports then fall back to the physical session_id, so a failure never blocks the build).""" try: from agent.prompt_cache_scope import resolve_prompt_cache_scope_safe return resolve_prompt_cache_scope_safe(agent) except Exception: logger.debug("prompt-cache scope resolution failed", exc_info=True) return None def _merge_nous_portal_messages_extra_body(agent, anthropic_kwargs: dict) -> dict: """Merge Portal ``tags`` / ``session_id`` onto an Anthropic Messages kwargs dict. The Nous profile is only consulted by the OpenAI-wire transport; ``session_id`` only — never ``provider_preferences`` (an OpenAI-wire routing object).""" if getattr(agent, "provider", None) not in {"nous", "nous-portal", "nousresearch"}: return anthropic_kwargs try: from providers import get_provider_profile nous_profile = get_provider_profile("nous") if nous_profile is not None: anthropic_kwargs.setdefault("extra_body", {}).update( nous_profile.build_extra_body(session_id=getattr(agent, "session_id", None))) except Exception as exc: # noqa: BLE001 — never block a turn on tagging logger.debug("Nous Portal extra_body merge failed: %s", exc) return anthropic_kwargs def _estimate_chunk_bytes(chunk: Any) -> int: """Cheap per-chunk size estimate for the stream diagnostic counters: delta string lengths plus a framing floor (~3x cheaper than ``len(repr(chunk))`` in the agent's hottest loop). Unknown shapes just keep the floor.""" size = 40 # SSE/JSON framing floor per chunk def _add(obj, *attrs): nonlocal size for attr in attrs: v = getattr(obj, attr, None) if isinstance(v, str): size += len(v) with contextlib.suppress(Exception): choices = getattr(chunk, "choices", None) if choices: delta = getattr(choices[0], "delta", None) if delta is not None: _add(delta, "content", "reasoning_content", "reasoning") for tc in getattr(delta, "tool_calls", None) or (): fn = getattr(tc, "function", None) if fn is not None: _add(fn, "arguments", "name") else: _add(getattr(chunk, "delta", None), "text", "partial_json") return size # ── Cross-turn stale-call circuit breaker (#58962) ───────────────────── # A session wedged against an unresponsive provider would otherwise hit the # stale detector on every call forever. ``agent._consecutive_stale_streams`` # is bumped on every stale kill and reset only when a call completes or the # provider is swapped (switch_model / try_activate_fallback / # restore_primary_runtime — the streak measured the OLD provider). Past the # give-up threshold, calls abort immediately with an actionable error. def _stale_streak(agent) -> int: try: return int(getattr(agent, "_consecutive_stale_streams", 0) or 0) except Exception: return 0 def _bump_stale_streak(agent) -> None: with contextlib.suppress(Exception): agent._consecutive_stale_streams = _stale_streak(agent) + 1 def _reset_stale_streak(agent) -> None: with contextlib.suppress(Exception): agent._consecutive_stale_streams = 0 _INTERRUPTED_WAIT_STALE_SECONDS = 30.0 def _record_interrupted_provider_wait(agent, elapsed: float, *, response_started: bool) -> bool: """Count a user-aborted pre-response stall toward the stale breaker: past the wait-notice interval an interrupt is evidence of an unresponsive attempt. Mid-response and early interrupts stay neutral.""" if response_started or elapsed < _INTERRUPTED_WAIT_STALE_SECONDS: return False _bump_stale_streak(agent) logger.warning("Interrupted provider wait counted as stale after %.0fs with no output; " "consecutive stale attempts=%d.", elapsed, _stale_streak(agent)) return True def _report_stale_nonstream_kill(agent, api_kwargs: dict, elapsed: float, stale_timeout: float, *, inline: bool = False, hint: Optional[str] = None) -> None: """Log + status message for a stale non-streaming kill, shared by the worker poll loop and the inline ``direct_api_call`` watchdog (their kill/state sequences differ deliberately: different locking models).""" model = api_kwargs.get("model", "unknown") logger.warning("%son-streaming API call stale for %.0fs (threshold %.0fs). " "model=%s context=~%s tokens. Killing connection.", "Inline n" if inline else "N", elapsed, stale_timeout, model, f"{estimate_request_context_tokens(api_kwargs):,}") try: agent._buffer_diagnostic_status( f"⚠️ No response from provider for {int(elapsed)}s (non-streaming, model: {model}). {hint or 'Aborting call.'}") except Exception: logger.debug("stale status buffering failed", exc_info=True) def _touch_stale_kill_activity(agent, elapsed: float) -> None: try: agent._touch_activity(f"stale non-streaming call killed after {int(elapsed)}s") except Exception: logger.debug("stale activity touch failed", exc_info=True) def _check_stale_giveup(agent) -> None: """Raise immediately when the consecutive-stale streak is past the give-up threshold — no network attempt, no stale-timeout wait.""" _giveup = env_int("HERMES_STREAM_STALE_GIVEUP", 5) _streak = _stale_streak(agent) if _giveup > 0 and _streak >= _giveup: raise RuntimeError( "Provider has been unresponsive (no response received) for " f"{_streak} consecutive stale attempts — aborting this call to " "avoid an indefinite stall. Switch models or start a new session, then retry." ) def _stream_env_stale_base() -> "tuple[float, bool]": """(HERMES_STREAM_STALE_TIMEOUT or the implicit 180s, explicit) — like ``AIAgent._resolved_api_call_stale_timeout_base``; an explicit env value is the user's deadline, so it is never capped to the run budget.""" return env_float("HERMES_STREAM_STALE_TIMEOUT", 180.0), "HERMES_STREAM_STALE_TIMEOUT" in os.environ def _configured_stale_base(agent) -> float: """Per-provider ``stale_timeout_seconds`` config, else HERMES_STREAM_STALE_TIMEOUT (180s).""" cfg = get_provider_stale_timeout(agent.provider, agent.model) return cfg if cfg is not None else _stream_env_stale_base()[0] def _local_stream_stale_timeout_default() -> float: """Local-provider stale ceiling: ``agent.local_stream_stale_timeout`` (900s) or HERMES_LOCAL_STREAM_STALE_TIMEOUT. Shared by the stream stale detector and the Responses first-event watchdog so both give a local server the same prefill grace.""" local_default = 900.0 with contextlib.suppress(Exception): from hermes_cli.config import load_config_readonly cfg = load_config_readonly() # read-only consumer — no deepcopy agent_cfg = cfg.get("agent") if isinstance(cfg, dict) else None value = agent_cfg.get("local_stream_stale_timeout") if isinstance(agent_cfg, dict) else None if isinstance(value, (int, float)): local_default = float(value) return env_float("HERMES_LOCAL_STREAM_STALE_TIMEOUT", local_default) def _scale_stale_timeout_for_context(base: float, est_tokens: int) -> float: """Large contexts: slow models think for minutes before the first token; scale the threshold or the detector kills healthy streams.""" if est_tokens > 100_000: return max(base, 300.0) if est_tokens > 50_000: return max(base, 240.0) return base def _cloud_stale_timeout(base: float, api_kwargs: dict) -> float: """Cloud stale-stream patience: ``base`` scaled for context size, then floored for known reasoning models. ``model`` (OpenAI/Anthropic) wins over ``modelId`` (Bedrock); Bedrock's dotted, region-prefixed profile id can't match the floor's slug regex directly, so it is normalized as a fallback.""" from agent.reasoning_timeouts import get_reasoning_stale_timeout_floor timeout = _scale_stale_timeout_for_context(base, estimate_request_context_tokens(api_kwargs)) floor = get_reasoning_stale_timeout_floor(api_kwargs.get("model") or api_kwargs.get("modelId") or "") if floor is None and api_kwargs.get("modelId"): floor = _bedrock_reasoning_stale_floor(api_kwargs["modelId"]) return timeout if floor is None else max(timeout, floor) def _derive_stream_stale_timeout(agent, api_kwargs: dict) -> float: """Stale-stream patience for a provider that is never a local endpoint (Bedrock): the OpenAI/Anthropic stale detector's budget minus its local branch.""" return _cloud_stale_timeout_for(agent, api_kwargs) def cap_to_run_budget(agent, timeout: float) -> float: """Cap an IMPLICIT stale timeout at half the remaining --run-budget (>= 60s), so one hung call can't outlive the run and the wrap-up notice stays reachable (#97968). Shared by the streaming and non-streaming resolvers; callers skip it for explicit user settings.""" run_budget = getattr(agent, "run_budget_seconds", None) started = getattr(agent, "_run_budget_started_at", None) if not run_budget or not started: return timeout remaining = float(run_budget) - (time.time() - float(started)) return min(timeout, max(60.0, remaining * 0.5)) def _cloud_stale_timeout_for(agent, api_kwargs: dict) -> float: """An explicit ``providers..stale_timeout_seconds`` is the operator's deadline and wins over every implicit floor — the context-size tier as well as the reasoning-model floor — so it can SHORTEN patience for a hung stream (#115024). Only the 180s default is scaled and floored.""" explicit = get_provider_stale_timeout(agent.provider, agent.model) if explicit is not None: return explicit base, explicit_env = _stream_env_stale_base() timeout = _cloud_stale_timeout(base, api_kwargs) return timeout if explicit_env else cap_to_run_budget(agent, timeout) def _bedrock_reasoning_stale_floor(model_id: object) -> "float | None": """Map a Bedrock inference-profile id to its reasoning stale-timeout floor. ``us.anthropic.claude-opus-4-6-v1:0`` -> strip the region prefix, then try the segment after the provider namespace (``claude-opus-4-6-v1:0``) and the id with the provider dot dashed (``deepseek-r1-v1:0``). The floor table mixes dashed and dotted versions while Bedrock always dashes, so each candidate is also tried with digit-dash-digit <-> digit-dot-digit swapped (version separators only). First non-None wins; None for unknown models. """ from agent.reasoning_timeouts import get_reasoning_stale_timeout_floor if not model_id or not isinstance(model_id, str): return None name = model_id.strip().lower() for prefix in ("global.", "us.", "eu.", "apac.", "ap.", "au.", "jp.", "ca.", "sa.", "me.", "af."): if name.startswith(prefix): name = name[len(prefix):] break base_candidates = [name] if "." in name: base_candidates.append(name.rsplit(".", 1)[1]) # claude-opus-4-6-v1:0 base_candidates.append(name.replace(".", "-", 1)) # deepseek-r1-v1:0 candidates = dict.fromkeys( form for cand in base_candidates for form in (cand, re.sub(r"(?<=\d)-(?=\d)", ".", cand), re.sub(r"(?<=\d)\.(?=\d)", "-", cand))) return next((f for f in map(get_reasoning_stale_timeout_floor, candidates) if f is not None), None) def _bedrock_converse_call(api_kwargs: dict, *, stream: bool, on_stream_denied=None): """Pop the Hermes routing keys and call ``converse`` / ``converse_stream`` (boto3 directly) with the shared recovery: a cachePoint rejection (Nova: toolConfig.tools, #97281) drops the marker and resends once inside the same attempt; a streaming IAM denial hands off to ``on_stream_denied(client, kwargs, exc)``; a stale connection evicts the cached client so the outer retry builds a fresh pool. Streaming returns the event stream; non-streaming an OpenAI-shaped SimpleNamespace.""" from agent.bedrock_adapter import (_get_bedrock_runtime_client, invalidate_runtime_client, is_stale_connection_error, is_streaming_access_denied_error, normalize_converse_response, recover_from_cache_point_rejection) region = api_kwargs.pop("__bedrock_region__", "us-east-1") api_kwargs.pop("__bedrock_converse__", None) client = _get_bedrock_runtime_client(region) method = client.converse_stream if stream else client.converse finish = (lambda raw: raw.get("stream", [])) if stream else normalize_converse_response try: raw_response = method(**api_kwargs) except Exception as exc: retry_kwargs = recover_from_cache_point_rejection(exc, api_kwargs) if retry_kwargs is not None: return finish(method(**retry_kwargs)) if on_stream_denied is not None and is_streaming_access_denied_error(exc): return on_stream_denied(client, api_kwargs, exc) if is_stale_connection_error(exc): invalidate_runtime_client(region) raise return finish(raw_response) def _dispatch_nonstreaming_api_request(agent, api_kwargs: dict, *, make_client): """Run one non-streaming LLM request for the active api_mode and return it. Shared by ``interruptible_api_call`` and ``direct_api_call``. ``make_client(reason, kind=...)`` builds the per-request client (``"openai"`` / ``"anthropic_messages"``) so callers can register it with their abort/close machinery; bedrock / MoA manage their own clients. Interrupt/abort/close semantics stay in callers. """ if agent.api_mode == "codex_responses": return agent._run_codex_stream(api_kwargs, client=make_client("codex_stream_request"), on_first_delta=getattr(agent, "_codex_on_first_delta", None)) if agent.api_mode == "anthropic_messages": # Request-local client so the stale/interrupt watchdog aborts sockets # from the stranger thread while the worker owns the SDK close (#67142). request_client = make_client("anthropic_messages_request", kind="anthropic_messages") return agent._anthropic_messages_create(api_kwargs, client=request_client) if agent.api_mode == "bedrock_converse": return _bedrock_converse_call(api_kwargs, stream=False) if agent.provider == "moa": # MoA is a virtual provider backed by the in-process MoAClient facade — never # rebuild a request-local client from the virtual metadata. After a client # replacement agent.client may be a native OpenAI client while provider stays # "moa": pop the MoA-internal key ONLY then (the facade consumes it; stripping # it there forces a duplicate fan-out). Only the facade exposes ``prepare()`` (#78382). _completions = getattr(getattr(agent.client, "chat", None), "completions", None) if not callable(getattr(_completions, "prepare", None)): api_kwargs.pop("_moa_prepared_request", None) return agent.client.chat.completions.create(**api_kwargs) request_client = make_client("chat_completion_request") # #93650: keep the bulk wire-format payload out of the SDK's GIL-holding # request transform. No-op unless this really is the OpenAI SDK, so the # MoA facade above and the suite's stand-in clients are unaffected. api_kwargs = bypass_chat_sdk_request_transform(api_kwargs, request_client) return request_client.chat.completions.create(**api_kwargs) def should_use_direct_api_call(agent) -> bool: """Whether an OpenAI-wire request should skip the interrupt worker. Gateway cron turns (#62151) and delegated children (#60203) run inside nested thread pools that wedge before the socket opens when the request is pushed onto yet another daemon worker. Running inline drops the deepest layer; interrupts still work because the inline path registers ``agent._active_request_abort``, which ``interrupt()`` invokes cross-thread (#72227). Cron also inlines Codex Responses (#69734): both Codex paths (non-stream, and streaming via ``_stream_codex_passthrough`` -> ``_interruptible_api_call``) reach ``direct_api_call``, whose client comes from ``make_client`` so the inline stale watchdog can abort it; the stale budget keeps the openai-codex floor/hard cap. Trade-off: the worker-only Codex TTFB/progress/idle watchdogs don't run inline, so a Codex call that never sends a first byte waits the full wall-clock stale budget (600-1200s on large contexts) instead of the ~120s TTFB cutoff. Delegated children and Native/Bedrock/MoA keep their workers: cancellation and client ownership differ. """ api_mode = getattr(agent, "api_mode", None) if getattr(agent, "provider", None) == "moa": return False if getattr(agent, "platform", None) == "cron": return api_mode in {"chat_completions", "codex_responses"} if api_mode != "chat_completions": return False # Delegated child — via the execution ContextVar set by _run_single_child, # with the agent's platform stamp as a fallback for callers that bypass it. with contextlib.suppress(Exception): from agent.delegation_context import is_delegated_child_context if is_delegated_child_context(): return True return getattr(agent, "platform", None) == "subagent" # How often an in-flight direct_api_call refreshes last_activity_ts. Must stay well # under the async-delegation idle stall threshold (450s) and below the 30s monitor sweep. _DIRECT_API_ACTIVITY_HEARTBEAT_SECONDS = 15.0 def _managed_local_load_notice(agent, api_kwargs: dict) -> "Optional[str]": """Live phase notice ("⏳ loading into memory — N%" / "⚙ processing prompt — P%") while the managed local server works before the first token; None when neither applies. Otherwise a cold load reads as a generic stall.""" try: base = str(getattr(agent, "base_url", "") or "") if not base: return None from urllib.parse import urlparse from hermes_cli.local_runtime.load_progress import get_loading_progress, get_prefill_progress from hermes_cli.local_runtime.supervisor import state_path state = json.loads(state_path().read_text(encoding="utf-8-sig")) managed = urlparse(str(state.get("base_url", ""))).netloc.lower() if not managed or urlparse(base).netloc.lower() != managed: return None model = str(api_kwargs.get("model", "")) progress = get_loading_progress().get(model) if progress is not None: return (f"⏳ loading {model} into memory — {progress['percent']}% " "(responses start once the model is loaded)") prefill = get_prefill_progress(model) if prefill is None: return None processed = int(prefill["processed"]) total = estimate_request_context_tokens(api_kwargs) if total and total >= processed: return f"⚙ processing prompt — {max(0, min(100, round(processed / total * 100)))}%" # Counter past the estimate (estimator undercounted): no honest denominator, label-only. return "⚙ processing prompt" except Exception: # noqa: BLE001 — a status nicety must never break a call return None def _resolve_direct_stale_timeout(agent, api_kwargs: dict) -> float: """Stale budget for the inline call via ``agent._compute_non_stream_stale_timeout``, plus the same openai-codex floor/hard cap the worker path applies (inline cron Codex, #69734). A non-numeric result (stub agent) leaves the watchdog disarmed; a resolver that *raises* propagates — swallowing into ``inf`` would reinstate the hang.""" resolver = getattr(agent, "_compute_non_stream_stale_timeout", None) value = resolver(api_kwargs) if callable(resolver) else None if isinstance(value, bool) or not isinstance(value, (int, float)): return float("inf") base_url = getattr(agent, "base_url", None) if getattr(agent, "api_mode", None) == "codex_responses" and not (base_url and is_local_endpoint(base_url)): return _bound_openai_codex_stale_timeout(float(value), estimate_request_context_tokens(api_kwargs)) return float(value) def _inline_nonstream_hard_timeout(stale_timeout: float): """Socket-level backstop for inline non-streaming calls (#85252): the keepalive client uses ``read=None`` and the stranger-thread abort must not ``close()`` the FD (#29507), so a hung provider otherwise waits until TCP dies. Returns an ``httpx.Timeout`` with read == stale budget, a float if httpx is unavailable, or ``None`` when the watchdog is disarmed (non-finite budget).""" if not math.isfinite(stale_timeout) or stale_timeout <= 0: return None conn_cap = min(stale_timeout, 60.0) try: import httpx as _httpx return _httpx.Timeout(connect=conn_cap, read=stale_timeout, write=conn_cap, pool=conn_cap) except Exception: return stale_timeout class _InlineRequest: """Lifecycle state for one inline non-streaming request (#75301). Every transition happens under ``lock``: ``done`` stops a late timer bumping the stale streak after unwind; ``cancelled`` lets an interrupt own the outcome so a racing timer can't misclassify the kill as staleness; ``stale`` is the one-shot transition.""" def __init__(self, agent, api_kwargs: dict, stale_timeout: float, call_start: float): self.agent = agent self.api_kwargs = api_kwargs self.stale_timeout = stale_timeout self.call_start = call_start self.client = None self.done = False self.stale = False self.cancelled = False self.lock = threading.Lock() self.abort_hook = self.abort # single bound object: identity-checked on cleanup self._hb_stop = threading.Event() self._hb = threading.Thread(target=self._activity_heartbeat, name="direct-api-activity-hb", daemon=True) self._watchdog = None def _activity_heartbeat(self) -> None: # Never put the API call itself on another worker thread — that is the nested-pool # deadlock this path exists to avoid (#60203). This ticker only refreshes the clock. while not self._hb_stop.wait(_DIRECT_API_ACTIVITY_HEARTBEAT_SECONDS): with contextlib.suppress(Exception): self.agent._touch_activity("waiting for non-streaming API response") def _on_stale(self) -> None: # Timer thread: aborts sockets only, never issues a request (keeps the no-worker # property). False = request finished or an interrupt owns the outcome; stay silent. if not self.abort("stale_call_kill"): return elapsed = time.time() - self.call_start _report_stale_nonstream_kill(self.agent, self.api_kwargs, elapsed, self.stale_timeout, inline=True) _touch_stale_kill_activity(self.agent, elapsed) def start_watchdogs(self) -> None: """Start the activity heartbeat and (for a finite budget) the stale timer.""" self._hb.start() if math.isfinite(self.stale_timeout) and self.stale_timeout > 0: self._watchdog = threading.Timer(self.stale_timeout, self._on_stale) self._watchdog.name = "direct-api-stale-watchdog" self._watchdog.daemon = True self._watchdog.start() def stop_watchdogs(self) -> None: if self._watchdog is not None: self._watchdog.cancel() self.mark_done() self._hb_stop.set() self._hb.join(timeout=2.0) def _abort_client(self, client, reason: str, log_msg: str) -> None: try: self.agent._abort_request_openai_client(client, reason=reason) except Exception: logger.debug(log_msg, exc_info=True) def abort(self, reason: str) -> bool: """Abort the inline request from a watchdog/interrupt thread. Returns True when this call owned the stale transition (the timer reports/bumps once, never after an interrupt or a completed request). Aborts under the lock (same contract as _RequestClientRegistry): once released the finally may cache the client and the NEXT call check it out.""" with self.lock: if self.done: return False if reason == "stale_call_kill": if self.cancelled: return False newly_stale = not self.stale if newly_stale: self.stale = True # Bump BEFORE releasing: a fast retry's reset must not be # overtaken by this older timer restoring the streak. _bump_stale_streak(self.agent) else: # Interrupt wins the lock -> owns the outcome; a later timer # must not count it as staleness. self.cancelled = True newly_stale = False if self.client is not None: self._abort_client(self.client, reason, f"Inline request abort failed ({reason})") return newly_stale def make_client(self, reason: str, kind: str = "openai"): # Only OpenAI-wire / Codex requests reach direct_api_call; ``kind`` exists # for signature parity with the dispatch helper. client = self.agent._create_request_openai_client(reason=reason, api_kwargs=self.api_kwargs) with self.lock: self.client = client stale_before_dispatch = self.stale if stale_before_dispatch: # Timer fired during client construction: the abort found no # socket, so dispatching now would open one AFTER the only # watchdog fired. Fail here instead. (Residual ms-scale window # before httpx opens its socket is accepted.) self._abort_client(client, "stale_call_kill", "Inline abort after late client registration failed") if stale_before_dispatch: raise TimeoutError( f"Non-streaming API call timed out before request dispatch (threshold: {int(self.stale_timeout)}s)") self.agent._active_request_abort = self.abort_hook return client def mark_done(self) -> None: with self.lock: self.done = True def pop_client(self): with self.lock: client, self.client = self.client, None return client def direct_api_call(agent, api_kwargs: dict): """Run a non-streaming LLM call inline on the conversation thread (cron turns, delegated children — see ``should_use_direct_api_call``): no interrupt worker, so the nested-pool deadlock cannot occur. An activity heartbeat keeps ``last_activity_ts`` advancing (else the stall monitor interrupts a healthy wait at ~450s). A stale-call watchdog bounds the request (#80759): the timer aborts in-flight sockets via the registered hook, and a per-call ``timeout`` equal to the stale budget is the backstop when the abort finds nothing (#85252). Both surface a retryable ``TimeoutError`` for the outer retry loop.""" _check_stale_giveup(agent) agent._touch_activity("waiting for non-streaming API response") # Resolve the budget BEFORE the heartbeat starts: the resolver may raise # (fail-closed), and a leaked heartbeat thread would mask real stalls forever. call_start = time.time() stale_timeout = _resolve_direct_stale_timeout(agent, api_kwargs) # Never override an explicit per-call timeout; otherwise pin read=stale_timeout so a # no-op abort can't leave the read=None socket hanging until TCP dies (#85252). hard_timeout = _inline_nonstream_hard_timeout(stale_timeout) if hard_timeout is not None and "timeout" not in api_kwargs: api_kwargs = {**api_kwargs, "timeout": hard_timeout} request = _InlineRequest(agent, api_kwargs, stale_timeout, call_start) request.start_watchdogs() # Only a clean return reports the reuse reason; errors/interrupts really # close the client so the retry builds a fresh pool. succeeded = False try: response = _dispatch_nonstreaming_api_request(agent, api_kwargs, make_client=request.make_client) except Exception: if getattr(agent, "_interrupt_requested", False): raise InterruptedError("Agent interrupted during API call") from None with request.lock: was_stale = request.stale if was_stale: # Our own abort caused the transport error: raise a retryable # TimeoutError, never InterruptedError ("the user wants to stop"). raise TimeoutError( f"Non-streaming API call timed out after {int(time.time() - call_start)}s with no response " f"(threshold: {int(stale_timeout)}s)") from None raise else: if getattr(agent, "_interrupt_requested", False): raise InterruptedError("Agent interrupted during API call") # Mark ``done`` under the lock so a timer firing between response # arrival and unwind is a no-op and cannot overwrite the reset below. # If a timer already won, the request still completed: return it (the # reset undoes the bump; the finally discards the poisoned client). request.mark_done() _reset_stale_streak(agent) succeeded = True return response finally: request.stop_watchdogs() if getattr(agent, "_active_request_abort", None) is request.abort_hook: agent._active_request_abort = None request_client = request.pop_client() if request_client is not None: agent._close_request_openai_client(request_client, reason="request_complete" if succeeded else "request_error_cleanup") class _RequestClientRegistry: """Per-request client / stream-handle registry shared by the request worker and the stranger threads (interrupt loop, stale detector) that may abort it. ``kind`` (``"openai"`` / ``"anthropic_messages"`` / ``"stream"``) routes :meth:`close_once` (#67142). ``"stream"`` registers a stream handle: under the MoA facade the singleton client has no per-request sockets, so interrupts must close the stream object itself (#57354). Thread-ownership rule (#29507): the owning worker pops + fully closes on its way out. A *stranger* thread only aborts the sockets — never ``client.close()`` — avoiding the FD-recycling race where a just-closed TLS FD was reassigned to ``kanban.db`` and the live SSL BIO wrote into the SQLite header. The abort happens under the lock: once released the worker may cache the client and the NEXT call check it out. Stream handles are safe to close from any thread. """ def __init__(self, agent): self.agent = agent self.client = None self.kind = "openai" self.owner_tid = None self.diag = None # per-attempt stream diagnostics (streaming path) self.lock = threading.Lock() def set_client(self, client, *, kind: str = "openai"): with self.lock: self.client, self.kind, self.owner_tid = client, kind, threading.get_ident() return client @staticmethod def _stream_close_callable(stream): for owner in (stream, getattr(stream, "response", None)): close = getattr(owner, "close", None) if callable(close): return close return None def set_stream_handle(self, stream): return stream if self._stream_close_callable(stream) is None else self.set_client(stream, kind="stream") def _close_stream_handle(self, stream, reason: str) -> None: close = self._stream_close_callable(stream) if close is None: return try: close() logger.info("Streaming response handle closed (%s)", reason) except Exception as exc: logger.debug("Streaming response handle close failed (%s): %s", reason, exc) def close_once(self, reason: str) -> None: with self.lock: request_client, request_kind, owner_tid = self.client, self.kind, self.owner_tid stranger_thread = ( request_kind != "stream" and request_client is not None and owner_tid is not None and owner_tid != threading.get_ident() ) if stranger_thread: abort = (self.agent._abort_request_anthropic_client if request_kind == "anthropic_messages" else self.agent._abort_request_openai_client) abort(request_client, reason=reason) return self.client = None self.owner_tid = None if request_client is None: return if request_kind == "stream": self._close_stream_handle(request_client, reason) elif request_kind == "anthropic_messages": self.agent._close_request_anthropic_client(request_client, reason=reason) else: self.agent._close_request_openai_client(request_client, reason=reason) # Silence budget for high-or-above reasoning effort on a Codex request. GPT-5-family models at # high effort think server-side for 100-170s before the first substantive SSE event even on a # ~6KB prompt (#112909), while the token-sized tiers below hand such a prompt 12s/120s/90s; the # watchdog killed healthy requests three times in a row and blamed the provider. Applies as a # floor to the IMPLICIT defaults only -- explicit env/config values keep winning, and the stale # timeout's run-budget cap is applied AFTER this floor (AIAgent._compute_non_stream_stale_timeout). HIGH_EFFORT_SILENCE_FLOOR_SECONDS = 300.0 # First-progress budget for a lifecycle-only stream on an official-Codex large request: the # stream opened but no substantive model event has arrived. Measured from the physical-attempt # start (a reconnect restarts it; lifecycle frames do not), and applied regardless of reasoning # effort. Equal to the high-effort floor today, but a separate knob so tuning one cannot silently # retune the other. CODEX_FIRST_PROGRESS_TIMEOUT_SECONDS = 300.0 def _high_effort_silence_floor(agent) -> float: """``HIGH_EFFORT_SILENCE_FLOOR_SECONDS`` when the wire reasoning config is enabled at ``high`` or any stronger :data:`~agent.reasoning_effort.EFFORT_LADDER` level (xhigh/max/ultra), else 0.""" from agent.reasoning_effort import EFFORT_LADDER cfg = getattr(agent, "reasoning_config", None) if not isinstance(cfg, dict) or cfg.get("enabled") is False: return 0.0 effort = str(cfg.get("effort") or "").strip().lower() if effort not in EFFORT_LADDER or EFFORT_LADDER.index(effort) < EFFORT_LADDER.index("high"): return 0.0 return HIGH_EFFORT_SILENCE_FLOOR_SECONDS @dataclass class _NonStreamWatchdogs: """Poll-loop thresholds for one non-streaming request.""" stale_timeout: float codex: bool # api_mode == codex_responses (codex watchdogs armed) est_tokens: int ttfb_enabled: bool ttfb_timeout: float idle_enabled: bool idle_timeout: float idle_requires_progress: bool progress_timeout: float = 0.0 def _resolve_nonstream_watchdogs(agent, api_kwargs: dict) -> _NonStreamWatchdogs: """Stale-call timeout plus the Codex Responses stream watchdogs. 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-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 (opt-in ceiling, default 0 = none), HERMES_CODEX_HARD_TIMEOUT_SECONDS. """ # The effort floor on the STALE timeout lives inside _compute_non_stream_stale_timeout so the # run-budget cap still bounds it; here the floor only raises the TTFB/idle implicit defaults. stale_timeout = agent._compute_non_stream_stale_timeout(api_kwargs) codex = agent.api_mode == "codex_responses" openai_codex_backend = _is_openai_codex_backend(agent) est_tokens = estimate_request_context_tokens(api_kwargs) effort_floor = _high_effort_silence_floor(agent) if codex else 0.0 codex_floor = 0.0 # Local Responses servers keep their configured local stale/TTFB grace: the hosted # large-context floor, hard ceiling and TTFB scale-up/cap below must not tighten it. base_url = getattr(agent, "base_url", None) local = bool(base_url) and is_local_endpoint(base_url) if codex and not local: codex_floor = openai_codex_stale_timeout_floor(est_tokens) stale_timeout = _bound_openai_codex_stale_timeout(stale_timeout, est_tokens) idle_default = max(effort_floor, next( (default for threshold, default in ((100_000, 180.0), (50_000, 120.0), (10_000, 60.0)) if est_tokens > threshold), 12.0)) # 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_explicit = env_float("HERMES_CODEX_TTFB_TIMEOUT_SECONDS", -1.0) != -1.0 ttfb_timeout = env_float("HERMES_CODEX_TTFB_TIMEOUT_SECONDS", 120.0) if ttfb_timeout <= 0: ttfb_enabled = False elif codex and not local: # Large requests legitimately spend tens of seconds in admission/prefill before the # first SSE event: scale the cutoff up to the idle default unless TTFB_STRICT is set. 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 codex-responses 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 # Opt-in ceiling (0 = off): a 120s default here silently undid the scale-up above (#91621). ttfb_cap = env_float("HERMES_CODEX_TTFB_MAX_SECONDS", 0.0) if ttfb_cap > 0 and ttfb_timeout > ttfb_cap: logger.info("Capping codex-responses no-event TTFB timeout from %.0fs to %.0fs " "(context=~%s tokens) per HERMES_CODEX_TTFB_MAX_SECONDS.", ttfb_timeout, ttfb_cap, f"{est_tokens:,}") ttfb_timeout = ttfb_cap elif not ttfb_explicit and local: # A local server prefills for minutes before its first event; the chat-completions # siblings already grant local endpoints the local stale ceiling, so the Responses # transport gets the same grace instead of the 120s hosted cutoff (#92302). local_ceiling = _local_stream_stale_timeout_default() if local_ceiling > ttfb_timeout: logger.info("Local provider detected (%s) — no-event TTFB watchdog raised from %.0fs to %.0fs " "(agent.local_stream_stale_timeout); set HERMES_CODEX_TTFB_TIMEOUT_SECONDS for an explicit cutoff.", base_url, ttfb_timeout, local_ceiling) ttfb_timeout = local_ceiling if ttfb_enabled and not ttfb_explicit: # High-effort thinking precedes the first event; the floor outranks the cap. ttfb_timeout = max(ttfb_timeout, effort_floor) # An operator-set idle timeout keeps first-event semantics; only the implicit # default defers arming until model progress. Sentinel: env_float returns the # default for unset AND unparseable values, so both count as implicit. idle_explicit = env_float("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", -1.0) != -1.0 idle_timeout = env_float("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", idle_default) progress_gated = codex and openai_codex_backend and codex_floor > 0 and not idle_explicit 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_requires_progress=progress_gated, # A lifecycle frame proves transport liveness, not model progress. Bound that phase # from the physical-attempt start; events cannot restart the grace period. progress_timeout=CODEX_FIRST_PROGRESS_TIMEOUT_SECONDS if progress_gated else 0.0) def _codex_silent_hang_hint(agent, api_kwargs: dict) -> Optional[str]: hint_fn = getattr(agent, "_codex_silent_hang_hint", None) with contextlib.suppress(Exception): if callable(hint_fn): return hint_fn(model=api_kwargs.get("model")) return None def interruptible_api_call(agent, api_kwargs: dict): """Run the API call on a worker thread so the caller can detect interrupts without waiting for the full HTTP round-trip. Each worker gets its own per-request client (interrupts close only that one); a stale-call detector kills the connection and raises so the main retry loop can back off / rotate credentials / fall back.""" # Nested-pool contexts (cron, delegated children) wedge on a worker thread # (#62151): run inline. See should_use_direct_api_call. if should_use_direct_api_call(agent): return direct_api_call(agent, api_kwargs) _check_stale_giveup(agent) # cross-turn stale breaker (#58962), non-streaming sibling from agent.chat_completion_nonstream import _NonStreamRequest return _NonStreamRequest(agent, api_kwargs).run() def _consume_ephemeral_reasoning_off(agent) -> bool: """Consume the one-shot "answer without thinking" continuation flag. Set by the length-continuation path when a request returned reasoning but NO visible content (thinking ate the output cap); continuation turns never replay prior reasoning, so thinking ON would re-burn the budget. When True the caller overrides the wire reasoning_config with ``{"enabled": False, "effort": "none"}`` for exactly the next call. Prompt-cache cost is bounded to ONE cold prefix write on config-sensitive providers (Anthropic, OpenAI) — far cheaper than four futile full-budget continuations. """ consumed = bool(getattr(agent, "_ephemeral_reasoning_off", False)) if consumed: agent._ephemeral_reasoning_off = False return consumed def _reasoning_config_for_wire(agent): """``agent.reasoning_config`` with the one-shot reasoning-off override applied. Once the route has answered a disable with "reasoning is mandatory" (``agent._reasoning_disable_rejected``), every disable — configured or the one-shot continuation override — is dropped for the rest of the session: the request goes out without a reasoning config and the route applies its own default. """ cfg = agent.reasoning_config ephemeral_off = _consume_ephemeral_reasoning_off(agent) if getattr(agent, "_reasoning_effort_rejected", False): # The route rejected the configured reasoning LEVEL itself (#100536: ``reasoning.effort: # max`` on an enabled config). Omit the reasoning fields for the rest of the session — # the route default — as the auxiliary ladder does; resending would 400 identically. agent._wire_reasoning_config = None return None if cfg is None: # Unset effort: the profile's default (custom/OpenAI-compatible: medium) rather than the # route's own, recorded below as what went out so a rejection of it lands in the branch above. from agent.reasoning_params import unset_reasoning_default cfg = unset_reasoning_default(agent) if getattr(agent, "_reasoning_disable_rejected", False): # The route rejects disables. Resend exactly what the session has # been sending — the user's own config — so the retry lands on the # same provider cache key as every prior request. Only a config that # is itself a disable changes, and that session has never sent # anything else, so nothing warm is lost: a route that said the # disable is *mandatory-on* gets the floor effort (closest to what # the user asked for); a relay that does not know the field gets # nothing (route default). if isinstance(cfg, dict) and ( cfg.get("enabled") is False or cfg.get("effort") == "none" ): if getattr(agent, "_reasoning_floor_required", False): from agent.auxiliary_reasoning_floor import REASONING_FLOOR_EFFORT floored = {**cfg, "enabled": True, "effort": REASONING_FLOOR_EFFORT} agent._wire_reasoning_config = floored return floored agent._wire_reasoning_config = None return None agent._wire_reasoning_config = cfg return cfg if ephemeral_off: cfg = {**(cfg or {}), "enabled": False, "effort": "none"} # What actually went out: the reasoning-rejection rung reads it to tell a rejected # disable (drop the disable) from a rejected level (drop the reasoning fields). agent._wire_reasoning_config = cfg return cfg def _alias_tool_search_bridge_for_xai(agent, transport, tools_for_api): """xAI chat-completions reserves ``tool_search`` and 400s when the bridge declares it (#95003): rename the wire declaration; ``normalize_response`` maps calls back via the transport's ``_last_wire_aliases`` (reset here so a stale map can't reverse-map a name this request never aliased). Deep-copy first (#27907).""" if transport is not None and hasattr(transport, "_last_wire_aliases"): transport._last_wire_aliases = {} is_xai_chat = agent.provider in {"xai", "xai-oauth"} or agent._base_url_hostname == "api.x.ai" if not (is_xai_chat and tools_for_api): return tools_for_api try: import copy as _copy_xai from agent.transports.chat_completions import _rename_tool_search_bridge_for_xai has_bridge = any( (t.get("function") or {}).get("name") == "tool_search" for t in tools_for_api if isinstance(t, dict) ) if has_bridge: tools_for_api = _copy_xai.deepcopy(tools_for_api) tools_for_api, alias_map = _rename_tool_search_bridge_for_xai(tools_for_api) if transport is not None: transport._last_wire_aliases = alias_map except Exception as exc: logger.warning("%s⚠️ Failed to alias tool_search bridge for xAI: %s", getattr(agent, "log_prefix", ""), exc) return tools_for_api def _consume_ephemeral_max_output(agent): """Pop the one-shot ephemeral output cap; whichever path builds the request consumes it.""" ephemeral_out = getattr(agent, "_ephemeral_max_output_tokens", None) if ephemeral_out is not None: agent._ephemeral_max_output_tokens = None return ephemeral_out def _build_anthropic_kwargs(agent, api_messages, tools_for_api, reasoning_config, request_overrides): ctx_len = getattr(agent, "context_compressor", None) ephemeral_out = _consume_ephemeral_max_output(agent) anthropic_kwargs = agent._get_transport().build_kwargs(model=agent.model, messages=agent._prepare_anthropic_messages_for_api(api_messages), tools=tools_for_api, max_tokens=ephemeral_out if ephemeral_out is not None else agent.max_tokens, reasoning_config=reasoning_config, is_oauth=agent._is_anthropic_oauth, preserve_dots=agent._anthropic_preserve_dots(), context_length=ctx_len.context_length if ctx_len else None, base_url=getattr(agent, "_anthropic_base_url", None), fast_mode=request_overrides.get("speed") == "fast", drop_context_1m_beta=bool(getattr(agent, "_oauth_1m_beta_disabled", False))) # Portal reads ``tags`` / ``session_id`` on its Messages route too, but the profile hook # is only consulted by the OpenAI-wire transport — merge here to keep sticky routing. return _merge_nous_portal_messages_extra_body(agent, anthropic_kwargs) def _build_bedrock_kwargs(agent, api_messages, tools_for_api): # Bedrock Converse — the adapter converts messages/tools and calls boto3 directly. return agent._get_transport().build_kwargs(model=agent.model, messages=api_messages, tools=tools_for_api, max_tokens=agent.max_tokens, region=getattr(agent, "_bedrock_region", None) or "us-east-1", guardrail_config=getattr(agent, "_bedrock_guardrail_config", None)) def _build_codex_kwargs(agent, api_messages, tools_for_api, reasoning_config, request_overrides, cache_scope_id): from agent.codex_responses_adapter import classify_responses_route from agent.native_compaction import native_compaction_context_management is_codex_backend, is_xai_responses, is_github_responses = classify_responses_route(agent) # Native server-side compaction (gpt-5.6 on direct OpenAI / ChatGPT Codex routes # only) — None on every other route/model, leaving the request unchanged. context_management = native_compaction_context_management(agent, is_codex_backend=is_codex_backend, is_xai_responses=is_xai_responses, is_github_responses=is_github_responses) # xAI's /responses endpoint 400s on ``pattern``/``format`` schema keywords and on # ``enum`` values containing ``/`` — strip them (#27197). Deep-copy first: the # sanitizers mutate in place and tools_for_api aliases agent.tools (#27907). if is_xai_responses: try: import copy as _copy from tools.schema_sanitizer import strip_pattern_and_format, strip_slash_enum tools_for_api = _copy.deepcopy(tools_for_api) tools_for_api, _ = strip_pattern_and_format(tools_for_api) tools_for_api, _ = strip_slash_enum(tools_for_api) except Exception as exc: logger.warning("%s⚠️ Failed to sanitize tool schemas for xAI: %s", getattr(agent, "log_prefix", ""), exc) ephemeral_out = _consume_ephemeral_max_output(agent) return agent._get_transport().build_kwargs(model=agent.model, messages=agent._prepare_messages_for_non_vision_model(api_messages), tools=tools_for_api, reasoning_config=reasoning_config, session_id=getattr(agent, "session_id", None), cache_scope_id=cache_scope_id, base_url=agent.base_url, max_tokens=ephemeral_out if ephemeral_out is not None else agent.max_tokens, timeout=agent._resolved_api_call_timeout(), request_overrides=request_overrides, provider=getattr(agent, "provider", None), is_github_responses=is_github_responses, is_codex_backend=is_codex_backend, is_xai_responses=is_xai_responses, github_reasoning_extra=agent._github_models_reasoning_extra_body() if is_github_responses else None, replay_encrypted_reasoning=bool(getattr(agent, "_codex_reasoning_replay_enabled", True)), context_management=context_management, text_verbosity=getattr(agent, "text_verbosity", None)) def _build_chat_completions_kwargs(agent, api_messages, tools_for_api, reasoning_config, request_overrides, cache_scope_id): transport = agent._get_transport() tools_for_api = _alias_tool_search_bridge_for_xai(agent, transport, tools_for_api) _is_qwen = agent._is_qwen_portal() _is_or = agent._is_openrouter_url() _host = agent._base_url_lower _is_gh = base_url_host_matches(_host, "models.github.ai") or base_url_host_matches(_host, "githubcopilot.com") _is_lmstudio = (agent.provider or "").strip().lower() == "lmstudio" # _fixed_temperature_for_model may return the OMIT_TEMPERATURE sentinel # (temperature omitted entirely), a numeric override, or None. _omit_temp, _fixed_temp = False, None with contextlib.suppress(Exception): from agent.auxiliary_client import _fixed_temperature_for_model, OMIT_TEMPERATURE _ft = _fixed_temperature_for_model(agent.model, agent.base_url) _omit_temp = _ft is OMIT_TEMPERATURE _fixed_temp = None if _omit_temp else _ft _prefs = _provider_preferences_for_agent(agent) _qwen_meta = {"sessionId": agent.session_id or "hermes", "promptId": str(uuid.uuid4())} if _is_qwen else None _profile = None with contextlib.suppress(Exception): from providers import get_provider_profile _profile = get_provider_profile(agent.provider) _ephemeral_out = _consume_ephemeral_max_output(agent) # Strip image parts for non-vision models on BOTH paths (registered # providers with profiles used to bypass it). _common = dict(model=agent.model, messages=agent._prepare_messages_for_non_vision_model(api_messages), tools=tools_for_api, base_url=agent.base_url, timeout=agent._resolved_api_call_timeout(), max_tokens=agent.max_tokens, ephemeral_max_output_tokens=_ephemeral_out, max_tokens_param_fn=agent._max_tokens_param, reasoning_config=reasoning_config, request_overrides=request_overrides, session_id=getattr(agent, "session_id", None), cache_scope_id=cache_scope_id, ollama_num_ctx=agent._ollama_num_ctx, provider_preferences=_prefs or None, openrouter_min_coding_score=agent.openrouter_min_coding_score, supports_reasoning=agent._supports_reasoning_extra_body(), qwen_session_metadata=_qwen_meta) if _profile: # Profiles handle per-provider quirks via hooks fed the context above. return transport.build_kwargs(provider_profile=_profile, **_common) # Legacy flag path: only for a provider absent from the providers/ registry. return transport.build_kwargs( **_common, model_lower=(agent.model or "").lower(), is_openrouter=_is_or, is_nous=base_url_host_matches(_host, "nousresearch.com"), is_qwen_portal=_is_qwen, is_github_models=_is_gh, is_nvidia_nim=base_url_host_matches(_host, "integrate.api.nvidia.com"), is_kimi=any(base_url_host_matches(agent.base_url, h) for h in ("api.kimi.com", "moonshot.ai", "moonshot.cn")), is_tokenhub=base_url_host_matches(_host, "tokenhub.tencentmaas.com"), is_lmstudio=_is_lmstudio, is_custom_provider=agent.provider == "custom", qwen_prepare_fn=agent._qwen_prepare_chat_messages if _is_qwen else None, qwen_prepare_inplace_fn=agent._qwen_prepare_chat_messages_inplace if _is_qwen else None, fixed_temperature=_fixed_temp, omit_temperature=_omit_temp, github_reasoning_extra=agent._github_models_reasoning_extra_body() if _is_gh else None, lmstudio_reasoning_options=agent._lmstudio_reasoning_options_cached() if _is_lmstudio else None, provider_name=agent.provider, ) def build_api_kwargs(agent, api_messages: list, tools_for_api: list | None = None) -> dict: """Build the keyword arguments dict for the active API mode. Wraps the per-api_mode builder so the conversation-affinity headers (OpenCode's ``x-opencode-session``, a custom provider's opt-in ``session_affinity_header``) ride on every request regardless of transport (chat_completions / codex_responses / anthropic_messages). No-op for every other provider. """ from agent.opencode_affinity import merge_session_affinity_headers kwargs = _build_api_kwargs_for_mode(agent, api_messages, tools_for_api) return merge_session_affinity_headers( kwargs, getattr(agent, "provider", None), getattr(agent, "base_url", None), getattr(agent, "session_id", None), ) def _build_api_kwargs_for_mode(agent, api_messages: list, tools_for_api: list | None = None) -> dict: # One-shot continuation override — consumed exactly once, on the FIRST # request this call builds (only one api_mode branch runs per invocation). reasoning_config = _reasoning_config_for_wire(agent) if tools_for_api is None: tools_for_api = agent.tools # The one place request_overrides are consumed: static /fast values are already pinned # in agent.request_overrides; auto/cold windows layer the fast override per request. request_overrides = effective_request_overrides(agent) if agent.api_mode == "anthropic_messages": return _build_anthropic_kwargs(agent, api_messages, tools_for_api, reasoning_config, request_overrides) if agent.api_mode == "bedrock_converse": return _build_bedrock_kwargs(agent, api_messages, tools_for_api) # Rotation-stable logical cache scope shared by every OpenAI-wire branch # (memoized on the agent); anthropic/bedrock above don't use it. cache_scope_id = _prompt_cache_scope_for_agent(agent) builder = _build_codex_kwargs if agent.api_mode == "codex_responses" else _build_chat_completions_kwargs return builder(agent, api_messages, tools_for_api, reasoning_config, request_overrides, cache_scope_id) def _model_dump_safe(obj): """``model_dump(warnings=False)`` (avoids pydantic serializer UserWarnings on generic-union SDK models), falling back for shims that reject the kwarg.""" try: return obj.model_dump(warnings=False) except TypeError: return obj.model_dump() def _dump_if_model(value): return _model_dump_safe(value) if hasattr(value, "model_dump") else value def _assistant_reasoning_text(agent, assistant_message) -> Optional[str]: """Structured reasoning, else inline ```` blocks embedded in content.""" reasoning_text = agent._extract_reasoning(assistant_message) if not reasoning_text: content = flatten_message_text(getattr(assistant_message, "content", None)) think_blocks = re.findall(r'(.*?)', content, flags=re.DOTALL) if think_blocks: reasoning_text = "\n\n".join(b.strip() for b in think_blocks if b.strip()) or None if reasoning_text and agent.verbose_logging: logging.debug(f"Captured reasoning ({len(reasoning_text)} chars): {reasoning_text}") # When streaming is active the reasoning was already displayed during the # stream (structured deltas or tag extraction); fire only for # non-streaming modes (gateway, batch, quiet). Anything not shown during # streaming is caught by the CLI post-response fallback. if reasoning_text and agent.reasoning_callback and not agent.stream_delta_callback and not agent._stream_callback: with contextlib.suppress(Exception): agent.reasoning_callback(reasoning_text) return _sanitize_surrogates(reasoning_text) if reasoning_text else reasoning_text def _assistant_content_for_storage(agent, assistant_message): # Sanitize surrogates (Kimi/GLM via Ollama emit code points that crash json.dumps), # strip inline tags at the storage boundary (they leaked to platforms and # polluted titles), then redact inlined credentials before the message enters # history / state.db / gateway delivery (no-op with HERMES_REDACT_SECRETS off). content = _sanitize_surrogates(flatten_message_text(getattr(assistant_message, "content", None))) if isinstance(content, str) and content: content = agent._strip_think_blocks(content).strip() if content: from agent.redact import redact_sensitive_text content = redact_sensitive_text(content) return content def _assistant_tool_call_dict(agent, tool_call, index: int) -> dict: raw_id = getattr(tool_call, "id", None) call_id = getattr(tool_call, "call_id", None) if not isinstance(call_id, str) or not call_id.strip(): call_id, _ = agent._split_responses_tool_id(raw_id) if not isinstance(call_id, str) or not call_id.strip(): if isinstance(raw_id, str) and raw_id.strip(): call_id = raw_id.strip() else: _fn = getattr(tool_call, "function", None) call_id = agent._deterministic_call_id(getattr(_fn, "name", "") if _fn else "", getattr(_fn, "arguments", "{}") if _fn else "{}", index) call_id = call_id.strip() response_item_id = getattr(tool_call, "response_item_id", None) if not isinstance(response_item_id, str) or not response_item_id.strip(): _, response_item_id = agent._split_responses_tool_id(raw_id) response_item_id = agent._derive_responses_function_call_id(call_id, response_item_id if isinstance(response_item_id, str) else None) # Arguments are deliberately NOT redacted: this dict is replayed to the model every # turn, so a ``***`` mask would break credential-dependent commands (#43083). tc_dict = {"id": call_id, "call_id": call_id, "response_item_id": response_item_id, "type": tool_call.type, "function": {"name": tool_call.function.name, "arguments": tool_call.function.arguments}} # Preserve extra_content (Gemini thought_signature) or Gemini 3 thinking # models 400 on the next request. # Tool-call arguments are intentionally NOT redacted here. This dict enters the in-memory conversation # history that is replayed to the model on every subsequent turn AND persisted to state.db, which is # itself replayed verbatim on session resume (get_messages_as_conversation). Masking a credential to # `***` here poisons that replay: the model reads back its own `PGPASSWORD='***' psql ...` call and # copies the placeholder into the next tool call, breaking every credential-dependent command on the # second turn (#43083). The masking also provided no real protection — the same secret still leaks # verbatim through tool OUTPUT (file contents, command output, diffs, the compaction block), none of # which this pass ever touched. Keeping secrets out of the replayable store is a separate # tokenization/vault concern, not something arg-redaction can deliver without breaking replay. # Storage-time redaction remains governed by the `security.redact_secrets` toggle. (#19798 introduced # this; #43083 removed it.) Preserve extra_content (e.g. Gemini thought_signature) so it is sent back on # subsequent API calls. Without this, Gemini 3 thinking models reject the request with a 400 error. extra = getattr(tool_call, "extra_content", None) if extra is not None: tc_dict["extra_content"] = _dump_if_model(extra) return tc_dict def build_assistant_message(agent, assistant_message, finish_reason: str) -> dict: """Build a normalized assistant message dict (reasoning, reasoning_details, optional tool_calls) shared by the tool-call and final-response paths. Textless turns are NOT padded here: ``repair_empty_non_final_messages`` is the single owner — write-time padding broke codex commentary turns and cannot survive ``_rows_to_conversation``.""" assistant_tool_calls = getattr(assistant_message, "tool_calls", None) reasoning_text = _assistant_reasoning_text(agent, assistant_message) msg = stamp_message_timestamp({"role": "assistant", "content": _assistant_content_for_storage(agent, assistant_message), "reasoning": reasoning_text, "finish_reason": finish_reason}) raw_reasoning_content = getattr(assistant_message, "reasoning_content", None) if raw_reasoning_content is None: model_extra = getattr(assistant_message, "model_extra", None) or {} if isinstance(model_extra, dict) and "reasoning_content" in model_extra: raw_reasoning_content = model_extra["reasoning_content"] if raw_reasoning_content is not None: msg["reasoning_content"] = _sanitize_surrogates(raw_reasoning_content) elif assistant_tool_calls and agent._needs_thinking_reasoning_pad(): # DeepSeek v4 / Kimi thinking modes 400 on a replayed tool-call message without # reasoning_content; pad with a single space (empty string is rejected too). # Without it, replaying the persisted message causes HTTP 400 ("The reasoning_content in the # thinking mode must be passed back to the API"). Include streamed reasoning text when captured; # otherwise pad with a single space — DeepSeek V4 Pro tightened validation and rejects empty string # ("The reasoning content in the thinking mode must be passed back to the API"). A space satisfies # non-empty checks everywhere without leaking fabricated reasoning. Refs #15250, #17400, #17341. msg["reasoning_content"] = reasoning_text or " " elif reasoning_text: # Streaming-only providers accumulate reasoning via deltas and never set # it on the message; replaying through a thinking model then 400s. # Promote ONLY when nothing set the field: SDK reasoning_content and the # tool-call pad win, and reasoning-less turns leave the field absent so # the replay-time leak guard and promotion tiers still apply. # Additive fallback (refs #16844, #16884). Streaming-only providers (glm, MiniMax, gpt-5.x via aigw, # Anthropic via openai-compat shims) accumulate reasoning through ``delta.reasoning_content`` chunks # but never land it on the message object as a top-level attribute, so neither branch above fires # and the chain-of-thought is stored only under the internal ``reasoning`` key. When the user later # replays that history through a DeepSeek-v4 / Kimi thinking model, the missing # ``reasoning_content`` causes HTTP 400 ("The reasoning_content in the thinking mode must be passed # back to the API."). Promote the already-sanitized streamed ``reasoning_text`` to # ``reasoning_content`` at write time, but ONLY when no prior branch already set it AND we actually # captured reasoning text. This preserves every existing behavior: - SDK-exposed # ``reasoning_content`` (OpenAI/Moonshot/DeepSeek SDK) still wins. msg["reasoning_content"] = reasoning_text if getattr(assistant_message, "reasoning_details", None): # Preserve reasoning_details exactly (opaque signature / # encrypted_content fields) for cross-turn reasoning continuity. preserved = [] for d in assistant_message.reasoning_details: if isinstance(d, dict): preserved.append(d) elif hasattr(d, "__dict__"): preserved.append(d.__dict__) elif hasattr(d, "model_dump"): preserved.append(_model_dump_safe(d)) if preserved: msg["reasoning_details"] = preserved # Provider-native carriers replayed verbatim on later turns: # anthropic_content_blocks keeps interleaved thinking + tool_use order # (reconstruction reorders signed blocks -> HTTP 400); codex_* items are # the encrypted reasoning / exact message items Responses prefix caching # needs. for attr in ("anthropic_content_blocks", "bedrock_content_blocks", "codex_reasoning_items", "codex_message_items"): value = getattr(assistant_message, attr, None) if value: msg[attr] = value if attr == "codex_reasoning_items": from agent.codex_responses_adapter import ( has_replayable_native_compaction_checkpoint, ) if has_replayable_native_compaction_checkpoint(agent, [msg]): note_checkpoint = getattr( agent.context_compressor, "note_native_compaction_checkpoint", None ) if callable(note_checkpoint): note_checkpoint() # The response priced the pre-checkpoint input, not the next # compacted request. A matching durable prefix is now stale. from agent.usage_anchor import set_usage_anchor set_usage_anchor(agent, None) # The next request drops every item before this checkpoint, so a repeat # read must serve content again, not an "unchanged" stub (#32106). # Without a task id the reset would clear every task's caches. if task_id := getattr(agent, "_current_task_id", None): from agent.conversation_compression import _reset_read_dedup_caches _reset_read_dedup_caches(task_id, session_id=getattr(agent, "session_id", None) or "") if assistant_tool_calls: msg["tool_calls"] = [_assistant_tool_call_dict(agent, tc, i) for i, tc in enumerate(assistant_tool_calls)] return msg def rewrite_prompt_model_identity(agent, model: str, provider: str) -> None: """Rewrite the cached prompt's ``Model:``/``Provider:`` lines after a provider switch. Not persisted: the stored row keeps the primary's labels so a restored primary replays a byte-identical prompt (prefix cache intact). Only the LAST occurrence of each line is touched — earlier matches may be user content (memory snapshots, context files).""" sp = getattr(agent, "_cached_system_prompt", None) if not isinstance(sp, str) or not sp: return for label, value in (("Model", model), ("Provider", provider)): if not value: continue matches = list(re.finditer(rf"(?m)^{label}: .*$", sp)) if matches: last = matches[-1] sp = f"{sp[:last.start()]}{label}: {value}{sp[last.end():]}" agent._cached_system_prompt = sp def _fallback_entry_key(fb: dict) -> tuple[str, str, str]: return (str(fb.get("provider") or "").strip().lower(), str(fb.get("model") or "").strip(), str(fb.get("base_url") or "").strip().rstrip("/")) def _fallback_entry_unavailable_without_network(agent, fb: dict) -> Optional[str]: """Return a skip reason for fallback entries known to be unusable locally.""" if (fb.get("provider") or "").strip().lower() != "nous": return None try: from hermes_cli.auth import get_provider_auth_state state = get_provider_auth_state("nous") or {} except Exception as exc: return f"nous_auth_unreadable:{type(exc).__name__}" has_token = any(isinstance(t, str) and t.strip() for t in (state.get("access_token"), state.get("refresh_token"))) return None if has_token else "nous_token_missing" _FALLBACK_REASON_LABELS = { FailoverReason.auth: "authentication failed", FailoverReason.auth_permanent: "authentication permanently failed", FailoverReason.billing: "billing or quota exhausted", FailoverReason.rate_limit: "rate limit", FailoverReason.upstream_rate_limit: "upstream model rate limit", FailoverReason.overloaded: "provider overloaded", FailoverReason.server_error: "provider server error", FailoverReason.timeout: "request timeout", FailoverReason.ssl_cert_verification: "TLS certificate verification failed", FailoverReason.context_overflow: "context window exceeded", FailoverReason.payload_too_large: "request payload too large", FailoverReason.image_too_large: "image payload too large", FailoverReason.model_not_found: "model not found", FailoverReason.provider_policy_blocked: "provider policy blocked the request", FailoverReason.content_policy_blocked: "content policy blocked the request", FailoverReason.format_error: "request format rejected", FailoverReason.role_alternation: "adjacent same-role messages rejected", FailoverReason.invalid_encrypted_content: "encrypted reasoning state rejected", FailoverReason.multimodal_tool_content_unsupported: "multimodal tool content unsupported", FailoverReason.thinking_signature: "thinking signature rejected", FailoverReason.long_context_tier: "long-context tier unavailable", FailoverReason.oauth_long_context_beta_forbidden: "OAuth long-context beta unavailable", FailoverReason.llama_cpp_grammar_pattern: "grammar pattern rejected", FailoverReason.unknown: "provider failure", } def _fallback_reason_text(reason: "FailoverReason | None") -> str: """Return a concise operator-facing explanation for a fallback switch.""" label = _FALLBACK_REASON_LABELS.get(reason) return label or str(getattr(reason, "value", None) or reason or "provider failure").replace("_", " ") def _is_anthropic_wire_url(url: str) -> bool: """Same Messages-only host match as determine_api_mode() / _detect_api_mode_for_url(): api.anthropic.com, a /anthropic suffix, or Kimi Code's api.kimi.com/coding (its /chat/completions 404s — #77256).""" from hermes_cli.providers import host_mandated_api_mode return host_mandated_api_mode(url) == "anthropic_messages" def _fallback_api_mode_hint(fb: dict, fb_provider: str, fb_base_url_hint: Optional[str]) -> tuple[bool, str]: """(explicit, api_mode) for a fallback entry from its ORIGINAL base_url: resolve_provider_client() rewrites a dual-surface /anthropic base to /v1, losing the Anthropic wire signal. An explicit ``api_mode`` always wins (even "chat_completions") and suppresses later re-detection; ``provider: anthropic`` without a base_url still resolves to anthropic_messages.""" from hermes_cli.runtime_provider import _get_named_custom_provider, _parse_api_mode # Entries accept the same ``api_mode`` / ``transport`` spellings as ``providers.``. explicit = _parse_api_mode(fb.get("api_mode") or fb.get("transport")) if explicit: return True, explicit # A named ``providers.`` block declares its wire once (``api_mode``/``transport``); a # fallback entry naming that provider inherits it instead of being re-detected from the host # (#33062, #81932: an Anthropic-Messages or Responses-only relay on a plain host was downgraded # to chat_completions while resolve_provider_client had already built the declared client). if fb_provider and fb_provider not in {"custom", "moa"}: declared = (_get_named_custom_provider(fb_provider) or {}).get("api_mode") if declared: return True, declared if fb_provider == "anthropic" or (fb_base_url_hint and _is_anthropic_wire_url(fb_base_url_hint)): return False, "anthropic_messages" return False, "chat_completions" def _fallback_api_mode_resolved(agent, fb_provider: str, fb_model: str, fb_base_url: str) -> str: """Re-detect api_mode from provider / resolved base URL / model when the hint pass landed on the chat_completions default (never called for an explicit api_mode).""" if fb_provider == "openai-codex": return "codex_responses" from hermes_cli.models import opencode_model_api_mode from hermes_cli.runtime_provider_custom import _opencode_family_for_custom opencode_family = _opencode_family_for_custom(fb_provider, fb_base_url) if opencode_family is not None: # OpenCode Zen/Go/free serve Responses-only (muse-spark, gpt-*, grok-*), anthropic_messages # (minimax, qwen) and chat_completions models behind one provider; the primary /model path # already re-derives per model — the fallback wire must agree (#102148). return opencode_model_api_mode(opencode_family, fb_model) if fb_provider in {"nous", "nous-portal", "nousresearch"}: # Portal is dual-wire: anthropic/* must land on /v1/messages (the swap rebuilds the native client). from hermes_cli.providers import nous_api_mode return nous_api_mode(fb_model) if _is_anthropic_wire_url(fb_base_url): # Named custom providers (cron-anthropic) resolve base_url from config; the hint pass never saw it. return "anthropic_messages" if agent._is_azure_openai_url(fb_base_url): return "chat_completions" # Azure serves gpt-5.x on /chat/completions — no Responses API. # Provider exceptions (Copilot gpt-5-mini) stay inside the requires-responses predicate. if agent._is_direct_openai_url(fb_base_url) or agent._provider_model_requires_responses_api(fb_model, provider=fb_provider): return "codex_responses" host = base_url_hostname(fb_base_url) if fb_provider == "bedrock" or (host.startswith("bedrock-runtime.") and base_url_host_matches(fb_base_url, "amazonaws.com")): return "bedrock_converse" return "chat_completions" def _rebind_fallback_credential_pool(agent, fb_provider: str, fb_model: str) -> None: """Rebind the credential pool when the provider changes (else rate_limit/billing/auth recovery mutates the wrong credentials and overwrites the fallback's base_url). Same-provider pool: kept.""" existing_pool = getattr(agent, "_credential_pool", None) if existing_pool is not None: pool_provider = (getattr(existing_pool, "provider", "") or "").strip().lower() if pool_provider and pool_provider != fb_provider: logger.info( "Fallback to %s/%s: clearing primary credential pool (pool_provider=%s) to prevent cross-provider contamination", fb_provider, fb_model, pool_provider) agent._credential_pool = agent._credential_pool_entry_id = None if getattr(agent, "_credential_pool", None) is None: try: from agent.credential_pool import load_pool fallback_pool = load_pool(fb_provider) if fallback_pool and fallback_pool.has_credentials(): agent._credential_pool = fallback_pool logger.info("Fallback to %s/%s: attached fallback credential pool", fb_provider, fb_model) except Exception as exc: logger.debug("Fallback to %s/%s: could not attach credential pool: %s", fb_provider, fb_model, exc) def _log_fallback_activated(agent, reason, old_model, old_provider, fb_model, fb_provider) -> None: """A billing switch is a WARNING naming the profile, both models and the remedy: the gateway persists the turn as a transient failure otherwise, and nothing in the log says the paid model was refused for credits or how to fix it (#115702). Other reasons stay INFO.""" if reason != FailoverReason.billing: logger.info("Fallback activated: %s → %s (%s)", old_model, fb_model, fb_provider) return from hermes_constants import get_hermes_home, profile_name_for_home profile = profile_name_for_home(get_hermes_home()) or "default" remedy = "hermes model" if profile == "default" else f"hermes -p {profile} model" logger.warning( "Profile %s: %s via %s refused for billing/credits — using fallback %s via %s. " "Top up credits, or run `%s` to pick a model this account can use.", profile, old_model, old_provider, fb_model, fb_provider, remedy, ) def _fallback_chain_exhausted(agent, reason: "FailoverReason | None") -> bool: """Chain exhausted (always False). A non-empty chain walked on a non-rate-limit failure arms a short cooldown so next turn's restore_primary_runtime stays gated instead of replaying the whole context across every provider again.""" from agent.fallback_cooldown import _RATE_LIMIT_FAILOVER_REASONS if agent._fallback_chain and reason not in _RATE_LIMIT_FAILOVER_REASONS: agent._rate_limited_until = max( getattr(agent, "_rate_limited_until", 0) or 0, time.monotonic() + _FALLBACK_EXHAUSTED_COOLDOWN_S) return False def _candidate_pool_exhausted(agent, fb_provider: str, fb_model: str) -> bool: """True when every credential the candidate would use sits in an exhaustion cooldown longer than the retry loop's longest wait (the 600s Retry-After cap): switching to it only fails the turn the same way the primary just did (#89401). A short throttle still gets its chance.""" pool = getattr(agent, "_credential_pool", None) if pool is None or (getattr(pool, "provider", "") or "").strip().lower() != fb_provider: try: from agent.credential_pool import load_pool pool = load_pool(fb_provider) except Exception: return False if pool is None or not pool.has_credentials() or pool.has_available(model=fb_model): return False until = pool.next_available_at(model=fb_model) return until is None or until - time.time() > 600 def _should_skip_fallback_candidate(agent, fb: dict, fb_key: tuple, fb_provider: str, fb_model: str, unavailable: set) -> bool: """True when the entry is already unavailable, malformed, locally unusable, or resolves to the backend that just failed (falling back to it would loop the failure).""" if fb_key in unavailable: logger.debug("Fallback skip: %s previously marked unavailable", fb_key) return True if not fb_provider or not fb_model: return True from agent.fallback_cooldown import _is_entitlement_rejected if _is_entitlement_rejected(agent, fb_provider, fb_model): logger.info("Fallback skip: %s/%s was rejected as unentitled for this account", fb_provider, fb_model) return True if _candidate_pool_exhausted(agent, fb_provider, fb_model): logger.warning("Fallback skip: %s/%s credential pool is exhausted (every entry in cooldown)", fb_provider, fb_model) return True local_skip_reason = _fallback_entry_unavailable_without_network(agent, fb) if local_skip_reason: unavailable.add(fb_key) logger.warning("Fallback skip: %s/%s is not locally usable (%s); suppressing for this session", fb_provider, fb_model, local_skip_reason) return True # Identity semantics (axes, shim aliases, credential surfaces, multi-endpoint pools) # are owned by agent.backend_identity — do not re-implement comparisons here. # Skip entries that resolve to the same backend that just failed — falling back to it loops the failure. # See #22548, #62984, #70893. from agent.backend_identity import BackendIdentity, should_skip_candidate current_ident = BackendIdentity.build(provider=getattr(agent, "provider", ""), model=getattr(agent, "model", ""), base_url=str(getattr(agent, "base_url", "") or "")) fb_ident = BackendIdentity.build(provider=fb_provider, model=fb_model, base_url=(fb.get("base_url") or "")) if should_skip_candidate(fb_ident, current_ident): logger.warning( "Fallback skip: chain entry %s/%s resolves to the same backend as the current one (%s)", fb_provider, fb_model, current_ident.base_url or current_ident.provider) return True return False def _update_fallback_context_compressor(agent) -> None: """Point compression limits at the fallback model's context window (not the primary's), respecting the explicit model.context_length config override.""" compressor = getattr(agent, "context_compressor", None) if not compressor: return from agent.model_metadata import get_model_context_length fb_context_length = get_model_context_length( agent.model, base_url=agent.base_url, api_key=agent.api_key if isinstance(agent.api_key, str) else "", # callable (Entra ID) → probes need str provider=agent.provider, config_context_length=getattr(agent, "_config_context_length", None), custom_providers=getattr(agent, "_custom_providers", None), ) compressor.update_model( # callable api_key preserved → call_llm model=agent.model, context_length=fb_context_length, base_url=agent.base_url, api_key=getattr(agent, "api_key", ""), provider=agent.provider, api_mode=agent.api_mode, ) # Fallback activation is an error path: refresh an EXISTING verdict eagerly (the ceiling was voided by # update_model()), but a session that never probed keeps its lazy compaction-time probe rather than # resolving an auxiliary client while the primary route is failing (#114707). if getattr(agent, "_compression_feasibility_checked", False) is True: from agent.conversation_compression import revalidate_compression_feasibility revalidate_compression_feasibility(agent) def _reresolve_fallback_reasoning_config(agent) -> None: """Per-model override > global reasoning_effort (YAML False = disabled); a config load failure keeps the current reasoning_config rather than killing the swap.""" try: # Re-resolve reasoning_config for the new fallback model (Closes #21256). Wrapped in try/except # because a config load failure must not kill the swap. from hermes_cli.config import load_config from hermes_constants import resolve_reasoning_config agent.reasoning_config = resolve_reasoning_config(load_config() or {}, agent.model) logger.info("Fallback %s: reasoning_config resolved: %s", agent.model, agent.reasoning_config) except Exception as _reasoning_err: logger.debug("Failed to resolve reasoning_config for fallback %s; keeping current: %s", agent.model, _reasoning_err) def _rescope_fallback_extra_body(agent, old_model: str, old_provider: str, old_base_url: str) -> None: """Drop the OLD provider's custom_providers-contributed extra_body keys, then merge the fallback provider's own. KEY-SCOPED: a key is dropped only if its value still equals what the old provider's config injected — a caller override of the same key won at init and differs, so it survives; keys the new provider redefines are re-added by the merge.""" try: from agent.agent_init import _custom_provider_extra_body_for_agent, _merge_custom_provider_extra_body custom_providers = getattr(agent, "_custom_providers", None) or [] old_provider_eb = _custom_provider_extra_body_for_agent(provider=old_provider, model=old_model, base_url=old_base_url, custom_providers=custom_providers) or {} overrides = dict(getattr(agent, "request_overrides", {}) or {}) existing_eb = overrides.get("extra_body") if isinstance(existing_eb, dict) and old_provider_eb: scrubbed = {k: v for k, v in existing_eb.items() if not (k in old_provider_eb and v == old_provider_eb[k])} if scrubbed: overrides["extra_body"] = scrubbed else: overrides.pop("extra_body", None) agent.request_overrides = overrides _merge_custom_provider_extra_body(agent, custom_providers) logger.info("Fallback %s: extra_body resolved: %s", agent.model, (getattr(agent, "request_overrides", {}) or {}).get("extra_body")) except Exception as _eb_err: logger.debug("Failed to resolve extra_body for fallback %s; keeping current: %s", agent.model, _eb_err) def _buffer_fallback_notice(agent, notice: str) -> None: """Buffer the switch notice for terminal failure AND retain it as a durable one-shot for _emit_pending_fallback_notice (a successful fallback clears retry chatter).""" agent._buffer_diagnostic_status(notice) pending = getattr(agent, "_pending_fallback_notice", None) if isinstance(pending, list): pending.append(notice) else: agent._pending_fallback_notice = [str(pending), notice] if pending else [notice] def try_activate_fallback(agent, reason: "FailoverReason | None" = None, reset_at=None) -> bool: """Switch to the next fallback model/provider in the chain; False when exhausted. Swaps client, model slug and provider in place so the retry loop continues on the new backend; client construction goes through resolve_provider_client (no duplicated provider→key mappings).""" from agent.fallback_cooldown import _arm_rate_limit_cooldown, switch_deferred_by_reset if switch_deferred_by_reset(agent, reason, reset_at): return False cooldown_seconds = _arm_rate_limit_cooldown(agent, reason, reset_at=reset_at) while True: if agent._fallback_index >= len(agent._fallback_chain): return _fallback_chain_exhausted(agent, reason) fb = agent._fallback_chain[agent._fallback_index] agent._fallback_index += 1 fb_key = _fallback_entry_key(fb) if getattr(agent, "_unavailable_fallback_keys", None) is None: agent._unavailable_fallback_keys = set() unavailable = agent._unavailable_fallback_keys fb_provider = (fb.get("provider") or "").strip().lower() fb_model = (fb.get("model") or "").strip() if _should_skip_fallback_candidate(agent, fb, fb_key, fb_provider, fb_model, unavailable): continue try: from agent.auxiliary_client import resolve_provider_client from hermes_cli.fallback_config import resolve_entry_api_key # Pass the entry's base_url/api_key so custom endpoints (Ollama Cloud) resolve instead # of falling through to OpenRouter defaults. fb_base_url_hint = (fb.get("base_url") or "").strip() or None fb_api_key_hint = resolve_entry_api_key(fb) fb_api_mode_explicit, fb_api_mode = _fallback_api_mode_hint(fb, fb_provider, fb_base_url_hint) # Ollama Cloud: OLLAMA_API_KEY from env when the entry has no key. Host match, not # substring — GHSA-76xc-57q6-vm5m. if fb_base_url_hint and base_url_host_matches(fb_base_url_hint, "ollama.com") and not fb_api_key_hint: from agent.secret_scope import get_secret fb_api_key_hint = get_secret("OLLAMA_API_KEY") or None # raw_codex=True: the main agent needs direct responses.stream() access for Codex providers. fb_client, _resolved_fb_model = resolve_provider_client( fb_provider, model=fb_model, raw_codex=True, explicit_base_url=fb_base_url_hint, explicit_api_key=fb_api_key_hint, api_mode=fb_api_mode) if fb_client is None: logger.warning("Fallback to %s failed: provider not configured", fb_provider) unavailable.add(fb_key) continue if fb_provider == "moa": # A MoA entry means the preset itself, exactly like ``provider: moa`` in config or # ``/model --provider moa``. The chokepoint's client is the preset's # aggregator: it only proves the preset resolves and the aggregator has credentials. # Installing it as the acting client with the virtual identity is a hybrid nobody # handles (#112525: preset name sent as model id → 404; #112623: every # ``provider == "moa"`` guard and key misfires and the next rebuild swaps in the # facade anyway). Bind the facade with the same pins every other MoA build site uses. fb_base_url, fb_api_mode = "moa://local", "chat_completions" else: try: from hermes_cli.model_normalize import normalize_model_for_provider fb_model = normalize_model_for_provider(fb_model, fb_provider) except Exception as _norm_err: logger.warning("Could not normalize fallback model %r for provider %r: %s", fb_model, fb_provider, _norm_err) fb_base_url = str(fb_client.base_url) from hermes_cli.providers import is_actual_route if is_actual_route(fb_provider, fb_base_url): fb_api_mode = "chat_completions" elif not fb_api_mode_explicit and fb_api_mode == "chat_completions": fb_api_mode = _fallback_api_mode_resolved(agent, fb_provider, fb_model, fb_base_url) old_model, old_provider, old_base_url = agent.model, agent.provider, agent.base_url # Clear the per-config context_length override so the fallback model's own context # window is resolved instead of the previous model's stale value. # See #22387. agent._config_context_length = None agent.model, agent.provider, agent.requested_provider = fb_model, fb_provider, fb_provider agent.base_url, agent.api_mode = fb_base_url, fb_api_mode # reasoning_content echo opt-in travels with the active provider; restore_primary_runtime reverts it. agent._reasoning_echo_flag = bool(fb.get("reasoning_echo", False)) if hasattr(agent, "_transport_cache"): agent._transport_cache.clear() from agent.turn_recovery import reset_codex_reasoning_replay reset_codex_reasoning_replay(agent) agent._fallback_activated = True _rebind_fallback_credential_pool(agent, fb_provider, fb_model) if fb_provider == "moa": from agent.moa_loop import bind_moa_runtime bind_moa_runtime(agent, fb_model) else: from agent.client_lifecycle import _swap_fallback_clients _swap_fallback_clients(agent, fb_client, fb_provider, fb_model, fb_base_url, fb_api_mode) from agent.agent_runtime_helpers import sync_credential_pool_entry_id sync_credential_pool_entry_id(agent) agent._use_prompt_caching, agent._use_native_cache_layout = agent._anthropic_prompt_cache_policy( provider=fb_provider, base_url=fb_base_url, api_mode=fb_api_mode, model=fb_model) agent._ensure_lmstudio_runtime_loaded() # LM Studio: preload before probing context length _update_fallback_context_compressor(agent) _reresolve_fallback_reasoning_config(agent) _rescope_fallback_extra_body(agent, old_model, old_provider, old_base_url) rewrite_prompt_model_identity(agent, fb_model, fb_provider) notice = ( f"⚠️ Model fallback: {old_model} via {old_provider} unavailable " f"({_fallback_reason_text(reason)}); using {fb_model} via {fb_provider}.") if cooldown_seconds is not None: remaining = max(0, math.ceil(agent._rate_limited_until - time.monotonic())) notice += f" Primary retry eligible in ~{remaining} s; recovery is not guaranteed." _buffer_fallback_notice(agent, notice) # ``_fallback_activated`` is also reused by `/model --once` restoration; separate # provenance so the restore path only emits a recovery notice after a real fallback. agent._provider_fallback_active = True agent._provider_fallback_route = (str(fb_model), str(fb_provider)) _log_fallback_activated(agent, reason, old_model, old_provider, fb_model, fb_provider) from hermes_cli.observability.shared_metrics_events import record_fallback record_fallback(from_provider=old_provider, to_provider=fb_provider, reason=reason) # The stale-call streak measured the OLD provider; carrying it over would # short-circuit the fresh fallback before its first stream attempt. _reset_stale_streak(agent) from agent.native_compaction import resolve_native_compaction_capabilities agent.runtime_capabilities = resolve_native_compaction_capabilities( model=agent.model, base_url=agent.base_url, provider=fb_provider, is_codex_backend=fb_provider == "openai-codex") return True except Exception as e: if fb_provider == "nous": unavailable.add(fb_key) logger.error("Failed to activate fallback %s: %s", fb_model, e) continue # try next in chain # Keys outside the Chat Completions schema that strict gateways (Fireworks-backed OpenCode # Go, Mistral, Moonshot/Kimi) reject with 422. The transport's convert_messages() drops them # in the main loop; the summary path calls chat.completions.create() directly, so mirror it. _SUMMARY_FOREIGN_MESSAGE_KEYS = PERSISTENCE_ONLY_MESSAGE_FIELDS | {"reasoning", "finish_reason", "tool_name", "codex_reasoning_items", "codex_message_items", "platform_message_id"} _EMPTY_SUMMARY_RESPONSE = "I reached the iteration limit and couldn't generate a summary." def _iteration_summary_api_messages(agent, messages: list) -> list: """Wire-ready messages for the summary call, mirroring the main loop's api_messages build (sidecar substitution, tool-call repair, thinking-only drop, underscore-key sweep). ``reasoning_details`` is kept: the anthropic_messages converter rebuilds signed thinking blocks from it, and the chat-completions transport already drops it on the wire for routes that do not replay it (``_chat_summary_attempt`` -> ``_build_api_kwargs``).""" needs_sanitize = agent._should_sanitize_tool_calls() sanitize_model = agent.model if needs_sanitize and agent.provider == "moa": # MoA: agent.model is the virtual preset; use the real aggregator so Gemini keeps thought_signature. agg_slot = getattr(getattr(agent, "client", None), "last_aggregator_slot", None) sanitize_model = (agg_slot or {}).get("model") or sanitize_model api_messages = [] for msg in messages: api_msg = msg.copy() agent._copy_reasoning_content_for_api(msg, api_msg) for key in _SUMMARY_FOREIGN_MESSAGE_KEYS: api_msg.pop(key, None) # Mirror of the transport's role-qualified strip: ``name`` is # schema-foreign on tool results only (strict providers reject with # "contains item with unknown key name"); it stays on user/assistant. if api_msg.get("role") == "tool": api_msg.pop("name", None) # api_content holds the exact bytes the main loop sent; substituting (not popping) # keeps the summary's prefix identical instead of re-prefilling the largest context. # Strict OpenAI-compatible gateways (Fireworks-backed OpenCode Go, Mistral, Moonshot/Kimi) reject # any message key outside the Chat Completions schema. The main loop drops these via # ChatCompletionsTransport.convert_messages(), but the summary path hand-builds messages and calls # chat.completions.create() directly, bypassing the transport — so mirror that sanitization here: # tool_name (SQLite FTS bookkeeping), the codex_* reasoning carriers, timestamp (preserved on # gateway user replay entries for the stale-confirmation expiry check — #47868 rejection class), and # every Hermes-internal underscore-prefixed scaffolding key. substitute_api_content(api_msg) if needs_sanitize: agent._sanitize_tool_calls_for_strict_api(api_msg, model=sanitize_model) api_messages.append(api_msg) effective_system = agent._cached_system_prompt or "" if agent.ephemeral_system_prompt: effective_system = (effective_system + "\n\n" + agent.ephemeral_system_prompt).strip() if effective_system: api_messages = [{"role": "system", "content": effective_system}] + api_messages for idx, pfm in enumerate(agent.prefill_messages or ()): api_messages.insert((1 if effective_system else 0) + idx, pfm.copy()) # Compression/resume can orphan a tool result whose parent tool_call was summarized away. api_messages = agent._sanitize_api_messages(api_messages) # Same send-path vision eviction as the main loop (#89296). from agent.context_compressor import evict_stale_outbound_tool_images evict_stale_outbound_tool_images(api_messages) # Same per-model image strip as turn_api_request.build_api_request: this path builds # api_messages by hand and calls _build_api_kwargs directly, so a model recorded in # agent._image_rejecting_models would otherwise get images here → 4xx → no summary. # Safe on the shallow row copies: the strip rebinds the row's ``content``, never the # nested list shared with history. strip_images_for_rejecting_model(agent, api_messages) # Thinking-only assistant turns 400 on Anthropic-family providers; _thinking_prefill must # survive until here so the drop pass recognizes stubs after reasoning is stripped. api_messages = agent._drop_thinking_only_and_merge_users(api_messages) for api_msg in api_messages: # underscore scaffolding: the transport's sweeper is bypassed here if isinstance(api_msg, dict): for internal_key in [k for k in api_msg if isinstance(k, str) and k.startswith("_")]: del api_msg[internal_key] # Same closing normalization as assemble_api_request so the summary's prefix stays # bit-identical to the main loop's (a diverging early row defeats prefix caching). for api_msg in api_messages: if isinstance(api_msg.get("content"), str): api_msg["content"] = api_msg["content"].strip() from agent.conversation_loop import _canonicalize_api_tool_calls, _clone_message_for_send _canonicalize_api_tool_calls(api_messages) # Third closing pass of the main path: lone surrogates -> U+FFFD (else the SDK's utf-8 # wire encode raises and burns the summary retries). The sanitizer is in-place and these # rows still share nested dicts with history, so clone first like the main path does. api_messages = [_clone_message_for_send(m) for m in api_messages] _sanitize_messages_surrogates(api_messages) return api_messages def _managed_summary_call(agent, api_request_id: str, request, callback, *, retry_count: int): from agent import relay_llm return relay_llm.execute_current( request, callback, name=str(getattr(agent, "provider", "") or "provider"), model_name=str(getattr(agent, "model", "") or ""), metadata={"api_mode": str(getattr(agent, "api_mode", "") or "chat_completions"), "api_request_id": api_request_id, "call_role": "iteration_summary", "retry_count": retry_count}, defer_logical_completion=True, ) def _summary_text(agent, response, **normalize_kwargs) -> str: if is_router_timeout_shim(response): # Router failure in a 200 envelope (#68396): an empty summary takes the retry slot. logger.warning("Iteration summary returned a router timeout shim; retrying") return "" normalized = agent._get_transport().normalize_response(response, **normalize_kwargs) if normalized.tool_calls: # No summary path executes tool calls; log so a tool-only response that falls into the # empty-summary retry is diagnosable. logger.warning("Iteration summary emitted tool calls; discarding them") return (normalized.content or "").strip() def _codex_summary_attempt(agent, api_messages: list, api_request_id: str): def _attempt(retry_count: int) -> str: codex_kwargs = agent._build_api_kwargs(api_messages) # The transport emits these three as one block (transports/codex.py build_kwargs); # strict Responses backends 400 on tool_choice/parallel_tool_calls without tools. codex_kwargs.pop("tools", None) codex_kwargs.pop("tool_choice", None) codex_kwargs.pop("parallel_tool_calls", None) # Route through the same seam as normal Codex turns: a direct _run_codex_stream # bypasses the stale/TTFB watchdogs, interrupt handling and client cleanup, so an # unattended cron summary could wedge forever (#70943). return _summary_text(agent, agent._interruptible_api_call(codex_kwargs)) return _attempt def _anthropic_summary_attempt(agent, api_messages: list, api_request_id: str): def _attempt(retry_count: int) -> str: ant_kw = agent._get_transport().build_kwargs( model=agent.model, messages=api_messages, tools=None, max_tokens=agent.max_tokens, reasoning_config=agent.reasoning_config, is_oauth=agent._is_anthropic_oauth, preserve_dots=agent._anthropic_preserve_dots(), base_url=getattr(agent, "_anthropic_base_url", None)) ant_kw = _merge_nous_portal_messages_extra_body(agent, ant_kw) response = _managed_summary_call( agent, api_request_id, ant_kw, agent._interruptible_api_call, retry_count=retry_count) return _summary_text(agent, response, strip_tool_prefix=agent._is_anthropic_oauth) return _attempt def _chat_summary_attempt(agent, api_messages: list, api_request_id: str): # Same kwargs builder as the main loop so the summary keeps the cached prefix (tools, # prompt_cache_key, xAI alias, Moonshot sanitization). Do not omit tools or force # tool_choice="none" here: SGLang renders the prompt with tools=None in that mode and the KV # prefix diverges. (cache_control breakpoint decoration is not re-applied on this path.) summary_kwargs = agent._build_api_kwargs(api_messages) # The summary now carries ``tools``; on cache-planned routes the main loop scrubbed a deep # copy, so ``agent.tools`` may still hold bytes the provider 400s on. sanitize_outbound_kwargs(agent, summary_kwargs) def _attempt(retry_count: int) -> str: # Use the ordinary request-local lifecycle: a summary can be interrupted # during a long prefill without closing the shared primary client. response = _managed_summary_call( agent, api_request_id, summary_kwargs, agent._interruptible_api_call, retry_count=retry_count) return _summary_text(agent, response) return _attempt _SUMMARY_ATTEMPT_BUILDERS = {"codex_responses": _codex_summary_attempt, "anthropic_messages": _anthropic_summary_attempt} def handle_max_iterations(agent, messages: list, api_call_count: int) -> str: """Request a summary when max iterations are reached. Returns the final response text.""" warning = f"⚠️ Reached maximum iterations ({agent.max_iterations}). Requesting summary..." if getattr(agent, "suppress_status_output", False): # Strict machine-readable mode (-Q, oneshot): keep diagnostics off stdout. quiet_mode is # NOT the gate — the interactive CLI runs quiet_mode=True by default and must see this. # Strict machine-readable mode (hermes chat -Q, oneshot, background review): keep diagnostics out of # stdout so wrappers receive only the final assistant content (#93220 class). logger.warning(warning) else: agent._safe_print(warning, diagnostic=True) summary_api_request_id = f"iteration-summary:{uuid.uuid4()}" summary_call_outcome = "failed" # Shared constant so compaction recognizers can identify this runtime nudge by its stable # content after SessionDB projection strips metadata flags. from agent.context_compressor import MAX_ITERATIONS_SUMMARY_REQUEST nudge = append_message(messages, {"role": "user", "content": MAX_ITERATIONS_SUMMARY_REQUEST}) try: api_messages = _iteration_summary_api_messages(agent, messages) build_attempt = _SUMMARY_ATTEMPT_BUILDERS.get(agent.api_mode, _chat_summary_attempt) attempt = build_attempt(agent, api_messages, summary_api_request_id) # One retry on an empty summary; a summary empty once its block is stripped is NOT retried. final_response = _EMPTY_SUMMARY_RESPONSE for retry_count in (0, 1): text = attempt(retry_count) if not text: continue if "" in text: text = re.sub(r'.*?\s*', '', text, flags=re.DOTALL).strip() if text: summary_call_outcome = "success" append_message(messages, {"role": "assistant", "content": text}) final_response = text break except InterruptedError: # Cancellation is not a summary failure: drop the unanswered nudge and let the # finalizer end the turn as interrupted so the pending message is requeued. summary_call_outcome = "cancelled" if messages and messages[-1] is nudge: messages.pop() raise except Exception as e: logger.warning("Failed to get summary response: %s", e) from agent.turn_failure_copy import site_copy final_response = site_copy("max_iterations_no_summary", limit=agent.max_iterations) finally: from agent import relay_llm relay_llm.complete_logical_call(summary_api_request_id, outcome=summary_call_outcome) return final_response def cleanup_task_resources(agent, task_id: str) -> None: """Per-turn VM + browser cleanup for a task. Skips ``cleanup_vm`` for persistent terminal envs (``_cleanup_inactive_envs`` reaps them after ``terminal.lifetime_seconds``) and ``cleanup_browser`` in headed mode (the inactivity reaper handles idle sessions).""" def _headed() -> bool: try: from tools.browser_tool_cloud import _is_headed_mode return _is_headed_mode() except Exception: return bool(os.environ.get("AGENT_BROWSER_HEADED")) for label, skip, skip_what, cleanup in ( ("VM", is_persistent_env, "cleanup_vm for persistent env", lambda: _ra().cleanup_vm(task_id)), ("browser", lambda _tid: _headed(), "cleanup_browser for headed session", lambda: _ra().cleanup_browser(task_id)), ): try: if skip(task_id): if agent.verbose_logging: logging.debug(f"Skipping per-turn {skip_what} {task_id}; idle reaper will handle it.") else: cleanup() except Exception as e: if agent.verbose_logging: logger.warning("Failed to cleanup %s for task %s: %s", label, task_id, e) def _build_partial_stream_stub(role, full_content, full_reasoning, model_name, usage_obj, *, dropped_tool_names=None, overflow_terminal=False, api_mode=None, clean_eof=False): """Stub for an SSE stream that ended without ``finish_reason`` after delivering content. Tagged ``PARTIAL_STREAM_STUB_ID`` + ``FINISH_REASON_LENGTH`` so the loop enters its continuation/retry path instead of accepting truncated output as a complete turn (#32086). ``overflow_terminal`` (``full_content=None``): the stream died on a context-overflow error. Seeding the recovered text as a continuation stub would grow every later request into the same overflow (#106260); the loop treats the marker as terminal and ends the turn via the recovery contract. ``api_mode="anthropic_messages"`` returns a Messages-shaped stub (``content`` block list + ``stop_reason="max_tokens"``) so AnthropicTransport validates it and the loop continues instead of entering the invalid-response retry ladder (#45908). Empty content keeps one empty text block: validate_response rejects an empty list for ``max_tokens``. ``clean_eof``: the stream ended with no transport exception and no ``finish_reason`` (server/intermediary closed cleanly). Only the two clean-EOF sites in ``_finish_chat_stream`` pass True; the stub built after a real transport exception keeps False so the loop can word the two failure modes differently (#102766). """ if api_mode == "anthropic_messages": return SimpleNamespace( id=PARTIAL_STREAM_STUB_ID, type="message", role=role, model=model_name, content=[SimpleNamespace(type="text", text=full_content or "")], stop_reason="max_tokens", stop_sequence=None, usage=usage_obj, _dropped_tool_names=dropped_tool_names or None, _overflow_terminal=overflow_terminal, _clean_eof=clean_eof, ) return SimpleNamespace( id=PARTIAL_STREAM_STUB_ID, model=model_name, choices=[SimpleNamespace( index=0, message=SimpleNamespace(role=role, content=full_content, tool_calls=None, reasoning_content=full_reasoning), finish_reason=FINISH_REASON_LENGTH, )], usage=usage_obj, _dropped_tool_names=dropped_tool_names or None, _overflow_terminal=overflow_terminal, _clean_eof=clean_eof, ) # SSE error events from proxies (OpenRouter's {"error":{"message":"Network # connection lost."}}) surface as SDK APIError without a status_code (unlike # APIStatusError). They mean the upstream stream died: retry with a fresh # connection like an httpx drop. _SSE_CONN_PHRASES = ("connection lost", "connection reset", "connection closed", "connection terminated", "network error", "network connection", "terminated", "peer closed", "broken pipe", "upstream connect error") def _rejects_stream_options(exc: BaseException) -> bool: """A 400/422 whose body names ``stream_options`` as an unknown/extra field: strict OpenAI-compatible endpoints (Azure AI Foundry MaaS, Pydantic ``extra_forbidden``) reject the usage extension outright (#9705). Distinct from "stream not supported", which flips the whole session to non-streaming.""" if getattr(exc, "status_code", None) not in (400, 422): return False body = f"{getattr(exc, 'body', '') or ''} {exc}".lower() return "stream_options" in body and any( k in body for k in ("extra", "not supported", "unrecognized", "unexpected", "unknown")) def _wait_stream_retry_backoff(agent, delay: float) -> None: """Sleep ``delay`` seconds in 0.1s steps, returning early as soon as the agent is interrupted (so /stop is never held hostage by a backoff; the retry loop's own interrupt check then ends the call).""" deadline = time.monotonic() + max(0.0, delay) while not getattr(agent, "_interrupt_requested", False): remaining = deadline - time.monotonic() if remaining <= 0: return time.sleep(min(0.1, remaining)) def _anthropic_connection_error_types() -> tuple: # An Anthropic error instance implies the SDK is already imported; never import # (or lazy-install) it from inside an error handler. anthropic = sys.modules.get("anthropic") return (anthropic.APIConnectionError,) if anthropic is not None else () def _is_sse_connection_error(exc: BaseException) -> bool: from openai import APIError as _APIError if not isinstance(exc, _APIError) or getattr(exc, "status_code", None): return False err_lower = str(exc).lower() return any(phrase in err_lower for phrase in _SSE_CONN_PHRASES) def _relay_stream_identity(agent, name_default: str) -> dict: """``session_id``/``name``/``model_name`` kwargs for ``relay_llm.stream``.""" return {"session_id": str(getattr(agent, "session_id", "") or ""), "name": str(getattr(agent, "provider", "") or name_default), "model_name": str(getattr(agent, "model", "") or "")} def _relay_stream_metadata(agent, api_mode: str) -> dict: call_role = ("delegated" if getattr(agent, "is_subagent", False) else "fallback" if int(getattr(agent, "_fallback_index", 0) or 0) > 0 else "primary") return {"api_mode": api_mode, "api_request_id": getattr(agent, "_current_api_request_id", None), "call_role": call_role} def _stream_final_text(response) -> str: with contextlib.suppress(Exception): choices = getattr(response, "choices", None) first_choice = choices[0] if isinstance(choices, (list, tuple)) and choices else None content = getattr(getattr(first_choice, "message", None), "content", None) if isinstance(content, str): return content with contextlib.suppress(Exception): content = getattr(response, "content", None) if isinstance(content, str): return content if isinstance(content, list): return "".join(t for t in (getattr(part, "text", None) for part in content) if isinstance(t, str)) return "" def _with_stream_emitters(agent, run): """Bracket ``run()`` with the agent's ``_emit_stream_start`` / ``_emit_stream_end`` hooks when present (end carries the final text on success, the error string on failure) and re-raise.""" start = getattr(agent, "_emit_stream_start", None) if start is not None: start() try: response = run() except Exception as exc: end = getattr(agent, "_emit_stream_end", None) if end is not None: end(final_text="", finished=False, error=str(exc)) raise end = getattr(agent, "_emit_stream_end", None) if end is not None: end(final_text=_stream_final_text(response), finished=True, error=None) return response def _stream_codex_passthrough(agent, api_kwargs: dict, on_first_delta): """Codex streams internally via _run_codex_stream (reached through _interruptible_api_call); park ``on_first_delta`` on the agent so it can pick it up, and bracket the call with the stream start/end emitters.""" agent._codex_on_first_delta = on_first_delta try: return _with_stream_emitters(agent, lambda: agent._interruptible_api_call(api_kwargs)) finally: agent._codex_on_first_delta = None def _finalize_bedrock_relay_events(events): """Relay finalizer for Bedrock: a stream without messageStop has no complete response to record, so return None and let the live consumer raise (#109988).""" from agent.bedrock_adapter import stream_converse_with_callbacks try: return stream_converse_with_callbacks({"stream": list(events)}) except EmptyStreamError: return None class _BedrockStream: """Bedrock Converse streaming: boto3 ``converse_stream()`` on a worker thread with real-time delta callbacks, polled by an interrupt / stale-event watchdog (same UX as the Anthropic and chat_completions streams).""" def __init__(self, agent, api_kwargs: dict, on_first_delta): self.agent = agent self.api_kwargs = api_kwargs self.on_first_delta = on_first_delta self.result = {"response": None, "error": None} self.first_delta_fired = False self.response_started = False # Liveness for the boto3 worker: ``for event in event_stream`` has NO read timeout, # so on_event stamps every event and the poll loop trips a watchdog on a long gap. self.started_at = time.time() self.last_event = self.started_at # Read (not popped): the worker's own pop inside _open_stream must # still resolve the same region. self.region = api_kwargs.get("__bedrock_region__", "us-east-1") # Same patience budget as the OpenAI/Anthropic stale detector. self.stale_timeout = _derive_stream_stale_timeout(agent, api_kwargs) def _model(self) -> str: return self.api_kwargs.get("modelId", "unknown") def _fire_first(self): self.response_started = True if not self.first_delta_fired and self.on_first_delta: self.first_delta_fired = True with contextlib.suppress(Exception): self.on_first_delta() def _after_first(self, fire): """Wrap a delta callback so the first delivered event also fires ``on_first_delta``.""" def _on(value): self._fire_first() fire(value) return _on def _open_stream(self, next_api_kwargs: dict[str, Any]): return _bedrock_converse_call(dict(next_api_kwargs), stream=True, on_stream_denied=self._fall_back_to_converse) def _fall_back_to_converse(self, client, final_kwargs: dict, exc: Exception): # InvokeModel-only IAM policies cannot stream; fall back inside the same Relay # attempt (one lifecycle boundary). from agent.bedrock_adapter import normalize_converse_response self.agent._disable_streaming = True self.agent._safe_print("\n⚠ AWS IAM denied bedrock:InvokeModelWithResponseStream — " "falling back to non-streaming InvokeModel.\n" " Grant that action to restore streaming output.\n", diagnostic=True) logger.info("bedrock: converse_stream denied by IAM (%s) — " "using non-streaming converse() for this session.", type(exc).__name__) return normalize_converse_response(client.converse(**final_kwargs)) def _worker(self): agent = self.agent stream = None try: from agent import relay_llm from agent.bedrock_adapter import stream_converse_with_callbacks intercepted_events = [] writer_token = {"value": None} def _stream_created(_stream: Any) -> None: writer_token["value"] = claim_stream_writer(agent) def _accept_event(_event: Any) -> bool: token = writer_token["value"] return token is None or stream_writer_is_current(agent, token) def _stamp_event() -> None: self.last_event = time.time() try: from agent.plugin_stream_hooks import has_reasoning_stream_observer_hooks plugin_reasoning_observer = has_reasoning_stream_observer_hooks() except Exception: logger.debug("plugin reasoning stream observer check failed", exc_info=True) plugin_reasoning_observer = False stream = relay_llm.stream(dict(self.api_kwargs), self._open_stream, **_relay_stream_identity(agent, "bedrock"), finalizer=lambda: _finalize_bedrock_relay_events(intercepted_events), on_stream_created=_stream_created, on_chunk=intercepted_events.append, chunk_adapter=lambda chunk: chunk, accept_chunk=_accept_event, completed_response_predicate=lambda response: bool(getattr(response, "choices", None)), metadata=_relay_stream_metadata(agent, "custom"), defer_logical_completion=True) wants_reasoning = agent.reasoning_callback or agent.stream_delta_callback or plugin_reasoning_observer try: streamed_response = stream_converse_with_callbacks({"stream": stream}, on_text_delta=self._after_first(agent._fire_stream_delta) if agent._has_stream_consumers() else None, on_tool_start=self._after_first(agent._fire_tool_gen_started), on_reasoning_delta=self._after_first(agent._fire_reasoning_delta) if wants_reasoning else None, on_interrupt_check=lambda: agent._interrupt_requested, on_event=_stamp_event) except EmptyStreamError: # IAM-denied fallback: no stream events, but converse() already completed. if stream.final_response is None: raise streamed_response = None self.result["response"] = stream.final_response or streamed_response except Exception as e: self.result["error"] = e finally: if stream is not None: stream.close() def _raise_if_interrupted(self, message: str, worker=None) -> None: if not self.agent._interrupt_requested: return _record_interrupted_provider_wait( self.agent, time.time() - self.started_at, response_started=self.response_started) if worker is not None: # Let the worker unwind Relay scopes before raising (#81521). _join_worker_for_relay_teardown(worker, label="Bedrock streaming") raise InterruptedError(message) def _on_stale(self, stale_elapsed: float) -> None: """No event past the stale timeout = wedged stream (the worker would block in the event loop forever).""" agent = self.agent logger.warning("Bedrock stream stale for %.0fs (threshold %.0fs) — no events " "received. region=%s model=%s. Aborting call.", stale_elapsed, self.stale_timeout, self.region, self._model()) agent._buffer_diagnostic_status(f"⚠️ No events from Bedrock for {int(stale_elapsed)}s (model: {self._model()}). Aborting...") _bump_stale_streak(agent) # Evict the region's cached client so the NEXT call gets a fresh pool. # This does NOT abort the in-flight botocore EventStream (no external # cancellation exists); the daemon worker keeps reading until its # socket errors, so THIS call ends via the TimeoutError below. try: from agent.bedrock_adapter import invalidate_runtime_client invalidate_runtime_client(self.region) except Exception as _inval_exc: logger.debug("bedrock: stale client eviction failed: %s", _inval_exc) self.last_event = time.time() # Raises RuntimeError past HERMES_STREAM_STALE_GIVEUP; otherwise end # THIS call with a TimeoutError and let the streak carry forward. _check_stale_giveup(agent) self.result["error"] = TimeoutError( f"Bedrock stream produced no events for {int(stale_elapsed)}s (threshold {int(self.stale_timeout)}s) " f"— aborting stalled stream so the retry/fallback path can recover.") def _poll(self): t = threading.Thread(target=_context_thread_target(self._worker), daemon=True) t.start() while t.is_alive(): t.join(timeout=0.3) self._raise_if_interrupted("Agent interrupted during Bedrock API call", worker=t) stale_elapsed = time.time() - self.last_event if stale_elapsed > self.stale_timeout: self._on_stale(stale_elapsed) break # The Bedrock callback returns a PARTIAL response on interrupt without raising # (on_interrupt_check), so the in-loop raise may never fire. Re-check (#59999 area). self._raise_if_interrupted("Agent interrupted during Bedrock API call (post-worker)") if self.result["error"] is not None: raise self.result["error"] # Success clears the cross-turn breaker (#58962). if self.result["response"] is not None: _reset_stale_streak(self.agent) return self.result["response"] def run(self): # Cross-turn stale-stream circuit breaker (#58962), as on the OpenAI/ # Anthropic path. _check_stale_giveup(self.agent) return _with_stream_emitters(self.agent, self._poll) class _ToolCallAccumulator: """Assemble streamed tool-call deltas into complete ``tool_calls`` entries (``acc``: slot index -> entry dict). Ollama-compatible endpoints reuse index 0 for every call in a parallel batch, distinguishing them only by id, so a new id at an already-seen raw index is redirected to a fresh slot.""" def __init__(self): self.acc: dict = {} self._notified: set = set() self._last_id_at_idx: dict = {} # raw_index -> last seen non-empty id self._active_slot_by_idx: dict = {} # raw_index -> current slot in acc # Argument deltas are collected per slot and joined once in ``materialize`` — # ``+=`` per chunk rebuilds the whole string every delta (quadratic on big args). self._argument_parts: dict[int, list[str]] = {} def materialize(self) -> dict: """Join buffered argument deltas into each entry's ``arguments``; idempotent. Returns ``acc``.""" for idx, parts in self._argument_parts.items(): self.acc[idx]["function"]["arguments"] = "".join(parts) return self.acc def feed(self, tc_delta) -> Optional[str]: """Merge one delta; return the tool name the first time it is complete.""" raw_idx = getattr(tc_delta, "index", None) if raw_idx is None: raw_idx = 0 tc_id = getattr(tc_delta, "id", None) delta_id = tc_id or "" if isinstance(tc_id, int): # Poolside sends integer ids tc_id = str(tc_id) self._active_slot_by_idx.setdefault(raw_idx, raw_idx) if delta_id and raw_idx in self._last_id_at_idx and delta_id != self._last_id_at_idx[raw_idx]: self._active_slot_by_idx[raw_idx] = max(self.acc, default=-1) + 1 if delta_id: self._last_id_at_idx[raw_idx] = delta_id idx = self._active_slot_by_idx[raw_idx] entry = self.acc.setdefault( idx, {"id": tc_id or "", "type": "function", "function": {"name": "", "arguments": ""}, "extra_content": None}, ) parts = self._argument_parts.setdefault(idx, []) if tc_id: entry["id"] = tc_id tc_function = getattr(tc_delta, "function", None) if tc_function: if getattr(tc_function, "name", None): # Assignment, not +=: names arrive complete and some providers (MiniMax via # NVIDIA NIM) resend the full name every chunk — += gives "read_fileread_file". entry["function"]["name"] = tc_function.name if getattr(tc_function, "arguments", None): parts.append(tc_function.arguments) extra = getattr(tc_delta, "extra_content", None) if extra is None and hasattr(tc_delta, "model_extra"): extra = (tc_delta.model_extra if isinstance(tc_delta.model_extra, dict) else {}).get("extra_content") if extra is not None: entry["extra_content"] = _dump_if_model(extra) name = entry["function"]["name"] if name and idx not in self._notified: self._notified.add(idx) return name return None class _StreamingCall(StreamingWaitMonitor): """One streaming request on the chat_completions / anthropic_messages wire. State shared between the request worker and the poll-loop monitor (heartbeat, stale kill, interrupt abort) lives on the instance, mutated from both threads.""" def __init__(self, agent, api_kwargs: dict, on_first_delta): self.agent = agent self.api_kwargs = api_kwargs self.on_first_delta = on_first_delta self.worker = None # request thread; None in inline mode self.result = {"response": None, "error": None, "partial_tool_names": []} self.clients = _RequestClientRegistry(agent) # Request-local cancel flag: the worker recognizes its own interrupt # force-close (RemoteProtocolError) and exits instead of retrying (#6600). self._request_cancelled = {"value": False} self.first_delta_fired = {"done": False} self.deltas_were_sent = {"yes": False} # for the partial-delivery fallback self.provider_tool_in_flight = {"yes": False} # Last REAL chunk; the monitor detects SSE-ping-only connections with it. self.last_chunk_time = {"t": time.time()} # Shared by the socket read timeout (``_stream_timeouts``) and the stale # detector (``_resolve_stale_timeout``); None until resolved. self._stream_stale_timeout = None self.stream_attempt_lock = threading.Lock() self.stream_attempt_state = {"current": 0, "cancelled": set(), "discarded_chunks": 0, "discarded_bytes": 0} self._stale_counted_attempts: set[int] = set() # breaker counts each attempt once self.managed_stream_holder = {"stream": None} # Per-attempt: single-writer token, request-local client, raw HTTP response (chat wire). self._writer_token = self._attempt_request_client = self._attempt_stream_response = None # The route ``api_kwargs`` was assembled for; a retry must not replay it on another one. self._request_route = self._live_route() # ── shared small helpers ──────────────────────────────────────────── def _live_route(self) -> tuple: agent = self.agent return tuple(str(getattr(agent, attr, "") or "") for attr in ("model", "provider", "base_url", "api_mode")) def _route_switched_under_request(self) -> bool: """True once ``/model`` (``switch_model``) re-pointed the agent while this request was in flight. The captured payload names the OLD model and is shaped for the OLD provider, but every (re)open builds its client from the LIVE agent, so a retry would send a foreign model slug to the new base_url (404 + a rate-limit hold, #112121). The turn loop rebuilds the request for the current route on its own next attempt, so hand the error back to it. """ live = self._live_route() if live == self._request_route: return False logger.warning( "Stream retry skipped: model/provider switched mid-request (%s via %s -> %s via %s); " "handing back to the turn loop to rebuild the request for the current route.", self._request_route[0], self._request_route[1] or self._request_route[2], live[0], live[1] or live[2], ) return True @staticmethod def _quiet(fn, *args) -> None: """Best-effort callback: never let a display hook break the stream.""" with contextlib.suppress(Exception): fn(*args) def _set_managed_stream(self, stream: Any) -> Any: self.managed_stream_holder["stream"] = stream return stream def _close_managed_stream(self) -> None: close = getattr(self.managed_stream_holder.pop("stream", None), "close", None) if callable(close): try: close() except Exception: logger.debug("Managed provider stream cleanup failed", exc_info=True) def _start_stream_attempt(self) -> int: with self.stream_attempt_lock: self.stream_attempt_state["current"] += 1 attempt_id = int(self.stream_attempt_state["current"]) self.provider_tool_in_flight["yes"] = False # Attempt-local like provider_tool_in_flight: a tool name from a stream that died # before any text must not label a later attempt's partial stub or its retry decision. self.result["partial_tool_names"] = [] return attempt_id def _cancel_current_stream_attempt(self, reason: str) -> None: with self.stream_attempt_lock: current = int(self.stream_attempt_state["current"]) if current: self.stream_attempt_state["cancelled"].add(current) if current: logger.debug("Marked stream attempt %s cancelled: %s", current, reason) def _stream_attempt_is_active(self, stream_attempt_id: int) -> bool: with self.stream_attempt_lock: state = self.stream_attempt_state return stream_attempt_id == int(state["current"]) and stream_attempt_id not in state["cancelled"] def _stream_attempt_was_cancelled(self, stream_attempt_id: int) -> bool: with self.stream_attempt_lock: return stream_attempt_id in self.stream_attempt_state["cancelled"] def _discard_stale_stream_chunk(self, stream_attempt_id: int, chunk) -> None: try: chunk_bytes = len(repr(chunk)) except Exception: chunk_bytes = 0 with self.stream_attempt_lock: state = self.stream_attempt_state state["discarded_chunks"] += 1 state["discarded_bytes"] += chunk_bytes discarded_chunks, discarded_bytes = state["discarded_chunks"], state["discarded_bytes"] first = discarded_chunks == 1 (logger.warning if first else logger.debug)( ("Discarding chunk from superseded stream attempt %s " if first else "Discarded stale stream chunk from attempt %s ") + "(discarded_chunks=%s discarded_bytes=%s)", stream_attempt_id, discarded_chunks, discarded_bytes, ) def _fire_first_delta(self): if not self.first_delta_fired["done"] and self.on_first_delta: self.first_delta_fired["done"] = True self._quiet(self.on_first_delta) def _emit_text(self, text: str) -> None: self._fire_first_delta() self.agent._fire_stream_delta(text) self.deltas_were_sent["yes"] = True def _visible_text_delivered(self) -> bool: """True when visible assistant text actually reached a stream consumer this attempt (``_fire_stream_delta`` records only scrubbed, delivered text; ``deltas_were_sent`` flips on any content delta, including whitespace/think-only ones nobody saw).""" return bool((getattr(self.agent, "_current_streamed_assistant_text", "") or "").strip()) def _emit_reasoning(self, text: str) -> None: self._fire_first_delta() self.agent._fire_reasoning_delta(text) def _emit_tool_started(self, name: str) -> None: self._fire_first_delta() self.agent._fire_tool_gen_started(name) def _route_suppressed_text(self, text: str) -> None: """Tool-call turns suppress content streaming (no chatty preamble), but reasoning tags inside it must still reach the display: route through the delta callback for tag extraction (the CLI drops non-reasoning text once the stream box is closed).""" if self.agent.stream_delta_callback: self._quiet(lambda: (self.agent.stream_delta_callback(text), self.agent._record_streamed_assistant_text(text))) def _new_diag(self) -> dict: diag = self.agent._stream_diag_init() self.clients.diag = diag return diag def _count_chunk(self, diag, chunk) -> None: """Stamp liveness for a real chunk; diagnostics are best-effort.""" self.last_chunk_time["t"] = time.time() self.agent._touch_activity("receiving stream response") with contextlib.suppress(Exception): diag["chunks"] = int(diag.get("chunks", 0)) + 1 if diag.get("first_chunk_at") is None: diag["first_chunk_at"] = self.last_chunk_time["t"] # Delta-length estimate: ~3x cheaper than repr() per chunk. diag["bytes"] = int(diag.get("bytes", 0)) + _estimate_chunk_bytes(chunk) @staticmethod def _mark_finish_seen(diag, finish_reason) -> None: """Record that this attempt saw a terminal finish/stop reason (#102766).""" if finish_reason and isinstance(diag, dict) and not diag.get("finish_reason_seen"): diag["finish_reason_seen"] = True # ── chat_completions wire ─────────────────────────────────────────── def _stream_timeouts(self) -> tuple[float, float, float]: """``(write, read, connect/pool)`` socket timeouts. Per-provider ``request_timeout_seconds`` wins over HERMES_API_TIMEOUT (1800s) and HERMES_STREAM_READ_TIMEOUT (120s); connect/pool cover the handshake, not inference: 30s, or capped at 60s when configured.""" cfg = get_provider_request_timeout(self.agent.provider, self.agent.model) base = cfg if cfg is not None else env_float("HERMES_API_TIMEOUT", 1800.0) if cfg is not None: return base, cfg, min(base, 60.0) read = env_float("HERMES_STREAM_READ_TIMEOUT", 120.0) stale = self._stream_stale_timeout if read == 120.0 and self.agent.base_url and is_local_endpoint(self.agent.base_url): read = base # local providers prefill for minutes logger.debug("Local provider detected (%s) — stream read timeout raised to %.0fs", self.agent.base_url, read) elif read == 120.0 and stale is not None and stale != float("inf") and stale > read: # Reasoning models pause mid-stream for minutes; the stale detector # tolerates that, so the raw read timeout must not fire first. read = stale logger.debug("Cloud reasoning stream — read timeout raised to %.0fs to match stale-stream detector", read) return base, read, 30.0 @staticmethod def _choiceless_chunk(chunk, finish_reason): """Chunk with empty ``choices`` -> ``(usage, finish_reason)``. Raises ProviderStreamError for providers (DeepInfra) that send validation errors as in-stream chunks (choices=None + error_type/error_message), which would otherwise surface as a misleading EmptyStreamError plus retries.""" usage = chunk.usage if hasattr(chunk, "usage") and chunk.usage else None # final usage chunk # Without this check the error is silently dropped and the stream ends empty → EmptyStreamError → # misleading "empty stream" message and pointless retries on the same bad request. (#65631) _err_type = getattr(chunk, "error_type", None) _err_msg = getattr(chunk, "error_message", None) if _err_type or _err_msg: _status = _status_code_from_payload({"code": _err_type, "message": _err_msg}) or _status_code_from_value(_err_type) body = _provider_error_body( {"code": _err_type or "provider_in_stream_error", "message": str(_err_msg or chunk)}, _status) raise ProviderStreamError(status_code=_status, body=body, raw_text=f"{_err_type}: {_err_msg}") # Nous Portal usage frames (choices=[] + lastOne=true, no [DONE]) are a # clean terminal, not a drop; relabelled upstreams send 1 / "true". # See #90848. last_one = getattr(chunk, "lastOne", None) if last_one is None and isinstance(getattr(chunk, "model_extra", None), dict): last_one = chunk.model_extra.get("lastOne") if last_one in (True, 1, "true") and finish_reason is None: finish_reason = "stop" return usage, finish_reason def _open_chat_stream(self, stream_kwargs: dict[str, Any]): # Native Gemini rejects OpenAI's usage-streaming extension; so do strict endpoints that # already 4xx'd on it this session (``_stream_options_unsupported``, see #9705). if not is_native_gemini_base_url(self.agent.base_url) and not getattr(self.agent, "_stream_options_unsupported", False): stream_kwargs["stream_options"] = {"include_usage": True} request_client = self._attempt_request_client = self.clients.set_client( self.agent._create_request_openai_client(reason="chat_completion_stream_request", api_kwargs=stream_kwargs)) self.last_chunk_time["t"] = time.time() self.agent._touch_activity("waiting for provider response (streaming)") # #93650: as above — the streaming path carries the same bulk # messages/tools payload and pays the same client-side walk. stream_kwargs = bypass_chat_sdk_request_transform(stream_kwargs, request_client) return request_client.chat.completions.create(**stream_kwargs) def _chat_stream_created(self, raw_stream: Any) -> None: response = self._attempt_stream_response = getattr(raw_stream, "response", None) self.agent._capture_rate_limits(response) self.agent._capture_credits(response) self.agent._capture_nous_model_switch(response) self.agent._stream_diag_capture_response(self.clients.diag, response) self.agent._check_openrouter_cache_status(response) self._writer_token = claim_stream_writer(self.agent) self._reabort_if_cancelled(response) def _reabort_if_cancelled(self, response: Any) -> None: """Interrupt/stale abort that raced ``create()``: the one-shot pool sweep ran while the connect/TLS window held no socket yet (``tcp_force_closed=0``), so nothing stopped the request once it came up and the serve kept generating into a dropped consumer (#98974). Response headers prove the socket exists now — shut it down (shutdown-only, never a cross-thread close) so the worker unwinds as after a stale kill.""" with self.stream_attempt_lock: current = int(self.stream_attempt_state["current"]) cancelled = self._request_cancelled["value"] or current in self.stream_attempt_state["cancelled"] if not cancelled: return self._shutdown_stale_attempt_socket(response) if self._attempt_request_client is not None: # Kind-aware: the anthropic_messages wire (incl. anthropic-compatible custom endpoints) # runs on a request-local Anthropic client with its own slot sweep. abort = (self.agent._abort_request_anthropic_client if self.agent.api_mode == "anthropic_messages" else self.agent._abort_request_openai_client) abort(self._attempt_request_client, reason="cancelled_attempt_late_connect") def _accept_chat_chunk(self, stream_attempt_id: int, chunk: Any) -> bool: with contextlib.suppress(Exception): choices = getattr(chunk, "choices", None) choice = choices[0] if choices else None delta = getattr(choice, "delta", None) # A stale-attempt fence can win while Relay hands back a tool-call chunk: record # the in-flight tool call (retry policy must not see a partial text response). if getattr(delta, "tool_calls", None): self.provider_tool_in_flight["yes"] = True # Marker-only finish chunk (no writable delta) always passes: the fence only stops # MORE text; fending the completion signal would mislabel a clean end as a drop. if getattr(choice, "finish_reason", None) and not any( getattr(delta, attr, None) for attr in ("content", "tool_calls", "reasoning_content", "reasoning")): return True if not self._stream_attempt_is_active(stream_attempt_id): return False if not self._writer_still_current("Streaming"): return False # Stamp BEFORE Relay processes the chunk so the watchdog can't cancel # a live stream mid-interceptor. self.last_chunk_time["t"] = time.time() return True def _writer_still_current(self, label: str) -> bool: """Single-writer fence: False (with a warning) once a newer stream claimed the writer slot.""" token = self._writer_token if token is None or stream_writer_is_current(self.agent, token): return True logger.warning( "%s attempt superseded by a newer stream; stopping consumption to preserve the " "single-writer invariant (model=%s).", label, self.api_kwargs.get("model", "unknown")) return False def _call_chat_completions(self, stream_attempt_id: int): """Stream a chat completions response.""" import httpx as _httpx base_timeout, read_timeout, conn_cap = self._stream_timeouts() content_parts: list = [] reasoning_parts: list = [] # Live-display accumulator for detail-derived reasoning text: de-gluing must # compare against what the display actually received, not ``reasoning_parts`` # (a provider that mirrors the same text in both fields would otherwise read # as already-glued on the first chunk and get a spurious break). detail_display_parts: list[str] = [] # OpenAI structured refusal (``delta.refusal``): the explanation streams here and # ``delta.content`` stays empty, so an un-accumulated refusal looks like an empty # stream and burns the empty-response retries (the non-streaming fix is #46013). refusal_parts: list[str] = [] reasoning_details: list = [] # OpenRouter replay data (signatures, encrypted blocks) pending_text_parts: list[str] = [] tool_calls = _ToolCallAccumulator() tool_calls_acc = tool_calls.acc finish_reason = model_name = usage_obj = None response_id = upstream_provider = None # the provider's own id / serving upstream, from the chunks role = "assistant" _diag = self._new_diag() self._writer_token = self._attempt_request_client = self._attempt_stream_response = None from agent.chat_completion_helpers_relay import RelayChatAccumulator relay_response = RelayChatAccumulator() # The raw channels, not the callback-delivered text: cron, subagents and callback-less # gateway turns register no stream callback, so nothing else sees a loop mid-stream. content_watch, reasoning_watch = RunawayStreamWatch(), RunawayStreamWatch() # Own watch: the display takes one reasoning representation per chunk, but the plain # field is persisted either way and need not mirror the details. detail_watch = RunawayStreamWatch() runaway = None def _open_stream(next_api_kwargs: dict[str, Any]): timeout = _httpx.Timeout(connect=conn_cap, read=read_timeout, write=base_timeout, pool=conn_cap) return self._open_chat_stream({**next_api_kwargs, "stream": True, "timeout": timeout}) def _flush_pending_stream_text(): pending_parts = list(pending_text_parts) pending_text_parts.clear() for text in pending_parts: (self._route_suppressed_text if tool_calls_acc else self._emit_text)(text) from agent import relay_llm stream = self._set_managed_stream(relay_llm.stream(self.api_kwargs, _open_stream, **_relay_stream_identity(self.agent, "provider"), finalizer=relay_response.finalize, on_stream_created=self._chat_stream_created, on_chunk=relay_response.observe, accept_chunk=lambda chunk: self._accept_chat_chunk(stream_attempt_id, chunk), completed_response_predicate=lambda value: hasattr(value, "choices"), metadata=_relay_stream_metadata(self.agent, "chat_completions"), defer_logical_completion=True)) if self.agent.provider == "moa": # Hermes interrupts the managed stream; Relay alone closes the provider stream. self.clients.set_stream_handle(stream) def _close_half_read_stream(reason: str) -> None: # A half-read SSE response stays checked out of the httpx pool and the finally # would cache the client WITH the leaked connection: close on the owner first. try: stream.close() except Exception: # Still checked out: poison the slot so the finally really closes the pool. if self._attempt_request_client is not None: self.agent._abort_request_openai_client(self._attempt_request_client, reason=reason) for chunk in _iter_provider_stream_chunks(stream, response=lambda: self._attempt_stream_response): self._count_chunk(_diag, chunk) if self.agent._interrupt_requested: _close_half_read_stream("interrupt_stream_close_failed") break if not self._stream_attempt_is_active(stream_attempt_id): self._discard_stale_stream_chunk(stream_attempt_id, chunk) continue if hasattr(chunk, "model") and chunk.model: model_name = chunk.model if response_id is None and isinstance(getattr(chunk, "id", None), str) and chunk.id: response_id = chunk.id if upstream_provider is None and isinstance(getattr(chunk, "provider", None), str) and chunk.provider: upstream_provider = chunk.provider # OpenRouter stamps who served _diag["serving_provider"] = upstream_provider.strip()[:64] # attribute a mid-stream drop (#90216) if not chunk.choices: usage, finish_reason = self._choiceless_chunk(chunk, finish_reason) usage_obj = usage or usage_obj self._mark_finish_seen(_diag, finish_reason) continue choice = chunk.choices[0] delta = choice.delta # Read finish_reason/usage BEFORE any content-shape `continue`: the SSE-echo # guard can swallow a merged finish chunk (vLLM standalone ':' tokens). finish_reason = _normalize_finish_reason(getattr(choice, "finish_reason", None)) or finish_reason self._mark_finish_seen(_diag, finish_reason) if hasattr(chunk, "usage") and chunk.usage: usage_obj = chunk.usage reasoning_text = getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None) # Same ``model_extra`` fallback as the non-streaming path: a reasoning-only stream # whose deltas carry only this field otherwise trips the empty-stream guard (#56516). if reasoning_text is None and isinstance(getattr(delta, "model_extra", None), dict): reasoning_text = delta.model_extra.get("reasoning_content") or delta.model_extra.get("reasoning") if reasoning_text: # Summary-part models omit the separator between markdown blocks; re-insert it. reasoning_text = separate_glued_reasoning_blocks( reasoning_parts[-1] if reasoning_parts else "", reasoning_text) reasoning_parts.append(reasoning_text) # Structured reasoning_details deltas carry the provider's replay data; the # non-streaming path already keeps them, so dropping them here lost # reasoning continuity on nearly every turn. Pydantic parks unknown fields # in ``model_extra``. rd_delta = getattr(delta, "reasoning_details", None) if rd_delta is None and isinstance(getattr(delta, "model_extra", None), dict): rd_delta = delta.model_extra.get("reasoning_details") detail_text_parts = [] for rd in rd_delta if isinstance(rd_delta, (list, tuple)) else (): detail_text_parts.append(streamed_reasoning_detail_text(rd)) append_streamed_reasoning_detail(reasoning_details, rd) # Details may carry the full text while ordinary reasoning is only # a sparse fragment or a mirror. Deliver one representation per # chunk, without rewriting either persisted/replayed field. # Summary-part boundaries need the same repair the plain path applies: # de-glue against the detail display's own accumulator (not # ``reasoning_parts`` — with mirrored fields that would insert a # spurious break on the first chunk), so the live box and the # persisted ``reasoning_content`` agree. detail_text = "".join(detail_text_parts) if detail_text: detail_text = separate_glued_reasoning_blocks( detail_display_parts[-1] if detail_display_parts else "", detail_text) detail_display_parts.append(detail_text) display_reasoning = detail_text or reasoning_text if display_reasoning: self._emit_reasoning(display_reasoning) if reasoning_watch.feed(reasoning_text) or detail_watch.feed(detail_text): runaway = "reasoning" break # Not routed to the live display: the transport promotes a sole-payload # refusal to content + ``content_filter`` and the loop surfaces it terminally. delta_refusal = getattr(delta, "refusal", None) if delta_refusal is None and isinstance(getattr(delta, "model_extra", None), dict): delta_refusal = delta.model_extra.get("refusal") if isinstance(delta_refusal, str) and delta_refusal: refusal_parts.append(delta_refusal) # Text (list-of-blocks deltas flattened once); possible echoed SSE is # buffered until it can be judged. delta_content = flatten_message_text(getattr(delta, "content", None), sep="") if delta_content: content_parts.append(delta_content) if content_watch.feed(delta_content): runaway = "content" break if tool_calls_acc: self._route_suppressed_text(delta_content) elif (pending_text_parts or _provider_stream_text_may_be_sse(delta_content) # A shim cannot follow text already released to the display, so the # whole-content re-join runs only until the first emitted delta. or (not self.deltas_were_sent["yes"] and router_timeout_shim_may_follow("".join(content_parts)))): pending_text_parts.append(delta_content) pending = "".join(pending_text_parts) if not (_provider_stream_text_may_be_sse(pending) or router_timeout_shim_may_follow(pending)): _flush_pending_stream_text() continue else: self._emit_text(delta_content) delta_tool_calls = getattr(delta, "tool_calls", None) if delta_tool_calls: _flush_pending_stream_text() for tc_delta in delta_tool_calls: name = tool_calls.feed(tc_delta) if name is not None: self._emit_tool_started(name) # Lets the stub-builder warn if streaming dies before the args # complete instead of silently discarding the action. self.result["partial_tool_names"].append(name) if runaway: _close_half_read_stream("runaway_stream_close_failed") self._log_runaway_cut(runaway) tool_calls.materialize() self._close_managed_stream() if self._stream_attempt_was_cancelled(stream_attempt_id): raise _httpx.RemoteProtocolError(f"stream attempt {stream_attempt_id} was superseded") if stream.final_response is not None: return self._adopt_final_response(stream.final_response) response = self._finish_chat_stream(stream, role, content_parts, reasoning_parts, tool_calls_acc, "length" if runaway else finish_reason, model_name, usage_obj, flush_pending=_flush_pending_stream_text, response_id=response_id, upstream_provider=upstream_provider, reasoning_details=reasoning_details, refusal_parts=refusal_parts) if runaway: # Cut, not finished: the length path ends the turn on this mark instead of continuing. response._runaway_repetition = True return response def _log_runaway_cut(self, channel: str) -> None: logger.warning( "%sStream cut: the %s turned into a runaway repetition loop (model=%s, provider=%s).", getattr(self.agent, "log_prefix", ""), channel, self.api_kwargs.get("model", "unknown"), getattr(self.agent, "provider", "") or "unknown") def _adopt_final_response(self, final_response): """Adapter returned a completed response for ``stream=True``: switch the session to non-streaming and replay its content as deltas.""" logger.info("Streaming request returned a final response object instead of an iterator; " "switching %s/%s to non-streaming for this session.", self.agent.provider or "unknown", self.agent.model or "unknown") self.agent._disable_streaming = True return self._replay_final_response(final_response) def _replay_final_response(self, final_response): """Replay a completed chat-completions response's reasoning/content as deltas.""" choices = final_response.choices message = getattr(choices[0] if isinstance(choices, (list, tuple)) and choices else None, "message", None) if message is not None: reasoning_text = getattr(message, "reasoning_content", None) or getattr(message, "reasoning", None) if reasoning_text is None and isinstance(getattr(message, "model_extra", None), dict): reasoning_text = message.model_extra.get("reasoning_content") or message.model_extra.get("reasoning") if isinstance(reasoning_text, str) and reasoning_text: self._emit_reasoning(reasoning_text) content = getattr(message, "content", None) if isinstance(content, str) and content: self._fire_first_delta() self.agent._fire_stream_delta(content) # not _emit_text: deltas_were_sent stays False here return final_response @staticmethod def _assemble_tool_calls(tool_calls_acc, finish_reason): """Materialize accumulated tool calls; flag truncated/unrepairable args.""" mock_tool_calls = [] has_truncated_tool_args = False for idx in sorted(tool_calls_acc): tc = tool_calls_acc[idx] arguments = tc["function"]["arguments"] if arguments and arguments.strip(): try: json.loads(arguments) except json.JSONDecodeError: # A dropped stream is never repaired: closing its prefix yields valid # JSON that silently lacks every key and digit not yet streamed. if finish_reason is None: has_truncated_tool_args = True else: # Repair before flagging (GLM via Ollama); "{}" = unrepairable. repaired = _repair_tool_call_arguments(arguments, tc["function"]["name"] or "?") if repaired != "{}": arguments = repaired else: has_truncated_tool_args = True # Parseable JSON does not prove that a dropped stream completed its # action. Treat degenerate argument loops as partial calls too. # A provider-confirmed call may legitimately write repetitive data. if finish_reason is None and is_repetition_dominated(arguments): logger.warning( "Tool call '%s' has repetition-dominated arguments without a " "finish_reason; treating as a dropped tool call.", tc["function"]["name"] or "?") has_truncated_tool_args = True elif finish_reason is None: # Name arrived, zero arg bytes, no finish_reason: unflagged this # becomes a "stop" turn executing "{}" with no retry. has_truncated_tool_args = True mock_tool_calls.append(SimpleNamespace( id=tc["id"], type=tc["type"], extra_content=tc.get("extra_content"), function=SimpleNamespace(name=tc["function"]["name"], arguments=arguments, args_repaired=arguments != tc["function"]["arguments"]))) return mock_tool_calls or None, has_truncated_tool_args def _finish_chat_stream(self, stream, role, content_parts, reasoning_parts, tool_calls_acc, finish_reason, model_name, usage_obj, *, flush_pending, response_id=None, upstream_provider=None, reasoning_details=None, refusal_parts=None): """Assemble the non-streaming-shaped response after the chunk loop. A stream ending with no finish_reason is a drop, not a completion: return a partial-stream stub so the loop fails fast instead of executing empty args or stamping "stop".""" full_content = "".join(content_parts) or None full_reasoning = "".join(reasoning_parts) or None if not full_reasoning and full_content: # Inline-reasoning providers (MiniMax-M3 streams … in content) send no # reasoning delta; fill the structured field from the raw content (#89647). from agent.agent_runtime_helpers import extract_reasoning full_reasoning = extract_reasoning(self.agent, SimpleNamespace(content=full_content)) mock_tool_calls, has_truncated_tool_args = self._assemble_tool_calls(tool_calls_acc, finish_reason) # Zero-chunk guard: nothing usable = upstream error / malformed SSE. if finish_reason is None and not content_parts and not reasoning_parts and not refusal_parts and not tool_calls_acc: raise EmptyStreamError( "Provider returned an empty stream with no finish_reason (possible upstream error or malformed SSE response).") if has_truncated_tool_args and finish_reason is None: # Partial args WITH finish_reason="length" is a real output cap; with NONE the # upstream dropped mid tool-call, and stamping "length" burns 3 useless retries. _dropped_names = [(tool_calls_acc[idx]["function"]["name"] or "?") for idx in sorted(tool_calls_acc)] logger.warning( "Clean EOF, no finish_reason: server ended the stream (no transport exception) while a tool " "call's arguments were still incomplete (tools=%s). The server or a proxy closed the stream " "cleanly; not an output-length truncation.", _dropped_names) return _build_partial_stream_stub( role, full_content, full_reasoning, model_name, usage_obj, dropped_tool_names=_dropped_names or None, clean_eof=True) if finish_reason is None and (content_parts or reasoning_parts) and not tool_calls_acc and usage_obj is None: # Text-only (or reasoning-only) drop: otherwise the partial text is stamped "stop" # and the next step is lost — for reasoning-only, the clean-stop promotion in # finish_text_response would then surface a truncated thought as the answer. # A usage object proves the provider finished (include_usage's final chunk). logger.warning( "Clean EOF, no finish_reason: server ended the stream (no transport exception) after delivering " "text with no tool calls. The server or a proxy closed the stream cleanly.") return _build_partial_stream_stub(role, full_content, full_reasoning, model_name, usage_obj, clean_eof=True) effective_finish_reason = "length" if has_truncated_tool_args else (finish_reason or "stop") provider_stream_error = _provider_stream_error_from_text( full_content or "", effective_finish_reason, response=getattr(stream, "response", None)) if provider_stream_error is not None: raise provider_stream_error message = SimpleNamespace(role=role, content=full_content, tool_calls=mock_tool_calls, reasoning_content=full_reasoning, # ``normalize_response`` reads ``message.refusal`` — same contract as the non-streaming object. refusal="".join(refusal_parts or ()) or None) if reasoning_details: # Only when present: _build_assistant_message's passthrough persists them # for replay, and non-reasoning providers keep the attribute absent. message.reasoning_details = reasoning_details # The provider's id when the chunks carried one (chatcmpl-/gen-...): it is what a provider needs to # look a request up. Fabricated only when the stream never sent one. response = SimpleNamespace(id=response_id or ("stream-" + str(uuid.uuid4())), model=model_name, usage=usage_obj, provider=upstream_provider, choices=[SimpleNamespace(index=0, message=message, finish_reason=effective_finish_reason)]) # A held router timeout shim (#68396) is rejected by validate_response and retried; # releasing its text here would show the provider failure as assistant output. if not is_router_timeout_shim(response): flush_pending() return response # ── anthropic_messages wire ───────────────────────────────────────── @staticmethod def _check_anthropic_message(message, *, tool_drop: bool = True): """Raise EmptyStreamError for a message the stream never completed: no content and no stop_reason (eventless -> retry), or with ``tool_drop`` a ``tool_use`` block and no stop_reason — the SSE closed mid tool call and its input is a partial snapshot (usually ``{}``), so raising blocks the empty-args execution (bounded retry, or stub/continuation after text).""" content = getattr(message, "content", None) if not content and getattr(message, "stop_reason", None) is None: raise EmptyStreamError( "Provider returned an empty stream with no stop_reason (possible upstream error or malformed event stream).") if tool_drop and getattr(message, "stop_reason", None) is None and any( getattr(block, "type", None) == "tool_use" for block in content or []): raise EmptyStreamError( "Stream ended with no stop_reason while a tool_use block was still incomplete; " "treating as a mid-tool-call stream drop (#80498).") return message def _call_anthropic(self, request_client): """Stream an Anthropic Messages API response; fires delta callbacks but returns the native Message from get_final_message(). Runs on the per-request ``request_client`` so the watchdog can abort this socket without closing the shared client mid-flight.""" has_tool_use = False # No message_stop -> EmptyStreamError; saw_stream_event only picks the message. saw_stream_event = False saw_message_stop = False self.last_chunk_time["t"] = time.time() _diag = self._new_diag() self._writer_token = self._attempt_stream_response = None self._attempt_request_client = request_client _stream_context = {"manager": None, "stream": None} base_final_message = None text_watch, thinking_watch = RunawayStreamWatch(), RunawayStreamWatch() runaway = None from agent import relay_llm from agent.anthropic_adapter import normalize_stream_usage, sanitize_anthropic_kwargs accumulator = relay_llm.AnthropicStreamAccumulator() def _open_anthropic_stream(next_api_kwargs: dict[str, Any]): final_kwargs = dict(next_api_kwargs) sanitize_anthropic_kwargs(final_kwargs, log_prefix=getattr(self.agent, "log_prefix", "")) manager = request_client.messages.stream(**final_kwargs) _stream_context["manager"] = manager return normalize_stream_usage(manager.__enter__()) def _anthropic_stream_created(raw_stream: Any) -> None: _stream_context["stream"] = raw_stream # Same wiring as the chat_completions wire: MessageStream exposes the httpx response, # so the interrupt/stale abort can shut down THIS attempt's socket (#98974). response = self._attempt_stream_response = getattr(raw_stream, "response", None) # Snapshot response diagnostics now so they survive a stream dying before the first event. self._quiet(lambda: self.agent._stream_diag_capture_response(_diag, response)) self._writer_token = claim_stream_writer(self.agent) self._reabort_if_cancelled(response) stream = self._set_managed_stream(relay_llm.stream(self.api_kwargs, _open_anthropic_stream, **_relay_stream_identity(self.agent, "anthropic"), finalizer=accumulator.finalize, on_stream_created=_anthropic_stream_created, on_chunk=accumulator.observe, accept_chunk=lambda _event: self._writer_still_current("Anthropic streaming"), metadata=_relay_stream_metadata(self.agent, "anthropic_messages"), defer_logical_completion=True)) try: for event in stream: saw_stream_event = True self._count_chunk(_diag, event) if self.agent._interrupt_requested: break event_type = getattr(event, "type", None) if event_type == "message_stop": saw_message_stop = True elif event_type == "message_delta": self._mark_finish_seen(_diag, getattr(getattr(event, "delta", None), "stop_reason", None)) elif event_type == "content_block_start": block = getattr(event, "content_block", None) if block and getattr(block, "type", None) == "tool_use": has_tool_use = True if getattr(block, "name", None): self._emit_tool_started(block.name) # Same as the chat_completions wire: a stream that dies inside the # tool args is retried (no tool has run yet) instead of stubbed. self.result["partial_tool_names"].append(block.name) elif event_type == "content_block_delta": delta = getattr(event, "delta", None) delta_type = getattr(delta, "type", None) if delta else None if delta_type == "text_delta": text = getattr(delta, "text", "") if text and not has_tool_use: self._emit_text(text) if text_watch.feed(text): runaway = "content" break elif delta_type == "thinking_delta" and getattr(delta, "thinking", ""): self._emit_reasoning(delta.thinking) if thinking_watch.feed(delta.thinking): runaway = "reasoning" break raw_stream = _stream_context["stream"] if not self.agent._interrupt_requested and raw_stream is not None and not runaway: if not saw_message_stop: raise EmptyStreamError( "Anthropic Messages stream ended before message_stop (possible upstream stream drop)." if saw_stream_event else "Provider returned an empty stream with no events (possible upstream error or malformed event stream).") base_final_message = raw_stream.get_final_message() # The SDK snapshot keeps only stop_reason/stop_sequence from message_delta; the # refusal's stop_details (category/explanation) survives only in our accumulator. _stop_details = accumulator.finalize().get("stop_details") if _stop_details is not None and getattr(base_final_message, "stop_details", None) is None: base_final_message.stop_details = _stop_details finally: try: self._close_managed_stream() finally: manager = _stream_context["manager"] if manager is not None: manager.__exit__(None, None, None) if self.agent._interrupt_requested: return None if runaway: self._log_runaway_cut(runaway) # No message_stop, so no final message: the SDK's snapshot is what streamed so far. message = accumulator.response(_stream_context["stream"].current_message_snapshot) message.stop_reason = "max_tokens" message._runaway_repetition = True # same hand-off as the chat_completions wire return message if base_final_message is not None: self._check_anthropic_message(base_final_message, tool_drop=False) if not stream.output_modified: return self._check_anthropic_message(base_final_message) return self._check_anthropic_message(accumulator.response(base_final_message)) # ── retry loop ────────────────────────────────────────────────────── def _retry_after_drop(self, e, attempt: int, max_retries: int, *, mid_tool_call: bool, reason: str) -> None: """Warn about the drop and tear down the request-local client. Shared clients are never closed from inside a request (FD-recycle hazard); the OpenAI primary is replaced lazily.""" self.agent._emit_stream_drop( error=e, attempt=attempt + 1, max_attempts=max_retries + 1, mid_tool_call=mid_tool_call, diag=self.clients.diag) if self.agent._is_provider_stream_parse_error(e): from agent.anthropic_adapter import buffer_anthropic_tool_input buffer_anthropic_tool_input(self.api_kwargs, getattr(self.agent, "_anthropic_base_url", None)) self._cancel_current_stream_attempt(reason) self.clients.close_once(reason) # Exponential backoff between stream-level reconnects (back-to-back retries # hammer a provider that just dropped us). Interruptible: /stop exits at once. from agent.retry_utils import jittered_backoff _wait_stream_retry_backoff( self.agent, jittered_backoff(attempt + 1, base_delay=1.0, max_delay=4.0, jitter_ratio=0.0)) # The backoff is not the dead attempt's silence: restart the stale clock so the # stale monitor cannot kill (and strike) a stream that has not reopened yet. self.last_chunk_time["t"] = time.time() def _maybe_disable_streaming(self, e) -> None: """Flip to non-streaming for failures streaming itself cannot survive, or that re-streaming can only repeat: the provider rejecting streams outright, AnthropicBedrock IAM lacking InvokeModelWithResponseStream, a custom anthropic_messages provider emitting SSE events out of order (#72833), or a gateway answering with contentless SSE keepalive frames (a degraded gateway answers every streaming request that way, so the retry must change channel to make progress).""" if _is_provider_stream_empty_frame_error(e): self.agent._disable_streaming = True logger.warning( "Provider stream returned an empty SSE frame (keepalive, no payload) before any " "delta — switching %s/%s to non-streaming for this session.", self.agent.provider or "unknown", self.agent.model or "unknown") # Durable channel, not _buffer_status: this recovery is expected to SUCCEED, and # buffered retry chatter is dropped on successful recovery. Fires at most once per # session (streaming is off from here on). self.agent._emit_warning( "⚠️ Provider stream returned an empty keepalive frame — retrying this turn " "without streaming (streaming stays off for this session).") return from agent.anthropic_adapter import _is_stream_unavailable_error if not _is_stream_unavailable_error(e): return _err_lower = str(e).lower() _is_stream_unsupported = "stream" in _err_lower and "not supported" in _err_lower if "unexpected event order" in _err_lower and not _is_stream_unsupported: # Custom anthropic_messages SSE out of order (#72833): re-streaming repeats it. # Bedrock keeps turn_recovery's sticky Converse switch instead. if self.agent.api_mode != "anthropic_messages" or self.agent.provider == "bedrock": return self.agent._disable_streaming = True self.agent._safe_print( "\n⚠ Provider sent Anthropic stream events out of order. Switching to non-streaming.\n", diagnostic=True, ) return # Remaining matches: stream rejected outright, or Bedrock IAM stream denial. self.agent._disable_streaming = True self.agent._safe_print( "\n⚠ Streaming is not supported for this model/provider. Switching to non-streaming.\n" " To avoid this delay, set display.streaming: false in config.yaml\n" if _is_stream_unsupported else "\n⚠ AWS IAM denied bedrock:InvokeModelWithResponseStream. Switching to non-streaming.\n" " Grant that action to restore streaming output.\n", diagnostic=True, ) def _handle_stream_error(self, e: Exception, attempt: int, max_retries: int) -> bool: """Classify a failed attempt: True = retry; False = stop with ``result["error"]`` set (unless our own interrupt force-closed the socket). Runs inside the ``except`` so ``logger.exception`` works.""" import httpx as _httpx # Our own interrupt force-close: no retry/fallback/"reconnecting" (the # poll loop raises InterruptedError). if self._request_cancelled["value"]: logger.debug("Streaming worker caught %s after request cancellation — exiting without retry.", type(e).__name__) return False _is_timeout = isinstance(e, (_httpx.ReadTimeout, _httpx.ConnectTimeout, _httpx.PoolTimeout)) # ReadError: abort/reset mid-body (stale-kill shutdown under a parked reader, # ECONNRESET) — the retry loop owns recovery. # anthropic.APIConnectionError: the Anthropic SDK wraps connect/read drops # (incl. stale-kill aborts) in its own type, not httpx's. _is_conn_err = isinstance(e, (_httpx.ConnectError, _httpx.ReadError, _httpx.RemoteProtocolError, ConnectionError, *_anthropic_connection_error_types())) _is_stream_parse_err = self.agent._is_provider_stream_parse_error(e) _is_empty_stream = isinstance(e, EmptyStreamError) _is_sse_conn_err = not _is_timeout and not _is_conn_err and _is_sse_connection_error(e) _is_transient = _is_timeout or _is_conn_err or _is_sse_conn_err or _is_stream_parse_err if not self.deltas_were_sent["yes"] and not getattr(self.agent, "_stream_options_unsupported", False) and _rejects_stream_options(e): # Nothing streamed yet: drop the usage extension for this session and re-open. self.agent._stream_options_unsupported = True self._compat_retries = 1 logger.info("Endpoint rejected stream_options (HTTP %s); retrying without it for this session.", getattr(e, "status_code", None)) self._cancel_current_stream_attempt("stream_options_rejected_retry") self.clients.close_once("stream_options_rejected_retry") return True if self.deltas_were_sent["yes"]: _partial_tool_in_flight = bool(self.result.get("partial_tool_names")) or self.provider_tool_in_flight["yes"] if not _partial_tool_in_flight and not self._visible_text_delivered(): # Deltas fired but nothing visible reached a consumer (whitespace/think-only # deltas, or no display consumer at all) and no tool call is in flight: from # the user's and the model's point of view NOTHING was delivered. The # "partial delivery" stub would be EMPTY and the loop would ask the model to # continue from nowhere, so it repeats the lost step (#112419). Classify as an # undelivered failure instead: same-prefix retry, then the main loop's # fallback/backoff — there is no text to duplicate. logger.warning( "Stream died after deltas but before any visible text was delivered (0 chars, " "no tool call in flight); treating as an undelivered stream failure: %s", e) self._quiet(self.agent._reset_stream_delivery_tracking) self.deltas_were_sent["yes"] = False self.first_delta_fired["done"] = False if self.deltas_were_sent["yes"]: # Died AFTER tokens were delivered: normally no retry (would duplicate # text). Exception: a tool call in flight — aborting discards it, so # retry TRANSIENT errors (a "reconnecting" marker + duplicated # preamble beats a failed action; no tool has executed yet). if not (_partial_tool_in_flight and _is_transient and attempt < max_retries): logger.warning("Streaming failed after partial delivery, not retrying: %s", e) self.result["error"] = e return False # Marker explains the re-streamed preamble (``_emit_stream_drop`` logs the WARNING); # reset the streamed-text buffer so it isn't double-recorded; fresh accumulators. if self.agent._warning_presentation_enabled(): self._quiet(self.agent._fire_stream_delta, "\n\n⚠ Connection dropped mid tool-call; reconnecting…\n\n") self._quiet(self.agent._reset_stream_delivery_tracking) self.deltas_were_sent["yes"] = False self.first_delta_fired["done"] = False self._retry_after_drop(e, attempt, max_retries, mid_tool_call=True, reason="stream_mid_tool_retry_cleanup") return True if _is_transient or _is_empty_stream: # Transient network / timeout error: retry with a fresh connection first. if attempt < max_retries: self._retry_after_drop(e, attempt, max_retries, mid_tool_call=False, reason="stream_retry_cleanup") return True # Exhausted: log full diagnostics (chain, headers, bytes/elapsed). self.agent._log_stream_retry(kind="exhausted", error=e, attempt=max_retries + 1, max_attempts=max_retries + 1, mid_tool_call=False, diag=self.clients.diag) # Empty stream: "connection failed" would send users chasing network issues. if _is_stream_parse_err or _is_empty_stream: _what = ("Provider returned malformed streaming data after" if _is_stream_parse_err else "Provider returned an empty response stream after") self.agent._buffer_diagnostic_status( f"❌ {_what} {max_retries + 1} attempts. The provider may be experiencing issues — try again in a moment.") else: from agent.stream_diag import buffer_connect_exhausted_notice buffer_connect_exhausted_notice(self.agent, e, attempts=max_retries + 1, base_url=self.agent.base_url) else: self._maybe_disable_streaming(e) logger.exception("Streaming failed before delivery: %s", e) if self._unmask_server_error_with_nonstreaming(e): return False # Propagate to the main retry loop (credential rotation, fallback, backoff). self.result["error"] = e return False def _unmask_server_error_with_nonstreaming(self, e: Exception) -> bool: """One non-streaming re-issue when a 5xx killed the stream before any delta. Some gateways validate the request only on their non-streaming path and crash opaquely ("500 something went wrong") when streaming — the real 4xx, with its actionable message, never reaches the user through stream retries. One non-streaming probe per 60s window (timestamped on the agent: each outer retry builds a fresh _StreamingCall, so an instance flag would re-probe every attempt) surfaces it: on success the response is delivered for this turn WITHOUT latching non-streaming (a transient gateway 500 must not permanently disable streaming); on a probe 4xx that error REPLACES the opaque 5xx; any other probe failure keeps the original error. The successful delivery is bracketed by its own stream start/end pair (the failed attempt already emitted a terminal end), and a response that cannot be replayed propagates ``e``. The probe runs on this worker thread while the stream monitor still polls, so its stale check is suspended for the probe's duration (the probe has its own non-streaming watchdog). Interrupts re-raise (the outer handler routes them), and a /stop that arrived before this point suppresses the probe entirely — the loop's pre-retry interrupt check owns that decision, so a pending stop must not buy one more request. True = handled (caller must not overwrite result); False = propagate ``e``. """ if getattr(self.agent, "_interrupt_requested", False): return False status = _extract_status_code(e) if status is None or status < 500 or self.deltas_were_sent["yes"]: return False if getattr(self.agent, "api_mode", "") not in ("", "chat_completions"): return False # replay handles chat-completions shapes only now = time.monotonic() last_probe = self.agent._stream_5xx_probe_ts if last_probe is not None and now - last_probe < _STREAM_5XX_PROBE_WINDOW_S: return False # one probe per 60s window self.agent._stream_5xx_probe_ts = now probe_kwargs = {k: v for k, v in self.api_kwargs.items() if k not in ("stream", "stream_options")} stale_timeout = self._stream_stale_timeout self._stream_stale_timeout = float("inf") # no chunks arrive during the probe try: probe = interruptible_api_call(self.agent, probe_kwargs) except (KeyboardInterrupt, InterruptedError): raise # the outer handler routes user interrupts; never swallow them except Exception as probe_err: probe_status = _extract_status_code(probe_err) if probe_status is not None and probe_status < 500: # The provider's REAL validation error beats the opaque 5xx. logger.info("Non-streaming unmask probe surfaced the underlying error: %s", probe_err) self.result["error"] = probe_err return True logger.info("Non-streaming unmask probe failed: %s", probe_err) return False finally: self._stream_stale_timeout = stale_timeout logger.info("Streaming 5xx re-issued non-streaming successfully for %s/%s " "(not latched: the 5xx may be transient).", self.agent.provider or "unknown", self.agent.model or "unknown") self._quiet(self.agent._buffer_status, "⚠ Streaming failed with a provider server error; the non-streaming retry succeeded.") try: # The failed attempt already emitted its terminal on_stream_end(finished=False), # so the recovered delivery opens and closes its OWN stream pair — consumers must # never see deltas after that error event. replayed = _with_stream_emitters(self.agent, lambda: self._replay_final_response(probe)) except Exception as replay_err: # A response we cannot replay must not escape into _call()'s except block. logger.exception("Non-streaming unmask probe response could not be replayed: %s", replay_err) return False self.result["response"] = replayed return True def _call_wire(self, stream_attempt_id: int): if self.agent.api_mode != "anthropic_messages": return self._call_chat_completions(stream_attempt_id) # Per-request client so the watchdog aborts its socket, not the shared one. request_client = self.clients.set_client( self.agent._create_request_anthropic_client(reason="anthropic_stream_request"), kind="anthropic_messages") return self._call_anthropic(request_client) def _call(self): _max_stream_retries = env_int("HERMES_STREAM_RETRIES", 2) # The one stream_options compatibility retry (#9705) is not a network retry and must not # consume the transient budget: on the last attempt (or HERMES_STREAM_RETRIES=0) the # handler returned True and the loop ended with neither a response nor an error set. self._compat_retries = 0 _stream_attempt = -1 try: while _stream_attempt < _max_stream_retries + self._compat_retries: _stream_attempt += 1 stream_attempt_id = self._start_stream_attempt() # Otherwise /stop closes the connection and the retry opens a # FRESH one, blocking up to a full read timeout per attempt. if self.agent._interrupt_requested: self._cancel_current_stream_attempt("interrupt_before_stream_retry") raise InterruptedError("Agent interrupted before stream retry") try: self.result["response"] = _with_stream_emitters( self.agent, lambda: self._call_wire(stream_attempt_id)) return # success except Exception as e: self._close_managed_stream() if not self._handle_stream_error(e, _stream_attempt, _max_stream_retries): return if self._route_switched_under_request(): self.result["error"] = e return except InterruptedError as e: # Fast pre-retry interrupt surfaces through the normal result channel. self.result["error"] = e return finally: self._close_managed_stream() # Reuse only after a clean stream; otherwise really close (fresh pool next). self.clients.close_once( "stream_request_complete" if self.result["response"] is not None else "stream_error_cleanup") # ── poll-loop monitor (heartbeat / stale kill / interrupt) ────────── def _run_call(self): try: self._call() finally: self._call_done.set() def _shutdown_stale_attempt_socket(self, response: Any) -> None: """Best-effort ``shutdown()`` on the killed attempt's socket (monitor thread). The pool sweep in ``close_once`` can miss a connection that is checked out for the in-flight body read. ``shutdown(SHUT_RDWR)`` is FD-safe from any thread — it wakes the owner's ``recv`` without releasing the descriptor — so the worker unwinds and releases its own response on the owner thread (``_call``'s ``except``/``finally``). Never ``close()`` here: releasing a live TLS descriptor from a stranger thread lets the kernel recycle it under the owner's SSL BIO, which is exactly what the shutdown-only rule in ``_abort_request_slot_client`` forbids (it covers request-local clients too, #30858). """ if response is None or response is not self._attempt_stream_response: return try: from agent.agent_runtime_helpers import ( _connection_candidates, _shutdown_socket, _socket_from_candidate, ) exts = getattr(response, "extensions", None) or {} direct = exts.get("network_stream") if isinstance(exts, dict) else None for start in (direct, getattr(response, "stream", None)): if start is None: continue for candidate in _connection_candidates(start): sock = _socket_from_candidate(candidate) if sock is None: continue _shutdown_socket(sock) logger.info("Shut down the stale stream's socket to unblock the reader " "(attempt superseded; model=%s).", self.api_kwargs.get("model", "unknown")) return logger.debug("Stale stream socket shutdown found no socket; pool sweep is the only abort") except Exception: logger.debug("Stale stream socket shutdown failed", exc_info=True) def _uncounted_stale_attempt(self) -> int: """The started attempt the circuit breaker (see ``_stale_streak()``) has not counted yet, else 0. Like the non-streaming and inline watchdogs, each attempt counts once: the stale timer re-fires every window while the worker has not dispatched yet or is still unwinding a kill, and none of those re-kills is another unresponsive attempt.""" with self.stream_attempt_lock: attempt = int(self.stream_attempt_state["current"]) return 0 if attempt in self._stale_counted_attempts else attempt def _count_stale_attempt(self) -> None: attempt = self._uncounted_stale_attempt() if attempt: self._stale_counted_attempts.add(attempt) _bump_stale_streak(self.agent) def _kill_stale_stream(self, elapsed: float) -> None: """SSE pings but no chunks: cancel the attempt and abort the request-local client so the retry loop opens a fresh one. The shared client is never closed from this (stranger) thread — earlier stale-killed workers may still be unwinding SSL BIOs (FD-recycle corruption); the OpenAI primary is replaced lazily.""" _est_ctx = estimate_request_context_tokens(self.api_kwargs) logger.warning( "Stream stale for %.0fs (threshold %.0fs) — no chunks received. model=%s context=~%s tokens. Killing connection.", elapsed, self._stream_stale_timeout, self.api_kwargs.get("model", "unknown"), f"{_est_ctx:,}", ) self.agent._buffer_diagnostic_status( f"⚠️ No response from provider for {int(elapsed)}s (model: {self.api_kwargs.get('model', 'unknown')}, " f"context: ~{_est_ctx:,} tokens). Reconnecting...") # Captured BEFORE the cancel/abort: the pool sweep can miss a checked-out # connection, so shut down the killed attempt's own socket too — still # shutdown-only, never close (see the helper). _killed_response = self._attempt_stream_response with contextlib.suppress(Exception): self._cancel_current_stream_attempt("stale_stream_kill") self.clients.close_once("stale_stream_kill") self._shutdown_stale_attempt_socket(_killed_response) self._count_stale_attempt() # Reset the timer so we don't kill repeatedly while the worker unwinds. self.last_chunk_time["t"] = time.time() self.agent._emit_diagnostic_wait(f"⚠ no output from provider for {int(elapsed)}s — reconnecting...") self.agent._touch_activity(f"stale stream detected after {int(elapsed)}s, reconnecting") def _abort_for_interrupt(self, stale_elapsed: float) -> None: """/stop seen by the monitor: mark cancelled, abort the request-local socket, wait for the worker, flag the interrupt.""" # Once per attempt: a stale kill that already counted this attempt wins. attempt = self._uncounted_stale_attempt() if attempt and stale_elapsed <= self._stream_stale_timeout and _record_interrupted_provider_wait( self.agent, stale_elapsed, response_started=self.deltas_were_sent["yes"]): self._stale_counted_attempts.add(attempt) # Mark cancelled BEFORE force-closing so the worker treats the forced # transport error as a cancel, not a network error (#6600). self._request_cancelled["value"] = True logger.debug("Force-closing streaming httpx client due to interrupt (not a network error).") # Same as the stale kill: the pool sweep can miss the connection checked out for the # in-flight body read, so shut down the attempt's own socket too (#98974). _killed_response = self._attempt_stream_response with contextlib.suppress(Exception): self._cancel_current_stream_attempt("stream_interrupt_abort") # Kind-aware: only the request-local socket; the shared _anthropic_client is never closed here. self.clients.close_once("stream_interrupt_abort") self._shutdown_stale_attempt_socket(_killed_response) # Let the worker unwind Relay-managed scopes first; raising first lets # turn teardown race a still-open scope and corrupt the LIFO stack. if self.worker is not None: _join_worker_for_relay_teardown(self.worker, label="Streaming") self._monitor_interrupted["yes"] = True # ── orchestration ─────────────────────────────────────────────────── def _resolve_stale_timeout(self) -> None: """Set ``_stream_stale_timeout``. Local endpoints (unless the env is set) get long but FINITE patience — 900s / ``agent.local_stream_stale_timeout`` / HERMES_LOCAL_STREAM_STALE_TIMEOUT — an infinite one stalled sessions on a crashed endpoint forever. Cloud values scale with context size and are floored for known reasoning models (else BrokenPipeError from the gateway).""" base = _configured_stale_base(self.agent) if base == 180.0 and self.agent.base_url and is_local_endpoint(self.agent.base_url): self._stream_stale_timeout = _local_stream_stale_timeout_default() logger.debug("Local provider detected (%s) — stale stream timeout set to %.0fs", self.agent.base_url, self._stream_stale_timeout) return self._stream_stale_timeout = _cloud_stale_timeout_for(self.agent, self.api_kwargs) def _partial_stream_stub(self): """Tokens already reached the platform: a finish_reason="length" stub fires the continuation machinery; tool_calls=None blocks executing incomplete calls. Content may be EMPTY (dropped tool call, overflow) — the loop skips appending an empty stub and only sends the nudge (placeholder text leaked into the stitched response). A text-only death with 0 visible chars never gets here: the error handler reclassifies it as undelivered (#112419).""" error = self.result["error"] _partial_text = (getattr(self.agent, "_current_streamed_assistant_text", "") or "").strip() or None _partial_names = list(self.result.get("partial_tool_names") or []) if _partial_names: # User-visible warning so the user and model both know what was attempted. _name_str = ", ".join(_partial_names[:3]) if len(_partial_names) > 3: _name_str += f", +{len(_partial_names) - 3} more" _warn = (f"\n\n⚠ Stream stalled mid tool-call ({_name_str}); the action was not executed. " f"Ask me to retry if you want to continue.") _partial_text = (_partial_text or "") + _warn # model/result bookkeeping, never gated if self.agent._warning_presentation_enabled(): self._quiet(self.agent._fire_stream_delta, _warn) # visible immediately logger.warning( "Partial stream dropped tool call(s) %s after %s chars of text; surfaced warning to user: %s", _partial_names, len(_partial_text or ""), error) # Classify the error before it is swallowed into the stub: the loop reads the # content-filter tag and falls back; a context overflow must not be continued at all. _cls = None with contextlib.suppress(Exception): from agent.error_classifier import classify_api_error _cls = classify_api_error( error, provider=str(getattr(self.agent, "provider", "") or ""), model=str(getattr(self.agent, "model", "") or "")) _reset_stale_streak(self.agent) # deltas fired => provider responsive: clear the breaker # #106260: continuing after a context-overflow error re-sends a larger request into the # same overflow. Return an EMPTY stub marked terminal so the loop ends the turn instead. # Scope is context_overflow ONLY: payload_too_large (413) has its own byte-scored recovery # owner (turn_overflow._recover_payload_too_large, #88960/#47339) that must not be bypassed. if _cls is not None and _cls.reason == FailoverReason.context_overflow: logger.warning( "Partial stream ended on a context-overflow error after %s chars; " "NOT seeding a continuation stub (transcript is already over budget): %s", len(_partial_text or ""), error, ) return _build_partial_stream_stub( "assistant", None, None, getattr(self.agent, "model", "unknown"), None, dropped_tool_names=_partial_names, overflow_terminal=True, api_mode=getattr(self.agent, "api_mode", None), ) if not _partial_names: logger.warning( "Partial stream delivered before error; returning length-truncated stub with %s chars of " "recovered content so the loop can continue from where the stream died: %s", len(_partial_text or ""), error) _stub = _build_partial_stream_stub("assistant", _partial_text, None, getattr(self.agent, "model", "unknown"), None, dropped_tool_names=_partial_names, api_mode=getattr(self.agent, "api_mode", None)) if _cls is not None and _cls.reason == FailoverReason.content_policy_blocked: _stub._content_filter_terminated = True return _stub def run(self): """Resolve the stale timeout, run the request (worker thread or inline), drive the heartbeat/stale/interrupt monitor, then translate the outcome.""" self._resolve_stale_timeout() # Delegated children and cron turns run the request INLINE (a worker inside # their nested pools wedges before the socket opens) but must still STREAM # (edge proxies kill silent POSTs). Only the poll loop moves to a monitor # thread, which never issues a request, so the no-worker deadlock fix holds. self._call_done = threading.Event() self._monitor_interrupted = {"yes": False} if should_use_direct_api_call(self.agent): self.worker = None monitor = threading.Thread( target=_context_thread_target(self._monitor_loop), name="stream-inline-monitor", daemon=True) monitor.start() try: self._run_call() finally: monitor.join(timeout=2.0) else: self.worker = threading.Thread(target=_context_thread_target(self._run_call), daemon=True) self.worker.start() self._monitor_loop() if self._monitor_interrupted["yes"]: raise InterruptedError("Agent interrupted during streaming API call") if self.agent._interrupt_requested: # worker returned early before the monitor saw the flag raise InterruptedError("Agent interrupted during streaming API call (post-worker)") if self.result["error"] is not None: if self.deltas_were_sent["yes"]: return self._partial_stream_stub() raise self.result["error"] if self.result["response"] is not None: _reset_stale_streak(self.agent) # provider proved responsive: clear the breaker # Propagate first-chunk timing for the ``post_api_request`` hook. if isinstance(self.clients.diag, dict) and self.clients.diag.get("first_chunk_at"): self.agent._last_api_first_chunk_at = float(self.clients.diag["first_chunk_at"]) return self.result["response"] def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=None): """Streaming variant of _interruptible_api_call: fires the delta callbacks per text token (tool-call turns suppress them) and returns a SimpleNamespace in the non-streaming response shape. codex_responses delegates to the already- streaming codex runner; cron turns and delegated children run inline.""" if agent._interrupt_requested: raise InterruptedError("Agent interrupted before streaming API call") if agent.api_mode == "codex_responses": return _stream_codex_passthrough(agent, api_kwargs, on_first_delta) if agent.api_mode == "bedrock_converse": return _BedrockStream(agent, api_kwargs, on_first_delta).run() # Cross-turn stale-stream circuit breaker (see ``_stale_streak()``). _check_stale_giveup(agent) return _StreamingCall(agent, api_kwargs, on_first_delta).run() __all__ = ["interruptible_api_call", "build_api_kwargs", "build_assistant_message", "try_activate_fallback", "handle_max_iterations", "cleanup_task_resources", "interruptible_streaming_api_call"]