a4db03cee9
Route commits no longer NULL the stored prompt, so _stored_prompt_matches_runtime is the only rebuild trigger. A model-only request that clears agent.provider skipped the compare and replayed the previous route's prompt verbatim. The builder omits empty trailer lines, so the rebuilt prompt matches from the next turn (no rebuild loop); prompts without identity lines keep reusing.
1809 lines
90 KiB
Python
1809 lines
90 KiB
Python
"""The agent conversation loop — extracted from ``run_agent.AIAgent``.
|
|
|
|
``run_conversation(agent, ...)`` drives one user turn (model call, tool dispatch,
|
|
retries, fallbacks, compression, post-turn hooks). Symbols that callers patch on
|
|
``run_agent`` (``handle_function_call``, ``_set_interrupt``, ``OpenAI``) resolve via
|
|
``_ra`` so those patches keep working."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
import json
|
|
import logging
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass, field, fields
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from agent.codex_responses_adapter import _summarize_user_message_for_log
|
|
from agent.fast_mode import begin_turn as begin_fast_mode_turn
|
|
from agent.message_metadata import append_message
|
|
from agent.message_sanitization import _repair_tool_call_arguments, _sanitize_surrogates
|
|
from agent.model_metadata import MINIMUM_CONTEXT_LENGTH, _estimate_tools_tokens_rough
|
|
from agent.process_bootstrap import _install_safe_stdio
|
|
from agent.prompt_builder import RUNTIME_ENVIRONMENT_END, RUNTIME_ENVIRONMENT_HEADING
|
|
from agent.prompt_caching import (
|
|
build_prompt_cache_plan,
|
|
effective_cache_ttl,
|
|
strip_anthropic_cache_control,
|
|
strip_anthropic_tool_cache_control,
|
|
)
|
|
from agent.repetition_guard import REPETITION_LOOP_INTERRUPTED, is_runaway_repetition
|
|
from agent.runtime_cwd import resolve_agent_cwd
|
|
from agent.surface_switch import (
|
|
identity_line_value, note_inert_pinned_tools, runtime_host_value, stage_surface_switch_note,
|
|
)
|
|
from agent.turn_context import PreflightCompressionTimedOut, build_turn_context
|
|
from agent.turn_retry_state import TurnRetryState
|
|
# Phase helpers of the turn loop, bound at import so a source-tree swap cannot load a
|
|
# skewed phase mid-turn.
|
|
from agent.turn_api_call import handle_api_interrupt, nous_rate_limit_guard, perform_api_call
|
|
from agent.turn_api_error import handle_api_error
|
|
from agent.turn_api_request import build_api_request
|
|
from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, failed_turn_notice, site_copy
|
|
from agent.turn_final_response import finish_text_response
|
|
from agent.turn_finalizer import finalize_turn
|
|
from agent.turn_iteration_prep import (
|
|
announce_api_call,
|
|
apply_retry_restarts,
|
|
begin_iteration,
|
|
prepare_iteration,
|
|
)
|
|
from agent.turn_loop_errors import handle_outer_loop_error
|
|
from agent.turn_preflight_gate import run_preflight_gate
|
|
from agent.turn_request_assembly import assemble_api_request
|
|
from agent.turn_response_check import check_api_response
|
|
from agent.turn_response_intake import normalize_model_response
|
|
from agent.turn_tool_round import run_tool_round
|
|
from hermes_logging import set_session_context
|
|
from tools.skill_provenance import set_current_write_origin
|
|
from utils import base_url_host_matches
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Must mirror _STALE_TOOL_CALL_MARKER_RE in hermes_state.py; kept local so importing
|
|
# hermes_state (module-level DEFAULT_DB_PATH) is not forced at load time.
|
|
_STALE_MARKER_RE = re.compile(r"^\[[A-Za-z_][A-Za-z0-9_.-]*\]$")
|
|
|
|
# Shared by _apply_active_turn_redirect and the api_messages ghost-row filter so both sites cannot drift.
|
|
_INTERRUPT_SCAFFOLD_MARKER = "[This response was interrupted by a user correction.]"
|
|
|
|
|
|
# One-time wrap-up notice appended when a wall-clock run budget (--run-budget) crosses 80%.
|
|
RUN_BUDGET_WRAPUP_NOTICE = (
|
|
"[SYSTEM NOTICE — run time budget nearly exhausted] Run time budget nearly exhausted. "
|
|
"Stop new discovery/verification work now. Produce the required final deliverable "
|
|
"(answer/JSON/summary) from the state you already have, completing only mandatory writes."
|
|
)
|
|
|
|
|
|
def _midturn_request_pressure_tokens(
|
|
agent: Any, api_messages: List[Dict[str, Any]], effective_system: str, approx_tokens: int
|
|
) -> int:
|
|
"""Token figure the mid-turn pre-API compression guard compares: the pruned
|
|
native-Responses estimate when native compaction eligibility is proven (the generic
|
|
estimate overstates the wire on compacted sessions, #96995), else messages+tools.
|
|
The system prompt is counted exactly once.
|
|
|
|
When the upcoming request is eligible for native Responses compaction the transport will
|
|
checkpoint-prune the payload before sending, so the generic durable-history estimate overstates the wire
|
|
by orders of magnitude on a compacted session and fires a 600s local compression the main request never
|
|
needed (#96995).
|
|
"""
|
|
try:
|
|
from agent.codex_responses_adapter import estimate_native_responses_preflight_tokens
|
|
native = estimate_native_responses_preflight_tokens(
|
|
agent, api_messages, system_prompt=effective_system or "",
|
|
tools=getattr(agent, "tools", None) or None,
|
|
)
|
|
if isinstance(native, int) and not isinstance(native, bool) and native >= 0:
|
|
return native
|
|
except Exception:
|
|
logger.debug(
|
|
"native Responses mid-turn estimate unavailable; using generic transcript estimate",
|
|
exc_info=True,
|
|
)
|
|
return approx_tokens + (_estimate_tools_tokens_rough(agent.tools) if agent.tools else 0)
|
|
|
|
|
|
def _review_input_budget_exhausted(agent: Any) -> bool:
|
|
"""True when a detached review fork has replayed its aggregate input budget.
|
|
|
|
Only forks with an explicit ``_review_input_token_budget`` are gated (#93057). Fires
|
|
at the top of the NEXT iteration, so the budget-crossing request completes first."""
|
|
budget = getattr(agent, "_review_input_token_budget", None)
|
|
if not isinstance(budget, int) or isinstance(budget, bool) or budget <= 0:
|
|
return False
|
|
used = getattr(agent, "session_input_tokens", 0)
|
|
return isinstance(used, int) and not isinstance(used, bool) and used >= budget
|
|
|
|
|
|
def _maybe_inject_run_budget_wrapup(agent: Any, messages: List[Dict[str, Any]]) -> bool:
|
|
"""Inject the one-time wall-clock wrap-up notice when past 80% of budget.
|
|
|
|
Appends to the NEWEST ``role:"tool"`` message (cache-safe, like /steer); latches
|
|
``_run_budget_wrapup_injected`` only on a successful append."""
|
|
budget = getattr(agent, "run_budget_seconds", None)
|
|
started = getattr(agent, "_run_budget_started_at", None)
|
|
if not budget or not started or getattr(agent, "_run_budget_wrapup_injected", False) or (
|
|
(time.time() - started) < 0.8 * float(budget)
|
|
):
|
|
return False
|
|
from agent.context_compressor import _DB_PERSISTED_MARKER
|
|
for msg in reversed(messages):
|
|
if isinstance(msg, dict) and msg.get("role") == "tool":
|
|
# Only the current tool-result tail is mutable; an older turn may already be
|
|
# cached (same contract as _maybe_inject_iteration_budget_warning).
|
|
if msg.get(_DB_PERSISTED_MARKER):
|
|
return False
|
|
existing = msg.get("content", "")
|
|
if isinstance(existing, str):
|
|
msg["content"] = existing + f"\n\n{RUN_BUDGET_WRAPUP_NOTICE}"
|
|
else: # multimodal content blocks — append a text block
|
|
try:
|
|
msg["content"] = [*(existing or []), {"type": "text", "text": RUN_BUDGET_WRAPUP_NOTICE}]
|
|
except Exception:
|
|
return False
|
|
agent._run_budget_wrapup_injected = True
|
|
logger.info(
|
|
"Run budget wrap-up notice injected (budget=%.0fs, elapsed=%.0fs)",
|
|
float(budget), time.time() - started,
|
|
)
|
|
return True
|
|
return False
|
|
|
|
|
|
def _restore_user_after_reference_handoff(
|
|
messages: List[Dict[str, Any]], user_message: Any
|
|
) -> bool:
|
|
"""Re-append this turn's real user ask when compaction left only a handoff (#80622).
|
|
Returns True when a restore append happened."""
|
|
if isinstance(user_message, str):
|
|
restorable = bool(user_message.strip())
|
|
else:
|
|
restorable = isinstance(user_message, list) and bool(user_message)
|
|
if not restorable:
|
|
return False
|
|
last = messages[-1] if messages else None
|
|
if isinstance(last, dict) and last.get("role") == "user" and last.get("content") == user_message:
|
|
return False
|
|
append_message(messages, {"role": "user", "content": user_message})
|
|
return True
|
|
|
|
|
|
def _should_skip_model_call_for_reference_handoff(
|
|
messages: List[Dict[str, Any]], user_message: Any
|
|
) -> bool:
|
|
"""Guard post-compaction continues against sole-handoff active turns (#80622)."""
|
|
from agent.context_compressor import reference_handoff_would_drive_next_model_call
|
|
# A restored ask is an actionable non-synthetic user row appended after the
|
|
# handoff — by construction the handoff no longer drives.
|
|
return reference_handoff_would_drive_next_model_call(messages) and not (
|
|
_restore_user_after_reference_handoff(messages, user_message)
|
|
)
|
|
|
|
|
|
# Fallback final_response for the sole-handoff skip (#80622); finalize_turn appends it as a
|
|
# fresh assistant row, so it must not replay the last assistant text.
|
|
# Deliberately NOT a replay of the last assistant text: finalize_turn's non-assistant-tail chokepoint
|
|
# (#43849) appends final_response as a fresh assistant row, so recovering the previous turn's prose here
|
|
# would duplicate it in the durable transcript AND re-deliver it to the user as if it were this turn's
|
|
# answer. A short status is honest and idempotent.
|
|
_HANDOFF_SKIP_FINAL_RESPONSE = (
|
|
"Context was compacted. The previous response is complete — awaiting your next message."
|
|
)
|
|
|
|
# Terminal final_response when compression timed out while the request was still oversized (#98722).
|
|
# Terminal final_response for a turn ended because context compression hit its host progress-aware timeout
|
|
# while the request was still oversized (#98722, salvaged from #98741). Sending the unchanged request would
|
|
# only bounce off the provider's overflow error and re-enter compression in the same turn.
|
|
_COMPRESSION_TIMEOUT_FINAL_RESPONSE = (
|
|
"Context compression timed out without reducing this conversation. No messages were "
|
|
"dropped. Start a fresh session with /new, or check auxiliary.compression before retrying /compress."
|
|
)
|
|
|
|
|
|
# Stable prefix ACP/TUI match on to treat the text as cancellation metadata, not assistant prose.
|
|
INTERRUPT_WAITING_FOR_MODEL_PREFIX = "Operation interrupted: waiting for model response ("
|
|
|
|
|
|
def _should_rearm_compression_budget(
|
|
compression_attempts: int, *, completed_compaction_pending: bool, prompt_tokens: int, threshold_tokens: int
|
|
) -> bool:
|
|
"""True once a provider proves a completed compaction worked: rough estimates cannot
|
|
rearm the anti-thrash budget, only the completed-compaction latch plus a positive
|
|
normalized prompt count below the threshold."""
|
|
return bool(
|
|
compression_attempts and completed_compaction_pending and 0 < prompt_tokens < threshold_tokens
|
|
)
|
|
|
|
|
|
# Modules whose presence in a traceback (without any API-call module) marks a
|
|
# deterministic local bug not worth retrying. NEVER add "conversation_loop" or
|
|
# "run_agent": every exception passes through them; _hit_local would be True (#66267)
|
|
_LOCAL_PROCESSING_MODULES = frozenset({
|
|
"agent_runtime_helpers",
|
|
"message_content",
|
|
"message_sanitization",
|
|
"chat_completion_helpers", # only local when NOT also an API-call module
|
|
})
|
|
_API_CALL_MODULES = frozenset({"chat_completion_helpers"})
|
|
|
|
# Max outer-loop exceptions per user turn before giving up; only exceptions that
|
|
# ESCAPE the inner retry/fallback machinery count, so this can be small (#92450).
|
|
_MAX_OUTER_LOOP_ERRORS = 8
|
|
|
|
|
|
def _is_interpreter_shutdown_error(exc: Exception) -> bool:
|
|
"""True for a fatal interpreter-shutdown RuntimeError. The RuntimeError type gate
|
|
stays here: a ValueError carrying similar text must not match (#93269)."""
|
|
if isinstance(exc, RuntimeError):
|
|
# ── Interpreter finalization: abandon immediately ── The process is exiting (TUI quit, SIGTERM,
|
|
# one-shot done) while this turn — typically the post-turn review fork's daemon thread — is
|
|
# mid-flight. Retries, credential rotation, and fallbacks are all futile ("cannot schedule new
|
|
# futures..."), and the buffered ⚠️/❌ retry trace spams the shell after the TUI already exited. End
|
|
# the turn with a single log line: no print, no traceback, no debug dump, no retry. Same class as
|
|
# cron delivery (#55924/#58720) and concurrent tool submission — shared predicate.
|
|
from tools.interpreter_shutdown import interpreter_shutting_down
|
|
return interpreter_shutting_down(exc)
|
|
return False
|
|
|
|
|
|
def _moa_client_consumes_prepared_request(client: Any) -> bool:
|
|
"""True when ``client`` is the in-process MoA facade (only ``MoAChatCompletions`` exposes
|
|
``prepare()``; other clients raise TypeError on ``_moa_prepared_request`` even while
|
|
``agent.provider`` stays ``"moa"``)."""
|
|
completions = getattr(getattr(client, "chat", None), "completions", None)
|
|
return callable(getattr(completions, "prepare", None))
|
|
|
|
|
|
_MIN_CONTINUATION_OVERLAP = 32
|
|
|
|
|
|
def _continuation_overlap_length(previous: str, continuation: str) -> int:
|
|
"""Return the longest continuation prefix that repeats the previous suffix."""
|
|
if len(previous) < _MIN_CONTINUATION_OVERLAP or len(continuation) < _MIN_CONTINUATION_OVERLAP:
|
|
return 0
|
|
|
|
prefix_lengths = [0] * len(continuation)
|
|
matched = 0
|
|
for index in range(1, len(continuation)):
|
|
while matched and continuation[index] != continuation[matched]:
|
|
matched = prefix_lengths[matched - 1]
|
|
if continuation[index] == continuation[matched]:
|
|
matched += 1
|
|
prefix_lengths[index] = matched
|
|
|
|
matched = 0
|
|
last_index = len(previous) - 1
|
|
for index, char in enumerate(previous):
|
|
while matched and char != continuation[matched]:
|
|
matched = prefix_lengths[matched - 1]
|
|
if char == continuation[matched]:
|
|
matched += 1
|
|
if matched == len(continuation):
|
|
if index == last_index:
|
|
return matched
|
|
matched = prefix_lengths[matched - 1]
|
|
return matched if matched >= _MIN_CONTINUATION_OVERLAP else 0
|
|
|
|
|
|
def _join_truncated_parts(parts: List[tuple[str, bool]]) -> str:
|
|
"""Join continuation fragments, deduping only interrupted-stream seams."""
|
|
joined = ""
|
|
previous_was_partial_stub = False
|
|
for part, is_partial_stub in parts:
|
|
if previous_was_partial_stub and joined and part:
|
|
# Overlap can't exceed len(part): scan only that tail of ``joined``.
|
|
part = part[_continuation_overlap_length(joined[-len(part):], part):]
|
|
if joined and not joined[-1].isspace() and part and not part[0].isspace():
|
|
joined += "\n"
|
|
joined += part
|
|
previous_was_partial_stub = is_partial_stub
|
|
return joined
|
|
|
|
|
|
def _moa_reference_metrics_for_hook(agent: Any) -> Any:
|
|
"""Per-advisor metrics for post_api_request, or None off the MoA path (a plugin only
|
|
sees the aggregator generation; this carries the per-slot advisor spend)."""
|
|
client = getattr(agent, "client", None)
|
|
getter = getattr(client, "last_reference_metrics", None)
|
|
if not callable(getter):
|
|
return None
|
|
try:
|
|
return getter()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
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 so role alternation
|
|
holds and cached messages stay byte-identical. INVARIANTS: raw chain-of-thought never enters
|
|
replayable content (inlined CoT reads as a prefill jailbreak and bricks the session with
|
|
empty-response storms); the interruption scaffold is replay text carried only in the user
|
|
correction's ``api_content``; an on-screen-empty placeholder is ``display_kind=hidden``."""
|
|
visible = agent._strip_think_blocks(getattr(agent, "_current_streamed_assistant_text", "") or "").strip()
|
|
|
|
checkpoint_parts = [_INTERRUPT_SCAFFOLD_MARKER]
|
|
if is_runaway_repetition(visible):
|
|
# Runaway shape only (a correct batch-style partial stays replayable): the looped bytes must
|
|
# reach neither the replayed correction nor the placeholder below (empty ``visible`` takes
|
|
# the hidden shape).
|
|
checkpoint_parts.append(REPETITION_LOOP_INTERRUPTED)
|
|
visible = ""
|
|
elif visible:
|
|
checkpoint_parts += ["Visible response before the interruption:", visible]
|
|
checkpoint = "\n\n".join(checkpoint_parts)
|
|
correction = f"[Context from the interrupted assistant response]\n{checkpoint}\n\n{text}"
|
|
|
|
# The live tail is normally user or tool, so an assistant placeholder + correction
|
|
# keeps strict alternation; if the tail is already assistant, the checkpoint is folded
|
|
# into the user correction instead of creating assistant→assistant. The placeholder
|
|
# preserves alternation only — scaffold bytes must never land in it, since api_content
|
|
# is substituted back into content on replay (#81841).
|
|
if not (messages and messages[-1].get("role") == "assistant"):
|
|
placeholder: Dict[str, Any] = {"role": "assistant", "content": visible or ""}
|
|
if not visible:
|
|
placeholder["display_kind"] = "hidden"
|
|
# Hidden row, but a non-empty neutral api_content so the pre-call sanitizer
|
|
# does not re-heal it every call (#88955). Never _INTERRUPT_SCAFFOLD_MARKER:
|
|
# as assistant text the model echoes it (#81841).
|
|
from agent.agent_runtime_helpers import _INTERRUPTED_PLACEHOLDER
|
|
placeholder["api_content"] = _INTERRUPTED_PLACEHOLDER
|
|
append_message(messages, placeholder)
|
|
# Transcript shows the user's own words; the provider replays the scaffolded form.
|
|
append_message(messages, {"role": "user", "content": text, "api_content": correction})
|
|
|
|
# Stateful scrubber for <memory-context> spans split across stream deltas (#5719). sanitize_context()
|
|
# alone can't survive chunk boundaries because the block regex needs both tags in one string.
|
|
# Stateful scrubber for reasoning/thinking tags in streamed deltas (#17924). Replaces the per-delta
|
|
# _strip_think_blocks regex that destroyed downstream state (e.g. MiniMax-M2.7 streaming '<think>' as
|
|
# delta1 and 'Let me check' as delta2 — the regex erased delta1, so downstream state machines never
|
|
# learned a block was open and leaked delta2 as content).
|
|
agent._current_streamed_assistant_text = ""
|
|
agent._stream_needs_break = True
|
|
|
|
|
|
def _is_copilot_provider(agent: Any) -> bool:
|
|
"""Delegate to ``AIAgent._is_copilot_provider``; the fallback keeps the ``github-copilot`` /
|
|
``github`` aliases so credential recovery is not skipped for them."""
|
|
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 (status 400 AND an
|
|
integrator/model-not-supported marker, so a wrong model name never triggers the
|
|
single-shot re-exchange). Caller enforces scoping/guard."""
|
|
lowered = (error_message or "").lower()
|
|
if status_code != 400 and "error code: 400" not in lowered:
|
|
return False
|
|
return any(marker in lowered for marker in (
|
|
"model_not_available_for_integrator",
|
|
"not available for integrator",
|
|
"model_not_supported",
|
|
"the requested model is not supported",
|
|
))
|
|
|
|
|
|
def _pressure_with_real_floor(compressor: Any, rough_tokens: int) -> int:
|
|
"""Floor the ROUGH pre-API pressure estimate at the last REAL prompt size.
|
|
|
|
Applied only on the fallback path -- when ``anchored_context_tokens`` has
|
|
no valid anchor (first request, transcript rewritten under the anchor,
|
|
provider never reported usage). A valid anchor is provider-exact and is
|
|
used as-is; in particular on MoA turns the anchor deliberately uses the
|
|
pre-fold aggregator usage while ``last_real_prompt_tokens`` holds the
|
|
folded figure, so flooring an anchored value would re-add fan-out tokens
|
|
the anchor exists to exclude.
|
|
|
|
On the rough path, non-ASCII text (Cyrillic, Greek, Polish, ...)
|
|
under-counts by up to ~2x, so a session can sit at the provider's real
|
|
context ceiling while the rough figure stays under the compaction
|
|
threshold -- on silent-clip providers (ollama /v1) that is a truncation
|
|
death spiral the reactive overflow handler never sees (observed live:
|
|
real prompts 64,842->64,995 against a 55,705 threshold). The provider's
|
|
last reported prompt_tokens is authoritative; never let the rough figure
|
|
fall below it. Skipped for exactly one turn after a compaction, when
|
|
last_real_prompt_tokens still holds the stale pre-compression value
|
|
(#36718's awaiting_real_usage_after_compression window).
|
|
"""
|
|
last_real = int(getattr(compressor, "last_real_prompt_tokens", 0) or 0)
|
|
if last_real > rough_tokens and not getattr(
|
|
compressor, "awaiting_real_usage_after_compression", False
|
|
):
|
|
return last_real
|
|
return rough_tokens
|
|
|
|
|
|
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."""
|
|
runtime_ctx = getattr(agent, "_ollama_num_ctx", None)
|
|
if (
|
|
not getattr(agent, "tools", None)
|
|
or not isinstance(runtime_ctx, int)
|
|
or not 0 < runtime_ctx < MINIMUM_CONTEXT_LENGTH
|
|
):
|
|
return None
|
|
|
|
model = getattr(agent, "model", "") or "the selected model"
|
|
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, getattr(agent, "provider", "") or "unknown",
|
|
getattr(agent, "base_url", "") or "unknown base URL", runtime_ctx, MINIMUM_CONTEXT_LENGTH,
|
|
request_tokens, len(getattr(agent, "tools", None) or []),
|
|
getattr(agent, "session_id", None) or "none",
|
|
)
|
|
return (
|
|
f"Ollama loaded `{model}` with only {runtime_ctx:,} tokens of runtime context, but Hermes "
|
|
f"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 _maybe_grow_local_window(agent: Any, compressor: Any,
|
|
request_tokens: int) -> Optional[int]:
|
|
"""Grow a managed local model's context window before compressing; returns the new
|
|
window when the ladder granted one, else None."""
|
|
provider = (getattr(agent, "provider", "") or "").strip().lower()
|
|
base_url = getattr(agent, "base_url", "") or ""
|
|
if provider not in ("llamacpp", "llama.cpp", "llama-cpp", "custom") or not (
|
|
"127.0.0.1" in base_url or "localhost" in base_url
|
|
):
|
|
return None
|
|
try:
|
|
from hermes_cli.local_runtime.growth import maybe_grow_window
|
|
current_window = int(getattr(compressor, "context_length", 0) or 0)
|
|
if current_window <= 0:
|
|
return None
|
|
return maybe_grow_window(
|
|
getattr(agent, "model", "") or "", base_url=base_url,
|
|
session_tokens=int(request_tokens), current_window=current_window,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 — growth must never break a turn
|
|
logger.debug("local window growth check failed: %s", exc)
|
|
return None
|
|
|
|
|
|
def _ra():
|
|
"""Lazy ``run_agent`` reference so patches on ``run_agent.*`` 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)
|
|
return format_nous_portal_entitlement_message(
|
|
account_info, capability=capability, in_chat=True
|
|
) or ""
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _print_guidance(agent, message: str) -> bool:
|
|
"""Print each line of ``message`` as a 💡 hint; False when there is nothing to print."""
|
|
if not message:
|
|
return False
|
|
for line in message.splitlines():
|
|
agent._vprint(f"{agent.log_prefix} 💡 {line}", force=True, diagnostic=True)
|
|
return True
|
|
|
|
|
|
def _print_nous_entitlement_guidance(agent, capability: str) -> bool:
|
|
return _print_guidance(agent, _nous_entitlement_message(capability))
|
|
|
|
|
|
def _system_prompt_for_hooks(api_kwargs: Any, request_messages: Any) -> Any:
|
|
"""System prompt as sent to the provider (``system`` / ``instructions`` / ``messages[0]``)
|
|
for observability hooks; None when the request carries none."""
|
|
system_prompt = api_kwargs.get("system")
|
|
if system_prompt is None:
|
|
system_prompt = api_kwargs.get("instructions")
|
|
if system_prompt is None and isinstance(request_messages, list) and request_messages:
|
|
first = request_messages[0]
|
|
if isinstance(first, dict) and first.get("role") == "system":
|
|
system_prompt = first.get("content")
|
|
return system_prompt
|
|
|
|
|
|
def _is_nous_inference_route(provider: str, base_url: str) -> bool:
|
|
return (provider or "").strip().lower() == "nous" or base_url_host_matches(
|
|
str(base_url or ""), "inference-api.nousresearch.com"
|
|
)
|
|
|
|
|
|
def _billing_or_entitlement_message(
|
|
*, capability: str, provider: str, base_url: str, model: str, unverified: bool = False
|
|
) -> 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"
|
|
|
|
# Anthropic Pro/Max OAuth surfaces "extra usage" exhaustion as a hard 400 — "add credits"
|
|
# does not apply. ``unverified`` (#82154): the same 400 is returned for a server-side
|
|
# content-filter rejection, so hedge and name the other cause.
|
|
if (provider or "").strip().lower() == "anthropic":
|
|
switch = (
|
|
"You can also switch to an Anthropic API key or another provider with "
|
|
"/model <model> --provider <provider>."
|
|
)
|
|
if unverified:
|
|
return "\n".join([
|
|
f"{provider_label} reported that your Claude subscription usage may be exhausted for "
|
|
f"{model_label} (included quota + extra-usage credits) — but this specific error is "
|
|
"not proof of a billing problem.",
|
|
"If https://claude.ai/settings/usage still shows quota remaining, this is probably NOT "
|
|
"a billing problem: on a Claude subscription (OAuth) token Anthropic returns this same "
|
|
"message when its content filter rejects part of the request — typically a phrase in "
|
|
"the system prompt.",
|
|
"If usage really is exhausted: wait for the billing cycle to reset, or add extra usage "
|
|
"at https://claude.ai/settings/usage",
|
|
switch,
|
|
# The exhaustion latch replays the stored error without a request.
|
|
"Retry with a fresh credential state: `hermes auth reset anthropic`. Until that "
|
|
"cooldown clears, this error can be replayed from cache without contacting the API.",
|
|
])
|
|
return "\n".join([
|
|
f"{provider_label} reported that your Claude subscription usage is exhausted for "
|
|
f"{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",
|
|
switch,
|
|
])
|
|
|
|
# Provider-agnostic billing URL so every text surface shows the same actionable link.
|
|
try:
|
|
from agent.billing_links import build_billing_block
|
|
_link = build_billing_block(provider=provider, base_url=base_url, model=model)
|
|
provider_label = _link.provider_label or provider_label
|
|
billing_url = _link.billing_url
|
|
except Exception:
|
|
billing_url = None
|
|
return "\n".join([
|
|
f"{provider_label} reported that billing, credits, or account entitlement is exhausted for {model_label}.",
|
|
"Add credits or update billing with that provider, then retry.",
|
|
*([f"{provider_label} billing: {billing_url}"] if billing_url else []),
|
|
"You can switch providers temporarily with /model <model> --provider <provider>.",
|
|
])
|
|
|
|
|
|
def _billing_block_dict(provider, base_url, model, message="", *, unverified: bool = False) -> Optional[dict]:
|
|
"""Best-effort structured billing descriptor (None if billing_links is unavailable)."""
|
|
try:
|
|
from agent.billing_links import build_billing_block
|
|
block = build_billing_block(
|
|
provider=provider, base_url=str(base_url), model=model, message=message
|
|
).to_dict()
|
|
except Exception:
|
|
return None
|
|
if block is not None and unverified:
|
|
block["unverified"] = True # every surface rendering the block can hedge too (#82154)
|
|
return block
|
|
|
|
|
|
def _billing_terminal_label(summary: str, unverified: bool) -> str:
|
|
"""Terminal-failure prefix for a billing-classified error; ``unverified`` (#82154) must
|
|
not assert exhaustion as fact."""
|
|
if unverified:
|
|
return (
|
|
"Provider reported usage/credit exhaustion (unverified — the same "
|
|
f"error can be a content-filter rejection, not billing): {summary}"
|
|
)
|
|
return f"Billing or credits exhausted: {summary}"
|
|
|
|
|
|
def _billing_failure_result(
|
|
*, classified, summary: str, messages, api_call_count: int, provider: str, base_url, model: str,
|
|
guidance: Optional[str] = None,
|
|
) -> dict:
|
|
"""Structured terminal result for a billing-classified failure — the single construction
|
|
point for the non-retryable abort and max-retries paths (#82154)."""
|
|
unverified = bool(getattr(classified, "billing_unverified", False))
|
|
if guidance is None:
|
|
guidance = _billing_or_entitlement_message(
|
|
capability="model access", provider=provider, base_url=str(base_url), model=model,
|
|
unverified=unverified,
|
|
)
|
|
final = _billing_terminal_label(summary, unverified) + (f"\n\n{guidance}" if guidance else "")
|
|
return {
|
|
"final_response": final, "messages": messages, "api_calls": api_call_count,
|
|
"completed": False, "failed": True, "error": summary,
|
|
"failure_reason": classified.reason.value,
|
|
# Classifier's own retry verdict so the UI shows Retry only when a re-run can differ.
|
|
"failure_retryable": bool(classified.retryable),
|
|
"billing_unverified": unverified,
|
|
"billing_block": _billing_block_dict(provider, base_url, model, guidance, unverified=unverified),
|
|
}
|
|
|
|
|
|
def _print_billing_or_entitlement_guidance(
|
|
agent, *, capability: str, provider: str, base_url: str, model: str, unverified: bool = False
|
|
) -> bool:
|
|
return _print_guidance(agent, _billing_or_entitlement_message(
|
|
capability=capability, provider=provider, base_url=base_url, model=model,
|
|
unverified=unverified,
|
|
))
|
|
|
|
|
|
def _bot_chat_prompt_stale(agent, stored_prompt: str | None) -> bool:
|
|
"""Bot Chat capability epoch check for a stored prompt.
|
|
|
|
The stored prompt embeds a capability fingerprint; a mismatch is a deliberate
|
|
once-per-change rebuild. Unstamped prompts never match; probe failures fail closed
|
|
to "reuse" so the cache is kept. Legacy upgrade: a Bot Chat prompt predating the
|
|
epoch mechanism gets ONE title-gated migration rebuild; the stamped result cannot
|
|
re-fire. A NULL or empty stored prompt already rebuilds every turn, so this probe
|
|
is not a gate there and must not run.
|
|
"""
|
|
if not stored_prompt:
|
|
return False
|
|
try:
|
|
from tools.bot_mode_probe import (
|
|
BOT_CHAT_TITLE,
|
|
stored_bot_chat_prompt_needs_upgrade,
|
|
stored_prompt_capability_stale,
|
|
)
|
|
home = None
|
|
try:
|
|
from agent.system_prompt import _agent_home
|
|
home = _agent_home(agent)
|
|
except Exception:
|
|
pass
|
|
if stored_prompt_capability_stale(stored_prompt, home):
|
|
return True
|
|
if not getattr(agent, "_bot_mode_protocol", True):
|
|
return False
|
|
title = str(getattr(agent, "_session_title_hint", "") or "").strip()
|
|
if not title and agent._session_db and agent.session_id:
|
|
try:
|
|
title = str(agent._session_db.get_session_title(agent.session_id) or "").strip()
|
|
except Exception:
|
|
title = ""
|
|
return title == BOT_CHAT_TITLE and bool(stored_bot_chat_prompt_needs_upgrade(stored_prompt, home))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _persist_system_prompt(agent, failure_message: str, *, persist_tools: bool = False) -> None:
|
|
"""Persist ``agent._cached_system_prompt`` to the session row; failures log at WARNING
|
|
(with ``failure_message``) because the gateway path (fresh AIAgent per turn) reads
|
|
this row every turn, so a silent failure breaks prefix-cache reuse."""
|
|
if not agent._session_db:
|
|
return
|
|
try:
|
|
agent._session_db.update_system_prompt(agent.session_id, agent._cached_system_prompt)
|
|
if persist_tools:
|
|
from tools.mcp_tool_agent import persist_agent_tool_names
|
|
persist_agent_tool_names(agent)
|
|
except Exception as exc:
|
|
logger.warning(failure_message, agent.session_id, exc)
|
|
|
|
|
|
def _restore_pinned_tools(agent, session_row) -> list:
|
|
"""Pin ``agent.tools`` to the session's persisted array (tools freeze); returns the names
|
|
this surface built BEFORE the pin merged a previous surface's tools back in."""
|
|
from tools.mcp_tool_agent import agent_tool_names, persist_agent_tool_names, restore_agent_tool_prefix
|
|
built_for_this_surface = agent_tool_names(agent)
|
|
saved_tools = session_row.get("tool_names") if session_row else None
|
|
try:
|
|
pin = json.loads(saved_tools) if saved_tools else None
|
|
except ValueError:
|
|
pin = None # a pin hash whose row an older build's cleanup swept resolves to itself
|
|
try:
|
|
if pin:
|
|
restore_agent_tool_prefix(agent, pin)
|
|
elif session_row is not None and not getattr(agent, "_persist_disabled", False):
|
|
# No usable pin (swept row, a session from before pins): pin what this turn sends,
|
|
# or every later hop re-derives tools[] until the next compaction.
|
|
persist_agent_tool_names(agent)
|
|
except Exception:
|
|
logger.debug("tool prefix restore skipped", exc_info=True)
|
|
return built_for_this_surface
|
|
|
|
|
|
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 on first
|
|
build. Row states ``missing``/``null``/``empty``/``present`` are logged and DB
|
|
failures log at WARNING so silent prefix-cache misses show in ``agent.log``."""
|
|
stored_prompt = None
|
|
stored_state = "missing"
|
|
session_row = None
|
|
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")
|
|
stored_state = "null" if raw_prompt is None else ("empty" if raw_prompt == "" else "present")
|
|
stored_prompt = raw_prompt or None
|
|
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):
|
|
# NULL/empty rows never reach this probe: they already rebuild below.
|
|
if _bot_chat_prompt_stale(agent, stored_prompt):
|
|
logger.info(
|
|
"Bot Chat capability epoch changed for session %s; rebuilding system prompt to "
|
|
"adopt the new capability surface (one-time prefix-cache break).",
|
|
agent.session_id,
|
|
)
|
|
agent._session_title_hint = "Bot Chat"
|
|
# The skills index cache (LRU + disk snapshot) does not watch the skills
|
|
# dir; a capability refresh must rebuild THROUGH it or new skills are lost.
|
|
try:
|
|
from agent.prompt_builder import clear_skills_system_prompt_cache
|
|
clear_skills_system_prompt_cache(clear_snapshot=True)
|
|
except Exception:
|
|
pass
|
|
agent._cached_system_prompt = agent._build_system_prompt(system_message)
|
|
stage_surface_switch_note(agent, agent._cached_system_prompt, conversation_history)
|
|
# Persist so the NEXT turn restores the new bytes verbatim (cache break is
|
|
# once per capability change). on_session_start not re-fired: continuation.
|
|
_persist_system_prompt(
|
|
agent,
|
|
"Session DB update_system_prompt failed after Bot Chat capability refresh "
|
|
"(session=%s): %s. The refresh will re-fire next turn.",
|
|
)
|
|
return
|
|
# Continuing session — reuse the exact system prompt from the
|
|
# previous turn so the Anthropic cache prefix matches.
|
|
agent._cached_system_prompt = stored_prompt
|
|
# The reused bytes may describe the surface this conversation STARTED on; correct that
|
|
# at the tail of the request instead of rebuilding the prompt in front of it (#104414).
|
|
announced_switch = stage_surface_switch_note(agent, stored_prompt, conversation_history)
|
|
# Same contract for tools[]: pin the array to the order this session already
|
|
# sent (tools freeze) instead of re-probing every check_fn on a fresh AIAgent.
|
|
# The pin holds ON the announcing turn too. tools[] is serialized AHEAD of the system
|
|
# prompt this branch just preserved, so dropping the previous surface's toolset would
|
|
# change the request at token 0 and re-prefill everything behind it — the exact cost
|
|
# #104414 is about, paid on the exact turn we are here to make cheap. The merge still
|
|
# ADDS what the new surface brought (a tui -> desktop switch pays a break no freeze can
|
|
# avoid), and what it carries FORWARD is named in the note instead, so a tool that can
|
|
# only answer ``tool_error("desktop only")`` here does not read as a live capability.
|
|
built_for_this_surface = _restore_pinned_tools(agent, session_row)
|
|
if announced_switch:
|
|
note_inert_pinned_tools(agent, built_for_this_surface)
|
|
# Prompt-section callbacks are new-session-only; recover their frozen bytes
|
|
# from the persisted prompt so a compression rebuild keeps them. The static
|
|
# prefix is not persisted either; rebuild it for the early cache breakpoint or
|
|
# fresh-per-turn gateway agents fall back to the single-breakpoint layout
|
|
# (reconstruct_static_prefix gates on _use_prompt_caching, fails open to legacy).
|
|
from agent.system_prompt import reconstruct_static_prefix, restore_plugin_prompt_sections
|
|
restore_plugin_prompt_sections(agent, stored_prompt)
|
|
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"):
|
|
# Continuing session with an unusable stored prompt: every turn now rebuilds
|
|
# and the prefix cache misses every time.
|
|
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,
|
|
)
|
|
|
|
# First turn of a new session (or recovering from a broken stored prompt). Rebuilding an
|
|
# EXISTING session's prompt (cwd drift, model switch) still keeps its pinned tools[]: this
|
|
# surface's own build (the -q footprint, its tool_search catalog) would otherwise be
|
|
# persisted over the pin below. Pinned first, so the prompt describes the tools sent.
|
|
built_for_this_surface = _restore_pinned_tools(agent, session_row)
|
|
agent._cached_system_prompt = agent._build_system_prompt(system_message)
|
|
|
|
# The rebuilt prompt describes the CURRENT surface, but a surface note left in the
|
|
# transcript by an earlier switch does not — retire it here too, or a rebuild for an
|
|
# unrelated reason (a model switch) would leave the newest interface statement in the
|
|
# request naming a surface the conversation has left (#104414).
|
|
stage_surface_switch_note(agent, agent._cached_system_prompt, conversation_history)
|
|
note_inert_pinned_tools(agent, built_for_this_surface)
|
|
|
|
# Persistence-disabled forks share their parent's session ID and are not real sessions.
|
|
if not getattr(agent, "_persist_disabled", False):
|
|
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)
|
|
|
|
# Cold-start credits seed (L3) fallback for the first-turn path; TUI/desktop seed at
|
|
# session open, so this is idempotent (skips when _credits_state exists). Fail-open.
|
|
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)
|
|
|
|
_persist_system_prompt(
|
|
agent,
|
|
"Session DB update_system_prompt failed for session %s: %s. Subsequent turns will "
|
|
"rebuild the system prompt and miss the prefix cache.",
|
|
persist_tools=True,
|
|
)
|
|
|
|
|
|
def _stored_prompt_matches_runtime(agent, prompt: str) -> bool:
|
|
"""Return False when the persisted runtime-identity lines are stale."""
|
|
# Model/provider identity, then cwd drift. A cwd change is a real content change (context
|
|
# files, the workspace snapshot and the coding posture are all resolved from it), so it
|
|
# still rebuilds; the runtime surface does not (agent/surface_switch.py).
|
|
# The builder omits an empty trailer line, so stored-but-now-empty is a route change too;
|
|
# the rebuilt prompt then carries no line and matches from the next turn on. Stored-empty
|
|
# (pre-trailer prompts) keeps reusing.
|
|
for label, attr in (("Model", "model"), ("Provider", "provider")):
|
|
stored = identity_line_value(prompt, label)
|
|
current = str(getattr(agent, attr, "") or "").strip()
|
|
if stored and stored != current:
|
|
return False
|
|
# A prompt stamped for another session (a /branch child copies its parent's bytes) must not
|
|
# tell the model a foreign Session ID. Checked only when the trailer is on: with it off, a
|
|
# "Session ID:" line in project text would read as a mismatch and rebuild every turn.
|
|
stored_sid = identity_line_value(prompt, "Session ID")
|
|
if stored_sid and getattr(agent, "pass_session_id", False) and stored_sid != agent.session_id:
|
|
return False
|
|
# Compare against resolve_agent_cwd() — the SAME resolver used to build the
|
|
# prompt — so TERMINAL_CWD sessions are not falsely rejected.
|
|
stored_cwd = runtime_host_value(prompt, "Current working directory")
|
|
if stored_cwd and stored_cwd != str(resolve_agent_cwd()):
|
|
return False
|
|
# Platform is deliberately NOT an identity field: a surface switch does not invalidate the
|
|
# stored bytes, it only makes their interface section out of date, and that is corrected by
|
|
# agent.surface_switch.stage_surface_switch_note without touching the cached prefix (#104414).
|
|
return True
|
|
|
|
|
|
# Named so _is_synthetic_compression_user_turn can recognize a crash-persisted nudge by
|
|
# content (SessionDB projection strips the _length_continuation_nudge tag).
|
|
_LENGTH_CONTINUATION_NETWORK_STUB = (
|
|
"[System: The previous response was cut off by a network error mid-stream — a transport "
|
|
"interruption, NOT a change in your capabilities. Your tools are still fully available; call "
|
|
"them as normal and ignore any earlier claim that you lack tool access. Continue the task "
|
|
"from where you left off. Do not restart or repeat prior text.]"
|
|
)
|
|
_LENGTH_CONTINUATION_OUTPUT_LIMIT = (
|
|
"[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.]"
|
|
)
|
|
# Pre-#74990 wording; kept so crash-persisted nudges from older sessions are still recognized.
|
|
_LEGACY_LENGTH_CONTINUATION_NETWORK_STUB = (
|
|
"[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.]"
|
|
)
|
|
# The dropped-tools variant interpolates tool names; matched by prefix.
|
|
_LENGTH_CONTINUATION_DROPPED_TOOLS_PREFIX = "[System: Your previous tool call "
|
|
|
|
|
|
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 (
|
|
f"{_LENGTH_CONTINUATION_DROPPED_TOOLS_PREFIX}({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. The cut was a transport "
|
|
"interruption, not a capability change — your tools remain fully available.]"
|
|
)
|
|
return _LENGTH_CONTINUATION_NETWORK_STUB if is_partial_stub else _LENGTH_CONTINUATION_OUTPUT_LIMIT
|
|
|
|
|
|
# Codex/Responses turns that returned only internal reasoning: a bare retry would be
|
|
# byte-identical, so the model repeats it.
|
|
_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).]"
|
|
)
|
|
|
|
|
|
# Re-prompt after an acknowledgment-only Codex/Responses reply.
|
|
_CODEX_ACK_CONTINUATION_NUDGE = (
|
|
"[System: Continue now. Execute the required tool calls and only send your final answer "
|
|
"after completing the task.]"
|
|
)
|
|
|
|
# Re-prompt after a collapsed fragment ended a turn that had done real tool work (#103483). Asks
|
|
# for the same answer again when it WAS complete, so a false positive costs one call, never the answer.
|
|
_DEGENERATE_FINAL_NUDGE = (
|
|
"[System: Your previous message ended the turn with a fragment that is not a usable answer. "
|
|
"If the task is unfinished, continue it and then give the complete answer. If that fragment "
|
|
"WAS your complete answer, send it again exactly as before.]"
|
|
)
|
|
|
|
# Re-prompt for finish_reason="tool_calls" with empty tool_calls (an interrupt mid-retry can persist it).
|
|
_DROPPED_TOOLCALL_NUDGE_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."
|
|
)
|
|
|
|
# Re-prompt for an empty response after tool calls (#9400); the metadata flag does not
|
|
# survive SessionDB projection, so it is matched by content.
|
|
_EMPTY_TOOL_RESPONSE_NUDGE = (
|
|
"You just executed tool calls but returned an empty response. Please process the tool "
|
|
"results above and continue with the task."
|
|
)
|
|
|
|
|
|
|
|
|
|
# Memo for send-path tool-call argument canonicalization (re-run on every historical call
|
|
# each iteration). Sound because canonicalization is pure; malformed strings raise before
|
|
# being stored, so the repair fallback is never memoized. The byte budget exists because
|
|
# argument strings can run 100KB+, so a count bound alone does not bound memory.
|
|
_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:
|
|
"""Canonical wire form of a tool-call arguments JSON string; raises on malformed input
|
|
(the caller falls back to ``_repair_tool_call_arguments``)."""
|
|
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))
|
|
_canon_args_cache_bytes -= len(evicted_key) + len(_CANON_ARGS_CACHE.pop(evicted_key))
|
|
except (StopIteration, KeyError, RuntimeError):
|
|
break
|
|
return canonical
|
|
|
|
|
|
def _clone_message_for_send(msg):
|
|
"""Structural clone (dicts/lists recursively, immutable leaves shared) of a history
|
|
message for the per-call API copy, so send-path rewrites never reach the persisted
|
|
transcript (#80498). Cheaper than deepcopy: messages are JSON-shaped and acyclic."""
|
|
if isinstance(msg, dict):
|
|
return {k: _clone_message_for_send(v) if isinstance(v, (dict, list)) else v for k, v in msg.items()}
|
|
if isinstance(msg, list):
|
|
return [_clone_message_for_send(v) if isinstance(v, (dict, list)) else v for v in msg]
|
|
return msg
|
|
|
|
|
|
def _canonicalize_api_tool_calls(api_messages) -> None:
|
|
"""Canonicalize tool-call argument JSON on the send-path copy (copy-on-write for the
|
|
dicts it touches; persisted history untouched)."""
|
|
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:
|
|
fn = tc["function"]
|
|
try:
|
|
args = _canonicalize_tool_call_arguments(fn["arguments"])
|
|
except Exception:
|
|
args = _repair_tool_call_arguments(fn["arguments"], fn.get("name", "?"))
|
|
# Copy-on-write as defense in depth: callers may pass shallow copies, and
|
|
# writing into a shared tc["function"] rewrote the stored turn with "{}"
|
|
# on the unrepairable path (#80498).
|
|
tc = {**tc, "function": {**fn, "arguments": args}}
|
|
new_tcs.append(tc)
|
|
am["tool_calls"] = new_tcs
|
|
|
|
|
|
def _invalid_tool_name_error_content(name: str, valid_tool_names) -> str:
|
|
"""Error content for an unknown tool name. A blank name is a model echoing tool-call
|
|
syntax seen in data (#47967) — dumping the catalog feeds that loop, so it gets a terse
|
|
error; a nonempty wrong name still gets the catalog to 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]:
|
|
"""Terminal turn result for a content-policy block (deterministic for the unchanged
|
|
prompt, so no retry); shared by the HTTP-200 and exception paths."""
|
|
return {
|
|
"final_response": final_response, "messages": messages, "api_calls": api_call_count,
|
|
"completed": False, "failed": True, "error": f"content_policy_blocked: {error_detail}",
|
|
"failure_reason": "content_policy_blocked", "failure_retryable": False,
|
|
}
|
|
|
|
|
|
def _partial_turn_result(
|
|
final_response: str, messages: List[Dict], api_call_count: int, **flags: Any
|
|
) -> Dict[str, Any]:
|
|
"""Incomplete-turn result whose ``error`` mirrors ``final_response``; ``flags`` add the
|
|
recovery-contract keys (``failed``, ``compression_deferred``, ...)."""
|
|
return {
|
|
"final_response": final_response, "messages": messages, "completed": False,
|
|
"api_calls": api_call_count, "error": final_response, "partial": True, **flags,
|
|
}
|
|
|
|
|
|
def _compression_deferred_result(agent, messages: List[Dict], api_call_count: int, reason: str = "lock") -> Dict[str, Any]:
|
|
"""Soft turn result for a transiently-deferred compression. Both reasons must end as
|
|
``compression_deferred``, never ``compression_exhausted`` — the gateway wipes the
|
|
session on exhaustion (#9893/#35809). ``failed`` stays False; the turn persists."""
|
|
session = agent.session_id or "none"
|
|
if reason == "transient_block":
|
|
block = getattr(agent, "_compression_blocked_transient", None)
|
|
logger.info(
|
|
"turn deferred: compression transiently blocked (%s) (session=%s) — not counting as "
|
|
"compression exhaustion", block if isinstance(block, str) else "unknown guard", session,
|
|
)
|
|
_final = (
|
|
"Context compression is temporarily paused after a recent failed attempt. Please retry "
|
|
"in a moment — compression will resume automatically (or run /compress to force a retry now)."
|
|
)
|
|
else:
|
|
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", session, holder if isinstance(holder, str) else "unconfirmed",
|
|
)
|
|
_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."
|
|
)
|
|
try:
|
|
agent._flush_status_buffer()
|
|
except Exception:
|
|
pass
|
|
return _partial_turn_result(
|
|
_final, messages, api_call_count,
|
|
failed=False, compression_deferred=True, session_id=agent.session_id,
|
|
)
|
|
|
|
|
|
def _provider_overflow_exhausted_result(
|
|
agent, messages: List[Dict], conversation_history, api_call_count: int,
|
|
request_pressure_tokens: int, max_compression_attempts: int,
|
|
) -> Dict[str, Any]:
|
|
"""Fail closed when a rebuilt request is still too large after recovery."""
|
|
agent._flush_status_buffer()
|
|
logger.error(
|
|
"%sContext compression failed after %d attempts; rebuilt request "
|
|
"remains over threshold at ~%s tokens.",
|
|
agent.log_prefix, max_compression_attempts, f"{request_pressure_tokens:,}",
|
|
)
|
|
# Host progress-aware timeout (#98722, salvaged from #98741): the provider proved the request does not
|
|
# fit, but this recovery pass spent the full wait budget without a committed summary. Re-sending the
|
|
# unchanged request would bounce off the same overflow error and re-enter compression in the same turn.
|
|
# End the turn with the typed recovery contract instead — transcript intact, no further doomed provider
|
|
# sends.
|
|
# Prior <3 retries (or an earlier successful tool batch) leave a tool-result tail. Closing it here
|
|
# matches interrupt aborts (#48879 / #52592) so the next user turn is not tool→user for strict
|
|
# providers.
|
|
agent._persist_session(messages, conversation_history)
|
|
return _partial_turn_result(
|
|
site_copy("context_overflow", model=agent.model),
|
|
messages, api_call_count, failed=True, compression_exhausted=True,
|
|
turn_exit_reason="context_compression_exhausted",
|
|
failure_reason="context_overflow", failure_retryable=False,
|
|
)
|
|
|
|
|
|
def _rewrite_system_content_blocks(system_message: dict, effective: str) -> bool:
|
|
"""Rewrite a cache-decorated system message in place, keeping its blocks (a bare string
|
|
over the ``[static prefix, volatile tail]`` list would drop both cache_control
|
|
breakpoints). Returns False when the shape cannot be safely patched."""
|
|
content = system_message.get("content")
|
|
if not isinstance(content, list) or not content or 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) and effective[len(head):]:
|
|
content[1]["text"] = effective[len(head):]
|
|
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: ``api_messages`` were
|
|
built pre-failover and are reused each retry. Returns the new ``active_system_prompt``."""
|
|
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 + "\n\n" + agent.ephemeral_system_prompt).strip() if agent.ephemeral_system_prompt else sp
|
|
if not _rewrite_system_content_blocks(api_messages[0], effective):
|
|
api_messages[0]["content"] = effective
|
|
return sp
|
|
|
|
|
|
def _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry):
|
|
"""After a successful fallback activation: sync the system message and arm
|
|
``restart_with_rebuilt_messages``. Callers also zero ``retry_count`` /
|
|
``compression_attempts`` and ``break`` the retry loop."""
|
|
active_system_prompt = _sync_failover_system_message(
|
|
agent, api_messages, active_system_prompt)
|
|
_retry.primary_recovery_attempted = False
|
|
_retry.restart_with_rebuilt_messages = True
|
|
return active_system_prompt
|
|
|
|
|
|
def _ensure_cached_system_prompt_static(agent, system_message=None) -> None:
|
|
"""Rebuild ``_cached_system_prompt_static`` when caching becomes active (#72626): sessions
|
|
restored under a cache-off primary would otherwise fall back to the legacy layout after
|
|
failover to a cache-on provider."""
|
|
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 attached by ``_attach_reference_guidance``."""
|
|
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]], List[Dict[str, Any]]]:
|
|
"""Strip and re-apply cache_control for the *current* provider policy — failover
|
|
``continue`` paths reuse ``api_messages`` (#72626). MoA guidance is peeled and rebased."""
|
|
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":
|
|
# Prepared MoA state is canonical: the synchronous acting-aggregator
|
|
# sender owns its destination-local cache plan after it resolves the slot.
|
|
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"]
|
|
# Direct attribute access, not getattr: the flags are always initialized on
|
|
# AIAgent, and a default would mask a real init bug as silent cache-off.
|
|
elif agent._use_prompt_caching:
|
|
_ensure_cached_system_prompt_static(agent, system_message=system_message)
|
|
static = getattr(agent, "_cached_system_prompt_static", None)
|
|
from agent.prompt_caching import envelope_tool_part_cache_markers_supported
|
|
plan = build_prompt_cache_plan(
|
|
messages,
|
|
planned_tools,
|
|
# Clamp per-destination: a configured 1h regresses to 5m on
|
|
# Qwen/Alibaba routes, whose context cache is 5m-only (#84733).
|
|
cache_ttl=effective_cache_ttl(agent._cache_ttl, provider=agent.provider, model=agent.model),
|
|
native_anthropic=agent._use_native_cache_layout,
|
|
static_system_prefix=static if isinstance(static, str) else None,
|
|
direct_native_tool_cache=getattr(
|
|
agent, "_direct_native_anthropic_tool_cache_capability", lambda: False
|
|
)(),
|
|
# LiteLLM-style envelope routes forward part-level markers into
|
|
# tool_result.content[] → non-retryable 400 (#89886).
|
|
tool_part_markers=envelope_tool_part_cache_markers_supported(
|
|
getattr(agent, "provider", ""), getattr(agent, "base_url", "")
|
|
),
|
|
)
|
|
messages, planned_tools = plan.messages, plan.tools
|
|
|
|
return messages, prepared, planned_tools
|
|
|
|
|
|
def _engine_overrides_hook(engine: Any, name: str) -> bool:
|
|
"""True when ``engine`` implements ContextEngine hook ``name`` itself.
|
|
|
|
Non-implementing engines must pay nothing per turn; ``hasattr`` is not enough because
|
|
the ABC defines a no-op default. Lazy import avoids a cycle with agent.context_engine."""
|
|
hook = getattr(engine, name, None)
|
|
if engine is None or not callable(hook):
|
|
return False
|
|
try:
|
|
from agent.context_engine import ContextEngine as _CE
|
|
return getattr(hook, "__func__", None) is not getattr(_CE, name)
|
|
except Exception:
|
|
return True
|
|
|
|
|
|
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, fail-open: any
|
|
exception or invalid return yields ``api_messages`` unchanged; history is never mutated."""
|
|
engine = getattr(agent, "context_compressor", None)
|
|
if not _engine_overrides_hook(engine, "select_context"):
|
|
return api_messages
|
|
|
|
session_label = getattr(agent, "session_id", None) or "-"
|
|
# Structural clones: the engine must not be able to write through nested
|
|
# containers into persisted history; only the request list is acted on (#80498).
|
|
try:
|
|
selected = engine.select_context(
|
|
api_messages,
|
|
conversation_messages=(
|
|
[_clone_message_for_send(m) for m in conversation_messages]
|
|
if conversation_messages is not None else None
|
|
),
|
|
incoming_message=(
|
|
_clone_message_for_send(incoming_message)
|
|
if isinstance(incoming_message, dict) else incoming_message
|
|
),
|
|
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
|
|
# Require a NON-EMPTY list of dicts: ``all([])`` is ``True``, so a ``[]`` from a
|
|
# buggy engine would otherwise replace the request instead of failing open.
|
|
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 (fail-open; the engine
|
|
gets a copy so it cannot mutate the persisted transcript)."""
|
|
engine = getattr(agent, "context_compressor", None)
|
|
if not _engine_overrides_hook(engine, "on_turn_complete"):
|
|
return
|
|
try:
|
|
# Structural clones: dict(m) would let a hook write into nested containers of the
|
|
# persisted transcript (#80498).
|
|
engine.on_turn_complete([_clone_message_for_send(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 _decode_inline_moa_turn(user_message, persist_user_message):
|
|
"""Decode a MoA preset encoded into ``user_message``; returns ``(user_message,
|
|
moa_config, persist_user_message)``, unchanged with ``moa_config=None`` otherwise."""
|
|
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:
|
|
if persist_user_message is None:
|
|
persist_user_message = _decoded_message
|
|
return _decoded_message, _decoded_moa_config, persist_user_message
|
|
except Exception:
|
|
pass
|
|
return user_message, None, persist_user_message
|
|
|
|
|
|
def _preflight_timeout_result(agent, exc, conversation_history) -> Dict[str, Any]:
|
|
"""Typed recovery result when turn-start preflight compression timed out (#98424): no
|
|
provider call was sent, and surfaces would otherwise hide the actionable guidance."""
|
|
logger.warning(
|
|
"Turn-start preflight compression timed out — ending turn with typed recovery result: %s", exc,
|
|
)
|
|
# Clear the tripwire slot note_turn_start registered (the early return skips the persist
|
|
# funnel). The user row is deliberately NOT persisted (#7100).
|
|
from agent.agent_runtime_helpers import note_turn_persisted
|
|
note_turn_persisted(agent)
|
|
# Not _COMPRESSION_TIMEOUT_FINAL_RESPONSE — that describes a different state
|
|
# (compression ran, could not reduce); the exception text carries the guidance.
|
|
return _partial_turn_result(
|
|
str(exc), list(conversation_history or []), 0,
|
|
failed=True, compression_exhausted=True, turn_exit_reason="context_compression_timeout",
|
|
failure_reason="context_overflow", failure_retryable=False,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class _LoopState:
|
|
"""Every local the turn loop threads through the phase helpers in ``agent/turn_*.py``.
|
|
|
|
Helpers take the loop locals they need as keyword arguments named like these fields and
|
|
return a verdict whose non-``action``/``result`` fields carry the same names;
|
|
:func:`_run_phase` passes and copies them back by name, so a new helper input/output
|
|
needs a field here and nothing else. Per-iteration slots are rebound by the phases
|
|
before any later phase reads them, exactly as the former inline locals were."""
|
|
|
|
# Fixed for the turn.
|
|
user_message: Any
|
|
system_message: Any
|
|
moa_config: Any
|
|
original_user_message: Any
|
|
conversation_history: Any
|
|
effective_task_id: Any
|
|
turn_id: Any
|
|
_should_review_memory: Any
|
|
_plugin_user_context: Any
|
|
_ext_prefetch_cache: Any
|
|
# Turn-scoped state (rebound by the phases).
|
|
messages: Any
|
|
active_system_prompt: Any
|
|
current_turn_user_idx: Any
|
|
_preflight_compression_blocked: Any
|
|
# Compression attempt cap shared by the pre-API gate, 413 handlers and post-tool compaction:
|
|
# a consecutive-ineffective-attempt backstop, rearmed only after a provider response
|
|
# reports a prompt below threshold.
|
|
max_compression_attempts: Any
|
|
api_call_count: int = 0
|
|
final_response: Any = None
|
|
interrupted: bool = False
|
|
failed: bool = False
|
|
codex_ack_continuations: int = 0
|
|
length_continue_retries: int = 0
|
|
# Per-turn backstop for the refunding restarts (redirect / rebuilt-for-fallback).
|
|
# Unlike ``retry_count`` (rebound to 0 each iteration) this accumulates for the whole
|
|
# turn so a runaway interrupt/redirect that keeps re-arming a restart flag cannot
|
|
# refund the iteration budget forever and hold the turn lease indefinitely.
|
|
restart_count: int = 0
|
|
_outer_error_count: int = 0 # outer-loop exceptions this turn (#92450), see _MAX_OUTER_LOOP_ERRORS
|
|
truncated_tool_call_retries: int = 0
|
|
truncated_response_parts: List[tuple[str, bool]] = field(default_factory=list)
|
|
compression_attempts: int = 0
|
|
_last_preflight_pressure: Optional[int] = None
|
|
# A provider overflow outweighs the rough-estimate calibration that defers preflight after
|
|
# compaction: stays armed until the rebuilt request is below the threshold.
|
|
_provider_overflow_recovery_pending: bool = False
|
|
# A compression host-timeout ended the turn; finalize reuses the gateway context-recovery
|
|
# contract (error/partial/compression_exhausted) (#98722).
|
|
_compression_timeout_exhausted: bool = False
|
|
_turn_exit_reason: str = "unknown" # diagnostic: why the loop ended
|
|
# Answer held back by a verification gate (best user-facing result if the continuation
|
|
# exhausts the budget) and whether it was streamed as interim; ``_response_was_previewed``
|
|
# is set ONLY if it becomes the final response (#65919).
|
|
_pending_verification_response: Any = None
|
|
_pending_verification_response_previewed: bool = False
|
|
# MoA guidance retained across a pre-API compression, rebased next iteration (no second fan-out).
|
|
pending_moa_prepared_request: Any = None
|
|
# Per-iteration slots.
|
|
request_logger: Any = None
|
|
api_messages: Any = None
|
|
tools_for_api: Any = None
|
|
_moa_prepared_request: Any = None
|
|
approx_tokens: Any = None
|
|
request_pressure_tokens: Any = None
|
|
total_chars: Any = None
|
|
thinking_spinner: Any = None
|
|
api_start_time: Any = None
|
|
retry_count: int = 0
|
|
max_retries: Any = None
|
|
_retry: Any = None
|
|
finish_reason: str = "stop"
|
|
response: Any = None # None when every retry failed
|
|
api_kwargs: Any = None # None until built; read by the except handlers
|
|
api_request_id: Any = None
|
|
_original_api_kwargs: Any = None
|
|
_llm_middleware_trace: Any = None
|
|
api_duration: Any = None
|
|
assistant_message: Any = None
|
|
|
|
|
|
# _LoopState fields seeded from TurnContext (same name minus the leading underscore).
|
|
_CTX_FIELDS = frozenset({
|
|
"user_message", "original_user_message", "conversation_history", "effective_task_id", "turn_id",
|
|
"_should_review_memory", "_plugin_user_context", "_ext_prefetch_cache", "messages",
|
|
"active_system_prompt", "current_turn_user_idx", "_preflight_compression_blocked",
|
|
})
|
|
# Keyword names each phase helper takes (minus ``agent``), cached per function object.
|
|
_PHASE_PARAMS: Dict[Any, tuple] = {}
|
|
# Verdict fields the loop latches (only ever sets True) instead of copying back:
|
|
# ``handle_api_error`` reports overflow recovery per call and must not clear an earlier arm.
|
|
_LATCHED_VERDICT_FIELDS = {"handle_api_error": frozenset({"_provider_overflow_recovery_pending"})}
|
|
|
|
|
|
def _run_phase(fn, agent, state: _LoopState, **extra):
|
|
"""Call phase helper ``fn`` with the loop locals it names, copy its verdict fields back.
|
|
|
|
``extra`` supplies non-state arguments (the caught exception). Returns the verdict so
|
|
the caller can act on ``.action`` / ``.result``."""
|
|
params = _PHASE_PARAMS.get(fn)
|
|
if params is None:
|
|
params = _PHASE_PARAMS[fn] = tuple(p for p in inspect.signature(fn).parameters if p != "agent")
|
|
verdict = fn(agent, **{n: extra[n] if n in extra else getattr(state, n) for n in params})
|
|
latched = _LATCHED_VERDICT_FIELDS.get(getattr(fn, "__name__", ""), ())
|
|
for f in fields(verdict):
|
|
if f.name in ("action", "result"):
|
|
continue
|
|
value = getattr(verdict, f.name)
|
|
if f.name not in latched:
|
|
setattr(state, f.name, value)
|
|
elif value:
|
|
setattr(state, f.name, True)
|
|
return verdict
|
|
|
|
|
|
def _run_api_retry_loop(agent, s: _LoopState) -> Optional[Dict[str, Any]]:
|
|
"""One API call with its retry/recovery loop (guard → build → call → check, error handlers).
|
|
|
|
Returns a turn result dict when a phase ends the turn, else None once the loop is left
|
|
(success, a restart armed on ``s._retry``, interrupt, or retries exhausted)."""
|
|
while s.retry_count < s.max_retries:
|
|
_ng = _run_phase(nous_rate_limit_guard, agent, s)
|
|
if _ng.action == "return":
|
|
return _ng.result
|
|
if _ng.action == "break":
|
|
return None
|
|
try:
|
|
_run_phase(build_api_request, agent, s)
|
|
if _run_phase(perform_api_call, agent, s).action == "break":
|
|
return None
|
|
_rc = _run_phase(check_api_response, agent, s)
|
|
if _rc.action == "return":
|
|
return _rc.result
|
|
if _rc.action == "break":
|
|
return None
|
|
except InterruptedError:
|
|
if _run_phase(handle_api_interrupt, agent, s).action == "break":
|
|
return None
|
|
except Exception as api_error:
|
|
_ae = _run_phase(handle_api_error, agent, s, api_error=api_error)
|
|
if _ae.action == "return":
|
|
return _ae.result
|
|
if _ae.action == "break":
|
|
return None
|
|
return None
|
|
|
|
|
|
def _run_conversation_turn(
|
|
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,
|
|
persist_user_platform_id: Optional[str] = None,
|
|
turn_author: Optional[Dict[str, Any]] = None,
|
|
moa_config: Optional[dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Run a complete conversation with tool calling until completion; returns the result dict.
|
|
|
|
``stream_callback``: per-text-delta callback (TTS). ``persist_user_message``: clean text to
|
|
store when ``user_message`` carries API-only synthetic prefixes; timestamp / platform id are
|
|
stored as metadata (platform id lets restart drain recovery dedup). ``persist_user_display_*``:
|
|
display-only event rendering; the model still receives the message unchanged."""
|
|
if moa_config is None:
|
|
user_message, moa_config, persist_user_message = _decode_inline_moa_turn(
|
|
user_message, persist_user_message
|
|
)
|
|
|
|
# The gateway caches agents across turns; compression state is per-turn, or a stale
|
|
# in-place boundary would make a later uncompressed result look compacted.
|
|
agent._last_compaction_in_place = agent._last_compression_attempt_recorded = False
|
|
agent._last_compression_attempt_in_place = None
|
|
begin_fast_mode_turn(agent, conversation_history)
|
|
|
|
# Adopt ~/.hermes/.env credential/base-url edits made since the last turn — a
|
|
# Settings save updates .env, not this worker's client (#67821). No-op if unchanged.
|
|
try:
|
|
agent._try_refresh_env_client_credentials()
|
|
except Exception:
|
|
logger.debug("per-turn env credential refresh failed", exc_info=True)
|
|
|
|
# Per-turn setup: build_turn_context mutates ``agent`` and returns the locals the loop reads.
|
|
try:
|
|
_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,
|
|
persist_user_platform_id=persist_user_platform_id,
|
|
turn_author=turn_author,
|
|
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 turns append per-call aggregated context to the API copy of the
|
|
# user message, so no byte-stable api_content sidecar can be stamped.
|
|
moa_active=bool(moa_config),
|
|
)
|
|
except PreflightCompressionTimedOut as _preflight_timeout_exc:
|
|
return _preflight_timeout_result(agent, _preflight_timeout_exc, conversation_history)
|
|
|
|
# Per-turn agent state (the gateway caches agents across turns, so none of this may
|
|
# leak into the next message): interim-commentary dedup spans the whole turn but not
|
|
# the next; a SessionDB append failure (and its classified cause) halts only this turn;
|
|
# a failed compression-tip adoption is reported only against its own turn; the
|
|
# thinking-only-truncation one-shot must not survive an interrupted turn; credential-
|
|
# pool refresh tallies cap same-entry refreshes on a persistent 401 (#26080); usage
|
|
# for on_turn_complete() stays None on turns that never reach a response.
|
|
agent._delivered_interim_texts = set()
|
|
agent._incremental_persistence_failed = False
|
|
agent._last_persistence_error_cause = None
|
|
agent._compression_adoption_failed = False
|
|
agent._ephemeral_reasoning_off = False
|
|
agent._auth_pool_refresh_counts = {}
|
|
agent._last_turn_usage = None
|
|
|
|
s = _LoopState(
|
|
system_message=system_message, moa_config=moa_config,
|
|
max_compression_attempts=getattr(agent, "max_compression_attempts", 3),
|
|
**{f.name: getattr(_ctx, f.name.lstrip("_")) for f in fields(_LoopState) if f.name in _CTX_FIELDS},
|
|
)
|
|
# Opt-in runtime: api_mode == codex_app_server hands the whole turn to the codex
|
|
# app-server subprocess (see agent/transports/codex_app_server_session.py).
|
|
if agent.api_mode == "codex_app_server":
|
|
codex_result = agent._run_codex_app_server_turn(
|
|
user_message=s.user_message, original_user_message=s.original_user_message,
|
|
messages=s.messages, effective_task_id=s.effective_task_id,
|
|
should_review_memory=s._should_review_memory,
|
|
)
|
|
from agent.turn_recovery import activate_codex_app_server_fallback
|
|
if not activate_codex_app_server_fallback(agent, codex_result):
|
|
return codex_result
|
|
# Fallback activation rewrote provider/model/api_mode: retry this same user turn on the generic
|
|
# loop below, keeping codex's projected rows and its failed API call in the turn's accounting.
|
|
s.api_call_count = int(codex_result.get("api_calls") or 0)
|
|
s.active_system_prompt = _sync_failover_system_message(agent, None, s.active_system_prompt)
|
|
|
|
while (s.api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call:
|
|
if _run_phase(begin_iteration, agent, s).action == "break":
|
|
break
|
|
_run_phase(prepare_iteration, agent, s)
|
|
_run_phase(assemble_api_request, agent, s)
|
|
_pg = _run_phase(run_preflight_gate, agent, s)
|
|
if _pg.action == "return":
|
|
return _pg.result
|
|
if _pg.action == "break":
|
|
break
|
|
if _pg.action == "continue":
|
|
continue
|
|
_run_phase(announce_api_call, agent, s)
|
|
|
|
s.api_start_time, s.retry_count, s.max_retries = time.time(), 0, agent._api_max_retries
|
|
s._retry, s.finish_reason, s.response, s.api_kwargs = TurnRetryState(), "stop", None, None
|
|
s.api_request_id = agent._current_api_request_id = f"{s.turn_id}:api:{s.api_call_count}"
|
|
|
|
early_result = _run_api_retry_loop(agent, s)
|
|
if early_result is not None:
|
|
return early_result
|
|
|
|
_rs = _run_phase(apply_retry_restarts, agent, s)
|
|
if _rs.action == "break":
|
|
break
|
|
if _rs.action == "continue":
|
|
continue
|
|
|
|
try:
|
|
_ri = _run_phase(normalize_model_response, agent, s)
|
|
if _ri.action == "return":
|
|
return _ri.result
|
|
if _ri.action == "continue":
|
|
continue
|
|
_v = _run_phase(
|
|
run_tool_round if s.assistant_message.tool_calls else finish_text_response, agent, s
|
|
)
|
|
if _v.action == "return":
|
|
return _v.result
|
|
if _v.action == "break":
|
|
break
|
|
if _v.action == "continue":
|
|
continue
|
|
except Exception as e:
|
|
if _run_phase(handle_outer_loop_error, agent, s, e=e).action == "break":
|
|
break
|
|
|
|
# Post-loop finalization lives in agent/turn_finalizer.finalize_turn.
|
|
result = finalize_turn(agent, **{
|
|
name: getattr(s, name)
|
|
for name in inspect.signature(finalize_turn).parameters if name != "agent"
|
|
})
|
|
if s._compression_timeout_exhausted:
|
|
# Reuse the gateway's context-recovery contract: transcript stays intact while
|
|
# future input can move to a clean session (#98722).
|
|
result.update(error=_COMPRESSION_TIMEOUT_FINAL_RESPONSE, partial=True, compression_exhausted=True)
|
|
return result
|
|
|
|
|
|
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,
|
|
persist_user_platform_id: Optional[str] = None,
|
|
moa_config: Optional[dict[str, Any]] = None,
|
|
turn_author: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Run one turn (see ``_run_conversation_turn``) and export the current-turn boundary.
|
|
|
|
Every envelope that leaves the loop — success, partial/error, interrupt, retry-exhausted,
|
|
tool-limit, preflight timeout, codex runtime — passes through here, so the
|
|
``{turn_id, current_turn_user_idx}`` pair is stamped beside the exact ``messages`` it
|
|
addresses, after every history rewrite including post-turn micro-compaction.
|
|
"""
|
|
from agent.turn_context import export_current_turn_boundary
|
|
from tools.vision_tools_history_budget import native_turn_images
|
|
|
|
# Images attached natively to this user turn stay visible to vision_analyze for the turn, so
|
|
# it does not embed the same pixels a second time into the same request (#76411).
|
|
with native_turn_images(user_message):
|
|
result = _run_conversation_turn(
|
|
agent,
|
|
user_message,
|
|
system_message=system_message,
|
|
conversation_history=conversation_history,
|
|
task_id=task_id,
|
|
stream_callback=stream_callback,
|
|
persist_user_message=persist_user_message,
|
|
persist_user_timestamp=persist_user_timestamp,
|
|
persist_user_display_kind=persist_user_display_kind,
|
|
persist_user_display_metadata=persist_user_display_metadata,
|
|
persist_user_platform_id=persist_user_platform_id,
|
|
moa_config=moa_config,
|
|
turn_author=turn_author,
|
|
)
|
|
result = export_current_turn_boundary(agent, result, user_message)
|
|
_close_durable_failed_turn(agent, result)
|
|
return result
|
|
|
|
|
|
def _close_durable_failed_turn(agent, result: Any) -> None:
|
|
"""Append a Hermes-authored assistant boundary when a failed turn left ``user`` as the
|
|
durable conversation tail (in place, on ``result["messages"]`` and in SessionDB).
|
|
|
|
The terminal-failure paths (content-policy refusal, ``_Trunc.end_turn``, retry exhaustion,
|
|
interrupt before any assistant text) persist the accepted user row and return without
|
|
reaching ``finalize_turn``; the next prompt then appends a second user row and
|
|
``repair_message_sequence`` merges the failed request into the new one. The gateway
|
|
compensates with ``_hmwa_close_failed_turn``; CLI, TUI/Desktop and ACP hosts hand
|
|
``result["messages"]`` straight back as history, so the seam is here.
|
|
|
|
Excluded: the context-pressure classes (``compression_exhausted``, ``compression_deferred``,
|
|
``failure_reason == "context_overflow"``) — appending to an already-oversized session is the
|
|
#1630 growth loop; their repair is rotation or a retry. Idempotence is keyed on the DURABLE
|
|
tail (``SessionDB.latest_conversation_role``), so a redelivery or a tail already closed by
|
|
another writer is a no-op, and the gateway's own closer then no-ops in turn.
|
|
"""
|
|
try:
|
|
if not isinstance(result, dict) or result.get("completed") is True:
|
|
return
|
|
if (
|
|
result.get("compression_exhausted") or result.get("compression_deferred")
|
|
or result.get("failure_reason") == "context_overflow"
|
|
):
|
|
return
|
|
messages = result.get("messages")
|
|
db, session_id = getattr(agent, "_session_db", None), getattr(agent, "session_id", None)
|
|
if not isinstance(messages, list) or not messages or db is None or not session_id:
|
|
return
|
|
if getattr(agent, "_persist_disabled", False) or db.latest_conversation_role(session_id) != "user":
|
|
return
|
|
# Scope the "did a tool run" scan to this turn when its boundary is proven; otherwise
|
|
# hedge over the whole list rather than under-report a possible side effect.
|
|
start = result.get("current_turn_user_idx")
|
|
turn_messages = messages[start:] if isinstance(start, int) and 0 <= start < len(messages) else messages
|
|
append_message(messages, {
|
|
"role": "assistant", "content": failed_turn_notice(turn_messages), "display_kind": FAILED_TURN_DISPLAY_KIND,
|
|
})
|
|
agent._flush_messages_to_session_db(messages)
|
|
except Exception:
|
|
logger.debug("failed-turn boundary not written", exc_info=True)
|
|
|
|
|
|
__all__ = ["run_conversation"]
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
import os # noqa: F401,E402
|
|
import random # noqa: F401,E402
|
|
import ssl # noqa: F401,E402
|
|
import sys # noqa: F401,E402
|
|
|
|
|
|
_PLUGIN_COMPAT_LAZY = {
|
|
'COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE': ('agent.conversation_compression', 'COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE'),
|
|
'COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE': ('agent.conversation_compression', 'COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE'),
|
|
'COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE': ('agent.conversation_compression', 'COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE'),
|
|
'COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE': ('agent.conversation_compression', 'COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE'),
|
|
'FailoverReason': ('agent.error_classifier', 'FailoverReason'),
|
|
'KawaiiSpinner': ('agent.display', 'KawaiiSpinner'),
|
|
'PARTIAL_STREAM_STUB_ID': ('hermes_constants', 'PARTIAL_STREAM_STUB_ID'),
|
|
'PRE_API_COMPRESSION_STATUS_TEMPLATE': ('agent.conversation_compression', 'PRE_API_COMPRESSION_STATUS_TEMPLATE'),
|
|
'adaptive_rate_limit_backoff': ('agent.retry_utils', 'adaptive_rate_limit_backoff'),
|
|
'anchored_context_tokens': ('agent.usage_anchor', 'anchored_context_tokens'),
|
|
'automatic_compaction_status_message': ('agent.context_engine', 'automatic_compaction_status_message'),
|
|
'capture_usage_anchor': ('agent.usage_anchor', 'capture_usage_anchor'),
|
|
'classify_api_error': ('agent.error_classifier', 'classify_api_error'),
|
|
'close_interrupted_tool_sequence': ('agent.message_sanitization', 'close_interrupted_tool_sequence'),
|
|
'coalesce_tool_call_id': ('agent.message_sanitization', 'coalesce_tool_call_id'),
|
|
'compose_user_api_content': ('agent.turn_context', 'compose_user_api_content'),
|
|
'compression_blocked_transiently': ('agent.conversation_compression', 'compression_blocked_transiently'),
|
|
'compression_skipped_due_to_lock': ('agent.conversation_compression', 'compression_skipped_due_to_lock'),
|
|
'context_compression_timed_out': ('agent.conversation_compression', 'context_compression_timed_out'),
|
|
'conversation_history_after_compression': ('agent.conversation_compression', 'conversation_history_after_compression'),
|
|
'env_var_enabled': ('utils', 'env_var_enabled'),
|
|
'estimate_messages_tokens_rough': ('agent.model_metadata', 'estimate_messages_tokens_rough'),
|
|
'estimate_request_tokens_rough': ('agent.model_metadata', 'estimate_request_tokens_rough'),
|
|
'estimate_usage_cost': ('agent.usage_pricing', 'estimate_usage_cost'),
|
|
'get_context_length_from_provider_error': ('agent.model_metadata', 'get_context_length_from_provider_error'),
|
|
'has_incomplete_scratchpad': ('agent.trajectory', 'has_incomplete_scratchpad'),
|
|
'is_output_cap_error': ('agent.model_metadata', 'is_output_cap_error'),
|
|
'is_repetition_dominated': ('agent.repetition_guard', 'is_repetition_dominated'),
|
|
'is_zai_coding_overload_error': ('agent.retry_utils', 'is_zai_coding_overload_error'),
|
|
'jittered_backoff': ('agent.retry_utils', 'jittered_backoff'),
|
|
'normalize_usage': ('agent.usage_pricing', 'normalize_usage'),
|
|
'parse_available_output_tokens_from_error': ('agent.model_metadata', 'parse_available_output_tokens_from_error'),
|
|
'reanchor_current_turn_user_idx': ('agent.turn_context', 'reanchor_current_turn_user_idx'),
|
|
'save_context_length': ('agent.model_metadata', 'save_context_length'),
|
|
'serialized_messages_bytes': ('agent.message_sanitization', 'serialized_messages_bytes'),
|
|
'splice_provider_projection': ('agent.provider_projection', 'splice_provider_projection'),
|
|
'zai_coding_overload_retry_ceiling': ('agent.retry_utils', 'zai_coding_overload_retry_ceiling'),
|
|
}
|
|
|
|
|
|
def __getattr__(name): # PEP 562 — lazy so no import cycles
|
|
target = _PLUGIN_COMPAT_LAZY.get(name)
|
|
if target is None:
|
|
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
|
import importlib
|
|
from hermes_cli.plugin_compat import warn_once
|
|
warn_once(__name__, name, *target)
|
|
return getattr(importlib.import_module(target[0]), target[1])
|
|
# ---- END PLUGIN-COMPAT ----
|