| """The agent conversation loop β extracted from ``run_agent.AIAgent``.
|
|
|
| This is the biggest single chunk pulled out of ``run_agent.py``: the
|
| roughly 3,900-line :func:`run_conversation` body that drives one user
|
| turn through the agent (model call, tool dispatch, retries, fallbacks,
|
| compression, post-turn hooks, background memory/skill review nudges).
|
|
|
| The function takes the parent ``AIAgent`` instance as its first
|
| argument (``agent``) and accesses its state via attribute lookup.
|
| ``_ra().AIAgent.run_conversation`` is now a thin forwarder.
|
|
|
| Symbols that production code or tests patch on ``run_agent`` directly
|
| (``handle_function_call``, ``_set_interrupt``, ``OpenAI``, ...) are
|
| resolved through :func:`_ra` so those patches keep working.
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| import logging
|
| import os
|
| import random
|
| import re
|
| import ssl
|
| import time
|
| from typing import Any, Dict, List, Optional
|
|
|
| from agent.codex_responses_adapter import _summarize_user_message_for_log
|
| from agent.conversation_compression import (
|
| COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE,
|
| COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE,
|
| COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE,
|
| COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE,
|
| PRE_API_COMPRESSION_STATUS_TEMPLATE,
|
| compression_skipped_due_to_lock,
|
| conversation_history_after_compression,
|
| )
|
| from agent.context_engine import automatic_compaction_status_message
|
| from agent.display import KawaiiSpinner
|
| from agent.error_classifier import FailoverReason, classify_api_error
|
| from agent.turn_context import (
|
| _compression_warrants_another_preflight_pass,
|
| build_turn_context,
|
| compose_user_api_content,
|
| reanchor_current_turn_user_idx,
|
| )
|
| from agent.turn_retry_state import TurnRetryState
|
| from agent.runtime_cwd import resolve_agent_cwd
|
| from agent.message_sanitization import (
|
| close_interrupted_tool_sequence,
|
| _repair_tool_call_arguments,
|
| _sanitize_messages_non_ascii,
|
| _sanitize_messages_surrogates,
|
| _sanitize_structure_non_ascii,
|
| _sanitize_structure_surrogates,
|
| _sanitize_surrogates,
|
| _sanitize_tools_non_ascii,
|
| _strip_images_from_messages,
|
| _strip_non_ascii,
|
| )
|
| from agent.model_metadata import (
|
| MINIMUM_CONTEXT_LENGTH,
|
| _estimate_tools_tokens_rough,
|
| estimate_messages_tokens_rough,
|
| estimate_request_tokens_rough,
|
| get_context_length_from_provider_error,
|
| is_output_cap_error,
|
| parse_available_output_tokens_from_error,
|
| save_context_length,
|
| )
|
| from agent.process_bootstrap import _install_safe_stdio
|
| from agent.prompt_caching import (
|
| build_prompt_cache_plan,
|
| strip_anthropic_cache_control,
|
| strip_anthropic_tool_cache_control,
|
| )
|
| from agent.retry_utils import (
|
| adaptive_rate_limit_backoff,
|
| is_zai_coding_overload_error,
|
| jittered_backoff,
|
| zai_coding_overload_retry_ceiling,
|
| )
|
| from agent.trajectory import has_incomplete_scratchpad
|
| from agent.usage_pricing import estimate_usage_cost, normalize_usage
|
| from hermes_constants import PARTIAL_STREAM_STUB_ID
|
| from hermes_logging import set_session_context
|
| from tools.skill_provenance import set_current_write_origin
|
| from utils import base_url_host_matches, env_var_enabled
|
|
|
| logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
| INTERRUPT_WAITING_FOR_MODEL_PREFIX = "Operation interrupted: waiting for model response ("
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _LOCAL_PROCESSING_MODULES = frozenset({
|
| "agent_runtime_helpers",
|
| "message_content",
|
| "message_sanitization",
|
| "chat_completion_helpers",
|
| })
|
| _API_CALL_MODULES = frozenset({
|
| "chat_completion_helpers",
|
| })
|
|
|
|
|
| def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text: str) -> None:
|
| """Append a provider-safe checkpoint and correction to the live turn.
|
|
|
| Incomplete provider reasoning blocks are not valid replay items (Anthropic
|
| signs them; Responses reasoning items require their following output).
|
| Preserve only the *visible* response text, demoted to ordinary text, then
|
| add the correction as a real user message. This keeps role alternation
|
| valid and leaves every previously cached message byte-for-byte unchanged.
|
|
|
| INVARIANT β raw chain-of-thought must never be serialized into replayable
|
| message content. Streamed reasoning is display-only state: it may be shown
|
| live, but it does not re-enter the transcript as assistant (or user) text.
|
| An assistant turn whose content inlines its own chain-of-thought reads to
|
| Anthropic's output classifier as reasoning-injection/prefill jailbreak,
|
| and because the poisoned checkpoint is persisted and replayed on every
|
| subsequent call, the session dies permanently with deterministic
|
| "Provider returned an empty response" storms that no retry, nudge, or
|
| empty-recovery branch can escape (July 2026: four sessions bricked this
|
| way; every reasoning-free checkpoint that week was untouched β same
|
| mechanism as the ~/.hermes/prefill.json incident, 20/20 blocked with
|
| assistant-exposed CoT vs 0/20 without). The interrupted reasoning was
|
| incomplete by definition; the model regenerates it on the retried turn.
|
| If a future path needs to preserve interrupted thinking, carry it in a
|
| provider-gated reasoning *field*, never in content.
|
| INVARIANT β the scaffolding is provider-replay text, not transcript text.
|
| ``[This response was interrupted by a user correction.]`` and its
|
| ``Visible response before the interruption:`` header exist so the MODEL
|
| understands its own reply was cut off. They are not prose the user wrote
|
| or the agent said. Persisting them into ``content`` painted the raw
|
| machinery as an assistant bubble on every reload (and merged it into the
|
| preceding tool-call bubble), which is what made a steered transcript
|
| unreadable. Carry the scaffolded form in the ``api_content`` sidecar --
|
| the exact bytes replayed to the provider -- and keep ``content`` clean.
|
| When nothing was on screen there is no clean form at all, so the row is
|
| marked ``display_kind="hidden"``: still replayed to the model, dropped by
|
| every transcript surface (desktop, TUI, CLI resume), exactly like the
|
| compaction-reference rows.
|
| """
|
| visible = agent._strip_think_blocks(
|
| getattr(agent, "_current_streamed_assistant_text", "") or ""
|
| ).strip()
|
|
|
| checkpoint_parts = ["[This response was interrupted by a user correction.]"]
|
| if visible:
|
| checkpoint_parts.extend(
|
| ["Visible response before the interruption:", visible]
|
| )
|
| checkpoint = "\n\n".join(checkpoint_parts)
|
|
|
|
|
|
|
|
|
|
|
| if messages and messages[-1].get("role") == "assistant":
|
| correction = (
|
| "[Context from the interrupted assistant response]\n"
|
| f"{checkpoint}\n\n"
|
| f"{text}"
|
| )
|
|
|
|
|
| messages.append(
|
| {"role": "user", "content": text, "api_content": correction}
|
| )
|
| else:
|
| entry: Dict[str, Any] = {
|
| "role": "assistant",
|
| "content": visible or checkpoint,
|
| "api_content": checkpoint,
|
| }
|
| if not visible:
|
|
|
|
|
| entry["display_kind"] = "hidden"
|
| messages.append(entry)
|
| messages.append({"role": "user", "content": text})
|
|
|
| agent._current_streamed_assistant_text = ""
|
| agent._stream_needs_break = True
|
|
|
|
|
| def _is_copilot_provider(agent: Any) -> bool:
|
| """Delegate to ``AIAgent._is_copilot_provider`` (single owner of the check).
|
|
|
| ``agent.provider`` is not always the normalized ``copilot`` slug β
|
| ``/model`` and profile configs can leave the alias ``github-copilot`` (or
|
| ``github``) in place, and a bare ``provider == "copilot"`` gate silently
|
| skips credential recovery for those spellings.
|
| """
|
| try:
|
| return bool(agent._is_copilot_provider())
|
| except Exception:
|
| return (getattr(agent, "provider", "") or "").strip().lower() in {
|
| "copilot",
|
| "github-copilot",
|
| "github",
|
| }
|
|
|
|
|
| def _is_stale_copilot_credential_error(status_code: Optional[int], error_message: str) -> bool:
|
| """Detect a Copilot 400 that is really a STALE / DEGRADED credential.
|
|
|
| Copilot surfaces a stale or degraded credential as an HTTP 400 rather than a
|
| clean 401. Two body markers indicate this class:
|
|
|
| - ``model_not_available_for_integrator`` β the request reached the
|
| restricted ``copilot-language-server`` integrator (the server's fallback
|
| when it receives a raw OAuth token instead of an exchanged API token),
|
| whose model allowlist omits enterprise-only models.
|
| - ``model_not_supported`` / "the requested model is not supported" β the
|
| cached bearer's Copilot entitlement rotated out from under a long-lived
|
| process.
|
|
|
| Matched narrowly (status 400 AND a specific marker) so a genuinely wrong
|
| model name β a real 400 β never triggers the single-shot re-exchange. The
|
| caller enforces copilot-provider scoping and the single-shot guard.
|
| """
|
| lowered = (error_message or "").lower()
|
| is_400 = status_code == 400 or "error code: 400" in lowered
|
| if not is_400:
|
| return False
|
| return (
|
| "model_not_available_for_integrator" in lowered
|
| or "not available for integrator" in lowered
|
| or "model_not_supported" in lowered
|
| or "the requested model is not supported" in lowered
|
| )
|
|
|
|
|
| def _image_error_max_dimension(error: Exception) -> Optional[int]:
|
| """Extract a provider-reported image dimension ceiling, if present."""
|
| parts = []
|
| for value in (
|
| error,
|
| getattr(error, "message", None),
|
| getattr(error, "body", None),
|
| ):
|
| if value:
|
| try:
|
| parts.append(str(value))
|
| except Exception:
|
| pass
|
| text = " ".join(parts).lower()
|
| if "image" not in text or "dimension" not in text or "max allowed size" not in text:
|
| return None
|
|
|
| match = re.search(r"max allowed size(?:\s+for [^:]+)?:\s*(\d{3,5})\s*pixels?", text)
|
| if not match:
|
| return None
|
| try:
|
| max_dimension = int(match.group(1))
|
| except ValueError:
|
| return None
|
| if 512 <= max_dimension <= 8000:
|
| return max_dimension
|
| return None
|
|
|
|
|
| def _ollama_context_limit_error(agent: Any, request_tokens: int) -> Optional[str]:
|
| """Return a user-facing error when Ollama is loaded with too little context."""
|
| if not getattr(agent, "tools", None):
|
| return None
|
|
|
| runtime_ctx = getattr(agent, "_ollama_num_ctx", None)
|
| if not isinstance(runtime_ctx, int) or runtime_ctx <= 0:
|
| return None
|
| if runtime_ctx >= MINIMUM_CONTEXT_LENGTH:
|
| return None
|
|
|
| model = getattr(agent, "model", "") or "the selected model"
|
| base_url = getattr(agent, "base_url", "") or "unknown base URL"
|
| provider = getattr(agent, "provider", "") or "unknown"
|
| tool_count = len(getattr(agent, "tools", None) or [])
|
|
|
| logger.warning(
|
| "Ollama runtime context too small for Hermes tool use: "
|
| "model=%s provider=%s base_url=%s runtime_context=%d "
|
| "minimum_context=%d estimated_request_tokens=%d tool_count=%d "
|
| "session=%s",
|
| model,
|
| provider,
|
| base_url,
|
| runtime_ctx,
|
| MINIMUM_CONTEXT_LENGTH,
|
| request_tokens,
|
| tool_count,
|
| getattr(agent, "session_id", None) or "none",
|
| )
|
|
|
| return (
|
| f"Ollama loaded `{model}` with only {runtime_ctx:,} tokens of runtime "
|
| f"context, but Hermes needs at least {MINIMUM_CONTEXT_LENGTH:,} tokens "
|
| "for reliable tool use.\n\n"
|
| "Increase the Ollama context for this model and restart/reload the "
|
| "model before trying again. A known-good starting point is 65,536 "
|
| "tokens. In Hermes config, set `model.ollama_num_ctx: 65536` "
|
| "(and `model.context_length: 65536` if you also override the displayed "
|
| "model context). If you manage the model through an Ollama Modelfile, "
|
| "set `PARAMETER num_ctx 65536` there instead."
|
| )
|
|
|
|
|
| def _ra():
|
| """Lazy reference to ``run_agent`` so callers can patch
|
| ``run_agent.handle_function_call`` / ``run_agent._set_interrupt`` /
|
| ``run_agent.OpenAI`` and have those patches reach this code path.
|
| """
|
| import run_agent
|
| return run_agent
|
|
|
|
|
| def _nous_entitlement_message(capability: str) -> str:
|
| try:
|
| from hermes_cli.nous_account import (
|
| format_nous_portal_entitlement_message,
|
| get_nous_portal_account_info,
|
| )
|
|
|
| account_info = get_nous_portal_account_info(force_fresh=True)
|
| message = format_nous_portal_entitlement_message(
|
| account_info,
|
| capability=capability,
|
| )
|
| return message or ""
|
| except Exception:
|
| return ""
|
|
|
|
|
| def _print_nous_entitlement_guidance(agent, capability: str) -> bool:
|
| message = _nous_entitlement_message(capability)
|
| if not message:
|
| return False
|
| for line in message.splitlines():
|
| agent._vprint(f"{agent.log_prefix} π‘ {line}", force=True)
|
| return True
|
|
|
|
|
| def _is_nous_inference_route(provider: str, base_url: str) -> bool:
|
| provider = (provider or "").strip().lower()
|
| if provider == "nous":
|
| return True
|
| base = str(base_url or "")
|
| return (
|
| base_url_host_matches(base, "inference-api.nousresearch.com")
|
| )
|
|
|
|
|
| def _billing_or_entitlement_message(
|
| *,
|
| capability: str,
|
| provider: str,
|
| base_url: str,
|
| model: str,
|
| ) -> str:
|
| if _is_nous_inference_route(provider, base_url):
|
| return _nous_entitlement_message(capability)
|
|
|
| provider_label = (provider or "").strip() or "the selected provider"
|
| model_label = (model or "").strip() or "the selected model"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (provider or "").strip().lower() == "anthropic":
|
| lines = [
|
| (
|
| f"{provider_label} reported that your Claude subscription usage is "
|
| f"exhausted for {model_label} (included quota + extra-usage credits)."
|
| ),
|
| "Options: wait for the billing cycle to reset, or add extra usage at "
|
| "https://claude.ai/settings/usage",
|
| "You can also switch to an Anthropic API key or another provider with "
|
| "/model <model> --provider <provider>.",
|
| ]
|
| return "\n".join(lines)
|
|
|
|
|
|
|
|
|
| try:
|
| from agent.billing_links import build_billing_block
|
|
|
| _link = build_billing_block(provider=provider, base_url=base_url, model=model)
|
| if _link.provider_label:
|
| provider_label = _link.provider_label
|
| billing_url = _link.billing_url
|
| except Exception:
|
| billing_url = None
|
|
|
| lines = [
|
| (
|
| f"{provider_label} reported that billing, credits, or account "
|
| f"entitlement is exhausted for {model_label}."
|
| ),
|
| "Add credits or update billing with that provider, then retry.",
|
| ]
|
| if billing_url:
|
| lines.append(f"{provider_label} billing: {billing_url}")
|
| lines.append("You can switch providers temporarily with /model <model> --provider <provider>.")
|
| return "\n".join(lines)
|
|
|
|
|
| def _billing_block_dict(provider, base_url, model, message="") -> Optional[dict]:
|
| """Best-effort structured billing descriptor (None if billing_links is unavailable)."""
|
| try:
|
| from agent.billing_links import build_billing_block
|
|
|
| return build_billing_block(
|
| provider=provider, base_url=str(base_url), model=model, message=message
|
| ).to_dict()
|
| except Exception:
|
| return None
|
|
|
|
|
| def _print_billing_or_entitlement_guidance(
|
| agent,
|
| *,
|
| capability: str,
|
| provider: str,
|
| base_url: str,
|
| model: str,
|
| ) -> bool:
|
| message = _billing_or_entitlement_message(
|
| capability=capability,
|
| provider=provider,
|
| base_url=base_url,
|
| model=model,
|
| )
|
| if not message:
|
| return False
|
| for line in message.splitlines():
|
| agent._vprint(f"{agent.log_prefix} π‘ {line}", force=True)
|
| return True
|
|
|
|
|
| def _try_refresh_nous_paid_entitlement_credentials(agent) -> bool:
|
| """Refresh Nous runtime credentials after a fresh paid-entitlement check."""
|
| try:
|
| from hermes_cli.nous_account import get_nous_portal_account_info
|
|
|
| account_info = get_nous_portal_account_info(force_fresh=True)
|
| if account_info.paid_service_access is not True:
|
| return False
|
| return agent._try_refresh_nous_client_credentials(
|
| force=True,
|
| )
|
| except Exception:
|
| return False
|
|
|
|
|
| def _restore_or_build_system_prompt(agent, system_message, conversation_history):
|
| """Restore the cached system prompt from the session DB or build it fresh.
|
|
|
| Mutates ``agent._cached_system_prompt`` and persists a freshly-built
|
| prompt back to the session DB on first build. Extracted from
|
| ``run_conversation`` so the prefix-cache restore path can be tested in
|
| isolation.
|
|
|
| Three-way state distinction for the stored row, surfaced via logs so
|
| silent prefix-cache misses are visible in ``agent.log``:
|
|
|
| * ``missing`` β no session row yet (legitimate first turn).
|
| * ``null`` β row exists, ``system_prompt`` column is NULL.
|
| Legacy session predating system-prompt persistence, or a migration
|
| leftover. Warns when ``conversation_history`` is non-empty.
|
| * ``empty`` β row exists, ``system_prompt`` column is the empty
|
| string. Indicates a previous-turn write that ran but stored
|
| nothing (silent persistence bug). Always warns.
|
| * ``present`` β row exists with a usable prompt β reused verbatim.
|
|
|
| Read or write failures against the session DB log at WARNING (not
|
| DEBUG) so persistent issues (disk full, schema drift, lock contention)
|
| surface without needing verbose mode. This used to be a debug-level
|
| log that silently broke prefix-cache reuse on the gateway path
|
| (which constructs a fresh ``AIAgent`` per turn and depends on this
|
| DB roundtrip).
|
| """
|
| stored_prompt = None
|
| stored_state = "missing"
|
| if conversation_history and agent._session_db:
|
| try:
|
| session_row = agent._session_db.get_session(agent.session_id)
|
| if session_row is not None:
|
| raw_prompt = session_row.get("system_prompt")
|
| if raw_prompt is None:
|
| stored_state = "null"
|
| elif raw_prompt == "":
|
| stored_state = "empty"
|
| else:
|
| stored_prompt = raw_prompt
|
| stored_state = "present"
|
| except Exception as exc:
|
| logger.warning(
|
| "Session DB get_session failed for system-prompt restore "
|
| "(session=%s): %s. Falling back to fresh build β prefix "
|
| "cache will miss for this turn.",
|
| agent.session_id, exc,
|
| )
|
|
|
| if stored_prompt and _stored_prompt_matches_runtime(agent, stored_prompt):
|
|
|
|
|
| agent._cached_system_prompt = stored_prompt
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| from agent.system_prompt import reconstruct_static_prefix
|
|
|
| reconstruct_static_prefix(agent, system_message=system_message)
|
| return
|
| if stored_prompt:
|
| stored_state = "stale_runtime"
|
| logger.info(
|
| "Stored system prompt for session %s has stale runtime identity; "
|
| "rebuilding for model=%s provider=%s.",
|
| agent.session_id,
|
| getattr(agent, "model", "") or "",
|
| getattr(agent, "provider", "") or "",
|
| )
|
|
|
| if conversation_history and stored_state in ("null", "empty"):
|
|
|
|
|
|
|
|
|
| logger.warning(
|
| "Stored system prompt for session %s is %s; rebuilding "
|
| "from scratch this turn. Prefix cache will miss until "
|
| "the rebuild persists. Investigate the previous turn's "
|
| "update_system_prompt write path.",
|
| agent.session_id, stored_state,
|
| )
|
|
|
|
|
|
|
| agent._cached_system_prompt = agent._build_system_prompt(system_message)
|
|
|
|
|
|
|
|
|
| try:
|
| from hermes_cli.lifecycle import invoke_hook as _invoke_hook
|
| _invoke_hook(
|
| "on_session_start",
|
| session_id=agent.session_id,
|
| model=agent.model,
|
| platform=getattr(agent, "platform", None) or "",
|
| )
|
| except Exception as exc:
|
| logger.warning("on_session_start hook failed: %s", exc)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| try:
|
| from agent.credits_tracker import seed_credits_at_session_start
|
|
|
| seed_credits_at_session_start(agent)
|
| except Exception:
|
| logger.debug("cold-start credits seed failed (fail-open)", exc_info=True)
|
|
|
|
|
|
|
|
|
|
|
| if agent._session_db:
|
| try:
|
| agent._session_db.update_system_prompt(agent.session_id, agent._cached_system_prompt)
|
| except Exception as exc:
|
| logger.warning(
|
| "Session DB update_system_prompt failed for session %s: "
|
| "%s. Subsequent turns will rebuild the system prompt and "
|
| "miss the prefix cache.",
|
| agent.session_id, exc,
|
| )
|
|
|
|
|
| def _stored_prompt_matches_runtime(agent, prompt: str) -> bool:
|
| """Return False when the persisted runtime-identity lines are stale."""
|
|
|
| def line_value(label: str) -> str:
|
| """Last matching line wins.
|
|
|
| Safe ONLY for fields emitted in the volatile tier at the very END of
|
| the prompt (Model / Provider / Platform). User-supplied project
|
| context (AGENTS.md / CLAUDE.md / .cursorrules) is embedded in the
|
| middle context tier, so a last-match scan lets project prose shadow
|
| any field emitted EARLIER β see ``host_info_value``.
|
| """
|
| prefix = f"{label}:"
|
| value = ""
|
| for line in prompt.splitlines():
|
| if line.startswith(prefix):
|
| value = line[len(prefix):].strip()
|
| return value
|
|
|
| def host_info_value(label: str) -> str:
|
| """Read a field from the prompt's own host-info block.
|
|
|
| The host-info block (``build_environment_hints``) sits in the STABLE
|
| tier, ahead of the embedded project context files. A bare scan of the
|
| whole prompt would therefore match a user's ``AGENTS.md`` that merely
|
| contains a line starting with the same label, comparing runtime state
|
| against project prose. That mismatch never clears, so the check would
|
| reject the stored prompt on EVERY turn β rebuilding the system prompt
|
| each message and destroying the prefix cache for the whole session,
|
| which is far worse than the staleness this function guards against.
|
|
|
| Anchor on the ``User home directory:`` line that immediately precedes
|
| the working-directory line in that block, and take the FIRST such
|
| occurrence, so only Hermes' own emitted block can satisfy the read.
|
| """
|
| prefix = f"{label}:"
|
| lines = prompt.splitlines()
|
| for idx, line in enumerate(lines):
|
| if not line.startswith("User home directory:"):
|
| continue
|
| for candidate in lines[idx + 1: idx + 4]:
|
| if candidate.startswith(prefix):
|
| return candidate[len(prefix):].strip()
|
| return ""
|
|
|
| stored_model = line_value("Model")
|
| current_model = str(getattr(agent, "model", "") or "").strip()
|
| if stored_model and current_model and stored_model != current_model:
|
| return False
|
|
|
| stored_provider = line_value("Provider")
|
| current_provider = str(getattr(agent, "provider", "") or "").strip()
|
| if stored_provider and current_provider and stored_provider != current_provider:
|
| return False
|
|
|
|
|
|
|
|
|
|
|
|
|
| stored_cwd = host_info_value("Current working directory")
|
| if stored_cwd:
|
| if stored_cwd != str(resolve_agent_cwd()):
|
| return False
|
|
|
|
|
|
|
|
|
| stored_platform = line_value("Platform")
|
| current_platform = str(getattr(agent, "platform", "") or "").strip()
|
| if stored_platform and current_platform and stored_platform != current_platform:
|
| return False
|
|
|
| return True
|
|
|
|
|
| def _get_continuation_prompt(is_partial_stub: bool, dropped_tools: Optional[List[str]] = None) -> str:
|
| if is_partial_stub and dropped_tools:
|
| tool_list = ", ".join(dropped_tools[:3])
|
| return (
|
| "[System: Your previous tool call "
|
| f"({tool_list}) was too large and "
|
| "the stream timed out before it "
|
| "could be delivered. Do NOT retry "
|
| "the same tool call with the same "
|
| "large content. Instead, break the "
|
| "content into multiple smaller tool "
|
| "calls (e.g. use multiple patch calls "
|
| "or write smaller files). Each tool "
|
| "call's arguments must be under ~8K "
|
| "tokens to avoid stream timeouts.]"
|
| )
|
| elif is_partial_stub:
|
| return (
|
| "[System: The previous response was cut off by a "
|
| "network error mid-stream. Continue exactly where "
|
| "you left off. Do not restart or repeat prior text. "
|
| "Finish the answer directly.]"
|
| )
|
| else:
|
| return (
|
| "[System: Your previous response was truncated by the output "
|
| "length limit. Continue exactly where you left off. Do not "
|
| "restart or repeat prior text. Finish the answer directly.]"
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _CODEX_INCOMPLETE_NUDGE = (
|
| "[System: Your previous response contained only internal reasoning and "
|
| "never produced a visible answer or tool call. Do not keep thinking. "
|
| "Produce your final answer as plain text now (or make the tool call "
|
| "you were planning).]"
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _CONTENT_POLICY_RECOVERY_HINT = (
|
| "Try rephrasing the request, narrowing the context, or "
|
| "adding a fallback provider with `hermes fallback add`."
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _CANON_ARGS_CACHE: Dict[str, str] = {}
|
| _CANON_ARGS_CACHE_MAX = 4096
|
|
|
|
|
|
|
|
|
| _CANON_ARGS_CACHE_MAX_BYTES = 32 * 1024 * 1024
|
| _canon_args_cache_bytes = 0
|
|
|
|
|
| def _canonicalize_tool_call_arguments(arg_str: str) -> str:
|
| """Return the canonical wire form of a tool-call arguments JSON string.
|
|
|
| Raises whatever ``json.loads`` raises on malformed input; the caller
|
| falls back to ``_repair_tool_call_arguments``, exactly as before.
|
| """
|
| global _canon_args_cache_bytes
|
| cached = _CANON_ARGS_CACHE.get(arg_str)
|
| if cached is not None:
|
| return cached
|
| canonical = json.dumps(
|
| json.loads(arg_str), separators=(",", ":"), sort_keys=True,
|
| )
|
| _CANON_ARGS_CACHE[arg_str] = canonical
|
| _canon_args_cache_bytes += len(arg_str) + len(canonical)
|
| while len(_CANON_ARGS_CACHE) > _CANON_ARGS_CACHE_MAX or (
|
| _canon_args_cache_bytes > _CANON_ARGS_CACHE_MAX_BYTES
|
| and len(_CANON_ARGS_CACHE) > 1
|
| ):
|
| try:
|
| evicted_key = next(iter(_CANON_ARGS_CACHE))
|
| evicted_val = _CANON_ARGS_CACHE.pop(evicted_key)
|
| _canon_args_cache_bytes -= len(evicted_key) + len(evicted_val)
|
| except (StopIteration, KeyError, RuntimeError):
|
| break
|
| return canonical
|
|
|
|
|
| def _canonicalize_api_tool_calls(api_messages) -> None:
|
| """Canonicalize tool-call argument JSON on the send-path message copy.
|
|
|
| Rewrites each message's ``tool_calls`` in place (copy-on-write for the
|
| tool-call dicts it canonicalizes; the persisted history is untouched).
|
| The pass still traverses every message and tool call each iteration;
|
| the memo above bounds the JSON parse/serialize work to one round-trip
|
| per UNIQUE argument string instead of one per string per iteration β
|
| the quadratic part of the cost. The remaining traversal is pointer
|
| chasing and dict copies, cheap next to a json.loads + json.dumps.
|
| """
|
| for am in api_messages:
|
| tcs = am.get("tool_calls")
|
| if not tcs:
|
| continue
|
| new_tcs = []
|
| for tc in tcs:
|
| if isinstance(tc, dict) and "function" in tc:
|
| try:
|
| tc = {**tc, "function": {
|
| **tc["function"],
|
| "arguments": _canonicalize_tool_call_arguments(
|
| tc["function"]["arguments"]
|
| ),
|
| }}
|
| except Exception:
|
| tc["function"]["arguments"] = _repair_tool_call_arguments(
|
| tc["function"]["arguments"],
|
| tc["function"].get("name", "?"),
|
| )
|
| new_tcs.append(tc)
|
| am["tool_calls"] = new_tcs
|
|
|
|
|
| def _invalid_tool_name_error_content(name: str, valid_tool_names) -> str:
|
| """Error-result content for a tool call whose name isn't a real tool.
|
|
|
| A blank/whitespace-only name is not a typo the model can fuzzy-correct
|
| toward a real tool β it is almost always a weak open model echoing
|
| tool-call XML/JSON it saw in file or tool output (#47967:
|
| <tool_call>/<invoke name=...> payloads in a file prime
|
| mimo/nemotron-class models to emit empty structured calls), or a model
|
| degrading at very large context (observed with gpt-5.6 past ~350K input).
|
| Dumping the full tool catalog in that case feeds the priming loop more
|
| names to mimic and inflates context 3-4x across retries, so send a terse
|
| error that tells the model in-context tool-call syntax is DATA, not a
|
| call to make. A genuinely-wrong-but-nonempty name (an actual typo) still
|
| gets the catalog so the model can self-correct.
|
| """
|
| if not (name or "").strip():
|
| return (
|
| "Tool call rejected: the tool name was empty. "
|
| "If tool-call XML or JSON appeared in file "
|
| "contents or tool output, that is data β do "
|
| "not re-emit it as a tool call. To call a "
|
| "tool, use a valid name from your tool list; "
|
| "otherwise reply in plain text."
|
| )
|
| available = ", ".join(sorted(valid_tool_names))
|
| return f"Tool '{name}' does not exist. Available tools: {available}"
|
|
|
|
|
| def _content_policy_blocked_result(
|
| messages: List[Dict],
|
| api_call_count: int,
|
| *,
|
| final_response: str,
|
| error_detail: str,
|
| ) -> Dict[str, Any]:
|
| """Build the terminal turn result for a content-policy block.
|
|
|
| A content-policy refusal is deterministic for the unchanged prompt, so the
|
| turn ends here (no retry). Both the HTTP-200 refusal handler and the
|
| exception-path handler return the identical shape β a failed, non-completed
|
| turn carrying the user-facing message and a ``content_policy_blocked:``
|
| prefixed error β so they funnel through this one builder.
|
| """
|
| return {
|
| "final_response": final_response,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "failed": True,
|
| "error": f"content_policy_blocked: {error_detail}",
|
| }
|
|
|
|
|
| def _compression_deferred_result(
|
| agent,
|
| messages: List[Dict],
|
| api_call_count: int,
|
| ) -> Dict[str, Any]:
|
| """Build the soft turn result for a lock-contended compression defer.
|
|
|
| Another path (a sibling turn, a background review fork, a manual
|
| ``/compress``) holds this session's compression lock, so every
|
| compression pass this turn no-oped and the request still does not fit.
|
| This is a TEMPORARY condition β the lock winner is actively shrinking
|
| the same session β so the turn must end as a soft defer
|
| (``compression_deferred``), never as ``compression_exhausted``: the
|
| gateway auto-resets (wipes) the session on exhaustion (#9893/#35809),
|
| which would destroy a session that the concurrent compressor is about
|
| to make healthy again.
|
|
|
| ``failed`` stays False so the gateway persists the user turn (transient
|
| branch) and retry-next-message semantics apply.
|
| """
|
| holder = getattr(agent, "_compression_skipped_due_to_lock", None)
|
| logger.info(
|
| "turn deferred: compression lock held by another path "
|
| "(session=%s holder=%s) β not counting as compression exhaustion",
|
| agent.session_id or "none",
|
| holder if isinstance(holder, str) else "unconfirmed",
|
| )
|
| try:
|
| agent._flush_status_buffer()
|
| except Exception:
|
| pass
|
| _final = (
|
| "Context compression is already running for this session. "
|
| "Please retry in a moment β your next message will be processed "
|
| "once the concurrent compression finishes."
|
| )
|
| return {
|
| "final_response": _final,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final,
|
| "partial": True,
|
| "failed": False,
|
| "compression_deferred": True,
|
| "session_id": agent.session_id,
|
| }
|
|
|
|
|
| def _rewrite_system_content_blocks(system_message: dict, effective: str) -> bool:
|
| """Rewrite a cache-decorated system message in place, keeping its blocks.
|
|
|
| ``apply_anthropic_cache_control`` runs once per call block, *before* the
|
| retry loop, and splits the system prompt into ``[static prefix, volatile
|
| tail]`` text blocks carrying the cache_control breakpoints. Assigning a bare
|
| string over that list drops both breakpoints, so the failover retry ships
|
| the whole system prompt uncached and re-bills it in full.
|
|
|
| ``rewrite_prompt_model_identity`` only touches the LAST ``Model:`` /
|
| ``Provider:`` lines, and those live in the volatile tail β so the static
|
| prefix stays byte-identical and its cache entry keeps matching. Returns
|
| False when the shape is not one we can safely patch, so the caller falls
|
| back to the plain-string assignment.
|
| """
|
| content = system_message.get("content")
|
| if not isinstance(content, list) or not content:
|
| return False
|
| if not all(
|
| isinstance(part, dict) and part.get("type") == "text" for part in content
|
| ):
|
| return False
|
| if len(content) == 1:
|
| content[0]["text"] = effective
|
| return True
|
| if len(content) == 2:
|
| head = content[0].get("text") or ""
|
| if head and effective.startswith(head):
|
| tail = effective[len(head):]
|
| if tail:
|
| content[1]["text"] = tail
|
| return True
|
| return False
|
|
|
|
|
| def _sync_failover_system_message(agent, api_messages, active_system_prompt):
|
| """Refresh the in-flight system message after a provider failover.
|
|
|
| ``try_activate_fallback`` rewrites the ``Model:``/``Provider:`` identity
|
| lines on ``agent._cached_system_prompt`` (see
|
| ``rewrite_prompt_model_identity``) so the agent reports the model that is
|
| actually answering. But the current call block's ``api_messages`` were
|
| built from the pre-failover prompt, and the retry loop rebuilds
|
| ``api_kwargs`` from that list each iteration β without this sync the
|
| whole turn (and every gateway turn, since fallback re-activates per
|
| message while the primary is down) ships the stale identity.
|
|
|
| Mutates ``api_messages[0]`` in place and returns the prompt to use as
|
| ``active_system_prompt`` for subsequent call-block rebuilds.
|
| """
|
| sp = getattr(agent, "_cached_system_prompt", None)
|
| if not isinstance(sp, str) or not sp:
|
| return active_system_prompt
|
| if api_messages and api_messages[0].get("role") == "system":
|
| effective = sp
|
| if agent.ephemeral_system_prompt:
|
| effective = (effective + "\n\n" + agent.ephemeral_system_prompt).strip()
|
| if not _rewrite_system_content_blocks(api_messages[0], effective):
|
| api_messages[0]["content"] = effective
|
| return sp
|
|
|
|
|
| def _ensure_cached_system_prompt_static(agent, system_message=None) -> None:
|
| """Rebuild ``_cached_system_prompt_static`` when caching becomes active.
|
|
|
| Sessions restored under a cache-off primary skip the static-prefix rebuild
|
| (gated on ``_use_prompt_caching`` at restore time). A later failover to a
|
| cache-on provider would otherwise redecorate with ``static_system_prefix=
|
| None`` and silently fall back to the legacy system-plus-3 layout (#72626).
|
|
|
| Thin wrapper over :func:`agent.system_prompt.reconstruct_static_prefix`,
|
| which memoizes failed rebuilds so this stays cheap on the retry-loop hot
|
| path (it runs at the top of every attempt).
|
| """
|
| from agent.system_prompt import reconstruct_static_prefix
|
|
|
| reconstruct_static_prefix(
|
| agent, system_message=system_message, log_label="failover redecoration"
|
| )
|
|
|
|
|
| def _peel_moa_guidance(
|
| messages: List[Dict[str, Any]],
|
| guidance: Any,
|
| ) -> List[Dict[str, Any]]:
|
| """Remove MoA reference guidance previously attached by ``_attach_reference_guidance``.
|
|
|
| Thin wrapper over :func:`agent.moa_loop.peel_reference_guidance` (kept
|
| adjacent to the attach so the forward/inverse shapes evolve together).
|
| Lazy import mirrors the module's other moa_loop touchpoints.
|
| """
|
| from agent.moa_loop import peel_reference_guidance
|
|
|
| return peel_reference_guidance(messages, guidance)
|
|
|
|
|
| def _redecorate_prompt_cache_for_provider(
|
| agent,
|
| api_messages: List[Dict[str, Any]],
|
| *,
|
| system_message=None,
|
| moa_prepared: Optional[Dict[str, Any]] = None,
|
| tools_for_api: Optional[List[Dict[str, Any]]] = None,
|
| ) -> tuple[List[Dict[str, Any]], Optional[Dict[str, Any]]] | tuple[List[Dict[str, Any]], Optional[Dict[str, Any]], List[Dict[str, Any]]]:
|
| """Strip and re-apply cache_control for the *current* provider policy.
|
|
|
| Decoration runs once per call block before the retry loop for the primary
|
| provider. ``try_activate_fallback`` refreshes ``_use_prompt_caching`` /
|
| ``_use_native_cache_layout`` but the nine failover ``continue`` paths reused
|
| the old ``api_messages`` (#72626). Mirror ``_reapply_reasoning_echo_for_provider``
|
| by reshaping at the top of each retry attempt.
|
|
|
| The source list is the mutated in-flight request (image shrink / ASCII /
|
| reasoning_details recoveries already applied), never a pristine
|
| pre-decoration snapshot. MoA guidance is peeled and rebased without
|
| decoration; the acting aggregator plans its resolved destination later.
|
| """
|
| messages: List[Dict[str, Any]] = [
|
| dict(m) if isinstance(m, dict) else m for m in (api_messages or [])
|
| ]
|
| prepared = moa_prepared
|
| guidance = prepared.get("guidance") if isinstance(prepared, dict) else None
|
| if guidance:
|
| messages = _peel_moa_guidance(messages, guidance)
|
|
|
| strip_anthropic_cache_control(messages)
|
| planned_tools = strip_anthropic_tool_cache_control(
|
| tools_for_api if tools_for_api is not None else getattr(agent, "tools", [])
|
| )
|
|
|
| if prepared is not None and getattr(agent, "provider", None) == "moa":
|
|
|
|
|
| completions = getattr(getattr(agent.client, "chat", None), "completions", None)
|
| rebase = getattr(completions, "rebase_prepared_request", None)
|
| if callable(rebase):
|
| prepared = rebase(prepared, messages)
|
| messages = prepared["messages"]
|
| if tools_for_api is None:
|
| return messages, prepared
|
| return messages, prepared, planned_tools
|
|
|
|
|
|
|
|
|
| if agent._use_prompt_caching:
|
| _ensure_cached_system_prompt_static(agent, system_message=system_message)
|
| static = getattr(agent, "_cached_system_prompt_static", None)
|
| direct_tool_cache = getattr(
|
| agent,
|
| "_direct_native_anthropic_tool_cache_capability",
|
| lambda: False,
|
| )()
|
| plan = build_prompt_cache_plan(
|
| messages,
|
| planned_tools,
|
| cache_ttl=agent._cache_ttl,
|
| native_anthropic=agent._use_native_cache_layout,
|
| static_system_prefix=static if isinstance(static, str) else None,
|
| direct_native_tool_cache=direct_tool_cache,
|
| )
|
| messages = plan.messages
|
| planned_tools = plan.tools
|
|
|
| if tools_for_api is None:
|
| return messages, prepared
|
| return messages, prepared, planned_tools
|
|
|
|
|
| def _apply_context_engine_selection(
|
| agent: Any,
|
| api_messages: List[Dict[str, Any]],
|
| conversation_messages: List[Dict[str, Any]],
|
| incoming_message: Optional[Dict[str, Any]],
|
| *,
|
| logger: Any,
|
| ) -> List[Dict[str, Any]]:
|
| """Run the optional per-turn ``ContextEngine.select_context()`` hook.
|
|
|
| Returns the (possibly replaced) request message list. The hook is for
|
| context *selection / routing* (retrieval, topic routing, role switching),
|
| which is distinct from compression and fires every turn independent of
|
| ``should_compress()``.
|
|
|
| Fail-open by design: a missing hook, any exception, or an invalid return
|
| value yields the unmodified ``api_messages``. The result is request-only β
|
| persisted conversation history is never mutated here.
|
| """
|
| engine = getattr(agent, "context_compressor", None)
|
| if engine is None or not hasattr(engine, "select_context"):
|
| return api_messages
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| try:
|
| from agent.context_engine import ContextEngine as _CE
|
| if getattr(engine.select_context, "__func__", None) is _CE.select_context:
|
| return api_messages
|
| except Exception:
|
| pass
|
|
|
| session_label = getattr(agent, "session_id", None) or "-"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _conv_copy = [dict(m) if isinstance(m, dict) else m for m in conversation_messages] \
|
| if conversation_messages is not None else None
|
| _incoming_copy = dict(incoming_message) if isinstance(incoming_message, dict) else incoming_message
|
| try:
|
| selected = engine.select_context(
|
| api_messages,
|
| conversation_messages=_conv_copy,
|
| incoming_message=_incoming_copy,
|
| budget_tokens=getattr(engine, "context_length", 0) or 0,
|
| )
|
| except Exception:
|
| logger.warning(
|
| "Context engine select_context hook failed; using unmodified "
|
| "request messages (session=%s)",
|
| session_label,
|
| exc_info=True,
|
| )
|
| return api_messages
|
|
|
| if selected is None:
|
| return api_messages
|
|
|
|
|
|
|
|
|
|
|
| if isinstance(selected, list) and selected and all(isinstance(m, dict) for m in selected):
|
| return selected
|
|
|
| logger.warning(
|
| "Context engine select_context returned an invalid value "
|
| "(not a non-empty list of dicts); ignoring (session=%s)",
|
| session_label,
|
| )
|
| return api_messages
|
|
|
|
|
| def _notify_context_engine_turn_complete(
|
| agent: Any,
|
| messages: List[Dict[str, Any]],
|
| *,
|
| usage: Optional[Dict[str, Any]] = None,
|
| logger: Any,
|
| **meta: Any,
|
| ) -> None:
|
| """Notify the active context engine that a user turn has finished.
|
|
|
| Calls the optional ``ContextEngine.on_turn_complete()`` observation hook
|
| once per turn, after the assistant/tool loop has produced the finalized
|
| transcript. The complement to ``select_context()`` (pre-request selection):
|
| this lets an engine ingest / index / summarize the completed turn.
|
|
|
| Fail-open: a missing or no-op hook, or any exception, is swallowed.
|
| ``messages`` is passed as a shallow copy so the engine cannot mutate the
|
| persisted transcript.
|
| """
|
| engine = getattr(agent, "context_compressor", None)
|
| hook = getattr(engine, "on_turn_complete", None)
|
| if engine is None or not callable(hook):
|
| return
|
|
|
|
|
|
|
|
|
| try:
|
| from agent.context_engine import ContextEngine as _CE
|
| if getattr(hook, "__func__", None) is _CE.on_turn_complete:
|
| return
|
| except Exception:
|
| pass
|
|
|
| try:
|
| hook(
|
| [dict(m) if isinstance(m, dict) else m for m in messages],
|
| usage=usage,
|
| **meta,
|
| )
|
| except Exception:
|
| logger.warning(
|
| "Context engine on_turn_complete hook failed (session=%s)",
|
| getattr(agent, "session_id", None) or "-",
|
| exc_info=True,
|
| )
|
|
|
|
|
| def run_conversation(
|
| agent,
|
| user_message: Any,
|
| system_message: str = None,
|
| conversation_history: List[Dict[str, Any]] = None,
|
| task_id: str = None,
|
| stream_callback: Optional[callable] = None,
|
| persist_user_message: Optional[Any] = None,
|
| persist_user_timestamp: Optional[float] = None,
|
| persist_user_display_kind: Optional[str] = None,
|
| persist_user_display_metadata: Optional[Dict[str, Any]] = None,
|
| moa_config: Optional[dict[str, Any]] = None,
|
| ) -> Dict[str, Any]:
|
| """
|
| Run a complete conversation with tool calling until completion.
|
|
|
| Args:
|
| user_message (str): The user's message/question
|
| system_message (str): Custom system message (optional, overrides ephemeral_system_prompt if provided)
|
| conversation_history (List[Dict]): Previous conversation messages (optional)
|
| task_id (str): Unique identifier for this task to isolate VMs between concurrent tasks (optional, auto-generated if not provided)
|
| stream_callback: Optional callback invoked with each text delta during streaming.
|
| Used by the TTS pipeline to start audio generation before the full response.
|
| When None (default), API calls use the standard non-streaming path.
|
| persist_user_message: Optional clean user message to store in
|
| transcripts/history when user_message contains API-only
|
| synthetic prefixes.
|
| persist_user_timestamp: Optional platform event timestamp to store
|
| as metadata on that persisted user message.
|
| persist_user_display_kind: Optional presentation type for a
|
| synthesized user turn (``auto_continue``, ``model_switch``, β¦).
|
| Display-only: transcript surfaces render the row as a timeline
|
| event instead of a user bubble, while the model still receives
|
| the message unchanged.
|
| persist_user_display_metadata: Optional payload for that event
|
| (e.g. a delegation's task count).
|
| or queuing follow-up prefetch work.
|
|
|
| Returns:
|
| Dict: Complete conversation result with final response and message history
|
| """
|
| if moa_config is None:
|
| try:
|
| from hermes_cli.moa_config import decode_moa_turn
|
|
|
| _decoded_message, _decoded_moa_config = decode_moa_turn(user_message)
|
| if _decoded_moa_config is not None:
|
| user_message = _decoded_message
|
| moa_config = _decoded_moa_config
|
| if persist_user_message is None:
|
| persist_user_message = _decoded_message
|
| except Exception:
|
| pass
|
|
|
|
|
|
|
|
|
| agent._last_compaction_in_place = False
|
| agent._last_compression_attempt_recorded = False
|
| agent._last_compression_attempt_in_place = None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _ctx = build_turn_context(
|
| agent,
|
| user_message,
|
| system_message,
|
| conversation_history,
|
| task_id,
|
| stream_callback,
|
| persist_user_message,
|
| persist_user_timestamp,
|
| persist_user_display_kind=persist_user_display_kind,
|
| persist_user_display_metadata=persist_user_display_metadata,
|
| restore_or_build_system_prompt=_restore_or_build_system_prompt,
|
| install_safe_stdio=_install_safe_stdio,
|
| sanitize_surrogates=_sanitize_surrogates,
|
| summarize_user_message_for_log=_summarize_user_message_for_log,
|
| set_session_context=set_session_context,
|
| set_current_write_origin=set_current_write_origin,
|
| ra=_ra,
|
|
|
|
|
| moa_active=bool(moa_config),
|
| )
|
| user_message = _ctx.user_message
|
| original_user_message = _ctx.original_user_message
|
| messages = _ctx.messages
|
| conversation_history = _ctx.conversation_history
|
| active_system_prompt = _ctx.active_system_prompt
|
| effective_task_id = _ctx.effective_task_id
|
| turn_id = _ctx.turn_id
|
| current_turn_user_idx = _ctx.current_turn_user_idx
|
| _should_review_memory = _ctx.should_review_memory
|
| _plugin_user_context = _ctx.plugin_user_context
|
| _ext_prefetch_cache = _ctx.ext_prefetch_cache
|
|
|
|
|
|
|
| agent._delivered_interim_texts = set()
|
|
|
|
|
| agent._incremental_persistence_failed = False
|
|
|
|
|
| api_call_count = 0
|
| final_response = None
|
| interrupted = False
|
| failed = False
|
| codex_ack_continuations = 0
|
| length_continue_retries = 0
|
| truncated_tool_call_retries = 0
|
| truncated_response_parts: List[str] = []
|
| compression_attempts = 0
|
|
|
|
|
|
|
|
|
|
|
|
|
| max_compression_attempts = getattr(agent, "max_compression_attempts", 3)
|
| _last_preflight_pressure: Optional[int] = None
|
| _preflight_compression_blocked = _ctx.preflight_compression_blocked
|
| _turn_exit_reason = "unknown"
|
|
|
|
|
|
|
|
|
| _pending_verification_response = None
|
|
|
|
|
|
|
|
|
|
|
| _pending_verification_response_previewed = False
|
|
|
|
|
|
|
| pending_moa_prepared_request = None
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._auth_pool_refresh_counts = {}
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._last_turn_usage = None
|
|
|
|
|
|
|
|
|
|
|
|
|
| if agent.api_mode == "codex_app_server":
|
| return agent._run_codex_app_server_turn(
|
| user_message=user_message,
|
| original_user_message=original_user_message,
|
| messages=messages,
|
| effective_task_id=effective_task_id,
|
| should_review_memory=_should_review_memory,
|
| )
|
|
|
| while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call:
|
| _redirect_text = agent._drain_pending_redirect()
|
| if _redirect_text:
|
| _apply_active_turn_redirect(agent, messages, _redirect_text)
|
| if isinstance(original_user_message, str):
|
| original_user_message = (
|
| f"{original_user_message}\n\n"
|
| f"User correction during the turn: {_redirect_text}"
|
| )
|
| agent._persist_session(messages, conversation_history)
|
|
|
|
|
| agent._checkpoint_mgr.new_turn()
|
|
|
|
|
| if agent._interrupt_requested:
|
| interrupted = True
|
| _turn_exit_reason = "interrupted_by_user"
|
| if not agent.quiet_mode:
|
| agent._safe_print("\nβ‘ Breaking out of tool loop due to interrupt...")
|
| break
|
|
|
| api_call_count += 1
|
| agent._api_call_count = api_call_count
|
| agent._touch_activity(f"starting API call #{api_call_count}")
|
|
|
|
|
|
|
|
|
| if agent._budget_grace_call:
|
| agent._budget_grace_call = False
|
| elif not agent.iteration_budget.consume():
|
| _turn_exit_reason = "budget_exhausted"
|
| if not agent.quiet_mode:
|
| agent._safe_print(f"\nβ οΈ Iteration budget exhausted ({agent.iteration_budget.used}/{agent.iteration_budget.max_total} iterations used)")
|
| break
|
|
|
|
|
| if agent.step_callback is not None:
|
| try:
|
| prev_tools = []
|
| for _idx, _m in enumerate(reversed(messages)):
|
| if _m.get("role") == "assistant" and _m.get("tool_calls"):
|
| _fwd_start = len(messages) - _idx
|
| _results_by_id = {}
|
| for _tm in messages[_fwd_start:]:
|
| if _tm.get("role") != "tool":
|
| break
|
| _tcid = _tm.get("tool_call_id")
|
| if _tcid:
|
| _results_by_id[_tcid] = _tm.get("content", "")
|
| prev_tools = [
|
| {
|
| "name": tc["function"]["name"],
|
| "result": _results_by_id.get(tc.get("id")),
|
| "arguments": tc["function"].get("arguments"),
|
| }
|
| for tc in _m["tool_calls"]
|
| if isinstance(tc, dict)
|
| ]
|
| break
|
| agent.step_callback(api_call_count, prev_tools)
|
| except Exception as _step_err:
|
| logger.debug("step_callback error (iteration %s): %s", api_call_count, _step_err)
|
|
|
|
|
|
|
| if (agent._skill_nudge_interval > 0
|
| and "skill_manage" in agent.valid_tool_names):
|
| agent._iters_since_skill += 1
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _pre_api_steer = agent._drain_pending_steer()
|
| if _pre_api_steer:
|
| _injected = False
|
| for _si in range(len(messages) - 1, -1, -1):
|
| _sm = messages[_si]
|
| if isinstance(_sm, dict) and _sm.get("role") == "tool":
|
| from agent.prompt_builder import format_steer_marker
|
| marker = format_steer_marker(_pre_api_steer)
|
| existing = _sm.get("content", "")
|
| if isinstance(existing, str):
|
| _sm["content"] = existing + marker
|
| else:
|
|
|
| try:
|
| blocks = list(existing) if existing else []
|
| blocks.append({"type": "text", "text": marker})
|
| _sm["content"] = blocks
|
| except Exception:
|
| pass
|
| _injected = True
|
| logger.debug(
|
| "Pre-API-call steer drain: injected into tool msg at index %d",
|
| _si,
|
| )
|
| break
|
| if not _injected:
|
|
|
|
|
| _lock = getattr(agent, "_pending_steer_lock", None)
|
| if _lock is not None:
|
| with _lock:
|
| if agent._pending_steer:
|
| agent._pending_steer = agent._pending_steer + "\n" + _pre_api_steer
|
| else:
|
| agent._pending_steer = _pre_api_steer
|
| else:
|
| existing = getattr(agent, "_pending_steer", None)
|
| agent._pending_steer = (existing + "\n" + _pre_api_steer) if existing else _pre_api_steer
|
|
|
|
|
|
|
|
|
|
|
|
|
| request_logger = getattr(agent, "logger", None) or logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
|
| _sanitize_cursor = getattr(agent, "_sanitize_args_cursor", None)
|
| if _sanitize_cursor is None:
|
| _sanitize_cursor = {}
|
| try:
|
| agent._sanitize_args_cursor = _sanitize_cursor
|
| except Exception:
|
| pass
|
| repaired_tool_calls = agent._sanitize_tool_call_arguments(
|
| messages,
|
| logger=request_logger,
|
| session_id=agent.session_id,
|
| cursor=_sanitize_cursor,
|
| )
|
| if repaired_tool_calls > 0:
|
| request_logger.info(
|
| "Sanitized %s corrupted tool_call arguments before request (session=%s)",
|
| repaired_tool_calls,
|
| agent.session_id or "-",
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| from agent.agent_runtime_helpers import repair_message_sequence_with_cursor
|
| repaired_seq = repair_message_sequence_with_cursor(agent, messages)
|
| if repaired_seq > 0:
|
| request_logger.info(
|
| "Repaired %s message-alternation violations before request (session=%s)",
|
| repaired_seq,
|
| agent.session_id or "-",
|
| )
|
|
|
| api_messages = []
|
| for idx, msg in enumerate(messages):
|
| api_msg = msg.copy()
|
|
|
|
|
|
|
|
|
|
|
|
|
| _api_content = api_msg.pop("api_content", None)
|
|
|
|
|
|
|
|
|
|
|
| api_msg.pop("display_kind", None)
|
| api_msg.pop("display_metadata", None)
|
|
|
|
|
|
|
|
|
|
|
| api_msg.pop("_row_id", None)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if idx == current_turn_user_idx and msg.get("role") == "user":
|
| if isinstance(_api_content, str) and _api_content:
|
|
|
|
|
|
|
|
|
|
|
| api_msg["content"] = _api_content
|
| else:
|
|
|
| _composed = compose_user_api_content(
|
| api_msg.get("content", ""),
|
| _ext_prefetch_cache,
|
| _plugin_user_context,
|
| )
|
| if _composed is not None:
|
| api_msg["content"] = _composed
|
| elif (
|
| isinstance(_api_content, str)
|
| and _api_content
|
| and msg.get("role") in ("user", "assistant")
|
| ):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| api_msg["content"] = _api_content
|
|
|
|
|
|
|
| agent._copy_reasoning_content_for_api(msg, api_msg)
|
|
|
|
|
|
|
| if "reasoning" in api_msg:
|
| api_msg.pop("reasoning")
|
|
|
| if "finish_reason" in api_msg:
|
| api_msg.pop("finish_reason")
|
|
|
| api_msg.pop("_thinking_prefill", None)
|
|
|
|
|
|
|
|
|
| if agent._should_sanitize_tool_calls():
|
|
|
|
|
|
|
|
|
| _sanitize_model = agent.model
|
| if agent.provider == "moa":
|
| if moa_config:
|
| _agg = moa_config.get("aggregator") or {}
|
| if _agg.get("model"):
|
| _sanitize_model = _agg["model"]
|
| if _sanitize_model == agent.model:
|
|
|
|
|
|
|
|
|
|
|
|
|
| _moa_client = getattr(agent, "client", None)
|
| _agg_slot = getattr(_moa_client, "last_aggregator_slot", None)
|
| if _agg_slot and _agg_slot.get("model"):
|
| _sanitize_model = _agg_slot["model"]
|
| agent._sanitize_tool_calls_for_strict_api(api_msg, model=_sanitize_model)
|
|
|
|
|
| api_messages.append(api_msg)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| effective_system = active_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
|
|
|
| if moa_config:
|
| try:
|
| from agent.message_content import flatten_message_text as _flatten_mt
|
| from agent.moa_loop import _preset_temperature, aggregate_moa_context
|
|
|
| _moa_context = aggregate_moa_context(
|
| user_prompt=(
|
| original_user_message
|
| if isinstance(original_user_message, str)
|
|
|
|
|
|
|
|
|
| else _flatten_mt(original_user_message)
|
| ),
|
| api_messages=api_messages,
|
| reference_models=moa_config.get("reference_models") or [],
|
| aggregator=moa_config.get("aggregator") or {},
|
| temperature=_preset_temperature(moa_config, "reference_temperature"),
|
| aggregator_temperature=_preset_temperature(moa_config, "aggregator_temperature"),
|
| reference_max_tokens=moa_config.get("reference_max_tokens"),
|
|
|
|
|
| reference_timeout=(
|
| float(moa_config["reference_timeout"])
|
| if moa_config.get("reference_timeout")
|
| else None
|
| ),
|
| degraded_reference_policy=str(
|
| moa_config.get("degraded_reference_policy") or "loud"
|
| ),
|
| agent=agent,
|
| )
|
| if _moa_context:
|
| for _msg in reversed(api_messages):
|
| if _msg.get("role") == "user":
|
| _base = _msg.get("content", "")
|
| if isinstance(_base, str):
|
| _msg["content"] = _base + "\n\n" + _moa_context
|
| elif isinstance(_base, list):
|
|
|
|
|
|
|
| _msg["content"] = [
|
| *_base,
|
| {"type": "text", "text": "\n\n" + _moa_context},
|
| ]
|
| break
|
| except Exception as _moa_exc:
|
| logger.warning("MoA context aggregation failed: %s", _moa_exc)
|
|
|
|
|
|
|
| if agent.prefill_messages:
|
| sys_offset = 1 if (api_messages and api_messages[0].get("role") == "system") else 0
|
| for idx, pfm in enumerate(agent.prefill_messages):
|
| api_messages.insert(sys_offset + idx, pfm.copy())
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _sel_incoming = (
|
| messages[current_turn_user_idx]
|
| if 0 <= current_turn_user_idx < len(messages)
|
| else None
|
| )
|
| api_messages = _apply_context_engine_selection(
|
| agent,
|
| api_messages,
|
| messages,
|
| _sel_incoming,
|
| logger=request_logger,
|
| )
|
|
|
|
|
|
|
|
|
|
|
| api_messages = agent._sanitize_api_messages(api_messages)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| api_messages = agent._drop_thinking_only_and_merge_users(
|
| api_messages,
|
| drop_codex_reasoning_items=agent.api_mode != "codex_responses",
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| for am in api_messages:
|
| if isinstance(am.get("content"), str):
|
| am["content"] = am["content"].strip()
|
| _canonicalize_api_tool_calls(api_messages)
|
|
|
|
|
|
|
|
|
|
|
| _sanitize_messages_surrogates(api_messages)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| tools_for_api = agent.tools
|
| if agent._use_prompt_caching and agent.provider != "moa":
|
| _static_system_prefix = getattr(agent, "_cached_system_prompt_static", None)
|
| _initial_cache_plan = build_prompt_cache_plan(
|
| api_messages,
|
| tools_for_api,
|
| cache_ttl=agent._cache_ttl,
|
| native_anthropic=agent._use_native_cache_layout,
|
| static_system_prefix=(
|
| _static_system_prefix
|
| if isinstance(_static_system_prefix, str)
|
| else None
|
| ),
|
| direct_native_tool_cache=agent._direct_native_anthropic_tool_cache_capability(),
|
| )
|
| api_messages = _initial_cache_plan.messages
|
| tools_for_api = _initial_cache_plan.tools
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _moa_prepared_request = None
|
| if agent.provider == "moa":
|
| _moa_completions = getattr(getattr(agent.client, "chat", None), "completions", None)
|
| if pending_moa_prepared_request is not None:
|
| _rebase_moa_request = getattr(_moa_completions, "rebase_prepared_request", None)
|
| if callable(_rebase_moa_request):
|
| _moa_prepared_request = _rebase_moa_request(
|
| pending_moa_prepared_request, api_messages
|
| )
|
| pending_moa_prepared_request = None
|
| if _moa_prepared_request is None:
|
| _prepare_moa_request = getattr(_moa_completions, "prepare", None)
|
| if callable(_prepare_moa_request):
|
| _moa_prepared_request = _prepare_moa_request(api_messages)
|
| if _moa_prepared_request is not None:
|
| api_messages = _moa_prepared_request["messages"]
|
|
|
|
|
|
|
|
|
|
|
|
|
| approx_tokens = estimate_messages_tokens_rough(api_messages)
|
| request_pressure_tokens = approx_tokens + (
|
| _estimate_tools_tokens_rough(agent.tools) if agent.tools else 0
|
| )
|
| total_chars = approx_tokens * 4
|
|
|
| _runtime_context_error = _ollama_context_limit_error(
|
| agent, request_pressure_tokens
|
| )
|
| if _runtime_context_error:
|
| final_response = _runtime_context_error
|
| failed = True
|
| _turn_exit_reason = "ollama_runtime_context_too_small"
|
| messages.append({"role": "assistant", "content": final_response})
|
| agent._emit_status("β Ollama runtime context is too small for Hermes tool use")
|
| api_call_count -= 1
|
| agent._api_call_count = api_call_count
|
| try:
|
| agent.iteration_budget.refund()
|
| except Exception:
|
| pass
|
| break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _compressor = agent.context_compressor
|
| _preflight_threshold = int(
|
| getattr(_compressor, "threshold_tokens", 0) or 0
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
| _previous_preflight_pressure = _last_preflight_pressure
|
| _last_preflight_pressure = None
|
| if (
|
| _previous_preflight_pressure is not None
|
| and request_pressure_tokens >= _preflight_threshold
|
| and not _compression_warrants_another_preflight_pass(
|
| _previous_preflight_pressure,
|
| request_pressure_tokens,
|
| _preflight_threshold,
|
| )
|
| ):
|
|
|
|
|
|
|
|
|
| _preflight_compression_blocked = True
|
| logger.warning(
|
| "Pre-API compression made insufficient progress: ~%s -> "
|
| "~%s request tokens; skipping additional preflight passes",
|
| f"{_previous_preflight_pressure:,}",
|
| f"{request_pressure_tokens:,}",
|
| )
|
| _defer_preflight = getattr(
|
| _compressor, "should_defer_preflight_to_real_usage", lambda _t: False
|
| )
|
| _compression_cooldown = getattr(
|
| _compressor, "get_active_compression_failure_cooldown", lambda: None
|
| )()
|
| if (
|
| agent.compression_enabled
|
| and len(messages) > 1
|
| and compression_attempts < max_compression_attempts
|
| and not _preflight_compression_blocked
|
| and not _defer_preflight(request_pressure_tokens)
|
| and not _compression_cooldown
|
| and _compressor.should_compress(request_pressure_tokens)
|
| ):
|
| if _moa_prepared_request is not None:
|
| pending_moa_prepared_request = _moa_prepared_request
|
| compression_attempts += 1
|
|
|
|
|
|
|
|
|
|
|
|
|
| _clear_warn = getattr(agent, "_clear_context_overflow_warn", None)
|
| if callable(_clear_warn):
|
| _clear_warn()
|
| logger.info(
|
| "Pre-API compression: ~%s request tokens >= %s threshold "
|
| "(context=%s, attempt=%s/%s)",
|
| f"{request_pressure_tokens:,}",
|
| f"{int(getattr(_compressor, 'threshold_tokens', 0) or 0):,}",
|
| f"{int(getattr(_compressor, 'context_length', 0) or 0):,}"
|
| if getattr(_compressor, "context_length", 0) else "unknown",
|
| compression_attempts,
|
| max_compression_attempts,
|
| )
|
| _pre_api_status = automatic_compaction_status_message(
|
| _compressor,
|
| phase="pre_api",
|
| default_message=PRE_API_COMPRESSION_STATUS_TEMPLATE.format(
|
| tokens=request_pressure_tokens
|
| ),
|
| approx_tokens=request_pressure_tokens,
|
| threshold_tokens=int(
|
| getattr(_compressor, "threshold_tokens", 0) or 0
|
| ),
|
| context_length=int(
|
| getattr(_compressor, "context_length", 0) or 0
|
| ),
|
| model=agent.model,
|
| attempt=compression_attempts,
|
| max_attempts=max_compression_attempts,
|
| )
|
| if _pre_api_status:
|
| agent._emit_status(_pre_api_status)
|
| _last_preflight_pressure = request_pressure_tokens
|
| _pre_api_input = messages
|
| messages, active_system_prompt = agent._compress_context(
|
| messages,
|
| system_message,
|
| approx_tokens=request_pressure_tokens,
|
| task_id=effective_task_id,
|
| )
|
| if messages is _pre_api_input and compression_skipped_due_to_lock(agent):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| compression_attempts -= 1
|
| _last_preflight_pressure = None
|
| if pending_moa_prepared_request is _moa_prepared_request:
|
| pending_moa_prepared_request = None
|
| else:
|
|
|
|
|
|
|
| agent._empty_content_retries = 0
|
| agent._thinking_prefill_retries = 0
|
| agent._last_content_with_tools = None
|
| agent._last_content_tools_all_housekeeping = False
|
| agent._mute_post_response = False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| conversation_history = conversation_history_after_compression(
|
| agent, messages, conversation_history
|
| )
|
| api_call_count -= 1
|
| agent._api_call_count = api_call_count
|
| agent.iteration_budget.refund()
|
| continue
|
| elif (
|
| agent.compression_enabled
|
| and len(messages) > 1
|
| and compression_attempts < max_compression_attempts
|
| and not _defer_preflight(request_pressure_tokens)
|
| and _compression_cooldown
|
| ):
|
|
|
|
|
|
|
|
|
|
|
|
|
| _block_reason = None
|
| try:
|
| _block_reason = _compressor.should_compress_info(
|
| request_pressure_tokens
|
| )[1]
|
| except Exception:
|
| _block_reason = None
|
| if _block_reason:
|
| agent._warn_context_overflow_blocked(
|
| _block_reason,
|
| request_pressure_tokens,
|
| int(getattr(_compressor, "threshold_tokens", 0) or 0),
|
| )
|
|
|
|
|
| thinking_spinner = None
|
|
|
| if not agent.quiet_mode:
|
| agent._vprint(f"\n{agent.log_prefix}π Making API call #{api_call_count}/{agent.max_iterations}...")
|
| agent._vprint(f"{agent.log_prefix} π Request size: {len(api_messages)} messages, ~{approx_tokens:,} tokens (~{total_chars:,} chars)")
|
| agent._vprint(f"{agent.log_prefix} π§ Available tools: {len(agent.tools) if agent.tools else 0}")
|
| else:
|
|
|
| face = random.choice(KawaiiSpinner.get_thinking_faces())
|
| verb = random.choice(KawaiiSpinner.get_thinking_verbs())
|
| if agent.thinking_callback:
|
|
|
|
|
| agent.thinking_callback(f"{face} {verb}...")
|
| elif not agent._has_stream_consumers() and agent._should_start_quiet_spinner():
|
|
|
|
|
| spinner_type = random.choice(['brain', 'sparkle', 'pulse', 'moon', 'star'])
|
| thinking_spinner = KawaiiSpinner(f"{face} {verb}...", spinner_type=spinner_type, print_fn=agent._print_fn)
|
| thinking_spinner.start()
|
|
|
|
|
| if agent.verbose_logging:
|
| logging.debug(f"API Request - Model: {agent.model}, Messages: {len(messages)}, Tools: {len(agent.tools) if agent.tools else 0}")
|
| logging.debug(f"Last message role: {messages[-1]['role'] if messages else 'none'}")
|
| logging.debug(f"Total message size: ~{approx_tokens:,} tokens")
|
|
|
| api_start_time = time.time()
|
| retry_count = 0
|
| max_retries = agent._api_max_retries
|
| _retry = TurnRetryState()
|
|
|
| finish_reason = "stop"
|
| response = None
|
| api_kwargs = None
|
| api_request_id = f"{turn_id}:api:{api_call_count}"
|
| agent._current_api_request_id = api_request_id
|
|
|
| while retry_count < max_retries:
|
|
|
|
|
|
|
|
|
|
|
| if agent.provider == "nous":
|
| try:
|
| from agent.nous_rate_guard import (
|
| nous_rate_limit_remaining,
|
| format_remaining as _fmt_nous_remaining,
|
| )
|
| _nous_remaining = nous_rate_limit_remaining()
|
| if _nous_remaining is not None and _nous_remaining > 0:
|
| _nous_msg = (
|
| f"Nous Portal rate limit active β "
|
| f"resets in {_fmt_nous_remaining(_nous_remaining)}."
|
| )
|
| agent._buffer_vprint(
|
| f"β³ {_nous_msg} Trying fallback..."
|
| )
|
| agent._buffer_status(f"β³ {_nous_msg}")
|
| if agent._try_activate_fallback():
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| continue
|
|
|
|
|
| agent._flush_status_buffer()
|
| agent._persist_session(messages, conversation_history)
|
| return {
|
| "final_response": (
|
| f"β³ {_nous_msg}\n\n"
|
| "No fallback provider available. "
|
| "Try again after the reset, or add a "
|
| "fallback provider in config.yaml."
|
| ),
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "failed": True,
|
| "error": _nous_msg,
|
| }
|
| except ImportError:
|
| pass
|
| except Exception:
|
| pass
|
|
|
| try:
|
| agent._reset_stream_delivery_tracking()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._reapply_reasoning_echo_for_provider(api_messages)
|
|
|
|
|
|
|
|
|
| api_messages, _moa_prepared_request, tools_for_api = (
|
| _redecorate_prompt_cache_for_provider(
|
| agent,
|
| api_messages,
|
| system_message=system_message,
|
| moa_prepared=_moa_prepared_request,
|
| tools_for_api=tools_for_api,
|
| )
|
| )
|
| if tools_for_api == agent.tools:
|
| api_kwargs = agent._build_api_kwargs(api_messages)
|
| else:
|
| api_kwargs = agent._build_api_kwargs(
|
| api_messages,
|
| tools_for_api=tools_for_api,
|
| )
|
| if agent._force_ascii_payload:
|
| _sanitize_structure_non_ascii(api_kwargs)
|
| if agent.api_mode == "codex_responses":
|
| api_kwargs = agent._get_transport().preflight_kwargs(
|
| api_kwargs,
|
| allow_stream=False,
|
| is_github_responses=agent._is_copilot_url(),
|
| sanitize_harmony_tokens=agent._is_codex_backend(),
|
| )
|
|
|
|
|
|
|
| if getattr(agent, "_is_user_initiated_turn", False) and agent._is_copilot_url():
|
| _xh = dict(api_kwargs.get("extra_headers") or {})
|
| _xh["x-initiator"] = "user"
|
| api_kwargs["extra_headers"] = _xh
|
| agent._is_user_initiated_turn = False
|
| try:
|
| from hermes_cli.middleware import apply_llm_request_middleware
|
|
|
| _llm_request_mw = apply_llm_request_middleware(
|
| api_kwargs,
|
| task_id=effective_task_id,
|
| turn_id=turn_id,
|
| api_request_id=api_request_id,
|
| session_id=agent.session_id or "",
|
| platform=agent.platform or "",
|
| model=agent.model,
|
| provider=agent.provider,
|
| base_url=agent.base_url,
|
| api_mode=agent.api_mode,
|
| api_call_count=api_call_count,
|
| )
|
| api_kwargs = _llm_request_mw.payload
|
| _original_api_kwargs = _llm_request_mw.original_payload
|
| _llm_middleware_trace = _llm_request_mw.trace
|
| except Exception:
|
| _original_api_kwargs = dict(api_kwargs)
|
| _llm_middleware_trace = []
|
|
|
| try:
|
| from hermes_cli.lifecycle import (
|
| has_hook,
|
| invoke_hook as _invoke_hook,
|
| )
|
| if has_hook("pre_api_request"):
|
| request_messages = api_kwargs.get("messages")
|
| if not isinstance(request_messages, list):
|
| request_messages = api_kwargs.get("input")
|
| if not isinstance(request_messages, list):
|
| request_messages = api_messages
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _request_payload = agent._api_request_payload_for_hook(api_kwargs)
|
| _invoke_hook(
|
| "pre_api_request",
|
| task_id=effective_task_id,
|
| turn_id=turn_id,
|
| api_request_id=api_request_id,
|
| session_id=agent.session_id or "",
|
| user_message=original_user_message,
|
| conversation_history=list(messages),
|
| platform=agent.platform or "",
|
| model=agent.model,
|
| provider=agent.provider,
|
| base_url=agent.base_url,
|
| api_mode=agent.api_mode,
|
| api_call_count=api_call_count,
|
| retry_count=retry_count,
|
| request_messages=list(request_messages)
|
| if isinstance(request_messages, list)
|
| else [],
|
| message_count=len(api_messages),
|
| tool_count=len(agent.tools or []),
|
| approx_input_tokens=approx_tokens,
|
| request_char_count=total_chars,
|
| max_tokens=agent.max_tokens,
|
| started_at=api_start_time,
|
| middleware_trace=list(_llm_middleware_trace),
|
| request=_request_payload,
|
| )
|
| except Exception:
|
| pass
|
|
|
| if env_var_enabled("HERMES_DUMP_REQUESTS"):
|
| agent._dump_api_request_debug(api_kwargs, reason="preflight")
|
|
|
|
|
|
|
|
|
| if _moa_prepared_request is not None and agent.provider == "moa":
|
| api_kwargs["_moa_prepared_request"] = _moa_prepared_request
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| def _stop_spinner():
|
| nonlocal thinking_spinner
|
| if thinking_spinner:
|
| thinking_spinner.stop("")
|
| thinking_spinner = None
|
| if agent.thinking_callback:
|
| agent.thinking_callback("")
|
|
|
| _use_streaming = True
|
|
|
|
|
|
|
| if getattr(agent, "_disable_streaming", False):
|
| _use_streaming = False
|
|
|
|
|
|
|
|
|
| elif (
|
| agent.provider in {"copilot-acp"}
|
| or str(agent.base_url or "").lower().startswith("acp://copilot")
|
| or str(agent.base_url or "").lower().startswith("acp+tcp://")
|
| ):
|
| _use_streaming = False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| elif agent.provider == "moa" and not agent._has_stream_consumers():
|
| _use_streaming = False
|
| elif not agent._has_stream_consumers():
|
|
|
|
|
|
|
| from unittest.mock import Mock
|
| if isinstance(getattr(agent, "client", None), Mock):
|
| _use_streaming = False
|
|
|
| def _perform_api_call(next_api_kwargs):
|
| if agent.api_mode == "codex_responses":
|
| next_api_kwargs = agent._get_transport().preflight_kwargs(
|
| next_api_kwargs,
|
| allow_stream=False,
|
| is_github_responses=agent._is_copilot_url(),
|
| sanitize_harmony_tokens=agent._is_codex_backend(),
|
| )
|
| if _use_streaming:
|
| return agent._interruptible_streaming_api_call(
|
| next_api_kwargs, on_first_delta=_stop_spinner
|
| )
|
| from agent import relay_llm
|
|
|
| return relay_llm.execute(
|
| next_api_kwargs,
|
| agent._interruptible_api_call,
|
| session_id=str(agent.session_id or ""),
|
| name=str(agent.provider or "provider"),
|
| model_name=str(agent.model or ""),
|
| metadata={
|
| "api_mode": agent.api_mode,
|
| "api_request_id": api_request_id,
|
| "call_role": (
|
| "delegated"
|
| if getattr(agent, "is_subagent", False)
|
| else "fallback"
|
| if int(getattr(agent, "_fallback_index", 0) or 0) > 0
|
| else "primary"
|
| ),
|
| "retry_count": retry_count,
|
| },
|
| defer_logical_completion=True,
|
| )
|
|
|
| from hermes_cli.middleware import run_llm_execution_middleware
|
|
|
| _model_request_active = getattr(agent, "_model_request_active", None)
|
| _redirect_lock = getattr(agent, "_pending_redirect_lock", None)
|
| if _redirect_lock is not None:
|
| with _redirect_lock:
|
| if _model_request_active is not None:
|
| _model_request_active.set()
|
| elif _model_request_active is not None:
|
| _model_request_active.set()
|
| _redirect_crossed_response = False
|
| try:
|
| response = run_llm_execution_middleware(
|
| api_kwargs,
|
| _perform_api_call,
|
| original_request=_original_api_kwargs,
|
| task_id=effective_task_id,
|
| turn_id=turn_id,
|
| api_request_id=api_request_id,
|
| session_id=agent.session_id or "",
|
| platform=agent.platform or "",
|
| model=agent.model,
|
| provider=agent.provider,
|
| base_url=agent.base_url,
|
| api_mode=agent.api_mode,
|
| api_call_count=api_call_count,
|
| middleware_trace=list(_llm_middleware_trace),
|
| )
|
| finally:
|
| if _redirect_lock is not None:
|
| with _redirect_lock:
|
| if _model_request_active is not None:
|
| _model_request_active.clear()
|
| _redirect_crossed_response = bool(
|
| agent._pending_redirect
|
| )
|
| else:
|
| if _model_request_active is not None:
|
| _model_request_active.clear()
|
| _redirect_crossed_response = agent._has_pending_redirect()
|
| if _redirect_crossed_response:
|
|
|
|
|
|
|
|
|
| if thinking_spinner:
|
| thinking_spinner.stop("")
|
| thinking_spinner = None
|
| if agent.thinking_callback:
|
| agent.thinking_callback("")
|
| if agent.clear_interrupt(preserve_redirect=True):
|
| _retry.restart_with_redirected_messages = True
|
| else:
|
| interrupted = True
|
| break
|
|
|
| api_duration = time.time() - api_start_time
|
|
|
|
|
|
|
| if thinking_spinner:
|
| thinking_spinner.stop("")
|
| thinking_spinner = None
|
| if agent.thinking_callback:
|
| agent.thinking_callback("")
|
|
|
| if not agent.quiet_mode:
|
| agent._vprint(f"{agent.log_prefix}β±οΈ API call completed in {api_duration:.2f}s")
|
|
|
| if agent.verbose_logging:
|
|
|
| resp_model = getattr(response, 'model', 'N/A') if response else 'N/A'
|
| logging.debug(f"API Response received - Model: {resp_model}, Usage: {response.usage if hasattr(response, 'usage') else 'N/A'}")
|
|
|
|
|
| response_invalid = False
|
| error_details = []
|
| if agent.api_mode == "codex_responses":
|
| _ct_v = agent._get_transport()
|
| if not _ct_v.validate_response(response):
|
| if response is None:
|
| response_invalid = True
|
| error_details.append("response is None")
|
| else:
|
|
|
|
|
|
|
| _codex_resp_status = str(getattr(response, "status", "") or "").strip().lower()
|
| if _codex_resp_status in {"failed", "cancelled"}:
|
| _codex_error_obj = getattr(response, "error", None)
|
| _codex_error_msg = (
|
| _codex_error_obj.get("message") if isinstance(_codex_error_obj, dict)
|
| else str(_codex_error_obj) if _codex_error_obj
|
| else f"Responses API returned status '{_codex_resp_status}'"
|
| )
|
| logger.warning(
|
| "Codex response status='%s' (error=%s). Routing to fallback. %s",
|
| _codex_resp_status, _codex_error_msg,
|
| agent._client_log_context(),
|
| )
|
| response_invalid = True
|
| error_details.append(f"response.status={_codex_resp_status}: {_codex_error_msg}")
|
| else:
|
|
|
|
|
| _out_text = getattr(response, "output_text", None)
|
| _out_text_stripped = _out_text.strip() if isinstance(_out_text, str) else ""
|
| if _out_text_stripped:
|
| logger.debug(
|
| "Codex response.output is empty but output_text is present "
|
| "(%d chars); deferring to normalization.",
|
| len(_out_text_stripped),
|
| )
|
| else:
|
| _resp_status = getattr(response, "status", None)
|
| _resp_incomplete = getattr(response, "incomplete_details", None)
|
| logger.warning(
|
| "Codex response.output is empty after stream backfill "
|
| "(status=%s, incomplete_details=%s, model=%s). %s",
|
| _resp_status, _resp_incomplete,
|
| getattr(response, "model", None),
|
| f"api_mode={agent.api_mode} provider={agent.provider}",
|
| )
|
| response_invalid = True
|
| error_details.append("response.output is empty")
|
| elif agent.api_mode == "anthropic_messages":
|
| _tv = agent._get_transport()
|
| if not _tv.validate_response(response):
|
| response_invalid = True
|
| if response is None:
|
| error_details.append("response is None")
|
| else:
|
| error_details.append("response.content invalid (not a non-empty list)")
|
| elif agent.api_mode == "bedrock_converse":
|
| _btv = agent._get_transport()
|
| if not _btv.validate_response(response):
|
| response_invalid = True
|
| if response is None:
|
| error_details.append("response is None")
|
| else:
|
| error_details.append("Bedrock response invalid (no output or choices)")
|
| else:
|
| _ctv = agent._get_transport()
|
| if not _ctv.validate_response(response):
|
| response_invalid = True
|
| if response is None:
|
| error_details.append("response is None")
|
| elif not hasattr(response, 'choices'):
|
| error_details.append("response has no 'choices' attribute")
|
| elif response.choices is None:
|
| error_details.append("response.choices is None")
|
| else:
|
| error_details.append("response.choices is empty")
|
|
|
| if response_invalid:
|
| agent._invoke_api_request_error_hook(
|
| task_id=effective_task_id,
|
| turn_id=turn_id,
|
| api_request_id=api_request_id,
|
| api_call_count=api_call_count,
|
| api_start_time=api_start_time,
|
| api_kwargs=api_kwargs,
|
| error_type="InvalidAPIResponse",
|
| error_message=", ".join(error_details) or "Invalid API response",
|
| status_code=getattr(getattr(response, "error", None), "code", None),
|
| retry_count=retry_count,
|
| max_retries=max_retries,
|
| retryable=True,
|
| reason="invalid_response",
|
| )
|
|
|
|
|
| if thinking_spinner:
|
| thinking_spinner.stop("")
|
| thinking_spinner = None
|
| if agent.thinking_callback:
|
| agent.thinking_callback("")
|
|
|
|
|
|
|
| retry_count += 1
|
|
|
|
|
|
|
|
|
| if agent._fallback_index < len(agent._fallback_chain):
|
| agent._buffer_status("β οΈ Empty/malformed response β switching to fallback...")
|
| if agent._try_activate_fallback():
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| continue
|
|
|
|
|
| error_msg = "Unknown"
|
| provider_name = "Unknown"
|
| if response and hasattr(response, 'error') and response.error:
|
| error_msg = str(response.error)
|
|
|
| if hasattr(response.error, 'metadata') and response.error.metadata:
|
| provider_name = response.error.metadata.get('provider_name', 'Unknown')
|
| elif response and hasattr(response, 'message') and response.message:
|
| error_msg = str(response.message)
|
|
|
|
|
| if provider_name == "Unknown" and response and hasattr(response, 'model') and response.model:
|
| provider_name = f"model={response.model}"
|
|
|
|
|
| if provider_name == "Unknown" and response:
|
|
|
| resp_attrs = {k: str(v)[:100] for k, v in vars(response).items() if not k.startswith('_')}
|
| if agent.verbose_logging:
|
| logging.debug(f"Response attributes for invalid response: {resp_attrs}")
|
|
|
|
|
| _resp_error_code = None
|
| if response and hasattr(response, 'error') and response.error:
|
| _code_raw = getattr(response.error, 'code', None)
|
| if _code_raw is None and isinstance(response.error, dict):
|
| _code_raw = response.error.get('code')
|
| if _code_raw is not None:
|
| try:
|
| _resp_error_code = int(_code_raw)
|
| except (TypeError, ValueError):
|
| pass
|
|
|
|
|
|
|
| if _resp_error_code == 524:
|
| _failure_hint = f"upstream provider timed out (Cloudflare 524, {api_duration:.0f}s)"
|
| elif _resp_error_code == 504:
|
| _failure_hint = f"upstream gateway timeout (504, {api_duration:.0f}s)"
|
| elif _resp_error_code == 429:
|
| _failure_hint = "rate limited by upstream provider (429)"
|
| elif _resp_error_code in {500, 502}:
|
| _failure_hint = f"upstream server error ({_resp_error_code}, {api_duration:.0f}s)"
|
| elif _resp_error_code in {503, 529}:
|
| _failure_hint = f"upstream provider overloaded ({_resp_error_code})"
|
| elif _resp_error_code is not None:
|
| _failure_hint = f"upstream error (code {_resp_error_code}, {api_duration:.0f}s)"
|
| elif api_duration < 10:
|
| _failure_hint = f"fast response ({api_duration:.1f}s) β likely rate limited"
|
| elif api_duration > 60:
|
| _failure_hint = f"slow response ({api_duration:.0f}s) β likely upstream timeout"
|
| else:
|
| _failure_hint = f"response time {api_duration:.1f}s"
|
|
|
| agent._buffer_vprint(f"β οΈ Invalid API response (attempt {retry_count}/{max_retries}): {', '.join(error_details)}")
|
| agent._buffer_vprint(f" π’ Provider: {provider_name}")
|
| cleaned_provider_error = agent._clean_error_message(error_msg)
|
| agent._buffer_vprint(f" π Provider message: {cleaned_provider_error}")
|
| agent._buffer_vprint(f" β±οΈ {_failure_hint}")
|
|
|
| if retry_count >= max_retries:
|
|
|
| if agent._has_pending_fallback():
|
| agent._buffer_status(f"β οΈ Max retries ({max_retries}) for invalid responses β trying fallback...")
|
| if agent._try_activate_fallback():
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| continue
|
|
|
| agent._flush_status_buffer()
|
| agent._emit_status(f"β Max retries ({max_retries}) exceeded for invalid responses. Giving up.")
|
| logger.error("%sInvalid API response after %d retries.", agent.log_prefix, max_retries)
|
| agent._persist_session(messages, conversation_history)
|
| _final_response = f"Invalid API response after {max_retries} retries: {_failure_hint}"
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final_response,
|
| "failed": True
|
| }
|
|
|
|
|
| wait_time = jittered_backoff(retry_count, base_delay=5.0, max_delay=120.0)
|
| agent._buffer_vprint(f"β³ Retrying in {wait_time:.1f}s ({_failure_hint})...")
|
| logger.warning("Invalid API response (retry %d/%d): %s | Provider: %s", retry_count, max_retries, ', '.join(error_details), provider_name)
|
|
|
|
|
| sleep_end = time.time() + wait_time
|
| _backoff_touch_counter = 0
|
| while time.time() < sleep_end:
|
| if agent._interrupt_requested:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if agent.clear_interrupt(preserve_redirect=True):
|
| _retry.restart_with_redirected_messages = True
|
| break
|
| agent._vprint(f"{agent.log_prefix}β‘ Interrupt detected during retry wait, aborting.", force=True)
|
| _interrupt_text = f"Operation interrupted during retry ({_failure_hint}, attempt {retry_count}/{max_retries})."
|
| close_interrupted_tool_sequence(messages, _interrupt_text)
|
| agent._persist_session(messages, conversation_history)
|
| agent.clear_interrupt()
|
| return {
|
| "final_response": _interrupt_text,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "interrupted": True,
|
| }
|
| time.sleep(0.2)
|
|
|
|
|
| _backoff_touch_counter += 1
|
| if _backoff_touch_counter % 150 == 0:
|
| agent._touch_activity(
|
| f"retry backoff ({retry_count}/{max_retries}), "
|
| f"{int(sleep_end - time.time())}s remaining"
|
| )
|
| if _retry.restart_with_redirected_messages:
|
| break
|
| continue
|
|
|
| agent._turn_received_provider_response = True
|
|
|
|
|
| if agent.api_mode == "codex_responses":
|
| status = getattr(response, "status", None)
|
| if isinstance(status, str):
|
| status = status.strip().lower()
|
| incomplete_details = getattr(response, "incomplete_details", None)
|
| incomplete_reason = None
|
| if isinstance(incomplete_details, dict):
|
| incomplete_reason = incomplete_details.get("reason")
|
| else:
|
| incomplete_reason = getattr(incomplete_details, "reason", None)
|
| if incomplete_reason is not None:
|
| incomplete_reason = str(incomplete_reason).strip().lower()
|
| if status == "incomplete" and incomplete_reason in {"max_output_tokens", "length"}:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| finish_reason = "incomplete"
|
| elif status == "incomplete" and incomplete_reason == "content_filter":
|
| finish_reason = "content_filter"
|
| else:
|
| finish_reason = "stop"
|
| elif agent.api_mode == "anthropic_messages":
|
| _tfr = agent._get_transport()
|
| finish_reason = _tfr.map_finish_reason(response.stop_reason)
|
| elif agent.api_mode == "bedrock_converse":
|
|
|
| _bt_fr = agent._get_transport()
|
| _bedrock_result = _bt_fr.normalize_response(response)
|
| finish_reason = _bedrock_result.finish_reason
|
| else:
|
| _cc_fr = agent._get_transport()
|
| _finish_result = _cc_fr.normalize_response(response)
|
| finish_reason = _finish_result.finish_reason
|
| assistant_message = _finish_result
|
| if agent._should_treat_stop_as_truncated(
|
| finish_reason,
|
| assistant_message,
|
| messages,
|
| ):
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Treating suspicious Ollama/GLM stop response as truncated",
|
| force=True,
|
| )
|
| finish_reason = "length"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if finish_reason == "content_filter":
|
| _refusal_transport = agent._get_transport()
|
| if agent.api_mode == "anthropic_messages":
|
| _refusal_result = _refusal_transport.normalize_response(
|
| response, strip_tool_prefix=agent._is_anthropic_oauth
|
| )
|
| else:
|
| _refusal_result = _refusal_transport.normalize_response(response)
|
| _refusal_text = (getattr(_refusal_result, "content", None) or "").strip()
|
|
|
|
|
| if not _refusal_text:
|
| _refusal_text = (agent._extract_reasoning(_refusal_result) or "").strip()
|
|
|
| agent._invoke_api_request_error_hook(
|
| task_id=effective_task_id,
|
| turn_id=turn_id,
|
| api_request_id=api_request_id,
|
| api_call_count=api_call_count,
|
| api_start_time=api_start_time,
|
| api_kwargs=api_kwargs,
|
| error_type="ContentPolicyBlocked",
|
| error_message=_refusal_text or "model declined to respond (content_filter)",
|
| status_code=None,
|
| retry_count=retry_count,
|
| max_retries=max_retries,
|
| retryable=False,
|
| reason=FailoverReason.content_policy_blocked.value,
|
| )
|
|
|
| if thinking_spinner:
|
| thinking_spinner.stop("")
|
| thinking_spinner = None
|
| if agent.thinking_callback:
|
| agent.thinking_callback("")
|
|
|
|
|
|
|
|
|
| if agent._has_pending_fallback():
|
| agent._buffer_status(
|
| "β οΈ Model declined to respond (safety refusal) β trying fallback..."
|
| )
|
| if agent._try_activate_fallback():
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| continue
|
|
|
| agent._flush_status_buffer()
|
| _refusal_log = (
|
| _refusal_text[:500] + "..."
|
| if len(_refusal_text) > 500
|
| else _refusal_text
|
| )
|
| logger.warning(
|
| "%sModel declined to respond (finish_reason=content_filter). "
|
| "model=%s provider=%s refusal=%s",
|
| agent.log_prefix, agent.model, agent.provider,
|
| _refusal_log or "(no text)",
|
| )
|
| agent._emit_status(
|
| "β οΈ The model declined to respond to this request (safety refusal)."
|
| )
|
|
|
| _refusal_detail = (
|
| f"Model's explanation: {_refusal_text}"
|
| if _refusal_text
|
| else "The model returned no explanation."
|
| )
|
| _refusal_response = (
|
| "β οΈ The model declined to respond to this request "
|
| "(safety refusal β not a Hermes/gateway failure).\n\n"
|
| f"{_refusal_detail}\n\n"
|
| f"{_CONTENT_POLICY_RECOVERY_HINT}"
|
| )
|
|
|
| agent._cleanup_task_resources(effective_task_id)
|
| agent._persist_session(messages, conversation_history)
|
| return _content_policy_blocked_result(
|
| messages,
|
| api_call_count,
|
| final_response=_refusal_response,
|
| error_detail=_refusal_text or "model declined (content_filter)",
|
| )
|
|
|
| if finish_reason == "length":
|
| if getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID:
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Stream interrupted by network error "
|
| f"(finish_reason='length' on partial-stream-stub)",
|
| force=True,
|
| )
|
| else:
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Response truncated "
|
| f"(finish_reason='length') - model hit max output tokens",
|
| force=True,
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _trunc_msg = None
|
| _trunc_transport = agent._get_transport()
|
| if agent.api_mode == "anthropic_messages":
|
| _trunc_result = _trunc_transport.normalize_response(
|
| response, strip_tool_prefix=agent._is_anthropic_oauth
|
| )
|
| else:
|
| _trunc_result = _trunc_transport.normalize_response(response)
|
| _trunc_msg = _trunc_result
|
|
|
| _trunc_content = getattr(_trunc_msg, "content", None) if _trunc_msg else None
|
| _trunc_has_tool_calls = bool(getattr(_trunc_msg, "tool_calls", None)) if _trunc_msg else False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _has_think_tags = bool(
|
| _trunc_content and re.search(
|
| r'<(?:think|thinking|reasoning|REASONING_SCRATCHPAD)[^>]*>',
|
| _trunc_content,
|
| re.IGNORECASE,
|
| )
|
| )
|
| _thinking_exhausted = (
|
| not _trunc_has_tool_calls
|
| and _has_think_tags
|
| and (
|
| (_trunc_content is not None and not agent._has_content_after_think_block(_trunc_content))
|
| or _trunc_content is None
|
| )
|
| )
|
|
|
| if _thinking_exhausted:
|
| _exhaust_error = (
|
| "Model used all output tokens on reasoning with none left "
|
| "for the response. Try lowering reasoning effort or "
|
| "increasing max_tokens."
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix}π Reasoning exhausted the output token budget β "
|
| f"no visible response was produced.",
|
| force=True,
|
| )
|
|
|
|
|
|
|
| _exhaust_response = (
|
| "β οΈ **Thinking Budget Exhausted**\n\n"
|
| "The model used all its output tokens on reasoning "
|
| "and had none left for the actual response.\n\n"
|
| "To fix this:\n"
|
| "β Lower reasoning effort: `/thinkon low` or `/thinkon minimal`\n"
|
| "β Or switch to a larger/non-reasoning model with `/model`"
|
| )
|
| agent._cleanup_task_resources(effective_task_id)
|
| agent._persist_session(messages, conversation_history)
|
| return {
|
| "final_response": _exhaust_response,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "partial": True,
|
| "error": _exhaust_error,
|
| }
|
|
|
| if agent.api_mode in {"chat_completions", "bedrock_converse", "anthropic_messages"}:
|
| assistant_message = _trunc_msg
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _cf_terminated = getattr(
|
| response, "_content_filter_terminated", False
|
| )
|
| if (
|
| _cf_terminated
|
| and agent._fallback_index < len(agent._fallback_chain)
|
| ):
|
| agent._vprint(
|
| f"{agent.log_prefix}π‘οΈ Content filter terminated "
|
| f"stream β activating fallback provider...",
|
| force=True,
|
| )
|
| agent._emit_status(
|
| "Content filter terminated stream; switching to fallback..."
|
| )
|
| if agent._try_activate_fallback():
|
|
|
|
|
|
|
|
|
| if truncated_response_parts:
|
| messages = agent._get_messages_up_to_last_assistant(messages)
|
| agent._session_messages = messages
|
| length_continue_retries = 0
|
| truncated_response_parts = []
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| _retry.restart_with_rebuilt_messages = True
|
| break
|
|
|
|
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ No fallback provider "
|
| f"configured β retrying with same provider "
|
| f"(may re-hit filter)...",
|
| force=True,
|
| )
|
| if assistant_message is not None and not _trunc_has_tool_calls:
|
| length_continue_retries += 1
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _is_empty_partial_stub = (
|
| getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID
|
| and not getattr(assistant_message, "content", None)
|
| )
|
| if not _is_empty_partial_stub:
|
| interim_msg = agent._build_assistant_message(assistant_message, finish_reason)
|
| messages.append(interim_msg)
|
| if assistant_message.content:
|
| truncated_response_parts.append(assistant_message.content)
|
|
|
| if length_continue_retries < 4:
|
| _is_partial_stream_stub = (
|
| getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID
|
| )
|
| _dropped_tools = getattr(
|
| response, "_dropped_tool_names", None
|
| )
|
|
|
| if _is_partial_stream_stub and _dropped_tools:
|
| _tool_list = ", ".join(_dropped_tools[:3])
|
| agent._vprint(
|
| f"{agent.log_prefix}β» Stream interrupted mid "
|
| f"tool-call ({_tool_list}) β requesting "
|
| f"chunked retry "
|
| f"({length_continue_retries}/4)..."
|
| )
|
| elif _is_partial_stream_stub:
|
| agent._vprint(
|
| f"{agent.log_prefix}β» Stream interrupted β "
|
| f"requesting continuation "
|
| f"({length_continue_retries}/4)..."
|
| )
|
| else:
|
| agent._vprint(
|
| f"{agent.log_prefix}β» Requesting continuation "
|
| f"({length_continue_retries}/4)..."
|
| )
|
|
|
| _continue_content = _get_continuation_prompt(
|
| _is_partial_stream_stub, _dropped_tools
|
| )
|
| continue_msg = {
|
| "role": "user",
|
| "content": _continue_content,
|
| }
|
| messages.append(continue_msg)
|
| agent._session_messages = messages
|
| _retry.restart_with_length_continuation = True
|
| break
|
|
|
| partial_response = agent._strip_think_blocks("".join(truncated_response_parts)).strip()
|
| agent._cleanup_task_resources(effective_task_id)
|
| agent._persist_session(messages, conversation_history)
|
| return {
|
| "final_response": partial_response or None,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "partial": True,
|
| "error": "Response remained truncated after 4 continuation attempts",
|
| }
|
|
|
| if agent.api_mode in {"chat_completions", "bedrock_converse", "anthropic_messages"}:
|
| assistant_message = _trunc_msg
|
| if assistant_message is not None and _trunc_has_tool_calls:
|
| _is_stub_stall = (
|
| getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID
|
| )
|
| if truncated_tool_call_retries < 4:
|
| truncated_tool_call_retries += 1
|
| if _is_stub_stall:
|
|
|
|
|
|
|
| agent._buffer_vprint(
|
| f"β οΈ Stream interrupted mid tool-call β "
|
| f"retrying ({truncated_tool_call_retries}/4)..."
|
| )
|
| else:
|
| agent._buffer_vprint(
|
| f"β οΈ Truncated tool call detected β "
|
| f"retrying API call "
|
| f"({truncated_tool_call_retries}/4)..."
|
| )
|
|
|
|
|
|
|
|
|
|
|
| _tc_boost_base = agent.max_tokens if agent.max_tokens else 4096
|
| _tc_boost = _tc_boost_base * (2 ** truncated_tool_call_retries)
|
| _tc_requested_cap = agent._requested_output_cap_from_api_kwargs(api_kwargs)
|
| if _tc_requested_cap is not None:
|
| _tc_boost = max(_tc_boost, _tc_requested_cap)
|
| _tc_boost_cap = max(32768, _tc_requested_cap or 0)
|
| agent._ephemeral_max_output_tokens = min(_tc_boost, _tc_boost_cap)
|
|
|
|
|
|
|
| continue
|
| agent._flush_status_buffer()
|
| if _is_stub_stall:
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Stream kept dropping mid tool-call after 4 retries β the action was not executed.",
|
| force=True,
|
| )
|
| else:
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Truncated tool call response detected again β refusing to execute incomplete tool arguments.",
|
| force=True,
|
| )
|
| agent._cleanup_task_resources(effective_task_id)
|
| _final_response = (
|
| "Stream repeatedly dropped mid tool-call (network); "
|
| "the tool was not executed"
|
| if _is_stub_stall
|
| else "Response truncated due to output length limit"
|
| )
|
|
|
|
|
|
|
| close_interrupted_tool_sequence(messages, _final_response)
|
| agent._persist_session(messages, conversation_history)
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "partial": True,
|
| "error": _final_response,
|
| }
|
|
|
|
|
| if len(messages) > 1:
|
| agent._vprint(f"{agent.log_prefix} βͺ Rolling back to last complete assistant turn")
|
| rolled_back_messages = agent._get_messages_up_to_last_assistant(messages)
|
|
|
| agent._cleanup_task_resources(effective_task_id)
|
| agent._persist_session(messages, conversation_history)
|
|
|
| return {
|
| "final_response": "Response truncated due to output length limit",
|
| "messages": rolled_back_messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "partial": True,
|
| "error": "Response truncated due to output length limit"
|
| }
|
| else:
|
|
|
| agent._flush_status_buffer()
|
| agent._vprint(f"{agent.log_prefix}β First response truncated - cannot recover", force=True)
|
| agent._persist_session(messages, conversation_history)
|
| return {
|
| "final_response": "First response truncated due to output length limit",
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "failed": True,
|
| "error": "First response truncated due to output length limit"
|
| }
|
|
|
|
|
| if hasattr(response, 'usage') and response.usage:
|
| canonical_usage = normalize_usage(
|
| response.usage,
|
| provider=agent.provider,
|
| api_mode=agent.api_mode,
|
| )
|
|
|
|
|
|
|
|
|
| aggregator_usage = canonical_usage
|
|
|
|
|
|
|
|
|
|
|
| _moa_ref_cost = None
|
| _moa_client = getattr(agent, "client", None)
|
| if _moa_client is not None and hasattr(_moa_client, "consume_reference_usage"):
|
| try:
|
| _ref_usage, _moa_ref_cost = _moa_client.consume_reference_usage()
|
| if _ref_usage is not None:
|
| canonical_usage = canonical_usage + _ref_usage
|
| except Exception as _moa_acct_exc:
|
| logger.debug("MoA reference usage accounting failed: %s", _moa_acct_exc)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if _moa_client is not None and hasattr(_moa_client, "consume_and_save_trace"):
|
| try:
|
| _agg_streamed_text = (
|
| getattr(agent, "_current_streamed_assistant_text", "") or ""
|
| )
|
| _moa_client.consume_and_save_trace(
|
| agent.session_id,
|
| aggregator_output_fallback=_agg_streamed_text or None,
|
| )
|
| except Exception as _moa_trace_exc:
|
| logger.debug("MoA trace flush failed: %s", _moa_trace_exc)
|
| prompt_tokens = canonical_usage.prompt_tokens
|
| completion_tokens = canonical_usage.output_tokens
|
| total_tokens = canonical_usage.total_tokens
|
|
|
|
|
|
|
|
|
| usage_dict = {
|
| "prompt_tokens": prompt_tokens,
|
| "completion_tokens": completion_tokens,
|
| "total_tokens": total_tokens,
|
| "input_tokens": canonical_usage.input_tokens,
|
| "output_tokens": canonical_usage.output_tokens,
|
| "cache_read_tokens": canonical_usage.cache_read_tokens,
|
| "cache_write_tokens": canonical_usage.cache_write_tokens,
|
| "reasoning_tokens": canonical_usage.reasoning_tokens,
|
| }
|
| agent.context_compressor.update_from_response(usage_dict)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._last_turn_usage = dict(usage_dict)
|
| elif getattr(
|
| agent.context_compressor,
|
| "awaiting_real_usage_after_compression",
|
| False,
|
| ):
|
|
|
|
|
|
|
|
|
|
|
| agent.context_compressor.update_from_response({})
|
|
|
| if hasattr(response, 'usage') and response.usage:
|
|
|
|
|
|
|
| if getattr(agent.context_compressor, "_context_probed", False):
|
| ctx = agent.context_compressor.context_length
|
| if getattr(agent.context_compressor, "_context_probe_persistable", False):
|
| save_context_length(agent.model, agent.base_url, ctx)
|
| agent._safe_print(f"{agent.log_prefix}πΎ Cached context length: {ctx:,} tokens for {agent.model}")
|
| agent.context_compressor._context_probed = False
|
| agent.context_compressor._context_probe_persistable = False
|
|
|
| agent.session_prompt_tokens += prompt_tokens
|
| agent.session_completion_tokens += completion_tokens
|
| agent.session_total_tokens += total_tokens
|
| agent.session_api_calls += 1
|
| agent.session_input_tokens += canonical_usage.input_tokens
|
| agent.session_output_tokens += canonical_usage.output_tokens
|
| agent.session_cache_read_tokens += canonical_usage.cache_read_tokens
|
| agent.session_cache_write_tokens += canonical_usage.cache_write_tokens
|
| agent.session_reasoning_tokens += canonical_usage.reasoning_tokens
|
|
|
|
|
| _cache_pct = ""
|
| if canonical_usage.cache_read_tokens and prompt_tokens:
|
| _cache_pct = f" cache={canonical_usage.cache_read_tokens}/{prompt_tokens} ({100*canonical_usage.cache_read_tokens/prompt_tokens:.0f}%)"
|
| logger.info(
|
| "API call #%d: model=%s provider=%s in=%d out=%d total=%d latency=%.1fs%s",
|
| agent.session_api_calls, agent.model, agent.provider or "unknown",
|
| prompt_tokens, completion_tokens, total_tokens,
|
| api_duration, _cache_pct,
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _agg_cost_model = agent.model
|
| _agg_cost_provider = agent.provider
|
| _agg_cost_base_url = agent.base_url
|
| _agg_slot = getattr(_moa_client, "last_aggregator_slot", None) if _moa_client is not None else None
|
| if _agg_slot and _agg_slot.get("model"):
|
| _agg_cost_model = _agg_slot["model"]
|
| _agg_cost_provider = _agg_slot.get("provider") or agent.provider
|
| _agg_cost_base_url = _agg_slot.get("base_url") or agent.base_url
|
| cost_result = estimate_usage_cost(
|
| _agg_cost_model,
|
| aggregator_usage,
|
| provider=_agg_cost_provider,
|
| base_url=_agg_cost_base_url,
|
| api_key=getattr(agent, "api_key", ""),
|
| )
|
| if cost_result.amount_usd is not None:
|
| agent.session_estimated_cost_usd += float(cost_result.amount_usd)
|
|
|
|
|
| if _moa_ref_cost is not None:
|
| try:
|
| agent.session_estimated_cost_usd += float(_moa_ref_cost)
|
| except (TypeError, ValueError):
|
| pass
|
| agent.session_cost_status = cost_result.status
|
| agent.session_cost_source = cost_result.source
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if agent._session_db and agent.session_id:
|
| try:
|
|
|
|
|
|
|
|
|
|
|
|
|
| if not agent._session_db_created:
|
| agent._ensure_db_session()
|
|
|
|
|
|
|
|
|
| _cost_delta = None
|
| if cost_result.amount_usd is not None:
|
| _cost_delta = float(cost_result.amount_usd)
|
| if _moa_ref_cost is not None:
|
| try:
|
| _cost_delta = (_cost_delta or 0.0) + float(_moa_ref_cost)
|
| except (TypeError, ValueError):
|
| pass
|
|
|
|
|
|
|
|
|
|
|
| agent._session_db.queue_token_counts(
|
| agent.session_id,
|
| input_tokens=canonical_usage.input_tokens,
|
| output_tokens=canonical_usage.output_tokens,
|
| cache_read_tokens=canonical_usage.cache_read_tokens,
|
| cache_write_tokens=canonical_usage.cache_write_tokens,
|
| reasoning_tokens=canonical_usage.reasoning_tokens,
|
| estimated_cost_usd=_cost_delta,
|
| cost_status=cost_result.status,
|
| cost_source=cost_result.source,
|
| billing_provider=agent.provider,
|
| billing_base_url=agent.base_url,
|
| billing_mode="subscription_included"
|
| if cost_result.status == "included" else None,
|
| model=agent.model,
|
| api_call_count=1,
|
| )
|
| except Exception as e:
|
|
|
|
|
|
|
| logger.debug(
|
| "Token persistence failed (session=%s, tokens=%d): %s",
|
| agent.session_id, total_tokens, e,
|
| )
|
|
|
| if agent.verbose_logging:
|
| logging.debug(f"Token usage: prompt={usage_dict['prompt_tokens']:,}, completion={usage_dict['completion_tokens']:,}, total={usage_dict['total_tokens']:,}")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| cached = canonical_usage.cache_read_tokens
|
| written = canonical_usage.cache_write_tokens
|
| prompt = usage_dict["prompt_tokens"]
|
| if (cached or written) and not agent.quiet_mode:
|
| hit_pct = (cached / prompt * 100) if prompt > 0 else 0
|
| agent._vprint(
|
| f"{agent.log_prefix} πΎ Cache: "
|
| f"{cached:,}/{prompt:,} tokens "
|
| f"({hit_pct:.0f}% hit, {written:,} written)"
|
| )
|
|
|
| _retry.has_retried_429 = False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if agent.provider == "nous":
|
| try:
|
| from agent.nous_rate_guard import clear_nous_rate_limit
|
| clear_nous_rate_limit()
|
| except Exception:
|
| pass
|
| from agent import relay_llm
|
|
|
| relay_llm.complete_logical_call(
|
| api_request_id,
|
| outcome="success",
|
| )
|
| agent._touch_activity(f"API call #{api_call_count} completed")
|
| break
|
|
|
| except InterruptedError:
|
| if thinking_spinner:
|
| thinking_spinner.stop("")
|
| thinking_spinner = None
|
| if agent.thinking_callback:
|
| agent.thinking_callback("")
|
| if agent._has_pending_redirect():
|
|
|
|
|
|
|
|
|
|
|
| if agent.clear_interrupt(preserve_redirect=True):
|
| _retry.restart_with_redirected_messages = True
|
| break
|
| api_elapsed = time.time() - api_start_time
|
| agent._vprint(f"{agent.log_prefix}β‘ Interrupted during API call.", force=True)
|
| interrupted = True
|
|
|
|
|
|
|
|
|
|
|
| _partial = agent._strip_think_blocks(
|
| getattr(agent, "_current_streamed_assistant_text", "") or ""
|
| ).strip()
|
| if _partial:
|
| messages.append({"role": "assistant", "content": _partial})
|
| final_response = _partial
|
| else:
|
| final_response = f"{INTERRUPT_WAITING_FOR_MODEL_PREFIX}{api_elapsed:.1f}s elapsed)."
|
| agent._persist_session(messages, conversation_history)
|
| break
|
|
|
| except Exception as api_error:
|
|
|
|
|
| if thinking_spinner:
|
| thinking_spinner.stop("")
|
| thinking_spinner = None
|
| if agent.thinking_callback:
|
| agent.thinking_callback("")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if isinstance(api_error, UnicodeEncodeError) and getattr(agent, '_unicode_sanitization_passes', 0) < 2:
|
| _err_str = str(api_error).lower()
|
| _is_ascii_codec = "'ascii'" in _err_str or "ascii" in _err_str
|
|
|
|
|
|
|
|
|
| _is_surrogate_error = (
|
| "surrogate" in _err_str
|
| or ("'utf-8'" in _err_str and not _is_ascii_codec)
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _surrogates_found = _sanitize_messages_surrogates(messages)
|
| if isinstance(api_messages, list):
|
| if _sanitize_messages_surrogates(api_messages):
|
| _surrogates_found = True
|
| if isinstance(api_kwargs, dict):
|
| if _sanitize_structure_surrogates(api_kwargs):
|
| _surrogates_found = True
|
| if isinstance(getattr(agent, "prefill_messages", None), list):
|
| if _sanitize_messages_surrogates(agent.prefill_messages):
|
| _surrogates_found = True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if _surrogates_found or _is_surrogate_error:
|
| agent._unicode_sanitization_passes += 1
|
| if _surrogates_found:
|
| agent._buffer_vprint(
|
| "β οΈ Stripped invalid surrogate characters from messages. Retrying..."
|
| )
|
| else:
|
| agent._buffer_vprint(
|
| "β οΈ Surrogate encoding error β retrying after full-payload sanitization..."
|
| )
|
| continue
|
| if _is_ascii_codec:
|
| agent._force_ascii_payload = True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _messages_sanitized = _sanitize_messages_non_ascii(messages)
|
| if isinstance(api_messages, list):
|
| _sanitize_messages_non_ascii(api_messages)
|
|
|
|
|
|
|
|
|
| if isinstance(api_kwargs, dict):
|
| _sanitize_structure_non_ascii(api_kwargs)
|
| _prefill_sanitized = False
|
| if isinstance(getattr(agent, "prefill_messages", None), list):
|
| _prefill_sanitized = _sanitize_messages_non_ascii(agent.prefill_messages)
|
|
|
| _tools_sanitized = False
|
| if isinstance(getattr(agent, "tools", None), list):
|
| _tools_sanitized = _sanitize_tools_non_ascii(agent.tools)
|
|
|
| _system_sanitized = False
|
| if isinstance(active_system_prompt, str):
|
| _sanitized_system = _strip_non_ascii(active_system_prompt)
|
| if _sanitized_system != active_system_prompt:
|
| active_system_prompt = _sanitized_system
|
| agent._cached_system_prompt = _sanitized_system
|
| _system_sanitized = True
|
| if isinstance(getattr(agent, "ephemeral_system_prompt", None), str):
|
| _sanitized_ephemeral = _strip_non_ascii(agent.ephemeral_system_prompt)
|
| if _sanitized_ephemeral != agent.ephemeral_system_prompt:
|
| agent.ephemeral_system_prompt = _sanitized_ephemeral
|
| _system_sanitized = True
|
|
|
| _headers_sanitized = False
|
| _default_headers = (
|
| agent._client_kwargs.get("default_headers")
|
| if isinstance(getattr(agent, "_client_kwargs", None), dict)
|
| else None
|
| )
|
| if isinstance(_default_headers, dict):
|
| _headers_sanitized = _sanitize_structure_non_ascii(_default_headers)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _credential_sanitized = False
|
| _raw_key = getattr(agent, "api_key", None) or ""
|
|
|
|
|
|
|
|
|
| if _raw_key and isinstance(_raw_key, str):
|
| _clean_key = _strip_non_ascii(_raw_key)
|
| if _clean_key != _raw_key:
|
| agent.api_key = _clean_key
|
| if isinstance(getattr(agent, "_client_kwargs", None), dict):
|
| agent._client_kwargs["api_key"] = _clean_key
|
|
|
|
|
|
|
| if getattr(agent, "client", None) is not None and hasattr(agent.client, "api_key"):
|
| agent.client.api_key = _clean_key
|
| _credential_sanitized = True
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ API key contained non-ASCII characters "
|
| f"(bad copy-paste?) β stripped them. If auth fails, "
|
| f"re-copy the key from your provider's dashboard.",
|
| force=True,
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._unicode_sanitization_passes += 1
|
| _any_sanitized = (
|
| _messages_sanitized
|
| or _prefill_sanitized
|
| or _tools_sanitized
|
| or _system_sanitized
|
| or _headers_sanitized
|
| or _credential_sanitized
|
| )
|
| if _any_sanitized:
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ System encoding is ASCII β stripped non-ASCII characters from request payload. Retrying...",
|
| force=True,
|
| )
|
| else:
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ System encoding is ASCII β enabling full-payload sanitization for retry...",
|
| force=True,
|
| )
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _err_body = ""
|
| try:
|
| _err_body = str(getattr(api_error, "body", None) or
|
| getattr(api_error, "message", None) or
|
| str(api_error))
|
| except Exception:
|
| pass
|
| _err_status = getattr(api_error, "status_code", None)
|
| _IMAGE_REJECTION_PHRASES = (
|
| "only 'text' content type is supported",
|
| "only text content type is supported",
|
| "image_url is not supported",
|
| "image content is not supported",
|
| "multimodal is not supported",
|
| "multimodal content is not supported",
|
| "multimodal input is not supported",
|
| "vision is not supported",
|
| "vision input is not supported",
|
| "does not support images",
|
| "does not support image input",
|
| "does not support multimodal",
|
| "does not support vision",
|
| "model does not support image",
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| "image_url'. expected",
|
|
|
|
|
|
|
| "unknown variant `image_url`, expected `text`",
|
| "unknown variant image_url, expected text",
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| "no endpoints found that support image input",
|
| )
|
| _err_lower = _err_body.lower()
|
| _looks_like_image_rejection = any(
|
| p in _err_lower for p in _IMAGE_REJECTION_PHRASES
|
| )
|
|
|
|
|
|
|
| _status_ok = _err_status is None or (400 <= int(_err_status) < 500)
|
| if (
|
| getattr(agent, "_vision_supported", True)
|
| and _looks_like_image_rejection
|
| and _status_ok
|
| ):
|
| agent._vision_supported = False
|
| _imgs_removed = _strip_images_from_messages(messages)
|
| if isinstance(api_messages, list):
|
| _strip_images_from_messages(api_messages)
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Server rejected image content β "
|
| f"switching to text-only mode for this session"
|
| + (". Stripped images from history and retrying." if _imgs_removed else "."),
|
| force=True,
|
| )
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| isinstance(api_error, RuntimeError)
|
| and "unexpected event order" in str(api_error).lower()
|
| and getattr(agent, "provider", "") == "bedrock"
|
| and agent.api_mode == "anthropic_messages"
|
| and not getattr(agent, "_bedrock_converse_fallback_attempted", False)
|
| ):
|
| agent._bedrock_converse_fallback_attempted = True
|
| agent.api_mode = "bedrock_converse"
|
| agent._bedrock_region = getattr(agent, "_bedrock_region", None) or "us-east-1"
|
| agent.client = None
|
| agent._client_kwargs = {}
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ AnthropicBedrock SDK streaming failed β "
|
| f"falling back to native Converse API for this session.",
|
| force=True,
|
| )
|
| continue
|
|
|
| status_code = getattr(api_error, "status_code", None)
|
| error_context = agent._extract_api_error_context(api_error)
|
|
|
|
|
| _compressor = getattr(agent, "context_compressor", None)
|
| _ctx_len = getattr(_compressor, "context_length", 200000) if _compressor else 200000
|
| classified = classify_api_error(
|
| api_error,
|
| provider=getattr(agent, "provider", "") or "",
|
| model=getattr(agent, "model", "") or "",
|
| approx_tokens=approx_tokens,
|
| context_length=_ctx_len,
|
| num_messages=len(api_messages) if api_messages else 0,
|
| )
|
| logger.debug(
|
| "Error classified: reason=%s status=%s retryable=%s compress=%s rotate=%s fallback=%s",
|
| classified.reason.value, classified.status_code,
|
| classified.retryable, classified.should_compress,
|
| classified.should_rotate_credential, classified.should_fallback,
|
| )
|
| agent._invoke_api_request_error_hook(
|
| task_id=effective_task_id,
|
| turn_id=turn_id,
|
| api_request_id=api_request_id,
|
| api_call_count=api_call_count,
|
| api_start_time=api_start_time,
|
| api_kwargs=api_kwargs,
|
| error_type=type(api_error).__name__,
|
| error_message=str(api_error),
|
| status_code=status_code,
|
| retry_count=retry_count,
|
| max_retries=max_retries,
|
| retryable=classified.retryable,
|
| reason=classified.reason.value,
|
| )
|
|
|
| if (
|
| classified.reason == FailoverReason.billing
|
| and _is_nous_inference_route(
|
| getattr(agent, "provider", "") or "",
|
| getattr(agent, "base_url", "") or "",
|
| )
|
| and not _retry.nous_paid_entitlement_refresh_attempted
|
| ):
|
| _retry.nous_paid_entitlement_refresh_attempted = True
|
| if _try_refresh_nous_paid_entitlement_credentials(agent):
|
| agent._vprint(
|
| f"{agent.log_prefix}π Nous paid access verified β "
|
| "refreshed runtime credentials and retrying request...",
|
| force=True,
|
| )
|
| continue
|
|
|
| recovered_with_pool, _retry.has_retried_429 = agent._recover_with_credential_pool(
|
| status_code=status_code,
|
| has_retried_429=_retry.has_retried_429,
|
| classified_reason=classified.reason,
|
| error_context=error_context,
|
| )
|
| if recovered_with_pool:
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| classified.reason == FailoverReason.image_too_large
|
| and not _retry.image_shrink_retry_attempted
|
| ):
|
| _retry.image_shrink_retry_attempted = True
|
| image_max_dimension = _image_error_max_dimension(api_error) or 8000
|
| if agent._try_shrink_image_parts_in_messages(
|
| api_messages,
|
| max_dimension=image_max_dimension,
|
| ):
|
| agent._vprint(
|
| f"{agent.log_prefix}π Image(s) exceeded provider size limit β "
|
| f"shrank and retrying...",
|
| force=True,
|
| )
|
| continue
|
| else:
|
| logger.info(
|
| "image-shrink recovery: no data-URL image parts found "
|
| "or shrink didn't reduce size; surfacing original error."
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| classified.reason == FailoverReason.multimodal_tool_content_unsupported
|
| and not _retry.multimodal_tool_content_retry_attempted
|
| ):
|
| _retry.multimodal_tool_content_retry_attempted = True
|
| if agent._try_strip_image_parts_from_tool_messages(api_messages):
|
| agent._vprint(
|
| f"{agent.log_prefix}π Provider rejected list-type tool content β "
|
| f"downgraded screenshots to text and retrying...",
|
| force=True,
|
| )
|
| continue
|
| else:
|
| logger.info(
|
| "multimodal-tool-content recovery: no list-type tool "
|
| "messages with image parts found; surfacing original error."
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| classified.reason == FailoverReason.oauth_long_context_beta_forbidden
|
| and agent.api_mode == "anthropic_messages"
|
| and agent._is_anthropic_oauth
|
| and not _retry.oauth_1m_beta_retry_attempted
|
| ):
|
| _retry.oauth_1m_beta_retry_attempted = True
|
| if not getattr(agent, "_oauth_1m_beta_disabled", False):
|
| agent._oauth_1m_beta_disabled = True
|
| try:
|
| agent._anthropic_client.close()
|
| except Exception:
|
| pass
|
| agent._rebuild_anthropic_client()
|
| agent._vprint(
|
| f"{agent.log_prefix}π OAuth subscription doesn't support "
|
| f"the 1M-context beta β disabled for this session and retrying...",
|
| force=True,
|
| )
|
| continue
|
|
|
| if (
|
| agent.api_mode == "codex_responses"
|
| and agent.provider in {"openai-codex", "xai-oauth"}
|
| and status_code == 401
|
| and not _retry.codex_auth_retry_attempted
|
| ):
|
| _retry.codex_auth_retry_attempted = True
|
| if agent._try_refresh_codex_client_credentials(force=True):
|
| _label = "xAI OAuth" if agent.provider == "xai-oauth" else "Codex"
|
| agent._buffer_vprint(f"π {_label} auth refreshed after 401. Retrying request...")
|
| continue
|
| if (
|
| agent.api_mode == "chat_completions"
|
| and agent.provider == "vertex"
|
| and status_code == 401
|
| and not _retry.vertex_auth_retry_attempted
|
| ):
|
| _retry.vertex_auth_retry_attempted = True
|
| if agent._try_refresh_vertex_client_credentials():
|
| agent._buffer_vprint("π Vertex AI token refreshed after 401. Retrying request...")
|
| continue
|
| if (
|
| agent.api_mode in ("chat_completions", "anthropic_messages")
|
| and agent.provider == "nous"
|
| and status_code == 401
|
| and not _retry.nous_auth_retry_attempted
|
| ):
|
| _retry.nous_auth_retry_attempted = True
|
| if agent._try_refresh_nous_client_credentials(force=True):
|
| print(f"{agent.log_prefix}π Nous agent key refreshed after 401. Retrying request...")
|
| continue
|
|
|
|
|
|
|
| from hermes_constants import display_hermes_home as _dhh_fn
|
| _dhh = _dhh_fn()
|
| _body_text = ""
|
| try:
|
| _body = getattr(api_error, "body", None) or getattr(api_error, "response", None)
|
| if _body is not None:
|
| _body_text = str(_body)[:200]
|
| except Exception:
|
| pass
|
| print(f"{agent.log_prefix}π Nous 401 β Portal authentication failed.")
|
| if _body_text:
|
| print(f"{agent.log_prefix} Response: {_body_text}")
|
| if not _print_nous_entitlement_guidance(agent, "Nous model access"):
|
| print(f"{agent.log_prefix} Most likely: Portal OAuth expired, account out of credits, or agent key revoked.")
|
| print(f"{agent.log_prefix} Troubleshooting:")
|
| print(f"{agent.log_prefix} β’ Re-authenticate: hermes auth add nous")
|
| print(f"{agent.log_prefix} β’ Check credits / billing: https://portal.nousresearch.com")
|
| print(f"{agent.log_prefix} β’ Verify stored credentials: {_dhh}/auth.json")
|
| print(f"{agent.log_prefix} β’ Switch providers temporarily: /model <model> --provider openrouter")
|
| if (
|
| _is_copilot_provider(agent)
|
| and status_code == 401
|
| and not _retry.copilot_auth_retry_attempted
|
| ):
|
| _retry.copilot_auth_retry_attempted = True
|
| if agent._try_refresh_copilot_client_credentials():
|
| agent._buffer_vprint("π Copilot credentials refreshed after 401. Retrying request...")
|
| continue
|
| if (
|
| agent.api_mode == "anthropic_messages"
|
| and status_code == 401
|
| and hasattr(agent, '_anthropic_api_key')
|
| and not _retry.anthropic_auth_retry_attempted
|
| ):
|
| _retry.anthropic_auth_retry_attempted = True
|
| from agent.anthropic_adapter import _is_oauth_token
|
| from agent.azure_identity_adapter import is_token_provider
|
| if agent._try_refresh_anthropic_client_credentials():
|
| print(f"{agent.log_prefix}π Anthropic credentials refreshed after 401. Retrying request...")
|
| continue
|
|
|
| key = agent._anthropic_api_key
|
| print(f"{agent.log_prefix}π Anthropic 401 β authentication failed.")
|
| if is_token_provider(key):
|
|
|
|
|
|
|
|
|
|
|
| print(f"{agent.log_prefix} Auth method: Microsoft Entra ID (httpx event hook)")
|
| print(f"{agent.log_prefix} Run `hermes doctor` for credential-chain diagnostics, or")
|
| print(f"{agent.log_prefix} `az login` if your developer session expired.")
|
| else:
|
| auth_method = "Bearer (OAuth/setup-token)" if _is_oauth_token(key) else "x-api-key (API key)"
|
| print(f"{agent.log_prefix} Auth method: {auth_method}")
|
| print(f"{agent.log_prefix} Token prefix: {key[:12]}..." if isinstance(key, str) and len(key) > 12 else f"{agent.log_prefix} Token: (empty or short)")
|
| print(f"{agent.log_prefix} Troubleshooting:")
|
| from hermes_constants import display_hermes_home as _dhh_fn
|
| _dhh = _dhh_fn()
|
| print(f"{agent.log_prefix} β’ Check ANTHROPIC_TOKEN in {_dhh}/.env for Hermes-managed OAuth/setup tokens")
|
| print(f"{agent.log_prefix} β’ Check ANTHROPIC_API_KEY in {_dhh}/.env for API keys or legacy token values")
|
| print(f"{agent.log_prefix} β’ For API keys: verify at https://platform.claude.com/settings/keys")
|
| print(f"{agent.log_prefix} β’ For Claude Code: run 'claude /login' to refresh, then retry")
|
| print(f"{agent.log_prefix} β’ Legacy cleanup: hermes config set ANTHROPIC_TOKEN \"\"")
|
| print(f"{agent.log_prefix} β’ Clear stale keys: hermes config set ANTHROPIC_API_KEY \"\"")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| classified.reason == FailoverReason.thinking_signature
|
| and not _retry.thinking_sig_retry_attempted
|
| ):
|
| _retry.thinking_sig_retry_attempted = True
|
| _api_stripped = 0
|
| for _m in api_messages:
|
| if isinstance(_m, dict) and "reasoning_details" in _m:
|
| _m.pop("reasoning_details", None)
|
| _api_stripped += 1
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Thinking block signature invalid, "
|
| f"stripped reasoning_details from api_messages for retry...",
|
| force=True,
|
| )
|
| logger.warning(
|
| "%sThinking block signature recovery: stripped "
|
| "reasoning_details from %d api_messages "
|
| "(canonical messages unchanged)",
|
| agent.log_prefix, _api_stripped,
|
| )
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| classified.reason == FailoverReason.invalid_encrypted_content
|
| and not _retry.invalid_encrypted_content_retry_attempted
|
| and agent.api_mode == "codex_responses"
|
| and bool(getattr(agent, "_codex_reasoning_replay_enabled", True))
|
| and any(
|
| isinstance(_m, dict)
|
| and _m.get("role") == "assistant"
|
| and isinstance(_m.get("codex_reasoning_items"), list)
|
| and _m.get("codex_reasoning_items")
|
| for _m in messages
|
| )
|
| ):
|
| _retry.invalid_encrypted_content_retry_attempted = True
|
| replay_stats = agent._disable_codex_reasoning_replay(messages)
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Encrypted reasoning replay was rejected by the provider β "
|
| f"disabled replay and stripped {replay_stats['items']} item(s) from "
|
| f"{replay_stats['messages']} message(s), retrying...",
|
| force=True,
|
| )
|
| logger.warning(
|
| "%sInvalid encrypted reasoning recovery: disabled replay and stripped %d items from %d messages",
|
| agent.log_prefix,
|
| replay_stats["items"],
|
| replay_stats["messages"],
|
| )
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| classified.reason == FailoverReason.llama_cpp_grammar_pattern
|
| and not _retry.llama_cpp_grammar_retry_attempted
|
| ):
|
| _retry.llama_cpp_grammar_retry_attempted = True
|
| try:
|
| from tools.schema_sanitizer import strip_pattern_and_format
|
| _, _stripped = strip_pattern_and_format(agent.tools)
|
| except Exception as _strip_exc:
|
| logger.warning(
|
| "%sllama.cpp grammar recovery: strip helper failed: %s",
|
| agent.log_prefix, _strip_exc,
|
| )
|
| _stripped = 0
|
| if _stripped:
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ llama.cpp rejected tool schema grammar β "
|
| f"stripped {_stripped} pattern/format keyword(s), retrying...",
|
| force=True,
|
| )
|
| logger.warning(
|
| "%sllama.cpp grammar recovery: stripped %d "
|
| "pattern/format keyword(s) from tool schemas",
|
| agent.log_prefix, _stripped,
|
| )
|
| continue
|
|
|
|
|
| logger.warning(
|
| "%sllama.cpp grammar error but no pattern/format "
|
| "keywords to strip β falling through to normal retry",
|
| agent.log_prefix,
|
| )
|
|
|
| retry_count += 1
|
| elapsed_time = time.time() - api_start_time
|
| agent._touch_activity(
|
| f"API error recovery (attempt {retry_count}/{max_retries})"
|
| )
|
|
|
| error_type = type(api_error).__name__
|
| error_msg = str(api_error).lower()
|
| _error_summary = agent._summarize_api_error(api_error)
|
| logger.warning(
|
| "API call failed (attempt %s/%s) error_type=%s %s summary=%s",
|
| retry_count,
|
| max_retries,
|
| error_type,
|
| agent._client_log_context(),
|
| _error_summary,
|
| )
|
|
|
| _provider = getattr(agent, "provider", "unknown")
|
| _base = getattr(agent, "base_url", "unknown")
|
| _model = getattr(agent, "model", "unknown")
|
| _status_code_str = f" [HTTP {status_code}]" if status_code else ""
|
| agent._buffer_vprint(f"β οΈ API call failed (attempt {retry_count}/{max_retries}): {error_type}{_status_code_str}")
|
| agent._buffer_vprint(f" π Provider: {_provider} Model: {_model}")
|
| agent._buffer_vprint(f" π Endpoint: {_base}")
|
| agent._buffer_vprint(f" π Error: {_error_summary}")
|
| if status_code and status_code < 500:
|
| _err_body = getattr(api_error, "body", None)
|
| _err_body_str = str(_err_body)[:300] if _err_body else None
|
| if _err_body_str:
|
| agent._buffer_vprint(f" π Details: {_err_body_str}")
|
| agent._buffer_vprint(f" β±οΈ Elapsed: {elapsed_time:.2f}s Context: {len(api_messages)} msgs, ~{approx_tokens:,} tokens")
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| agent._is_openrouter_url()
|
| and "support tool use" in error_msg
|
| ):
|
| agent._buffer_vprint(
|
| f" π‘ No OpenRouter providers for {_model} support tool calling with your current settings."
|
| )
|
| if agent.providers_allowed:
|
| agent._buffer_vprint(
|
| " Your provider_routing.only restriction is filtering out tool-capable providers."
|
| )
|
| agent._buffer_vprint(
|
| " Try removing the restriction or adding providers that support tools for this model."
|
| )
|
| agent._buffer_vprint(
|
| f" Check which providers support tools: https://openrouter.ai/models/{_model}"
|
| )
|
|
|
|
|
| if agent._interrupt_requested:
|
|
|
|
|
|
|
| if agent.clear_interrupt(preserve_redirect=True):
|
| _retry.restart_with_redirected_messages = True
|
| break
|
| agent._vprint(f"{agent.log_prefix}β‘ Interrupt detected during error handling, aborting retries.", force=True)
|
| _interrupt_text = f"Operation interrupted: handling API error ({error_type}: {agent._clean_error_message(str(api_error))})."
|
| close_interrupted_tool_sequence(messages, _interrupt_text)
|
| agent._persist_session(messages, conversation_history)
|
| agent.clear_interrupt()
|
| return {
|
| "final_response": _interrupt_text,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "interrupted": True,
|
| }
|
|
|
|
|
|
|
|
|
| status_code = getattr(api_error, "status_code", None)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _overflow_reasons = {
|
| FailoverReason.long_context_tier,
|
| FailoverReason.payload_too_large,
|
| FailoverReason.context_overflow,
|
| }
|
| _is_output_cap_error = (
|
| is_output_cap_error(error_msg)
|
| or parse_available_output_tokens_from_error(error_msg) is not None
|
| )
|
| if (
|
| classified.reason in _overflow_reasons
|
| and not getattr(agent, "compression_enabled", True)
|
| and not _is_output_cap_error
|
| ):
|
| agent._flush_status_buffer()
|
| agent._vprint(
|
| f"{agent.log_prefix}β Context overflow, but auto-compaction is disabled "
|
| f"(compression.enabled: false).",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} π‘ Run /compress to compact manually, /new to start fresh, "
|
| f"switch to a larger-context model, or reduce attachments.",
|
| force=True,
|
| )
|
| logger.error(
|
| f"{agent.log_prefix}Context overflow ({classified.reason.value}) with "
|
| f"auto-compaction disabled β not compressing."
|
| )
|
| agent._persist_session(messages, conversation_history)
|
| _final_response = (
|
| "Context overflow and auto-compaction is disabled "
|
| "(compression.enabled: false). Run /compress to compact manually, "
|
| "/new to start fresh, or switch to a larger-context model."
|
| )
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final_response,
|
| "partial": True,
|
| "failed": True,
|
| "compaction_disabled": True,
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if classified.reason == FailoverReason.long_context_tier:
|
| _reduced_ctx = 200000
|
| compressor = agent.context_compressor
|
| old_ctx = compressor.context_length
|
| if old_ctx > _reduced_ctx:
|
| compressor.update_model(
|
| model=agent.model,
|
| context_length=_reduced_ctx,
|
| base_url=agent.base_url,
|
| api_key=getattr(agent, "api_key", ""),
|
| provider=agent.provider,
|
| api_mode=agent.api_mode,
|
| )
|
|
|
|
|
| if hasattr(compressor, "_context_probed"):
|
| compressor._context_probed = True
|
|
|
|
|
|
|
|
|
| compressor._context_probe_persistable = False
|
| agent._buffer_vprint(
|
| f"β οΈ Anthropic long-context tier "
|
| f"requires extra usage β reducing context: "
|
| f"{old_ctx:,} β {_reduced_ctx:,} tokens"
|
| )
|
|
|
| compression_attempts += 1
|
| if compression_attempts <= max_compression_attempts:
|
| original_len = len(messages)
|
| messages, active_system_prompt = agent._compress_context(
|
| messages, system_message,
|
| approx_tokens=approx_tokens,
|
| task_id=effective_task_id,
|
| )
|
| conversation_history = conversation_history_after_compression(
|
| agent, messages, conversation_history
|
| )
|
| if len(messages) < original_len or old_ctx > _reduced_ctx:
|
| agent._buffer_status(
|
| COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE.format(
|
| new_ctx=_reduced_ctx, old_ctx=old_ctx
|
| )
|
| )
|
| time.sleep(2)
|
| _retry.restart_with_compressed_messages = True
|
| break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| is_rate_limited = classified.reason in {
|
| FailoverReason.rate_limit,
|
| FailoverReason.billing,
|
| FailoverReason.upstream_rate_limit,
|
| }
|
| _is_transport_failure = classified.reason in {
|
| FailoverReason.timeout,
|
| FailoverReason.overloaded,
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _is_zai_coding_overload = is_zai_coding_overload_error(
|
| base_url=str(_base), model=_model, error=api_error
|
| )
|
| if _is_zai_coding_overload:
|
| max_retries = max(max_retries, zai_coding_overload_retry_ceiling())
|
| _should_fallback = (
|
| is_rate_limited
|
| or (_is_transport_failure and retry_count >= 2)
|
| )
|
| if _should_fallback and agent._fallback_index < len(agent._fallback_chain):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _is_upstream = classified.reason == FailoverReason.upstream_rate_limit
|
| pool_may_recover = (
|
| False if _is_upstream
|
| else _ra()._pool_may_recover_from_rate_limit(
|
| agent._credential_pool,
|
| )
|
| )
|
| if not pool_may_recover:
|
| if _is_upstream:
|
| _upstream_name = (classified.error_context or {}).get(
|
| "upstream_provider", "aggregator"
|
| )
|
| agent._buffer_status(
|
| f"β οΈ Upstream {_upstream_name} rate-limited β "
|
| "switching to fallback model..."
|
| )
|
| elif classified.reason == FailoverReason.billing:
|
| agent._buffer_status(
|
| "β οΈ Billing or credits exhausted β switching to fallback provider..."
|
| )
|
| elif _is_transport_failure:
|
| agent._buffer_status(
|
| "β οΈ Provider unreachable β switching to fallback provider..."
|
| )
|
| else:
|
| agent._buffer_status("β οΈ Rate limited β switching to fallback provider...")
|
| if agent._try_activate_fallback(reason=classified.reason):
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| classified.is_auth
|
| and not _retry.auth_failover_attempted
|
| and agent._fallback_index < len(agent._fallback_chain)
|
| ):
|
| _retry.auth_failover_attempted = True
|
| agent._buffer_status(
|
| "π Authentication failed and could not be refreshed β "
|
| "switching to fallback provider..."
|
| )
|
| if agent._try_activate_fallback(reason=classified.reason):
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| is_rate_limited
|
| and agent.provider == "nous"
|
| and classified.reason == FailoverReason.rate_limit
|
| and not recovered_with_pool
|
| ):
|
| _genuine_nous_rate_limit = False
|
| try:
|
| from agent.nous_rate_guard import (
|
| is_genuine_nous_rate_limit,
|
| record_nous_rate_limit,
|
| )
|
| _err_resp = getattr(api_error, "response", None)
|
| _err_hdrs = (
|
| getattr(_err_resp, "headers", None)
|
| if _err_resp else None
|
| )
|
| _genuine_nous_rate_limit = is_genuine_nous_rate_limit(
|
| headers=_err_hdrs,
|
| last_known_state=agent._rate_limit_state,
|
| )
|
| if _genuine_nous_rate_limit:
|
| record_nous_rate_limit(
|
| headers=_err_hdrs,
|
| error_context=error_context,
|
| )
|
| else:
|
| logger.info(
|
| "Nous 429 looks like upstream capacity "
|
| "(no exhausted bucket in headers or "
|
| "last-known state) -- not tripping "
|
| "cross-session breaker."
|
| )
|
| except Exception:
|
| pass
|
| if _genuine_nous_rate_limit:
|
|
|
|
|
|
|
|
|
|
|
|
|
| retry_count = max(0, max_retries - 1)
|
| continue
|
|
|
|
|
|
|
|
|
| is_payload_too_large = (
|
| classified.reason == FailoverReason.payload_too_large
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| status_code == 413
|
| and isinstance(agent.base_url, str)
|
| and "models.inference.ai.azure.com" in agent.base_url
|
| ):
|
| agent._vprint(
|
| f"{agent.log_prefix} π‘ GitHub Models free tier (models.inference.ai.azure.com) caps every",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} request at ~8K tokens. Hermes' system prompt + tool schemas baseline",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} exceeds that floor, so this endpoint cannot run an agentic loop.",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} Use the `copilot` provider with a Copilot subscription token (`hermes",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} setup` β GitHub Copilot), or pick any other provider.",
|
| force=True,
|
| )
|
|
|
| if is_payload_too_large:
|
| compression_attempts += 1
|
| if compression_attempts > max_compression_attempts:
|
|
|
| agent._flush_status_buffer()
|
| agent._vprint(f"{agent.log_prefix}β Max compression attempts ({max_compression_attempts}) reached for payload-too-large error.", force=True)
|
| agent._vprint(f"{agent.log_prefix} π‘ Try /new to start a fresh conversation, or /compress to retry compression.", force=True)
|
| logger.error("%s413 compression failed after %d attempts.", agent.log_prefix, max_compression_attempts)
|
| agent._persist_session(messages, conversation_history)
|
| _final_response = f"Request payload too large: max compression attempts ({max_compression_attempts}) reached."
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final_response,
|
| "partial": True,
|
| "failed": True,
|
| "compression_exhausted": True,
|
| }
|
| agent._buffer_status(f"β οΈ Request payload too large (413) β compression attempt {compression_attempts}/{max_compression_attempts}...")
|
|
|
| original_len = len(messages)
|
| original_tokens = estimate_messages_tokens_rough(messages)
|
| _overflow_input = messages
|
| messages, active_system_prompt = agent._compress_context(
|
| messages, system_message, approx_tokens=approx_tokens,
|
| task_id=effective_task_id,
|
| )
|
| if messages is _overflow_input and compression_skipped_due_to_lock(agent):
|
|
|
|
|
|
|
|
|
|
|
|
|
| compression_attempts -= 1
|
| agent._persist_session(messages, conversation_history)
|
| return _compression_deferred_result(
|
| agent, messages, api_call_count
|
| )
|
| conversation_history = conversation_history_after_compression(
|
| agent, messages, conversation_history
|
| )
|
|
|
|
|
|
|
|
|
|
|
| new_tokens = estimate_messages_tokens_rough(messages)
|
| approx_tokens = new_tokens
|
|
|
| if len(messages) < original_len or (new_tokens > 0 and new_tokens < original_tokens * 0.95):
|
| if len(messages) < original_len:
|
| agent._buffer_status(COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE.format(before=original_len, after=len(messages)))
|
| else:
|
| agent._buffer_status(COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE.format(before=original_tokens, after=new_tokens))
|
| time.sleep(2)
|
| _retry.restart_with_compressed_messages = True
|
| break
|
| else:
|
| if agent._try_strip_image_parts_from_tool_messages(
|
| api_messages,
|
| remember_model=False,
|
| ):
|
| agent._buffer_status(
|
| "π Compression could not reduce the request further β "
|
| "removed retained vision payloads and retrying..."
|
| )
|
| continue
|
|
|
|
|
|
|
| agent._flush_status_buffer()
|
| agent._vprint(f"{agent.log_prefix}β Payload too large and cannot compress further.", force=True)
|
| agent._vprint(f"{agent.log_prefix} π‘ Try /new to start a fresh conversation, or /compress to retry compression.", force=True)
|
| logger.error("%s413 payload too large. Cannot compress further.", agent.log_prefix)
|
| agent._persist_session(messages, conversation_history)
|
| _final_response = "Request payload too large (413). Cannot compress further."
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final_response,
|
| "partial": True,
|
| "failed": True,
|
| "compression_exhausted": True,
|
| }
|
|
|
|
|
|
|
|
|
|
|
| is_context_length_error = (
|
| classified.reason == FailoverReason.context_overflow
|
| )
|
|
|
| if is_context_length_error:
|
| compressor = agent.context_compressor
|
| old_ctx = compressor.context_length
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| available_out = parse_available_output_tokens_from_error(error_msg)
|
| if available_out is not None:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| request_input_estimate = estimate_request_tokens_rough(
|
| api_messages, tools=agent.tools or None,
|
| )
|
| local_available_out = old_ctx - request_input_estimate
|
| if local_available_out > 0:
|
| safe_out = max(1, min(available_out, local_available_out) - 64)
|
| else:
|
|
|
|
|
|
|
|
|
| safe_out = max(1, available_out - 64)
|
| agent._ephemeral_max_output_tokens = safe_out
|
| agent._buffer_vprint(
|
| f"β οΈ Output cap too large for current prompt β "
|
| f"retrying with max_tokens={safe_out:,} "
|
| f"(provider_available={available_out:,}, "
|
| f"estimated_request_tokens={request_input_estimate:,}; "
|
| f"context_length unchanged at {old_ctx:,})"
|
| )
|
|
|
|
|
| compression_attempts += 1
|
| if compression_attempts > max_compression_attempts:
|
| agent._flush_status_buffer()
|
| agent._vprint(f"{agent.log_prefix}β Max compression attempts ({max_compression_attempts}) reached.", force=True)
|
| agent._vprint(f"{agent.log_prefix} π‘ Try /new to start a fresh conversation, or /compress to retry compression.", force=True)
|
| logger.error("%sContext compression failed after %d attempts.", agent.log_prefix, max_compression_attempts)
|
| agent._persist_session(messages, conversation_history)
|
| _final_response = f"Context length exceeded: max compression attempts ({max_compression_attempts}) reached."
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final_response,
|
| "partial": True,
|
| "failed": True,
|
| "compression_exhausted": True,
|
| }
|
| _retry.restart_with_compressed_messages = True
|
| break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if is_output_cap_error(error_msg):
|
| agent._flush_status_buffer()
|
| agent._vprint(
|
| f"{agent.log_prefix}β The provider rejected the request because "
|
| f"max_tokens exceeds its output cap for this model.",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} π‘ Lower model.max_tokens in your config.yaml to "
|
| f"at or below the model's max-output limit. "
|
| f"(This is an output-cap error, not a context overflow β "
|
| f"compression cannot fix it.)",
|
| force=True,
|
| )
|
| logger.error(
|
| f"{agent.log_prefix}Output-cap error not routed into compression "
|
| f"(max_tokens over provider cap): {error_msg[:200]}"
|
| )
|
| agent._persist_session(messages, conversation_history)
|
| _final_response = (
|
| "max_tokens exceeds the provider's output cap for this model. "
|
| "Lower model.max_tokens in config.yaml."
|
| )
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final_response,
|
| "partial": True,
|
| "failed": True,
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| new_ctx = get_context_length_from_provider_error(error_msg, old_ctx)
|
| _provider_lower = (getattr(agent, "provider", "") or "").lower()
|
| _base_lower = (getattr(agent, "base_url", "") or "").rstrip("/").lower()
|
| is_minimax_provider = (
|
| _provider_lower in {"minimax", "minimax-cn"}
|
| or _base_lower.startswith((
|
| "https://api.minimax.io/anthropic",
|
| "https://api.minimaxi.com/anthropic",
|
| ))
|
| )
|
| minimax_delta_only_overflow = (
|
| is_minimax_provider
|
| and new_ctx is None
|
| and "context window exceeds limit (" in error_msg
|
| )
|
|
|
| if new_ctx is not None:
|
| agent._buffer_vprint(f"Context limit detected from API: {new_ctx:,} tokens (was {old_ctx:,})")
|
| compressor.update_model(
|
| model=agent.model,
|
| context_length=new_ctx,
|
| base_url=agent.base_url,
|
| api_key=getattr(agent, "api_key", ""),
|
| provider=agent.provider,
|
| api_mode=agent.api_mode,
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
| save_context_length(agent.model, agent.base_url, new_ctx)
|
|
|
|
|
|
|
| if hasattr(compressor, "_context_probed"):
|
| compressor._context_probed = True
|
| compressor._context_probe_persistable = True
|
| agent._buffer_vprint(f"β οΈ Context length exceeded β using provider limit: {old_ctx:,} β {new_ctx:,} tokens")
|
| elif minimax_delta_only_overflow:
|
| agent._buffer_vprint(
|
| f"Provider reported overflow amount only; "
|
| f"keeping context_length at {old_ctx:,} tokens and compressing."
|
| )
|
| else:
|
| agent._buffer_vprint(
|
| f"β οΈ Context length exceeded, but provider did not report a max context length; "
|
| f"keeping context_length at {old_ctx:,} tokens and compressing."
|
| )
|
|
|
| compression_attempts += 1
|
| if compression_attempts > max_compression_attempts:
|
| agent._flush_status_buffer()
|
| agent._vprint(f"{agent.log_prefix}β Max compression attempts ({max_compression_attempts}) reached.", force=True)
|
| agent._vprint(f"{agent.log_prefix} π‘ Try /new to start a fresh conversation, or /compress to retry compression.", force=True)
|
| logger.error("%sContext compression failed after %d attempts.", agent.log_prefix, max_compression_attempts)
|
| agent._persist_session(messages, conversation_history)
|
| _final_response = f"Context length exceeded: max compression attempts ({max_compression_attempts}) reached."
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final_response,
|
| "partial": True,
|
| "failed": True,
|
| "compression_exhausted": True,
|
| }
|
| agent._buffer_status(COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE.format(tokens=approx_tokens, attempt=compression_attempts, cap=max_compression_attempts))
|
|
|
| original_len = len(messages)
|
| original_tokens = estimate_messages_tokens_rough(messages)
|
| _overflow_input = messages
|
| messages, active_system_prompt = agent._compress_context(
|
| messages, system_message, approx_tokens=approx_tokens,
|
| task_id=effective_task_id,
|
| )
|
| if messages is _overflow_input and compression_skipped_due_to_lock(agent):
|
|
|
|
|
|
|
|
|
|
|
|
|
| compression_attempts -= 1
|
| agent._persist_session(messages, conversation_history)
|
| return _compression_deferred_result(
|
| agent, messages, api_call_count
|
| )
|
| conversation_history = conversation_history_after_compression(
|
| agent, messages, conversation_history
|
| )
|
|
|
|
|
|
|
|
|
|
|
| new_tokens = estimate_messages_tokens_rough(messages)
|
| approx_tokens = new_tokens
|
|
|
| if len(messages) < original_len or (new_tokens > 0 and new_tokens < original_tokens * 0.95) or (new_ctx and new_ctx < old_ctx):
|
| if len(messages) < original_len:
|
| agent._buffer_status(COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE.format(before=original_len, after=len(messages)))
|
| elif new_tokens > 0 and new_tokens < original_tokens * 0.95:
|
| agent._buffer_status(COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE.format(before=original_tokens, after=new_tokens))
|
| time.sleep(2)
|
| _retry.restart_with_compressed_messages = True
|
| break
|
| else:
|
|
|
| agent._flush_status_buffer()
|
| agent._vprint(f"{agent.log_prefix}β Context length exceeded and cannot compress further.", force=True)
|
| agent._vprint(f"{agent.log_prefix} π‘ The conversation has accumulated too much content. Try /new to start fresh, or /compress to manually trigger compression.", force=True)
|
| logger.error("%sContext length exceeded: %s tokens. Cannot compress further.", agent.log_prefix, f"{new_tokens:,}")
|
| agent._persist_session(messages, conversation_history)
|
| _final_response = f"Context length exceeded ({new_tokens:,} tokens). Cannot compress further."
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "completed": False,
|
| "api_calls": api_call_count,
|
| "error": _final_response,
|
| "partial": True,
|
| "failed": True,
|
| "compression_exhausted": True,
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| is_local_validation_error = (
|
| isinstance(api_error, (ValueError, TypeError))
|
| and not isinstance(
|
| api_error, (UnicodeEncodeError, json.JSONDecodeError)
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| and not isinstance(api_error, ssl.SSLError)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| and not (
|
| isinstance(api_error, TypeError)
|
| and "nonetype" in str(api_error).lower()
|
| and "not iterable" in str(api_error).lower()
|
| )
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| is_client_error = (
|
| is_local_validation_error
|
| or (
|
| not classified.retryable
|
| and not classified.should_compress
|
| and classified.reason not in {
|
| FailoverReason.rate_limit,
|
| FailoverReason.overloaded,
|
| FailoverReason.context_overflow,
|
| FailoverReason.payload_too_large,
|
| FailoverReason.long_context_tier,
|
| FailoverReason.thinking_signature,
|
| }
|
| )
|
| ) and not is_context_length_error
|
|
|
| if is_client_error:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| _is_copilot_provider(agent)
|
| and not _retry.copilot_stale_cred_retry_attempted
|
| and _is_stale_copilot_credential_error(
|
| status_code, str(getattr(api_error, "message", "") or api_error)
|
| )
|
| ):
|
| _retry.copilot_stale_cred_retry_attempted = True
|
| if agent._try_recover_stale_copilot_credential():
|
| agent._buffer_vprint(
|
| "π Copilot credential re-exchanged after "
|
| "model_not_available 400. Retrying request..."
|
| )
|
| retry_count = 0
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
| if agent._has_pending_fallback():
|
| if classified.reason == FailoverReason.content_policy_blocked:
|
| agent._buffer_status("β οΈ Provider safety filter blocked this request β trying fallback...")
|
| elif classified.reason == FailoverReason.ssl_cert_verification:
|
| agent._buffer_status("β οΈ TLS certificate verification failed β trying fallback...")
|
| else:
|
| agent._buffer_status(f"β οΈ Non-retryable error (HTTP {status_code}) β trying fallback...")
|
| if agent._try_activate_fallback():
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| continue
|
| if api_kwargs is not None:
|
| agent._dump_api_request_debug(
|
| api_kwargs, reason="non_retryable_client_error", error=api_error,
|
| )
|
|
|
|
|
| agent._flush_status_buffer()
|
|
|
|
|
|
|
|
|
|
|
|
|
| _nonretryable_summary = agent._summarize_api_error(api_error)
|
| if classified.reason == FailoverReason.content_policy_blocked:
|
| agent._emit_status(
|
| f"β Provider safety filter blocked this request: "
|
| f"{_nonretryable_summary}"
|
| )
|
| elif classified.reason == FailoverReason.ssl_cert_verification:
|
| agent._emit_status(
|
| f"β TLS certificate verification failed: "
|
| f"{_nonretryable_summary}"
|
| )
|
| else:
|
| agent._emit_status(
|
| f"β Non-retryable error (HTTP {status_code}): "
|
| f"{_nonretryable_summary}"
|
| )
|
| agent._vprint(f"{agent.log_prefix}β Non-retryable client error (HTTP {status_code}). Aborting.", force=True)
|
| agent._vprint(f"{agent.log_prefix} π Provider: {_provider} Model: {_model}", force=True)
|
| agent._vprint(f"{agent.log_prefix} π Endpoint: {_base}", force=True)
|
|
|
| if classified.is_auth or classified.reason == FailoverReason.billing:
|
| if classified.reason == FailoverReason.billing and _print_billing_or_entitlement_guidance(
|
| agent,
|
| capability="model access",
|
| provider=_provider,
|
| base_url=str(_base),
|
| model=_model,
|
| ):
|
| pass
|
| elif _provider == "nous" and _print_nous_entitlement_guidance(
|
| agent,
|
| "Nous model access",
|
| ):
|
| pass
|
| elif _provider in {"openai-codex", "xai-oauth", "nous"} and status_code == 401:
|
| if _provider == "openai-codex":
|
| agent._vprint(f"{agent.log_prefix} π‘ Codex OAuth token was rejected (HTTP 401). Your token may have been", force=True)
|
| agent._vprint(f"{agent.log_prefix} refreshed by another client (Codex CLI, VS Code). To fix:", force=True)
|
| agent._vprint(f"{agent.log_prefix} 1. Run `codex` in your terminal to generate fresh tokens.", force=True)
|
| agent._vprint(f"{agent.log_prefix} 2. Then run `hermes auth` to re-authenticate.", force=True)
|
| elif _provider == "xai-oauth":
|
| agent._vprint(f"{agent.log_prefix} π‘ xAI OAuth token was rejected (HTTP 401). To fix:", force=True)
|
| agent._vprint(f"{agent.log_prefix} re-authenticate with xAI Grok OAuth (SuperGrok / Premium+) from `hermes model`.", force=True)
|
| else:
|
| agent._vprint(f"{agent.log_prefix} π‘ Nous Portal OAuth token was rejected (HTTP 401). Your token may be", force=True)
|
| agent._vprint(f"{agent.log_prefix} expired, revoked, or your account may be out of credits. To fix:", force=True)
|
| agent._vprint(f"{agent.log_prefix} 1. Re-authenticate: hermes portal", force=True)
|
| agent._vprint(f"{agent.log_prefix} 2. Check your portal account: https://portal.nousresearch.com", force=True)
|
|
|
|
|
| if isinstance(_model, str) and _model.endswith(":free"):
|
| agent._vprint(f"{agent.log_prefix} β οΈ Note: `{_model}` looks like an OpenRouter slug (`:free` suffix).", force=True)
|
| agent._vprint(f"{agent.log_prefix} Nous Portal won't recognize that model name. Either switch to a", force=True)
|
| agent._vprint(f"{agent.log_prefix} Nous catalog model, or run `/model openrouter:{_model}` to use OpenRouter.", force=True)
|
| else:
|
| agent._vprint(f"{agent.log_prefix} π‘ Your API key was rejected by the provider. Check:", force=True)
|
| agent._vprint(f"{agent.log_prefix} β’ Is the key valid? Run: hermes setup", force=True)
|
| agent._vprint(f"{agent.log_prefix} β’ Does your account have access to {_model}?", force=True)
|
| if base_url_host_matches(str(_base), "openrouter.ai"):
|
| agent._vprint(f"{agent.log_prefix} β’ Check credits: https://openrouter.ai/settings/credits", force=True)
|
| else:
|
| agent._vprint(f"{agent.log_prefix} π‘ This type of error won't be fixed by retrying.", force=True)
|
|
|
|
|
|
|
|
|
|
|
| if classified.reason == FailoverReason.content_policy_blocked:
|
| agent._vprint(
|
| f"{agent.log_prefix} π‘ The provider's safety filter rejected this specific prompt.",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} β’ Try rephrasing the request, narrowing the context, or splitting into smaller steps.",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} β’ Configure a fallback provider so future blocks route automatically:",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} hermes fallback add (interactive picker β same as `hermes model`)",
|
| force=True,
|
| )
|
|
|
|
|
|
|
|
|
| if classified.reason == FailoverReason.ssl_cert_verification:
|
| agent._vprint(
|
| f"{agent.log_prefix} π‘ The TLS certificate chain could not be verified. This fails the same",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} way on every retry β fix the environment, then try again:",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} β’ Corporate TLS-inspecting proxy? Point Python at its CA bundle:",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} export SSL_CERT_FILE=/path/to/corp-ca.pem (also REQUESTS_CA_BUNDLE)",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} β’ Missing/stale system CA store? Install/refresh it:",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} pip install --upgrade certifi (macOS: run 'Install Certificates.command')",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} β’ Self-signed local endpoint (llama.cpp, LM Studio, vLLM)? Use http://",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} for localhost, or add the server's cert to your trust store.",
|
| force=True,
|
| )
|
| logger.error("%sNon-retryable client error: %s", agent.log_prefix, api_error)
|
|
|
|
|
|
|
|
|
|
|
| if status_code == 400 and (approx_tokens > 50000 or len(api_messages) > 80):
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Skipping session persistence "
|
| f"for large failed session to prevent growth loop.",
|
| force=True,
|
| )
|
| else:
|
| agent._persist_session(messages, conversation_history)
|
| if classified.reason == FailoverReason.content_policy_blocked:
|
| _policy_response = (
|
| "β οΈ The model provider's safety filter blocked this request "
|
| "(not a Hermes/gateway failure).\n\n"
|
| f"Provider message: {_nonretryable_summary}\n\n"
|
| f"{_CONTENT_POLICY_RECOVERY_HINT}"
|
| )
|
| return _content_policy_blocked_result(
|
| messages,
|
| api_call_count,
|
| final_response=_policy_response,
|
| error_detail=_nonretryable_summary,
|
| )
|
|
|
|
|
|
|
|
|
| if classified.reason == FailoverReason.billing:
|
| _ce_guidance = _billing_or_entitlement_message(
|
| capability="model access",
|
| provider=_provider,
|
| base_url=str(_base),
|
| model=_model,
|
| )
|
| _ce_final = f"Billing or credits exhausted: {_nonretryable_summary}"
|
| if _ce_guidance:
|
| _ce_final += f"\n\n{_ce_guidance}"
|
| _ce_block = _billing_block_dict(_provider, _base, _model, _ce_guidance)
|
| return {
|
| "final_response": _ce_final,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "failed": True,
|
| "error": _nonretryable_summary,
|
| "failure_reason": classified.reason.value,
|
| "billing_block": _ce_block,
|
| }
|
| return {
|
| "final_response": _nonretryable_summary,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "failed": True,
|
| "error": _nonretryable_summary,
|
| }
|
|
|
| if retry_count >= max_retries:
|
|
|
|
|
|
|
|
|
| if not _retry.primary_recovery_attempted and agent._try_recover_primary_transport(
|
| api_error, retry_count=retry_count, max_retries=max_retries,
|
| ):
|
| _retry.primary_recovery_attempted = True
|
| retry_count = 0
|
|
|
|
|
|
|
|
|
| _retry.has_retried_429 = False
|
| agent._fallback_index = 0
|
| agent._fallback_activated = False
|
| continue
|
|
|
| if agent._has_pending_fallback():
|
| agent._buffer_status(f"β οΈ Max retries ({max_retries}) exhausted β trying fallback...")
|
| if agent._try_activate_fallback():
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| retry_count = 0
|
| compression_attempts = 0
|
| _retry.primary_recovery_attempted = False
|
| continue
|
|
|
| agent._flush_status_buffer()
|
| _final_summary = agent._summarize_api_error(api_error)
|
| _billing_guidance = ""
|
| if classified.reason == FailoverReason.billing:
|
| agent._emit_status(f"β Billing or credits exhausted β {_final_summary}")
|
| _billing_guidance = _billing_or_entitlement_message(
|
| capability="model access",
|
| provider=_provider,
|
| base_url=str(_base),
|
| model=_model,
|
| )
|
| _print_billing_or_entitlement_guidance(
|
| agent,
|
| capability="model access",
|
| provider=_provider,
|
| base_url=str(_base),
|
| model=_model,
|
| )
|
| elif is_rate_limited:
|
| agent._emit_status(f"β Rate limited after {max_retries} retries β {_final_summary}")
|
| else:
|
| agent._emit_status(f"β API failed after {max_retries} retries β {_final_summary}")
|
| agent._vprint(f"{agent.log_prefix} π Final error: {_final_summary}", force=True)
|
|
|
|
|
|
|
|
|
|
|
|
|
| _is_stream_drop = (
|
| not getattr(api_error, "status_code", None)
|
| and any(p in error_msg for p in (
|
| "connection lost", "connection reset",
|
| "connection closed", "network connection",
|
| "network error", "terminated",
|
| ))
|
| )
|
| if _is_stream_drop:
|
| agent._vprint(
|
| f"{agent.log_prefix} π‘ The provider's stream "
|
| f"connection keeps dropping. This often happens "
|
| f"when the model tries to write a very large "
|
| f"file in a single tool call.",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} Try asking the model "
|
| f"to use execute_code with Python's open() for "
|
| f"large files, or to write the file in smaller "
|
| f"sections.",
|
| force=True,
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| from agent.thinking_timeout_guidance import (
|
| is_thinking_timeout,
|
| )
|
| _is_thinking_timeout = is_thinking_timeout(
|
| classified,
|
| _model,
|
| error_msg,
|
| )
|
| if _is_thinking_timeout:
|
| agent._vprint(
|
| f"{agent.log_prefix} π‘ The model's thinking "
|
| f"phase exceeded the upstream proxy's idle "
|
| f"timeout before the first content token "
|
| f"arrived. This is a known issue with "
|
| f"reasoning models behind cloud gateways "
|
| f"(NVIDIA NIM, OpenAI, Anthropic, DeepSeek).",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} Workarounds in priority order:",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} 1. Set "
|
| f"`providers.{_provider}.models.{_model}.stale_timeout_seconds: 900` "
|
| f"in `~/.hermes/config.yaml` to extend the per-call "
|
| f"timeout. (Hermes's built-in floor is 600s for "
|
| f"known reasoning models β if you still see this "
|
| f"after raising, the upstream cap is even shorter.)",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} 2. Lower `reasoning_budget` or set "
|
| f"`reasoning_effort: medium` on this model if the provider supports it.",
|
| force=True,
|
| )
|
| agent._vprint(
|
| f"{agent.log_prefix} 3. Use a smaller / faster reasoning "
|
| f"model if the task doesn't require deep thinking.",
|
| force=True,
|
| )
|
|
|
| logger.error(
|
| "%sAPI call failed after %s retries. %s | provider=%s model=%s msgs=%s tokens=~%s",
|
| agent.log_prefix, max_retries, _final_summary,
|
| _provider, _model, len(api_messages), f"{approx_tokens:,}",
|
| )
|
| if api_kwargs is not None:
|
| agent._dump_api_request_debug(
|
| api_kwargs, reason="max_retries_exhausted", error=api_error,
|
| )
|
| agent._persist_session(messages, conversation_history)
|
| _billing_block = None
|
| if classified.reason == FailoverReason.billing:
|
| _final_response = f"Billing or credits exhausted: {_final_summary}"
|
| if _billing_guidance:
|
| _final_response += f"\n\n{_billing_guidance}"
|
|
|
|
|
| _billing_block = _billing_block_dict(_provider, _base, _model, _billing_guidance)
|
| else:
|
| _final_response = f"API call failed after {max_retries} retries: {_final_summary}"
|
| if _is_thinking_timeout:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| from agent.thinking_timeout_guidance import (
|
| build_thinking_timeout_guidance,
|
| )
|
| _final_response += build_thinking_timeout_guidance(
|
| provider=_provider,
|
| model=_model,
|
| )
|
| elif _is_stream_drop:
|
| _final_response += (
|
| "\n\nThe provider's stream connection keeps "
|
| "dropping β this often happens when generating "
|
| "very large tool call responses (e.g. write_file "
|
| "with long content). Try asking me to use "
|
| "execute_code with Python's open() for large "
|
| "files, or to write in smaller sections."
|
| )
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "failed": True,
|
| "error": _final_summary,
|
|
|
|
|
|
|
|
|
|
|
| "failure_reason": classified.reason.value,
|
|
|
|
|
| "billing_block": _billing_block,
|
| }
|
|
|
|
|
| _retry_after = None
|
| if is_rate_limited:
|
| _resp_headers = getattr(getattr(api_error, "response", None), "headers", None)
|
| if _resp_headers and hasattr(_resp_headers, "get"):
|
| _ra_raw = _resp_headers.get("retry-after") or _resp_headers.get("Retry-After")
|
| if _ra_raw:
|
| try:
|
|
|
|
|
|
|
|
|
|
|
| _retry_after = min(float(_ra_raw), 600)
|
| except (TypeError, ValueError):
|
| pass
|
| wait_time = _retry_after if _retry_after else jittered_backoff(retry_count, base_delay=2.0, max_delay=60.0)
|
| _backoff_policy = None
|
| if (is_rate_limited or _is_zai_coding_overload) and not _retry_after:
|
| wait_time, _backoff_policy = adaptive_rate_limit_backoff(
|
| retry_count,
|
| base_url=str(_base),
|
| model=_model,
|
| error=api_error,
|
| default_wait=wait_time,
|
| )
|
| if is_rate_limited or _is_zai_coding_overload:
|
| _policy_note = ""
|
| if _backoff_policy == "zai_coding_overload_long":
|
| _policy_note = " (Z.AI Coding overload adaptive long backoff)"
|
| elif _backoff_policy == "zai_coding_overload_short":
|
| _policy_note = " (Z.AI Coding overload short retry)"
|
| _wait_reason = "Provider overloaded" if _is_zai_coding_overload and not is_rate_limited else "Rate limited"
|
| _rate_limit_status = f"β±οΈ {_wait_reason}. Waiting {wait_time:.1f}s (attempt {retry_count + 1}/{max_retries}){_policy_note}..."
|
|
|
|
|
|
|
| if _backoff_policy == "zai_coding_overload_long":
|
| agent._emit_status(_rate_limit_status)
|
| else:
|
| agent._buffer_status(_rate_limit_status)
|
| else:
|
| agent._buffer_status(f"β³ Retrying in {wait_time:.1f}s (attempt {retry_count}/{max_retries})...")
|
| logger.warning(
|
| "Retrying API call in %ss (attempt %s/%s) %s policy=%s error=%s",
|
| wait_time,
|
| retry_count,
|
| max_retries,
|
| agent._client_log_context(),
|
| _backoff_policy or "default",
|
| api_error,
|
| )
|
|
|
|
|
| sleep_end = time.time() + wait_time
|
| _backoff_touch_counter = 0
|
| while time.time() < sleep_end:
|
| if agent._interrupt_requested:
|
|
|
|
|
|
|
| if agent.clear_interrupt(preserve_redirect=True):
|
| _retry.restart_with_redirected_messages = True
|
| break
|
| agent._vprint(f"{agent.log_prefix}β‘ Interrupt detected during retry wait, aborting.", force=True)
|
| _interrupt_text = f"Operation interrupted: retrying API call after error (retry {retry_count}/{max_retries})."
|
| close_interrupted_tool_sequence(messages, _interrupt_text)
|
| agent._persist_session(messages, conversation_history)
|
| agent.clear_interrupt()
|
| return {
|
| "final_response": _interrupt_text,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "interrupted": True,
|
| }
|
| time.sleep(0.2)
|
|
|
|
|
| _backoff_touch_counter += 1
|
| if _backoff_touch_counter % 150 == 0:
|
| agent._touch_activity(
|
| f"error retry backoff ({retry_count}/{max_retries}), "
|
| f"{int(sleep_end - time.time())}s remaining"
|
| )
|
| if _retry.restart_with_redirected_messages:
|
|
|
|
|
|
|
| break
|
|
|
| if _retry.restart_with_redirected_messages:
|
|
|
|
|
|
|
| api_call_count -= 1
|
| agent.iteration_budget.refund()
|
| _retry.restart_with_redirected_messages = False
|
| continue
|
|
|
|
|
| if interrupted:
|
| _turn_exit_reason = "interrupted_during_api_call"
|
| break
|
|
|
| if _retry.restart_with_compressed_messages:
|
| api_call_count -= 1
|
| agent.iteration_budget.refund()
|
|
|
|
|
|
|
| retry_count += 1
|
| _retry.restart_with_compressed_messages = False
|
|
|
|
|
|
|
|
|
|
|
|
|
| current_turn_user_idx = reanchor_current_turn_user_idx(
|
| messages, user_message
|
| )
|
| agent._persist_user_message_idx = current_turn_user_idx
|
| continue
|
|
|
| if _retry.restart_with_rebuilt_messages:
|
|
|
|
|
|
|
|
|
|
|
| api_call_count -= 1
|
| agent.iteration_budget.refund()
|
| _retry.restart_with_rebuilt_messages = False
|
| continue
|
|
|
| if _retry.restart_with_length_continuation:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _boost_base = agent.max_tokens if agent.max_tokens else 4096
|
| _boost = _boost_base * (2 ** length_continue_retries)
|
| _requested_cap = agent._requested_output_cap_from_api_kwargs(api_kwargs)
|
| if _requested_cap is not None:
|
| _boost = max(_boost, _requested_cap)
|
| _boost_cap = max(32768, _requested_cap or 0)
|
| agent._ephemeral_max_output_tokens = min(_boost, _boost_cap)
|
| continue
|
|
|
|
|
|
|
|
|
| if response is None:
|
| _turn_exit_reason = "all_retries_exhausted_no_response"
|
| print(f"{agent.log_prefix}β All API retries exhausted with no successful response.")
|
| agent._persist_session(messages, conversation_history)
|
| break
|
|
|
| try:
|
| _transport = agent._get_transport()
|
| _normalize_kwargs = {}
|
| if agent.api_mode == "anthropic_messages":
|
| _normalize_kwargs["strip_tool_prefix"] = agent._is_anthropic_oauth
|
| normalized = _transport.normalize_response(response, **_normalize_kwargs)
|
| assistant_message = normalized
|
| finish_reason = normalized.finish_reason
|
|
|
|
|
|
|
|
|
| if assistant_message.content is not None and not isinstance(assistant_message.content, str):
|
| raw = assistant_message.content
|
| if isinstance(raw, dict):
|
| assistant_message.content = raw.get("text", "") or raw.get("content", "") or json.dumps(raw)
|
| elif isinstance(raw, list):
|
|
|
| parts = []
|
| for part in raw:
|
| if isinstance(part, str):
|
| parts.append(part)
|
| elif isinstance(part, dict) and part.get("type") == "text":
|
| parts.append(part.get("text", ""))
|
| elif isinstance(part, dict) and "text" in part:
|
| parts.append(str(part["text"]))
|
| assistant_message.content = "\n".join(parts)
|
| else:
|
| assistant_message.content = str(raw)
|
|
|
| try:
|
| from hermes_cli.lifecycle import (
|
| has_hook,
|
| invoke_hook as _invoke_hook,
|
| )
|
| if has_hook("post_api_request"):
|
| _assistant_tool_calls = (
|
| getattr(assistant_message, "tool_calls", None) or []
|
| )
|
| _assistant_text = assistant_message.content or ""
|
| _api_ended_at = api_start_time + api_duration
|
| _invoke_hook(
|
| "post_api_request",
|
| task_id=effective_task_id,
|
| turn_id=turn_id,
|
| api_request_id=api_request_id,
|
| session_id=agent.session_id or "",
|
| platform=agent.platform or "",
|
| model=agent.model,
|
| provider=agent.provider,
|
| base_url=agent.base_url,
|
| api_mode=agent.api_mode,
|
| api_call_count=api_call_count,
|
| api_duration=api_duration,
|
| started_at=api_start_time,
|
| ended_at=_api_ended_at,
|
| finish_reason=finish_reason,
|
| message_count=len(api_messages),
|
| response_model=getattr(response, "model", None),
|
| response=agent._api_response_payload_for_hook(
|
| response,
|
| assistant_message,
|
| finish_reason=finish_reason,
|
| ),
|
| usage=agent._usage_summary_for_api_request_hook(response),
|
| assistant_message=assistant_message,
|
| assistant_content_chars=len(_assistant_text),
|
| assistant_tool_call_count=len(_assistant_tool_calls),
|
| )
|
| except Exception:
|
| pass
|
|
|
|
|
| if assistant_message.content and not agent.quiet_mode:
|
| if agent.verbose_logging:
|
| agent._vprint(f"{agent.log_prefix}π€ Assistant: {assistant_message.content}")
|
| else:
|
| agent._vprint(f"{agent.log_prefix}π€ Assistant: {assistant_message.content[:100]}{'...' if len(assistant_message.content) > 100 else ''}")
|
|
|
|
|
|
|
| if (assistant_message.content and agent.tool_progress_callback):
|
| _think_text = assistant_message.content.strip()
|
|
|
| _think_text = re.sub(
|
| r'</?(?:REASONING_SCRATCHPAD|think|reasoning)>', '', _think_text
|
| ).strip()
|
|
|
|
|
| first_line = _think_text.split('\n')[0][:80] if _think_text else ""
|
| if first_line and getattr(agent, '_delegate_depth', 0) > 0:
|
| try:
|
| agent.tool_progress_callback("_thinking", first_line)
|
| except Exception:
|
| pass
|
| elif _think_text:
|
| try:
|
| agent.tool_progress_callback("reasoning.available", "_thinking", _think_text[:500], None)
|
| except Exception:
|
| pass
|
|
|
|
|
|
|
| if has_incomplete_scratchpad(assistant_message.content or ""):
|
| agent._incomplete_scratchpad_retries += 1
|
|
|
| agent._buffer_vprint("β οΈ Incomplete <REASONING_SCRATCHPAD> detected (opened but never closed)")
|
|
|
| if agent._incomplete_scratchpad_retries <= 2:
|
| agent._buffer_vprint(f"π Retrying API call ({agent._incomplete_scratchpad_retries}/2)...")
|
|
|
| continue
|
| else:
|
|
|
| agent._flush_status_buffer()
|
| agent._vprint(f"{agent.log_prefix}β Max retries (2) for incomplete scratchpad. Saving as partial.", force=True)
|
| agent._incomplete_scratchpad_retries = 0
|
|
|
| rolled_back_messages = agent._get_messages_up_to_last_assistant(messages)
|
| agent._cleanup_task_resources(effective_task_id)
|
| agent._persist_session(messages, conversation_history)
|
|
|
| return {
|
| "final_response": "Incomplete REASONING_SCRATCHPAD after 2 retries",
|
| "messages": rolled_back_messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "partial": True,
|
| "error": "Incomplete REASONING_SCRATCHPAD after 2 retries"
|
| }
|
|
|
|
|
| agent._incomplete_scratchpad_retries = 0
|
|
|
| if agent.api_mode == "codex_responses" and finish_reason == "incomplete":
|
| agent._codex_incomplete_retries += 1
|
|
|
| interim_msg = agent._build_assistant_message(assistant_message, finish_reason)
|
| interim_has_content = bool((interim_msg.get("content") or "").strip())
|
| interim_has_reasoning = bool(interim_msg.get("reasoning", "").strip()) if isinstance(interim_msg.get("reasoning"), str) else False
|
| interim_has_codex_reasoning = bool(interim_msg.get("codex_reasoning_items"))
|
| interim_has_codex_message_items = bool(interim_msg.get("codex_message_items"))
|
|
|
| if (
|
| interim_has_content
|
| or interim_has_reasoning
|
| or interim_has_codex_reasoning
|
| or interim_has_codex_message_items
|
| ):
|
| last_msg = messages[-1] if messages else None
|
|
|
|
|
|
|
|
|
|
|
|
|
| last_interim_visible = (
|
| agent._interim_assistant_visible_text(last_msg)
|
| if isinstance(last_msg, dict)
|
| else ""
|
| )
|
| current_interim_visible = agent._interim_assistant_visible_text(interim_msg)
|
| if last_interim_visible or current_interim_visible:
|
| same_visible_output = last_interim_visible == current_interim_visible
|
| else:
|
|
|
|
|
| same_visible_output = (
|
| (last_msg.get("content") or "") == (interim_msg.get("content") or "")
|
| and (last_msg.get("reasoning") or "") == (interim_msg.get("reasoning") or "")
|
| ) if isinstance(last_msg, dict) else False
|
| visible_duplicate = (
|
| isinstance(last_msg, dict)
|
| and last_msg.get("role") == "assistant"
|
| and last_msg.get("finish_reason") == "incomplete"
|
| and same_visible_output
|
| )
|
| if visible_duplicate:
|
|
|
|
|
|
|
| for _key in (
|
| "content",
|
| "reasoning",
|
| "reasoning_content",
|
| "reasoning_details",
|
| "codex_reasoning_items",
|
| "codex_message_items",
|
| ):
|
| if _key in interim_msg:
|
| last_msg[_key] = interim_msg[_key]
|
| else:
|
| messages.append(interim_msg)
|
| agent._emit_interim_assistant_message(interim_msg)
|
|
|
| if agent._codex_incomplete_retries < 3:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| interim_replayable = (
|
| interim_has_content
|
| or interim_has_codex_reasoning
|
| or interim_has_codex_message_items
|
| )
|
| if not interim_replayable:
|
| _last_msg = messages[-1] if messages else None
|
| _already_nudged = (
|
| isinstance(_last_msg, dict)
|
| and _last_msg.get("role") == "user"
|
| and _last_msg.get("content") == _CODEX_INCOMPLETE_NUDGE
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _last_is_assistant = (
|
| isinstance(_last_msg, dict)
|
| and _last_msg.get("role") == "assistant"
|
| )
|
| if not _already_nudged and _last_is_assistant:
|
| messages.append({
|
| "role": "user",
|
| "content": _CODEX_INCOMPLETE_NUDGE,
|
| })
|
| if not agent.quiet_mode:
|
| agent._vprint(f"{agent.log_prefix}β» Codex response incomplete; continuing turn ({agent._codex_incomplete_retries}/3)")
|
|
|
|
|
|
|
|
|
|
|
| agent._emit_wait_notice(
|
| f"β» model returned reasoning with no final answer β "
|
| f"asking it to continue "
|
| f"({agent._codex_incomplete_retries}/3)"
|
| )
|
| agent._session_messages = messages
|
| continue
|
|
|
| agent._codex_incomplete_retries = 0
|
| agent._persist_session(messages, conversation_history)
|
| return {
|
| "final_response": "Codex response remained incomplete after 3 continuation attempts",
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "partial": True,
|
| "error": "Codex response remained incomplete after 3 continuation attempts",
|
| }
|
| elif hasattr(agent, "_codex_incomplete_retries"):
|
| agent._codex_incomplete_retries = 0
|
|
|
|
|
| if assistant_message.tool_calls:
|
| if not agent.quiet_mode:
|
| agent._vprint(f"{agent.log_prefix}π§ Processing {len(assistant_message.tool_calls)} tool call(s)...")
|
|
|
| if agent.verbose_logging:
|
| for tc in assistant_message.tool_calls:
|
| raw_args = tc.function.arguments
|
| args_preview = raw_args[:200] if isinstance(raw_args, str) else repr(raw_args)[:200]
|
| logging.debug("Tool call: %s with args: %s...", tc.function.name, args_preview)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._uniquify_tool_call_ids(assistant_message.tool_calls)
|
|
|
|
|
|
|
| for tc in assistant_message.tool_calls:
|
| if tc.function.name not in agent.valid_tool_names:
|
| repaired = agent._repair_tool_call(tc.function.name)
|
| if repaired:
|
| print(f"{agent.log_prefix}π§ Auto-repaired tool name: '{tc.function.name}' -> '{repaired}'")
|
| tc.function.name = repaired
|
| invalid_tool_calls = [
|
| tc.function.name for tc in assistant_message.tool_calls
|
| if tc.function.name not in agent.valid_tool_names
|
| ]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _mixed_invalid_batch = bool(invalid_tool_calls) and any(
|
| tc.function.name in agent.valid_tool_names
|
| for tc in assistant_message.tool_calls
|
| )
|
| if _mixed_invalid_batch:
|
| agent._invalid_tool_retries = 0
|
| invalid_name = invalid_tool_calls[0]
|
| invalid_preview = invalid_name[:80] + "..." if len(invalid_name) > 80 else invalid_name
|
| _n_valid = sum(
|
| 1 for tc in assistant_message.tool_calls
|
| if tc.function.name in agent.valid_tool_names
|
| )
|
| agent._buffer_vprint(
|
| f"β οΈ Unknown tool '{invalid_preview}' in batch β erroring that call, "
|
| f"executing {_n_valid} valid call(s)"
|
| )
|
| elif invalid_tool_calls:
|
|
|
| agent._invalid_tool_retries += 1
|
|
|
|
|
| invalid_name = invalid_tool_calls[0]
|
| invalid_preview = invalid_name[:80] + "..." if len(invalid_name) > 80 else invalid_name
|
| agent._buffer_vprint(f"β οΈ Unknown tool '{invalid_preview}' β sending error to model for agent-correction ({agent._invalid_tool_retries}/3)")
|
|
|
| if agent._invalid_tool_retries >= 3:
|
| agent._flush_status_buffer()
|
| agent._vprint(f"{agent.log_prefix}β Max retries (3) for invalid tool calls exceeded. Stopping as partial.", force=True)
|
| agent._invalid_tool_retries = 0
|
| _final_response = f"Model generated invalid tool call: {invalid_preview}"
|
|
|
|
|
|
|
|
|
| close_interrupted_tool_sequence(messages, _final_response)
|
| agent._persist_session(messages, conversation_history)
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "partial": True,
|
| "error": _final_response
|
| }
|
|
|
| assistant_msg = agent._build_assistant_message(assistant_message, finish_reason)
|
| messages.append(assistant_msg)
|
| for tc in assistant_message.tool_calls:
|
| _tc_name = tc.function.name
|
| if _tc_name not in agent.valid_tool_names:
|
|
|
|
|
| content = _invalid_tool_name_error_content(
|
| _tc_name, agent.valid_tool_names
|
| )
|
| else:
|
| content = "Skipped: another tool call in this turn used an invalid name. Please retry this tool call."
|
| messages.append({
|
| "role": "tool",
|
| "name": tc.function.name,
|
| "tool_call_id": tc.id,
|
| "content": content,
|
| })
|
| continue
|
|
|
| agent._invalid_tool_retries = 0
|
|
|
|
|
|
|
| invalid_json_args = []
|
| for tc in assistant_message.tool_calls:
|
| args = tc.function.arguments
|
| if isinstance(args, (dict, list)):
|
| tc.function.arguments = json.dumps(args)
|
| continue
|
| if args is not None and not isinstance(args, str):
|
| tc.function.arguments = str(args)
|
| args = tc.function.arguments
|
|
|
| if not args or not args.strip():
|
| tc.function.arguments = "{}"
|
| continue
|
| try:
|
| json.loads(args)
|
| except json.JSONDecodeError as e:
|
| if (
|
| _mixed_invalid_batch
|
| and tc.function.name not in agent.valid_tool_names
|
| ):
|
|
|
|
|
|
|
| continue
|
| invalid_json_args.append((tc.function.name, str(e)))
|
|
|
| if invalid_json_args:
|
|
|
|
|
|
|
|
|
|
|
|
|
| _truncated = any(
|
| not (tc.function.arguments or "").rstrip().endswith(("}", "]"))
|
| for tc in assistant_message.tool_calls
|
| if tc.function.name in {n for n, _ in invalid_json_args}
|
| )
|
| if _truncated:
|
| agent._vprint(
|
| f"{agent.log_prefix}β οΈ Truncated tool call arguments detected "
|
| f"(finish_reason={finish_reason!r}) β refusing to execute.",
|
| force=True,
|
| )
|
| agent._invalid_json_retries = 0
|
| agent._cleanup_task_resources(effective_task_id)
|
| _final_response = "Response truncated due to output length limit"
|
|
|
|
|
| close_interrupted_tool_sequence(messages, _final_response)
|
| agent._persist_session(messages, conversation_history)
|
| return {
|
| "final_response": _final_response,
|
| "messages": messages,
|
| "api_calls": api_call_count,
|
| "completed": False,
|
| "partial": True,
|
| "error": _final_response,
|
| }
|
|
|
|
|
| agent._invalid_json_retries += 1
|
|
|
| tool_name, error_msg = invalid_json_args[0]
|
| agent._buffer_vprint(f"β οΈ Invalid JSON in tool call arguments for '{tool_name}': {error_msg}")
|
|
|
| if agent._invalid_json_retries < 3:
|
| agent._buffer_vprint(f"π Retrying API call ({agent._invalid_json_retries}/3)...")
|
|
|
| continue
|
| else:
|
|
|
|
|
| agent._buffer_vprint("β οΈ Injecting recovery tool results for invalid JSON...")
|
| agent._invalid_json_retries = 0
|
|
|
|
|
| recovery_assistant = agent._build_assistant_message(assistant_message, finish_reason)
|
| messages.append(recovery_assistant)
|
|
|
|
|
| invalid_names = {name for name, _ in invalid_json_args}
|
| for tc in assistant_message.tool_calls:
|
| if tc.function.name in invalid_names:
|
| err = next(e for n, e in invalid_json_args if n == tc.function.name)
|
| tool_result = (
|
| f"Error: Invalid JSON arguments. {err}. "
|
| f"For tools with no required parameters, use an empty object: {{}}. "
|
| f"Please retry with valid JSON."
|
| )
|
| else:
|
| tool_result = "Skipped: other tool call in this response had invalid JSON."
|
| messages.append({
|
| "role": "tool",
|
| "name": tc.function.name,
|
| "tool_call_id": tc.id,
|
| "content": tool_result,
|
| })
|
| continue
|
|
|
|
|
| agent._invalid_json_retries = 0
|
|
|
|
|
| assistant_message.tool_calls = agent._cap_delegate_task_calls(
|
| assistant_message.tool_calls
|
| )
|
| assistant_message.tool_calls = agent._deduplicate_tool_calls(
|
| assistant_message.tool_calls
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
| _invalid_batch_calls = []
|
| if _mixed_invalid_batch:
|
| _invalid_batch_calls = [
|
| tc for tc in assistant_message.tool_calls
|
| if tc.function.name not in agent.valid_tool_names
|
| ]
|
|
|
| assistant_msg = agent._build_assistant_message(assistant_message, finish_reason)
|
|
|
| turn_content = assistant_message.content or ""
|
|
|
|
|
|
|
|
|
| _HOUSEKEEPING_TOOLS = frozenset({
|
| "memory", "todo", "skill_manage", "session_search",
|
| })
|
| _all_housekeeping = all(
|
| tc.function.name in _HOUSEKEEPING_TOOLS
|
| for tc in assistant_message.tool_calls
|
| )
|
|
|
|
|
|
|
|
|
| if assistant_message.tool_calls and not _all_housekeeping:
|
| agent._last_content_with_tools = None
|
| agent._last_content_tools_all_housekeeping = False
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._mute_post_response = False
|
|
|
|
|
|
|
|
|
|
|
| if turn_content and agent._has_content_after_think_block(turn_content):
|
| agent._last_content_with_tools = turn_content
|
|
|
|
|
|
|
|
|
|
|
| agent._last_content_tools_all_housekeeping = _all_housekeeping
|
| if _all_housekeeping and agent._has_stream_consumers():
|
| agent._mute_post_response = True
|
| elif agent._should_emit_quiet_tool_messages():
|
| clean = agent._strip_think_blocks(turn_content).strip()
|
| if clean:
|
| agent._vprint(f" β π¬ {clean}")
|
|
|
|
|
|
|
| _had_prefill = False
|
| while (
|
| messages
|
| and isinstance(messages[-1], dict)
|
| and messages[-1].get("_thinking_prefill")
|
| ):
|
| messages.pop()
|
| _had_prefill = True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if _had_prefill:
|
| agent._thinking_prefill_retries = 0
|
| agent._empty_content_retries = 0
|
|
|
|
|
|
|
| agent._post_tool_empty_retried = False
|
|
|
|
|
|
|
| agent._dropped_toolcall_retries = 0
|
|
|
| previous_msg = messages[-1] if messages else None
|
| current_interim_visible = agent._interim_assistant_visible_text(assistant_msg)
|
| previous_interim_visible = (
|
| agent._interim_assistant_visible_text(previous_msg)
|
| if isinstance(previous_msg, dict)
|
| else ""
|
| )
|
| duplicate_previous_interim = (
|
| bool(current_interim_visible)
|
| and isinstance(previous_msg, dict)
|
| and previous_msg.get("role") == "assistant"
|
| and previous_msg.get("finish_reason") == "incomplete"
|
| and previous_interim_visible == current_interim_visible
|
| )
|
| messages.append(assistant_msg)
|
|
|
|
|
|
|
|
|
|
|
|
|
| if _invalid_batch_calls:
|
| for tc in _invalid_batch_calls:
|
| messages.append({
|
| "role": "tool",
|
| "name": tc.function.name,
|
| "tool_call_id": tc.id,
|
| "content": _invalid_tool_name_error_content(
|
| tc.function.name, agent.valid_tool_names
|
| ),
|
| })
|
| assistant_message.tool_calls = [
|
| tc for tc in assistant_message.tool_calls
|
| if tc.function.name in agent.valid_tool_names
|
| ]
|
|
|
| _tool_turn_persisted = None
|
| try:
|
|
|
|
|
|
|
|
|
| _tool_turn_persisted = agent._flush_messages_to_session_db(
|
| messages, conversation_history
|
| )
|
| except Exception as exc:
|
| _tool_turn_persisted = False
|
| logger.warning(
|
| "Incremental tool-call persistence failed before execution "
|
| "(session=%s): %s",
|
| agent.session_id or "none",
|
| exc,
|
| )
|
|
|
| if _tool_turn_persisted is False:
|
|
|
|
|
|
|
|
|
| _turn_exit_reason = "session_persistence_failed"
|
| final_response = ""
|
| failed = True
|
| break
|
|
|
|
|
|
|
|
|
| if not duplicate_previous_interim:
|
| agent._emit_interim_assistant_message(assistant_msg)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if agent.stream_delta_callback:
|
| try:
|
| agent.stream_delta_callback(None)
|
| except Exception:
|
| pass
|
|
|
| agent._execute_tool_calls(assistant_message, messages, effective_task_id, api_call_count)
|
|
|
| if getattr(agent, "_incremental_persistence_failed", False):
|
|
|
|
|
|
|
| _turn_exit_reason = "session_persistence_failed"
|
| final_response = ""
|
| failed = True
|
| break
|
|
|
| if agent._tool_guardrail_halt_decision is not None:
|
| decision = agent._tool_guardrail_halt_decision
|
| _turn_exit_reason = "guardrail_halt"
|
| final_response = agent._toolguard_controlled_halt_response(decision)
|
| agent._emit_status(
|
| f"β οΈ Tool guardrail halted {decision.tool_name}: {decision.code}"
|
| )
|
| messages.append({"role": "assistant", "content": final_response})
|
|
|
|
|
|
|
|
|
|
|
| if final_response:
|
| agent._safe_print(f"\n{final_response}\n")
|
| if agent.stream_delta_callback:
|
| try:
|
| agent.stream_delta_callback(final_response)
|
| agent.stream_delta_callback(None)
|
| except Exception:
|
| pass
|
| break
|
|
|
|
|
|
|
|
|
| truncated_tool_call_retries = 0
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._stream_needs_break = True
|
|
|
|
|
|
|
|
|
| _tc_names = {tc.function.name for tc in assistant_message.tool_calls}
|
| if _tc_names == {"execute_code"}:
|
| agent.iteration_budget.refund()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _compressor = agent.context_compressor
|
| if _compressor.last_prompt_tokens > 0:
|
|
|
|
|
|
|
|
|
|
|
| _real_tokens = _compressor.last_prompt_tokens
|
| elif _compressor.last_prompt_tokens == -1:
|
|
|
|
|
|
|
| _real_tokens = 0
|
| else:
|
|
|
|
|
|
|
|
|
| _real_tokens = estimate_request_tokens_rough(
|
| messages, tools=agent.tools or None
|
| )
|
|
|
| if (
|
| agent.compression_enabled
|
| and compression_attempts < max_compression_attempts
|
| and _compressor.should_compress(_real_tokens)
|
| ):
|
| compression_attempts += 1
|
|
|
|
|
|
|
|
|
|
|
|
|
| _clear_warn = getattr(agent, "_clear_context_overflow_warn", None)
|
| if callable(_clear_warn):
|
| _clear_warn()
|
| agent._safe_print(" β³ compacting contextβ¦")
|
| _post_tool_input = messages
|
| messages, active_system_prompt = agent._compress_context(
|
| messages, system_message,
|
| approx_tokens=agent.context_compressor.last_prompt_tokens,
|
| task_id=effective_task_id,
|
| )
|
| if (
|
| messages is _post_tool_input
|
| and compression_skipped_due_to_lock(agent)
|
| ):
|
|
|
|
|
|
|
|
|
|
|
|
|
| compression_attempts -= 1
|
| else:
|
| conversation_history = conversation_history_after_compression(
|
| agent, messages, conversation_history
|
| )
|
| elif agent.compression_enabled:
|
|
|
|
|
|
|
|
|
|
|
| _block_reason = None
|
| _info = getattr(_compressor, "should_compress_info", None)
|
| if _info is not None:
|
| try:
|
| _block_reason = _info(_real_tokens)[1]
|
| except Exception:
|
| _block_reason = None
|
| if _block_reason:
|
| agent._warn_context_overflow_blocked(
|
| _block_reason,
|
| _real_tokens,
|
| int(getattr(_compressor, "threshold_tokens", 0) or 0),
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _prune = getattr(_compressor, "prune_tool_results_only", None)
|
| if callable(_prune):
|
| try:
|
| _pruned_msgs, _pruned_n = _prune(
|
| messages, current_tokens=_real_tokens
|
| )
|
| except Exception:
|
| logger.debug(
|
| "proactive tool-result prune failed; skipping",
|
| exc_info=True,
|
| )
|
| _pruned_msgs, _pruned_n = messages, 0
|
|
|
|
|
| if _pruned_n and _pruned_msgs is not messages:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| messages = _pruned_msgs
|
|
|
|
|
| agent._session_messages = messages
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._touch_activity(f"tool results posted, continuing iteration #{api_call_count}")
|
|
|
| continue
|
|
|
| else:
|
|
|
|
|
|
|
|
|
|
|
| final_response = assistant_message.content or ""
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._mute_post_response = False
|
|
|
|
|
| if not agent._has_content_after_think_block(final_response):
|
|
|
|
|
|
|
|
|
|
|
| _partial_streamed = (
|
| getattr(agent, "_current_streamed_assistant_text", "") or ""
|
| )
|
| if agent._has_content_after_think_block(_partial_streamed):
|
| _turn_exit_reason = "partial_stream_recovery"
|
| _recovered = agent._strip_think_blocks(_partial_streamed).strip()
|
| logger.info(
|
| "Partial stream content delivered (%d chars) "
|
| "β using as final response",
|
| len(_recovered),
|
| )
|
| agent._emit_status(
|
| "β» Stream interrupted β using delivered content "
|
| "as final response"
|
| )
|
| final_response = _recovered
|
|
|
|
|
|
|
|
|
| agent._response_was_previewed = False
|
| break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| fallback = getattr(agent, '_last_content_with_tools', None)
|
| if fallback and getattr(agent, '_last_content_tools_all_housekeeping', False):
|
| _turn_exit_reason = "fallback_prior_turn_content"
|
| logger.info("Empty follow-up after tool calls β using prior turn content as final response")
|
| agent._emit_status("β» Empty response after tool calls β using earlier content as final answer")
|
| agent._last_content_with_tools = None
|
| agent._last_content_tools_all_housekeeping = False
|
| agent._empty_content_retries = 0
|
|
|
|
|
|
|
|
|
| final_response = agent._strip_think_blocks(fallback).strip()
|
| agent._response_was_previewed = True
|
| break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _prior_was_tool = any(
|
| m.get("role") == "tool"
|
| for m in messages[-5:]
|
| )
|
|
|
|
|
|
|
|
|
|
|
| _has_inline_thinking = bool(
|
| re.search(
|
| r'<think>|<thinking>|<reasoning>',
|
| final_response or "",
|
| re.IGNORECASE,
|
| )
|
| )
|
| if (
|
| _prior_was_tool
|
| and not getattr(agent, "_post_tool_empty_retried", False)
|
| and not _has_inline_thinking
|
| ):
|
| agent._post_tool_empty_retried = True
|
|
|
|
|
| agent._last_content_with_tools = None
|
| agent._last_content_tools_all_housekeeping = False
|
| logger.info(
|
| "Empty response after tool calls β nudging model "
|
| "to continue processing"
|
| )
|
| agent._buffer_status(
|
| "β οΈ Model returned empty after tool calls β "
|
| "nudging to continue"
|
| )
|
|
|
|
|
|
|
|
|
|
|
| _nudge_msg = agent._build_assistant_message(assistant_message, finish_reason)
|
| _nudge_msg["content"] = "(empty)"
|
| _nudge_msg["_empty_recovery_synthetic"] = True
|
| messages.append(_nudge_msg)
|
| messages.append({
|
| "role": "user",
|
| "content": (
|
| "You just executed tool calls but returned an "
|
| "empty response. Please process the tool "
|
| "results above and continue with the task."
|
| ),
|
| "_empty_recovery_synthetic": True,
|
| })
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _has_structured = bool(
|
| getattr(assistant_message, "reasoning", None)
|
| or getattr(assistant_message, "reasoning_content", None)
|
| or getattr(assistant_message, "reasoning_details", None)
|
| or _has_inline_thinking
|
| )
|
| if _has_structured and agent._thinking_prefill_retries < 2:
|
| agent._thinking_prefill_retries += 1
|
| logger.info(
|
| "Thinking-only response (no visible content) β "
|
| "prefilling to continue (%d/2)",
|
| agent._thinking_prefill_retries,
|
| )
|
| agent._buffer_status(
|
| f"β» Thinking-only response β prefilling to continue "
|
| f"({agent._thinking_prefill_retries}/2)"
|
| )
|
| interim_msg = agent._build_assistant_message(
|
| assistant_message, "incomplete"
|
| )
|
| interim_msg["_thinking_prefill"] = True
|
| messages.append(interim_msg)
|
| agent._session_messages = messages
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _truly_empty = not agent._strip_think_blocks(
|
| final_response
|
| ).strip()
|
| _prefill_exhausted = (
|
| _has_structured
|
| and agent._thinking_prefill_retries >= 2
|
| )
|
| if _truly_empty and (not _has_structured or _prefill_exhausted) and agent._empty_content_retries < 3:
|
| agent._empty_content_retries += 1
|
| logger.warning(
|
| "Empty response (no content or reasoning) β "
|
| "retry %d/3 (model=%s)",
|
| agent._empty_content_retries, agent.model,
|
| )
|
| agent._buffer_status(
|
| f"β οΈ Empty response from model β retrying "
|
| f"({agent._empty_content_retries}/3)"
|
| )
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if _truly_empty and agent._fallback_chain:
|
| logger.warning(
|
| "Empty response after %d retries β "
|
| "attempting fallback (model=%s, provider=%s)",
|
| agent._empty_content_retries, agent.model,
|
| agent.provider,
|
| )
|
| agent._buffer_status(
|
| "β οΈ Model returning empty responses β "
|
| "switching to fallback provider..."
|
| )
|
| if agent._try_activate_fallback():
|
| active_system_prompt = _sync_failover_system_message(
|
| agent, api_messages, active_system_prompt)
|
| agent._empty_content_retries = 0
|
| agent._buffer_status(
|
| f"β» Switched to fallback: {agent.model} "
|
| f"({agent.provider})"
|
| )
|
| logger.info(
|
| "Fallback activated after empty responses: "
|
| "now using %s on %s",
|
| agent.model, agent.provider,
|
| )
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
| agent._flush_status_buffer()
|
| _turn_exit_reason = "empty_response_exhausted"
|
| reasoning_text = agent._extract_reasoning(assistant_message)
|
| agent._drop_trailing_empty_response_scaffolding(messages)
|
| assistant_msg = agent._build_assistant_message(assistant_message, finish_reason)
|
| assistant_msg["content"] = "(empty)"
|
|
|
|
|
|
|
|
|
|
|
| assistant_msg["_empty_terminal_sentinel"] = True
|
| messages.append(assistant_msg)
|
|
|
| if reasoning_text:
|
| reasoning_preview = reasoning_text[:500] + "..." if len(reasoning_text) > 500 else reasoning_text
|
| logger.warning(
|
| "Reasoning-only response (no visible content) "
|
| "after exhausting retries and fallback. "
|
| "Reasoning: %s", reasoning_preview,
|
| )
|
| agent._emit_status(
|
| "β οΈ Model produced reasoning but no visible "
|
| "response after all retries. Returning empty."
|
| )
|
| else:
|
| logger.warning(
|
| "Empty response (no content or reasoning) "
|
| "after %d retries. No fallback available. "
|
| "model=%s provider=%s",
|
| agent._empty_content_retries, agent.model,
|
| agent.provider,
|
| )
|
| agent._emit_status(
|
| "β Model returned no content after all retries"
|
| + (" and fallback attempts." if agent._fallback_chain else
|
| ". No fallback providers configured.")
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if reasoning_text:
|
| final_response = (
|
| "β οΈ The model produced only internal reasoning and "
|
| "no final answer, despite retries"
|
| + (" and fallback" if agent._fallback_chain else "")
|
| + ". Its last reasoning, which may contain the "
|
| "answer:\n\n" + reasoning_preview
|
| )
|
| else:
|
| final_response = "(empty)"
|
| break
|
|
|
|
|
| agent._empty_content_retries = 0
|
| agent._thinking_prefill_retries = 0
|
|
|
|
|
|
|
|
|
| agent._emit_pending_fallback_notice()
|
| agent._clear_status_buffer()
|
|
|
| from agent.agent_runtime_helpers import (
|
| intent_ack_continuation_mode,
|
| )
|
|
|
| _ack_mode = intent_ack_continuation_mode(agent)
|
| if (
|
| _ack_mode != "off"
|
| and agent.valid_tool_names
|
| and codex_ack_continuations < 2
|
| and agent._looks_like_codex_intermediate_ack(
|
| user_message=user_message,
|
| assistant_content=final_response,
|
| messages=messages,
|
| require_workspace=(_ack_mode == "codex_only"),
|
| )
|
| ):
|
| codex_ack_continuations += 1
|
| interim_msg = agent._build_assistant_message(assistant_message, "incomplete")
|
| messages.append(interim_msg)
|
| agent._emit_interim_assistant_message(interim_msg)
|
|
|
| continue_msg = {
|
| "role": "user",
|
| "content": (
|
| "[System: Continue now. Execute the required tool calls and only "
|
| "send your final answer after completing the task.]"
|
| ),
|
| }
|
| messages.append(continue_msg)
|
| agent._session_messages = messages
|
|
|
|
|
|
|
| final_response = None
|
| continue
|
|
|
| codex_ack_continuations = 0
|
|
|
| if truncated_response_parts:
|
| final_response = "".join(truncated_response_parts) + final_response
|
| truncated_response_parts = []
|
| length_continue_retries = 0
|
|
|
| final_response = agent._strip_think_blocks(final_response).strip()
|
|
|
| final_msg = agent._build_assistant_message(assistant_message, finish_reason)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| finish_reason == "tool_calls"
|
| and not assistant_message.tool_calls
|
| and getattr(agent, "_dropped_toolcall_retries", 0) < 3
|
| ):
|
| agent._dropped_toolcall_retries = getattr(agent, "_dropped_toolcall_retries", 0) + 1
|
| logger.warning(
|
| "finish_reason=tool_calls with empty tool_calls array "
|
| "(narration only) β re-prompting to emit the call "
|
| "(retry %d/3, model=%s provider=%s)",
|
| agent._dropped_toolcall_retries, agent.model, agent.provider,
|
| )
|
| agent._emit_status(
|
| "β» Model signaled a tool call but sent none β "
|
| f"re-prompting ({agent._dropped_toolcall_retries}/3)"
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| final_msg["_dropped_toolcall_nudge"] = True
|
| messages.append(final_msg)
|
| messages.append({
|
| "role": "user",
|
| "content": (
|
| "Your previous turn indicated a tool call but none was "
|
| "included. Do not narrate a plan or restate intent β issue "
|
| "the actual tool call now to continue the task."
|
| ),
|
| "_dropped_toolcall_nudge": True,
|
| })
|
| agent._session_messages = messages
|
| final_response = None
|
| continue
|
|
|
|
|
|
|
|
|
| agent._dropped_toolcall_retries = 0
|
|
|
|
|
|
|
|
|
|
|
|
|
| while (
|
| messages
|
| and isinstance(messages[-1], dict)
|
| and (
|
| messages[-1].get("_thinking_prefill")
|
| or messages[-1].get("_empty_recovery_synthetic")
|
| or messages[-1].get("_empty_terminal_sentinel")
|
| or messages[-1].get("_dropped_toolcall_nudge")
|
| )
|
| ):
|
| messages.pop()
|
|
|
| try:
|
| from agent.verification_stop import (
|
| build_verify_on_stop_nudge,
|
| verify_on_stop_enabled,
|
| )
|
|
|
| if verify_on_stop_enabled():
|
| _verify_nudge = build_verify_on_stop_nudge(
|
| session_id=getattr(agent, "session_id", None),
|
| changed_paths=getattr(agent, "_turn_file_mutation_paths", set()),
|
| attempts=getattr(agent, "_verification_stop_nudges", 0),
|
| )
|
| else:
|
| _verify_nudge = None
|
| except Exception:
|
| logger.debug("verification stop-loop check failed", exc_info=True)
|
| _verify_nudge = None
|
|
|
| if _verify_nudge:
|
| agent._verification_stop_nudges = (
|
| getattr(agent, "_verification_stop_nudges", 0) + 1
|
| )
|
| final_msg["finish_reason"] = "verification_required"
|
|
|
|
|
|
|
|
|
|
|
| agent._emit_interim_assistant_message(final_msg)
|
| messages.append(final_msg)
|
| try:
|
| agent._flush_messages_to_session_db(messages, conversation_history)
|
| except Exception:
|
| logger.debug("verify-on-stop interim flush failed", exc_info=True)
|
| messages.append({
|
| "role": "user",
|
| "content": _verify_nudge,
|
| "_verification_stop_synthetic": True,
|
| })
|
| agent._session_messages = messages
|
|
|
|
|
|
|
| logger.debug("verification stop-loop nudge issued (attempt %d)",
|
| agent._verification_stop_nudges)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| _pending_verification_response = final_response
|
| _pending_verification_response_previewed = (
|
| agent._interim_content_was_streamed(final_response or "")
|
| )
|
| final_response = None
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
| _verify_nudge2 = None
|
| _edited = sorted(getattr(agent, "_turn_file_mutation_paths", set()) or [])
|
| _attempt = getattr(agent, "_pre_verify_nudges", 0)
|
| try:
|
| from agent.verify_hooks import max_verify_nudges
|
| from hermes_cli.lifecycle import has_hook
|
| from hermes_cli.plugins import get_pre_verify_continue_message
|
|
|
| if _edited and has_hook("pre_verify") and _attempt < max_verify_nudges():
|
|
|
| coding = getattr(agent, "_resolved_is_coding", None)
|
| if coding is None:
|
| from agent.coding_context import is_coding_context
|
| coding = bool(is_coding_context(platform=getattr(agent, "platform", "") or ""))
|
| agent._resolved_is_coding = coding
|
| _verify_nudge2 = get_pre_verify_continue_message(
|
| session_id=getattr(agent, "session_id", None) or "",
|
| platform=getattr(agent, "platform", "") or "",
|
| model=getattr(agent, "model", "") or "",
|
| coding=coding,
|
| attempt=_attempt,
|
| final_response=final_response,
|
| changed_paths=_edited,
|
| )
|
| except Exception:
|
| logger.debug("pre_verify hook check failed", exc_info=True)
|
| _verify_nudge2 = None
|
|
|
| if _verify_nudge2:
|
| agent._pre_verify_nudges = _attempt + 1
|
| final_msg["finish_reason"] = "verify_hook_continue"
|
|
|
|
|
|
|
|
|
|
|
| agent._emit_interim_assistant_message(final_msg)
|
| messages.append(final_msg)
|
| try:
|
| agent._flush_messages_to_session_db(messages, conversation_history)
|
| except Exception:
|
| logger.debug("pre_verify interim flush failed", exc_info=True)
|
| messages.append({
|
| "role": "user",
|
| "content": _verify_nudge2,
|
| "_pre_verify_synthetic": True,
|
| })
|
| agent._session_messages = messages
|
| logger.debug("pre_verify nudge issued (attempt %d)",
|
| agent._pre_verify_nudges)
|
| _pending_verification_response = final_response
|
| _pending_verification_response_previewed = (
|
| agent._interim_content_was_streamed(final_response or "")
|
| )
|
| final_response = None
|
| continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| try:
|
| from agent.kanban_stop import build_kanban_stop_nudge
|
|
|
| _kanban_nudge = build_kanban_stop_nudge(
|
| messages=messages,
|
| attempts=getattr(agent, "_kanban_stop_nudges", 0),
|
| )
|
| except Exception:
|
| logger.debug("kanban stop-loop check failed", exc_info=True)
|
| _kanban_nudge = None
|
|
|
| if _kanban_nudge:
|
| agent._kanban_stop_nudges = (
|
| getattr(agent, "_kanban_stop_nudges", 0) + 1
|
| )
|
| final_msg["finish_reason"] = "kanban_terminal_required"
|
| final_msg["_kanban_stop_synthetic"] = True
|
| messages.append(final_msg)
|
| messages.append({
|
| "role": "user",
|
| "content": _kanban_nudge,
|
| "_kanban_stop_synthetic": True,
|
| })
|
| agent._session_messages = messages
|
| logger.info(
|
| "kanban stop-loop nudge issued (attempt %d) task=%s",
|
| agent._kanban_stop_nudges,
|
| os.environ.get("HERMES_KANBAN_TASK", ""),
|
| )
|
| agent._emit_status(
|
| "β οΈ Kanban worker tried to exit without "
|
| "kanban_complete/kanban_block β nudging to finish"
|
| )
|
|
|
|
|
|
|
|
|
| _pending_verification_response = final_response
|
| _pending_verification_response_previewed = (
|
| agent._interim_content_was_streamed(final_response or "")
|
| )
|
| final_response = None
|
| continue
|
|
|
| messages.append(final_msg)
|
|
|
| _turn_exit_reason = f"text_response(finish_reason={finish_reason})"
|
| if not agent.quiet_mode:
|
| agent._safe_print(f"π Conversation completed after {api_call_count} OpenAI-compatible API call(s)")
|
| break
|
|
|
| except Exception as e:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| tb_module_names: set[str] = set()
|
| _tb = e.__traceback__
|
| while _tb is not None:
|
| _fname = os.path.splitext(os.path.basename(_tb.tb_frame.f_code.co_filename))[0]
|
| tb_module_names.add(_fname)
|
| _tb = _tb.tb_next
|
|
|
| _hit_local = bool(tb_module_names & _LOCAL_PROCESSING_MODULES)
|
| _hit_api = bool(tb_module_names & _API_CALL_MODULES)
|
|
|
| _is_local_processing_error = _hit_local and not _hit_api
|
|
|
| if _is_local_processing_error:
|
| error_msg = (
|
| f"Error during local message processing after "
|
| f"OpenAI-compatible API call #{api_call_count}: {str(e)}"
|
| )
|
| else:
|
| error_msg = f"Error during OpenAI-compatible API call #{api_call_count}: {str(e)}"
|
| try:
|
| print(f"β {error_msg}")
|
| except (OSError, ValueError):
|
| logger.error(error_msg)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| logger.exception("Outer loop error in API call #%d", api_call_count)
|
|
|
|
|
|
|
|
|
| for idx in range(len(messages) - 1, -1, -1):
|
| msg = messages[idx]
|
| if not isinstance(msg, dict):
|
| break
|
| if msg.get("role") == "tool":
|
| continue
|
| if msg.get("role") == "assistant" and msg.get("tool_calls"):
|
| answered_ids = {
|
| m["tool_call_id"]
|
| for m in messages[idx + 1:]
|
| if isinstance(m, dict) and m.get("role") == "tool"
|
| }
|
| for tc in msg["tool_calls"]:
|
| if not tc or not isinstance(tc, dict): continue
|
| if tc["id"] not in answered_ids:
|
| err_msg = {
|
| "role": "tool",
|
| "name": _ra().AIAgent._get_tool_call_name_static(tc),
|
| "tool_call_id": tc["id"],
|
| "content": f"Error executing tool: {error_msg}",
|
| }
|
| messages.append(err_msg)
|
| break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| if (
|
| _is_local_processing_error
|
| or api_call_count >= agent.max_iterations - 1
|
| ):
|
| if _is_local_processing_error:
|
| _turn_exit_reason = f"local_processing_error({error_msg[:80]})"
|
| final_response = f"I apologize, but I encountered an error while processing the model response: {error_msg}"
|
| else:
|
| _turn_exit_reason = f"error_near_max_iterations({error_msg[:80]})"
|
| final_response = f"I apologize, but I encountered repeated errors: {error_msg}"
|
|
|
|
|
| messages.append({"role": "assistant", "content": final_response})
|
| break
|
|
|
|
|
|
|
|
|
| from agent.turn_finalizer import finalize_turn
|
| return finalize_turn(
|
| agent,
|
| final_response=final_response,
|
| api_call_count=api_call_count,
|
| interrupted=interrupted,
|
| failed=failed,
|
| messages=messages,
|
| conversation_history=conversation_history,
|
| effective_task_id=effective_task_id,
|
| turn_id=turn_id,
|
| user_message=user_message,
|
| original_user_message=original_user_message,
|
| _should_review_memory=_should_review_memory,
|
| _turn_exit_reason=_turn_exit_reason,
|
| _pending_verification_response=_pending_verification_response,
|
| _pending_verification_response_previewed=_pending_verification_response_previewed,
|
| )
|
|
|
|
|
|
|
| __all__ = ["run_conversation"]
|
|
|
