diff --git a/.github/workflows/fleet-ci-fail-alert.yml b/.github/workflows/fleet-ci-fail-alert.yml new file mode 100644 index 0000000000000..304499daff241 --- /dev/null +++ b/.github/workflows/fleet-ci-fail-alert.yml @@ -0,0 +1,76 @@ +# Fleet-added (Apollo). NOT upstream — do not expect on NousResearch/hermes-agent. +# Fires a single LOUD ping to the fleet #alerts channel ONLY when a real CI +# workflow concludes in failure. The failure-filter lives HERE (`if: failure`) +# because the Hermes webhook adapter filters by event TYPE only, not payload +# conclusion — a deliver_only route on raw check_suite/status would also fire on +# SUCCESS and re-create the flood this replaces. Zero-LLM: the receiver route is +# deliver_only, so this is a pure notification, never an agent run. +name: Fleet CI-fail alert + +on: + workflow_run: + workflows: + - "Tests" + - "Lint (ruff + ty)" + - "Fleet SAST (semgrep)" + - "Fleet Secret Scan (gitleaks)" + - "OSV-Scanner" + - "Supply Chain Audit" + types: + - completed + +permissions: + contents: read + +jobs: + notify-on-failure: + name: notify-on-failure + runs-on: ubuntu-latest + # Only on a genuine failure of the upstream workflow. Success/cancelled/skipped → no-op. + if: ${{ github.event.workflow_run.conclusion == 'failure' }} + steps: + - name: Sign and POST CI-failure event + env: + CI_FAIL_WEBHOOK_SECRET: ${{ secrets.CI_FAIL_WEBHOOK_SECRET }} + WEBHOOK_URL: https://hooks.angventures.io/webhooks/ci-fail + WF_NAME: ${{ github.event.workflow_run.name }} + WF_BRANCH: ${{ github.event.workflow_run.head_branch }} + WF_SHA: ${{ github.event.workflow_run.head_sha }} + WF_URL: ${{ github.event.workflow_run.html_url }} + WF_ACTOR: ${{ github.event.workflow_run.actor.login }} + REPO: ${{ github.repository }} + WF_EVENT: ${{ github.event.workflow_run.event }} + run: | + set -euo pipefail + if [ -z "${CI_FAIL_WEBHOOK_SECRET:-}" ]; then + echo "::error::CI_FAIL_WEBHOOK_SECRET is not set; cannot sign the alert." >&2 + exit 1 + fi + # Build the JSON body. event_type=ci_failure is what the receiver route + # filters on (X-GitHub-Event header is also set to the same value). + BODY="$(jq -nc \ + --arg event_type "ci_failure" \ + --arg repo "$REPO" \ + --arg workflow "$WF_NAME" \ + --arg branch "$WF_BRANCH" \ + --arg sha "$WF_SHA" \ + --arg run_url "$WF_URL" \ + --arg actor "$WF_ACTOR" \ + --arg trigger "$WF_EVENT" \ + '{event_type:$event_type, repo:$repo, workflow:$workflow, branch:$branch, sha:$sha, run_url:$run_url, actor:$actor, trigger:$trigger}')" + # Generic HMAC-SHA256 hex over the raw body → X-Webhook-Signature. + SIG="$(printf '%s' "$BODY" | openssl dgst -sha256 -hmac "$CI_FAIL_WEBHOOK_SECRET" | sed 's/^.* //')" + CODE="$(curl -sS -o /tmp/resp.txt -w '%{http_code}' \ + -X POST "$WEBHOOK_URL" \ + -H "Content-Type: application/json" \ + -H "X-GitHub-Event: ci_failure" \ + -H "X-GitHub-Delivery: ci-fail-${WF_SHA}-${GITHUB_RUN_ID}" \ + -H "X-Webhook-Signature: ${SIG}" \ + --data "$BODY")" + echo "Receiver responded HTTP $CODE" + cat /tmp/resp.txt || true + # 2xx = delivered/accepted. Non-2xx fails the step so the miss is visible. + case "$CODE" in + 2*) echo "CI-failure alert delivered." ;; + *) echo "::error::CI-fail alert POST returned HTTP $CODE" >&2; exit 1 ;; + esac diff --git a/.github/workflows/fleet-sast.yml b/.github/workflows/fleet-sast.yml new file mode 100644 index 0000000000000..0036ea216dd1c --- /dev/null +++ b/.github/workflows/fleet-sast.yml @@ -0,0 +1,41 @@ +# Fleet-added (Apollo/greploop posture-B floor). NOT upstream — do not expect on NousResearch/hermes-agent. +# Provides the `sast` deterministic-floor component. Diff-aware where it helps, full-tree gate at ERROR. +name: Fleet SAST (semgrep) + +on: + pull_request: + push: + branches: [main] + +permissions: + contents: read + +concurrency: + group: fleet-sast-${{ github.ref }} + cancel-in-progress: true + +jobs: + sast: + name: sast + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install semgrep + run: pip install "semgrep==1.46.0" + + - name: Run semgrep (ERROR severity gate) + run: | + semgrep \ + --config=p/python \ + --config=p/secrets \ + --severity=ERROR \ + --error \ + --quiet \ + hermes/ diff --git a/.github/workflows/fleet-secret-scan.yml b/.github/workflows/fleet-secret-scan.yml new file mode 100644 index 0000000000000..e02505857b00d --- /dev/null +++ b/.github/workflows/fleet-secret-scan.yml @@ -0,0 +1,61 @@ +# Fleet-added (Apollo/greploop posture-B floor). NOT upstream — do not expect on NousResearch/hermes-agent. +# Provides the `secret_scan` deterministic-floor component. DIFF-SCOPED on PRs: scans only the PR +# commit range, NOT full history (the fork inherits ~768 generic-api-key false positives in upstream +# test fixtures/docs — a full-tree scan would pin this RED forever). On push to main, scans the push range. +name: Fleet Secret Scan (gitleaks) + +on: + pull_request: + push: + branches: [main] + +permissions: + contents: read + +concurrency: + group: fleet-secret-scan-${{ github.ref }} + cancel-in-progress: true + +jobs: + secret_scan: + name: secret_scan + runs-on: ubuntu-latest + steps: + - name: Checkout (full history for range scan) + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Compute scan range + id: range + run: | + if [ "${{ github.event_name }}" = "pull_request" ]; then + BASE="${{ github.event.pull_request.base.sha }}" + HEAD="${{ github.event.pull_request.head.sha }}" + else + BASE="${{ github.event.before }}" + HEAD="${{ github.sha }}" + fi + # Guard against the all-zero "before" SHA on first push / new branch. + if ! git cat-file -e "${BASE}^{commit}" 2>/dev/null; then + BASE="$(git rev-parse "${HEAD}~1" 2>/dev/null || echo "${HEAD}")" + fi + echo "base=${BASE}" >> "$GITHUB_OUTPUT" + echo "head=${HEAD}" >> "$GITHUB_OUTPUT" + echo "scanning ${BASE}..${HEAD}" + + - name: Install gitleaks + run: | + VER=8.18.4 + curl -fsSL "https://github.com/gitleaks/gitleaks/releases/download/v${VER}/gitleaks_${VER}_linux_x64.tar.gz" -o gl.tgz + tar xzf gl.tgz gitleaks + sudo mv gitleaks /usr/local/bin/ + + - name: Gitleaks (diff-scoped) + run: | + gitleaks detect \ + --source . \ + --no-banner \ + --redact \ + --log-opts="${{ steps.range.outputs.base }}..${{ steps.range.outputs.head }}" \ + --exit-code 1 diff --git a/.gitleaks.toml b/.gitleaks.toml new file mode 100644 index 0000000000000..60ac6819439e0 --- /dev/null +++ b/.gitleaks.toml @@ -0,0 +1,24 @@ +# Gitleaks configuration for the hermes-agent fork. +# +# Extends the stock gitleaks ruleset and allowlists the obviously-fake credential +# *fixtures* the test suite uses (mocked clients / redaction tests, never real). +# They surfaced only because the diff-scoped scan re-reads them as "added" lines +# when a large test file is split/moved (PR #66). +# +# gitleaks `stopwords` allowlist a finding when its detected SECRET contains one +# of these substrings (case-insensitive). gitleaks strips the literal "..." from +# the secret it evaluates, so we match the distinctive ALPHANUMERIC token of each +# fake fixture. These tails are specific to the known fakes; a real high-entropy +# key does not contain them (verified with an sk-ant-api03 injection negative +# test that is still CAUGHT with this config). + +[extend] +useDefault = true + +[allowlist] +description = "Fake test-fixture API keys (mocked clients / redaction tests, never real)." +stopwords = [ + "1234567890", # test-key-1234567890 (AIAgent constructor stub) / test-k...7890 + "2mno", # gsk_ab...2mno / sk-ant...2mno (redaction fixtures) + "mnop", # sk-or-...mnop (key-masking fixture) +] diff --git a/agent/agent_init.py b/agent/agent_init.py index e1594b4585d48..9f287e51b9dcf 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -92,6 +92,58 @@ def _normalized_custom_base_url(value: Any) -> str: return value.strip().rstrip("/") +def _resolve_per_model_threshold( + per_model: Any, model: Any +) -> Optional[float]: + """Resolve a per-model compression threshold from the config map. + + ``per_model`` is the ``compression.per_model_threshold`` config value -- a + mapping of model-id -> threshold fraction. ``model`` is the active model id. + + Lookup is exact-match first, then case-insensitive. The threshold is the + fraction of the model's context window that must be consumed before + compression triggers; valid values are in (0, 1]. Out-of-range or + non-numeric values are ignored (return ``None``) so resolution falls + through to the built-in default and then the global config threshold. + + Returns the matched fraction as a float, or ``None`` when there is no + usable per-model entry for ``model``. + """ + if not isinstance(per_model, dict) or not per_model: + return None + if not isinstance(model, str) or not model: + return None + + raw = per_model.get(model) + if raw is None: + lowered = model.lower() + for key, value in per_model.items(): + if isinstance(key, str) and key.lower() == lowered: + raw = value + break + if raw is None: + return None + + try: + threshold = float(raw) + except (TypeError, ValueError): + logger.warning( + "Ignoring non-numeric compression.per_model_threshold for %r: %r", + model, + raw, + ) + return None + if not (0.0 < threshold <= 1.0): + logger.warning( + "Ignoring out-of-range compression.per_model_threshold for %r: %r " + "(must be in (0, 1])", + model, + threshold, + ) + return None + return threshold + + def _custom_provider_model_matches(agent_model: str, entry: Dict[str, Any]) -> bool: provider_model = str(entry.get("model", "") or "").strip().lower() if not provider_model: @@ -939,6 +991,9 @@ def init_agent( agent._fallback_chain = [] agent._fallback_index = 0 agent._fallback_activated = getattr(agent, "_fallback_activated", False) + # Tracks the last announced (old_model, new_model) fallback transition so a + # re-entrant fallback chain announces once per distinct transition (I5). + agent._last_fallback_announced = getattr(agent, "_last_fallback_announced", None) # Legacy attribute kept for backward compat (tests, external callers) agent._fallback_model = agent._fallback_chain[0] if agent._fallback_chain else None if agent._fallback_chain and not agent.quiet_mode: @@ -1059,6 +1114,12 @@ def init_agent( agent._codex_reasoning_replay_enabled = True agent._memory_write_origin = "assistant_tool" agent._memory_write_context = "foreground" + # Bridge sub-session routing suffix. None for normal/main agents; a forked + # agent that shares the parent's session_id (e.g. the background-review + # fork) sets this to a short tag (e.g. "review") so the claude-bridge + # provider emits a DISTINCT routing key and gets its own Claude CLI session + # instead of contaminating the parent's. See PRD bridge-subsession-routing. + agent._bridge_route_suffix = None # Cached system prompt -- built once per session, only rebuilt on compression agent._cached_system_prompt: Optional[str] = None @@ -1243,43 +1304,63 @@ def init_agent( if not isinstance(_compression_cfg, dict): _compression_cfg = {} compression_threshold = float(_compression_cfg.get("threshold", 0.50)) - # Per-model/route compaction-threshold override. Codex gpt-5.5 raises to - # 85% (the Codex backend caps the window at 272K, so the default 50% would - # compact at ~136K — half the usable context). Gated by an opt-out config - # flag so the user can fall back to the global threshold; when the override - # fires we stash a one-time notification (replayed on the first turn) that - # tells the user what changed and how to revert. + # Stable copy of the GLOBAL config threshold, captured before the per-model / + # built-in override below reassigns ``compression_threshold``. Threaded into + # the compressor so update_model's re-resolution has the correct global + # fallback tier (2026-06-19 compaction-thrash fix). + _global_compression_threshold = compression_threshold + # Per-model threshold override. Resolution order (first hit wins): + # 1. config ``compression.per_model_threshold`` map (user-tunable) + # 2. built-in per-model default / codex gpt-5.5 autoraise + # (``_compression_threshold_for_model``) + # 3. global ``compression.threshold`` (set above) + # 4. hardcoded 0.50 default (the ``.get`` fallback above) + # The threshold is a fraction of the model's context window in (0, 1]; + # the absolute trigger (context_length * threshold) is already model-aware + # and re-derived on every model switch in ContextCompressor.update_model. + # + # Codex gpt-5.5 autoraise (upstream): the Codex backend caps the window at + # 272K, so the default 50% would compact at ~136K — half the usable + # context. ``_compression_threshold_for_model`` raises it to 85% unless the + # ``compression.codex_gpt55_autoraise`` opt-out flag is set; when it fires + # we stash a one-time notification (replayed on the first turn). + _per_model = _compression_cfg.get("per_model_threshold") + _cfg_model_thresh = _resolve_per_model_threshold(_per_model, agent.model) _codex_gpt55_autoraise = str( _compression_cfg.get("codex_gpt55_autoraise", True) ).lower() in {"true", "1", "yes"} agent._compression_threshold_autoraised = None - try: - from agent.auxiliary_client import ( - _compression_threshold_for_model as _cthresh_fn, - _is_codex_gpt55 as _is_codex_gpt55_fn, - ) - _model_cthresh = _cthresh_fn( - agent.model, - agent.provider, - allow_codex_gpt55_autoraise=_codex_gpt55_autoraise, - ) - if _model_cthresh is not None: - _prev_threshold = compression_threshold - compression_threshold = _model_cthresh - # Notify only for the Codex gpt-5.5 autoraise (the Arcee Trinity - # override is a long-standing silent default). Skip the notice when - # the user's global threshold already meets/exceeds the raised - # value, since nothing actually changed for them. - if ( - _is_codex_gpt55_fn(agent.model, agent.provider) - and _model_cthresh > _prev_threshold + 1e-9 - ): - agent._compression_threshold_autoraised = { - "from": _prev_threshold, - "to": _model_cthresh, - } - except Exception: - pass + if _cfg_model_thresh is not None: + # Explicit user config for this model wins over every built-in default. + compression_threshold = _cfg_model_thresh + else: + try: + from agent.auxiliary_client import ( + _compression_threshold_for_model as _cthresh_fn, + _is_codex_gpt55 as _is_codex_gpt55_fn, + ) + _model_cthresh = _cthresh_fn( + agent.model, + agent.provider, + allow_codex_gpt55_autoraise=_codex_gpt55_autoraise, + ) + if _model_cthresh is not None: + _prev_threshold = compression_threshold + compression_threshold = _model_cthresh + # Notify only for the Codex gpt-5.5 autoraise (the Arcee Trinity + # override is a long-standing silent default). Skip the notice + # when the user's global threshold already meets/exceeds the + # raised value, since nothing actually changed for them. + if ( + _is_codex_gpt55_fn(agent.model, agent.provider) + and _model_cthresh > _prev_threshold + 1e-9 + ): + agent._compression_threshold_autoraised = { + "from": _prev_threshold, + "to": _model_cthresh, + } + except Exception: + pass compression_enabled = str(_compression_cfg.get("enabled", True)).lower() in {"true", "1", "yes"} compression_target_ratio = float(_compression_cfg.get("target_ratio", 0.20)) compression_protect_last = int(_compression_cfg.get("protect_last_n", 20)) @@ -1512,6 +1593,16 @@ def init_agent( provider=agent.provider, api_mode=agent.api_mode, abort_on_summary_failure=compression_abort_on_summary_failure, + # Thread the compression-threshold config so update_model can + # re-resolve the DESTINATION model's threshold on a mid-session + # fallback/switch (2026-06-19 compaction-thrash fix). The + # threshold_percent above is already the init-resolved value for + # agent.model; these add the raw inputs for re-resolution and do + # not change init behavior. _global_threshold is the global config + # threshold BEFORE any per-model/built-in override (captured above). + per_model_threshold=_per_model, + global_threshold_percent=_global_compression_threshold, + codex_gpt55_autoraise=_codex_gpt55_autoraise, ) agent.compression_enabled = compression_enabled @@ -1557,10 +1648,15 @@ def init_agent( if isinstance(t, dict) } for _schema in agent.context_compressor.get_tool_schemas(): - _tname = _schema.get("name", "") + if _schema.get("type") == "function" and isinstance(_schema.get("function"), dict): + _function_schema = _schema["function"] + _wrapped = _schema + else: + _function_schema = _schema + _wrapped = {"type": "function", "function": _schema} + _tname = _function_schema.get("name", "") if _tname and _tname in _existing_tool_names: continue # already registered via plugin/cache path - _wrapped = {"type": "function", "function": _schema} agent.tools.append(_wrapped) if _tname: agent.valid_tool_names.add(_tname) @@ -1596,6 +1692,9 @@ def init_agent( agent.session_cache_read_tokens = 0 agent.session_cache_write_tokens = 0 agent.session_reasoning_tokens = 0 + # Per-call snapshot for the most recent successful provider response. + # Cumulative session_* counters are still the source of truth for totals. + agent.last_turn_usage = None agent.session_estimated_cost_usd = 0.0 agent.session_cost_status = "unknown" agent.session_cost_source = "none" diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 884866dc1173a..9385c574c872c 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -845,6 +845,10 @@ def try_recover_primary_transport( if hasattr(agent, "_transport_cache"): agent._transport_cache.clear() agent.api_key = rt["api_key"] + # Restore the primary's reasoning effort if a fallback entry + # overrode it (#21256). Guarded for snapshots predating the key. + if "reasoning_config" in rt: + agent.reasoning_config = rt["reasoning_config"] if agent.api_mode == "anthropic_messages": from agent.anthropic_adapter import build_anthropic_client @@ -1010,6 +1014,12 @@ def restore_primary_runtime(agent) -> bool: agent.api_key = rt["api_key"] agent._client_kwargs = dict(rt["client_kwargs"]) agent._use_prompt_caching = rt["use_prompt_caching"] + # Restore the primary's reasoning effort (a fallback entry may have + # overridden it via reasoning_effort). Older snapshots predate this + # key — only restore when it was captured, to avoid clobbering the + # live value with None. See #21256. + if "reasoning_config" in rt: + agent.reasoning_config = rt["reasoning_config"] # Default to native layout when the restored snapshot predates the # native-vs-proxy split (older sessions saved before this PR). agent._use_native_cache_layout = rt.get( @@ -1049,6 +1059,27 @@ def restore_primary_runtime(agent) -> bool: # ── Reset fallback chain for the new turn ── agent._fallback_activated = False agent._fallback_index = 0 + # Clear the fallback-announce dedupe so a LATER fallback episode (after + # this primary recovery) re-announces instead of being suppressed as a + # repeat of the prior transition (I5: once per episode, not once ever). + agent._last_fallback_announced = None + + # Re-sync auxiliary-routing globals back to the restored PRIMARY so a + # post-recovery aux task (compression, etc.) routes to the primary + # model again — the mirror of the sync done in try_activate_fallback. + # ``turn_context`` also sets these at turn start, but restoration can + # run mid-turn, so keep them in lockstep here too. + try: + from agent.auxiliary_client import set_runtime_main + _pri_aux_key = agent.api_key if isinstance(getattr(agent, "api_key", ""), str) else "" + set_runtime_main( + agent.provider, agent.model, + base_url=agent.base_url, + api_key=_pri_aux_key, + api_mode=agent.api_mode, + ) + except Exception: # noqa: BLE001 — never let aux-routing sync break restore + pass logger.info( "Primary runtime restored for new turn: %s (%s)", @@ -1673,6 +1704,11 @@ def switch_model(agent, new_model, new_provider, api_key='', base_url='', api_mo "compressor_context_length": _cc.context_length if _cc else 0, "compressor_api_mode": getattr(_cc, "api_mode", agent.api_mode) if _cc else agent.api_mode, "compressor_threshold_tokens": _cc.threshold_tokens if _cc else 0, + # Snapshot the primary's reasoning effort so a fallback entry that + # overrides it (fallback_model[N].reasoning_effort) can be reverted + # when the primary is restored. See per-entry reasoning support in + # try_activate_fallback (#21256). + "reasoning_config": getattr(agent, "reasoning_config", None), } if api_mode == "anthropic_messages": agent._primary_runtime.update({ diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 01ea45d7be24d..685722a237053 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -281,6 +281,45 @@ def _compression_threshold_for_model( return _CODEX_GPT55_COMPACTION_THRESHOLD return None + +def resolve_compression_threshold( + per_model: Any, + model: Optional[str], + provider: Optional[str] = None, + *, + global_threshold: float = 0.50, + allow_codex_gpt55_autoraise: bool = True, +) -> float: + """Resolve the compression threshold fraction for a model, applying the + canonical precedence chain (first hit wins): + + 1. config ``compression.per_model_threshold`` map (user-tunable) + 2. built-in per-model / route default (``_compression_threshold_for_model``) + 3. global ``compression.threshold`` (``global_threshold``) + + This is the single source of truth shared by ``agent_init`` (session start) + and ``ContextCompressor.update_model`` (model switch / fallback) so the + destination model's configured threshold survives a mid-session route swap. + Returns a fraction in (0, 1]. + + ``_resolve_per_model_threshold`` (the config-map lookup) lives in + ``agent.agent_init``; it is imported lazily here to avoid an import cycle + (``agent_init`` imports this module). + """ + from agent.agent_init import _resolve_per_model_threshold + + cfg_thresh = _resolve_per_model_threshold(per_model, model) + if cfg_thresh is not None: + return cfg_thresh + + builtin = _compression_threshold_for_model( + model, provider, allow_codex_gpt55_autoraise=allow_codex_gpt55_autoraise, + ) + if builtin is not None: + return builtin + + return global_threshold + # Default auxiliary models for direct API-key providers (cheap/fast for side tasks) def _get_aux_model_for_provider(provider_id: str) -> str: """Return the cheap auxiliary model for a provider. @@ -3186,8 +3225,24 @@ def _resolve_auto(main_runtime: Optional[Dict[str, Any]] = None) -> Tuple[Option # on aggregators (OpenRouter, Nous) who previously got routed to a # cheap provider-side default. Explicit per-task overrides set via # config.yaml (auxiliary..provider) still win over this. - main_provider = str(runtime_provider or _read_main_provider() or "") - main_model = str(runtime_model or _read_main_model() or "") + # + # provider + model are a MATCHED PAIR and must be resolved from the + # SAME source. Resolving them with two independent ``or`` fallbacks + # (``runtime_provider or global``, ``runtime_model or global``) could + # cross a provider from one source with a model from another — e.g. + # after a mid-session route swap the caller's ``main_runtime`` carries + # the NEW provider (openai-codex) but the global ``_RUNTIME_MAIN_MODEL`` + # still holds the OLD model (claude-opus-4-8), yielding a Codex route + + # an Opus model id and a hard ``400: model ... not supported when using + # Codex``. Take the pair atomically: if the caller's runtime dict names + # a provider, its model wins (even when blank); only fall through to the + # process-global pair when the runtime dict has no provider at all. + if runtime_provider: + main_provider = str(runtime_provider) + main_model = str(runtime_model or "") + else: + main_provider = str(_read_main_provider() or "") + main_model = str(_read_main_model() or "") if (main_provider and main_model and main_provider not in {"auto", ""}): resolved_provider = main_provider @@ -3404,13 +3459,14 @@ def resolve_provider_client( # with grok-4.3 configured gets grok-4.3 for title generation # instead of silently dropping to whatever Step-2 fallback (#31845). # - # Each provider branch below sees a non-empty ``model`` whenever the - # user has *anything* configured — no provider-specific empty-model - # guards needed. When the user has NOTHING configured (fresh install, - # main_model also empty), the branches still hit their own - # missing-credentials returns and ``_resolve_auto`` falls through to - # the Step-2 chain as before. - if not model: + # Each explicit provider branch below sees a non-empty ``model`` whenever + # the user has *anything* configured — no provider-specific empty-model + # guards needed. The ``auto`` branch is the exception: it must let + # ``_resolve_auto(main_runtime=...)`` choose the matched provider/model pair + # from the live runtime. Falling back to ``_read_main_model()`` here can + # cross a stale config/default model (e.g. Opus) onto a newly failed-over + # provider client (e.g. Codex), producing provider+model mismatches. + if provider != "auto" and not model: model = _get_aux_model_for_provider(provider) or _read_main_model() or model def _needs_codex_wrap(client_obj, base_url_str: str, model_str: str) -> bool: @@ -4651,7 +4707,18 @@ def _get_cached_client( _client_cache[cache_key] = (client, default_model, bound_loop) else: client, default_model, _ = _client_cache[cache_key] - return client, model or default_model + # Re-apply model name normalization on the raw caller-passed `model` so + # the caller's explicit override still wins (preserving existing + # caller-wins semantics) but the namespace-stripping that + # `resolve_provider_client` performed via `_normalize_resolved_model` + # isn't silently bypassed. Without this, namespaced model names like + # "my-provider/claude-haiku-4-5" leak past the resolver's normalization + # and arrive at the wire unstripped — the Anthropic API rejects them + # with a 404 "model not found". `default_model` is the resolver's + # already-normalized `final_model`, so any normalization failure falls + # back to it safely. + normalized_caller_model = _normalize_resolved_model(model, provider) if model else None + return client, normalized_caller_model or default_model # Aliases that target direct REST APIs not modeled as first-class providers @@ -4704,9 +4771,39 @@ def _resolve_task_provider_model( cfg_api_key = str(task_config.get("api_key", "")).strip() or None cfg_api_mode = str(task_config.get("api_mode", "")).strip() or None + # 'auto' is a sentinel meaning "inherit from main runtime / auto-detect", not + # a literal model id. Without this, a config of `auxiliary..model: auto` + # propagates the literal string "auto" to the wire, where the provider returns + # a 200 OK with an error-text body (e.g. "the model 'auto' does not exist"), + # which downstream consumers like ContextCompressor accept as the task output. + # The provider-side 'auto' is handled in _resolve_auto() via main_runtime + # fallback, so dropping cfg_model to None here lets that path do its job. + if cfg_model and cfg_model.lower() == "auto": + cfg_model = None + resolved_model = model or cfg_model resolved_api_mode = cfg_api_mode + # When an explicit provider is given and no task-config api_mode override, + # fall back to the provider profile's declared api_mode. Without this, + # plugin providers that declare api_mode="anthropic_messages" (i.e. their + # upstream speaks the Anthropic Messages API, not OpenAI Chat Completions) + # silently land on the OpenAI-wire transport and 404 from + # `chat/completions` calls. The URL-suffix detection in + # `_endpoint_speaks_anthropic_messages` only catches /anthropic-suffixed + # gateways; it misses plain-base-URL Anthropic-API endpoints (e.g. local + # proxies on 127.0.0.1:PORT). Reading the profile here closes the gap so + # any provider plugin can declare api_mode and have it honored regardless + # of upstream URL shape. + if provider and not resolved_api_mode: + try: + from providers import get_provider_profile as _gpf_resolve + _profile = _gpf_resolve(provider) + if _profile and getattr(_profile, "api_mode", None): + resolved_api_mode = _profile.api_mode + except Exception: + pass + # Convenience aliases for direct API-key endpoints that aren't first-class # providers (e.g. ``provider: openai`` → custom + api.openai.com/v1). # Applied to both explicit args and config-derived values. When the user diff --git a/agent/background_review.py b/agent/background_review.py index d9f6ea5950de2..b8bc563031cd5 100644 --- a/agent/background_review.py +++ b/agent/background_review.py @@ -404,6 +404,9 @@ def _bg_review_auto_deny(command, description, **kwargs): max_iterations=16, quiet_mode=True, platform=agent.platform, + chat_id=getattr(agent, "_chat_id", None) or "", + chat_name=getattr(agent, "_chat_name", None) or "", + chat_type=getattr(agent, "_chat_type", None) or "", provider=agent.provider, api_mode=_parent_api_mode, base_url=_parent_runtime.get("base_url") or None, @@ -416,6 +419,16 @@ def _bg_review_auto_deny(command, description, **kwargs): ) review_agent._memory_write_origin = "background_review" review_agent._memory_write_context = "background_review" + # Bridge sub-session isolation: mark this fork so the claude-bridge + # provider routing key gets a distinct "-review" suffix. Without + # this, the fork shares the parent's session_id (pinned below for + # prefix-cache parity) → identical bridge routing key → the fork's + # restricted-toolset turn contaminates the main agent's resumed + # Claude CLI session (the "I only have memory/skill tools" bug). + # Object attribute (not a contextvar) so it rides the kwargs-build + # chain by reference regardless of thread. See PRD + # bridge-subsession-routing D-1/D-2. + review_agent._bridge_route_suffix = "review" review_agent._memory_store = agent._memory_store review_agent._memory_enabled = agent._memory_enabled review_agent._user_profile_enabled = agent._user_profile_enabled diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 1ee1702b45e82..a4f910cde59e8 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -28,9 +28,11 @@ from hermes_cli.timeouts import get_provider_request_timeout, get_provider_stale_timeout from hermes_constants import PARTIAL_STREAM_STUB_ID, FINISH_REASON_LENGTH from agent.error_classifier import FailoverReason -from agent.model_metadata import is_local_endpoint +from agent.model_metadata import is_local_endpoint, _ceil_chars_to_tokens from agent.message_sanitization import ( _sanitize_surrogates, + _SurrogateSplicer, + _splice_surrogates, _repair_tool_call_arguments, ) from tools.terminal_tool import is_persistent_env @@ -39,6 +41,44 @@ logger = logging.getLogger(__name__) +def _repair_anthropic_message_surrogates(message: Any) -> Any: + """Repair surrogate-pair artifacts in the native Anthropic final message. + + The live stream callback path splices deltas before display, but the + Anthropic SDK returns its own accumulated ``Message`` object. If an SDK or + compatible proxy preserves adjacent surrogate halves in that final object, + downstream validation/history can still inherit invalid UTF-8. Repair known + text-bearing fields in place while preserving the SDK object shape. + """ + content = getattr(message, "content", None) + if not isinstance(content, list): + return message + + for block in content: + for attr in ("text", "thinking"): + if isinstance(block, dict): + value = block.get(attr) + if isinstance(value, str): + block[attr] = _splice_surrogates(value) + continue + value = getattr(block, attr, None) + if isinstance(value, str): + fixed = _splice_surrogates(value) + if fixed != value: + try: + setattr(block, attr, fixed) + except Exception: + try: + object.__setattr__(block, attr, fixed) + except Exception: + logger.debug( + "Unable to repair Anthropic stream surrogate field %s", + attr, + exc_info=True, + ) + return message + + def _ra(): """Lazy ``run_agent`` reference. @@ -54,7 +94,7 @@ def estimate_request_context_tokens(api_payload: Any) -> int: """Estimate context/load tokens from an API payload, dict or messages list. The stale-call detectors historically assumed a Chat Completions request: - they pulled ``api_kwargs["messages"]`` and ran a cheap char/4 estimate. + they pulled ``api_kwargs["messages"]`` and ran a cheap char-based estimate. Codex / Responses API requests carry the conversational payload in ``input`` (with additional load in ``instructions`` and ``tools``), so the legacy estimator reported ~0 tokens for every Codex turn and the @@ -80,7 +120,7 @@ def _message_chars(messages: Any) -> int: return sum(_chars(item) for item in messages) if isinstance(api_payload, list): - return _message_chars(api_payload) // 4 + return _ceil_chars_to_tokens(_message_chars(api_payload)) if isinstance(api_payload, dict): messages = api_payload.get("messages") @@ -88,7 +128,7 @@ def _message_chars(messages: Any) -> int: total_chars = _message_chars(messages) if "tools" in api_payload: total_chars += _chars(api_payload.get("tools")) - return total_chars // 4 + return _ceil_chars_to_tokens(total_chars) if "input" in api_payload: total_chars = ( @@ -96,11 +136,13 @@ def _message_chars(messages: Any) -> int: + _chars(api_payload.get("instructions")) + _chars(api_payload.get("tools")) ) - return total_chars // 4 + return _ceil_chars_to_tokens(total_chars) - return sum(_chars(value) for value in api_payload.values()) // 4 + return _ceil_chars_to_tokens( + sum(_chars(value) for value in api_payload.values()) + ) - return _chars(api_payload) // 4 + return _ceil_chars_to_tokens(_chars(api_payload)) def _is_openai_codex_backend(agent) -> bool: @@ -338,6 +380,49 @@ def _call(): ) _ttfb_timeout = _ttfb_cap + # Ensure the fast-reconnect no-byte TTFB watchdog can actually fire well + # before the blunt wall-clock stale timer. When the TTFB cutoff (default + # 120s) is >= the stale timeout (default 90s), the stale detector always + # wins first: the connection is killed at the stale threshold and the much + # cheaper ~2s reconnect never gets a chance. A wedged + # chatgpt.com/backend-api/codex socket (observed: ReadError/Broken pipe with + # zero stream events) then burns the full stale timeout on every retry — + # e.g. 90s x 3 ~= 4.5min — before the fallback model kicks in. + # + # The no-byte watchdog is already disabled for large requests (>= 25k + # tokens) so we never kill a legitimate long prefill. For everything below + # that we can afford an aggressive fast-reconnect cutoff: small-context + # admission/prefill on the subscription Codex backend clears in a few + # seconds, so a ~40s cutoff reconnects 2-3x within the stale budget instead + # of waiting it out once. We also clamp to stay strictly below the stale + # timeout. Operators can disable via HERMES_CODEX_TTFB_BELOW_STALE=0 or tune + # the target via HERMES_CODEX_TTFB_FAST_RECONNECT_SECONDS. + if ( + _ttfb_enabled + and _codex_watchdog_enabled + and _env_float("HERMES_CODEX_TTFB_BELOW_STALE", 1.0) > 0 + ): + _ttfb_fast = _env_float("HERMES_CODEX_TTFB_FAST_RECONNECT_SECONDS", 40.0) + # Clamp below the stale timer (if finite) so the fast-reconnect cutoff + # is guaranteed to fire first; keep a small margin. + if _stale_timeout != float("inf"): + _ttfb_margin = _env_float( + "HERMES_CODEX_TTFB_BELOW_STALE_MARGIN_SECONDS", 10.0 + ) + _ttfb_fast = min(_ttfb_fast, max(_stale_timeout - _ttfb_margin, 5.0)) + # Only ever lower the cutoff — never raise an operator-set tighter value. + if _ttfb_fast > 0 and _ttfb_timeout > _ttfb_fast: + logger.info( + "Lowering codex no-byte TTFB cutoff from %.0fs to %.0fs so it " + "fires before the %.0fs stale timer (fast reconnect instead of " + "waiting out the stale timeout). Set HERMES_CODEX_TTFB_BELOW_STALE=0 " + "to disable or HERMES_CODEX_TTFB_FAST_RECONNECT_SECONDS to tune.", + _ttfb_timeout, + _ttfb_fast, + _stale_timeout, + ) + _ttfb_timeout = _ttfb_fast + _codex_idle_enabled = _codex_watchdog_enabled _codex_idle_timeout = _env_float( "HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", @@ -346,6 +431,39 @@ def _call(): if _codex_idle_timeout <= 0: _codex_idle_enabled = False + # Progress-stall watchdog. The stream-idle detector above is satisfied by + # ANY SSE frame — including content-free keepalive / ``response.in_progress`` + # frames. The chatgpt.com/backend-api/codex backend has a failure mode where + # it keeps the socket "alive" with periodic keepalives but never emits a real + # delta or completes, so _last_event_ts stays fresh forever and the idle + # watchdog never trips: the call burns the full blunt stale timeout (observed: + # 90s × 6 retries ≈ 9 min before fallback). ``_codex_stream_last_progress_ts`` + # (set by run_codex_stream only on real deltas / function calls / output items + # / terminal events) lets us catch "events flowing but zero forward progress" + # and reconnect at the fast cutoff instead. Reuses the fast-reconnect target + # (default 40s, clamped below the stale timer) so it fires before the blunt + # kill. Set HERMES_CODEX_PROGRESS_STALE_TIMEOUT_SECONDS=0 to disable, or tune + # it directly; otherwise it defaults to the same fast cutoff as the no-byte + # watchdog. + _codex_progress_enabled = _codex_watchdog_enabled + _codex_progress_default = _ttfb_timeout if _ttfb_enabled else _codex_idle_timeout + _codex_progress_timeout = _env_float( + "HERMES_CODEX_PROGRESS_STALE_TIMEOUT_SECONDS", + _codex_progress_default, + ) + if _codex_progress_timeout <= 0: + _codex_progress_enabled = False + elif _stale_timeout != float("inf"): + # Never let the progress-stall cutoff meet or exceed the blunt stale + # timer (it would then never win); keep the same margin as the no-byte + # fast-reconnect clamp. + _progress_margin = _env_float( + "HERMES_CODEX_TTFB_BELOW_STALE_MARGIN_SECONDS", 10.0 + ) + _codex_progress_timeout = min( + _codex_progress_timeout, max(_stale_timeout - _progress_margin, 5.0) + ) + if _codex_watchdog_enabled: # Reset before the worker starts so a marker left over from a previous # call on this agent can't be misread as first-byte for this one. @@ -469,6 +587,54 @@ def _call(): ) break + # Progress-stall detector: the Codex stream is emitting frames (keepalive + # / in_progress refresh _codex_stream_last_event_ts, so the idle detector + # above stays quiet) but has produced ZERO real forward progress — no text + # / reasoning delta, no function call, no output item, no terminal event — + # for longer than the fast-reconnect cutoff. This is the keepalive-only + # hang on chatgpt.com/backend-api/codex: without this branch the call + # rides the socket all the way to the blunt wall-clock stale timeout + # (observed: 90s × 6 retries ≈ 9 min) instead of reconnecting in ~1s. + # We only arm once at least one event has arrived (_last_event_ts set) so + # this never races the no-byte TTFB detector for the first-byte case. + _last_progress_ts = getattr(agent, "_codex_stream_last_progress_ts", None) + _last_event_ts_for_progress = getattr(agent, "_codex_stream_last_event_ts", None) + if ( + _codex_progress_enabled + and _last_event_ts_for_progress is not None + and _last_progress_ts is None + and _elapsed > _codex_progress_timeout + ): + logger.warning( + "Codex stream emitted events but no forward progress for %.0fs " + "(threshold %.0fs, model=%s, context=~%s tokens). Backend is " + "sending keepalives without producing output. Killing connection " + "so the retry loop can reconnect.", + _elapsed, + _codex_progress_timeout, + api_kwargs.get("model", "unknown"), + f"{_est_tokens_for_codex_watchdog:,}", + ) + agent._buffer_status( + f"⚠️ Codex stream stalled with no progress for {int(_elapsed)}s " + f"(keepalives only, model: {api_kwargs.get('model', 'unknown')}). " + f"Reconnecting." + ) + try: + _close_request_client_once("codex_progress_stall_kill") + except Exception: + pass + agent._touch_activity( + f"codex stream killed after {int(_elapsed)}s with no forward progress" + ) + t.join(timeout=2.0) + if result["error"] is None and result["response"] is None: + result["error"] = TimeoutError( + f"Codex stream emitted events but made no forward progress for " + f"{int(_elapsed)}s (threshold: {int(_codex_progress_timeout)}s)" + ) + break + # Stale-call detector: kill the connection if no response # arrives within the configured timeout. if _elapsed > _stale_timeout: @@ -753,6 +919,12 @@ def build_api_kwargs(agent, api_messages: list) -> dict: reasoning_config=agent.reasoning_config, request_overrides=agent.request_overrides, session_id=getattr(agent, "session_id", None), + # Bridge sub-session isolation: forks that share the parent's + # session_id (background-review) set _bridge_route_suffix so the + # claude-bridge routing key is distinct. None for normal agents. + # Only the chat_completions (profile) path reaches the bridge; the + # codex path is excluded by design. See PRD bridge-subsession-routing. + bridge_route_suffix=getattr(agent, "_bridge_route_suffix", None), provider_profile=_profile, ollama_num_ctx=agent._ollama_num_ctx, # Context forwarded to profile hooks: @@ -1042,7 +1214,116 @@ def build_assistant_message(agent, assistant_message, finish_reason: str) -> dic -def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool: +def _primary_cooldown_seconds(error_context: Optional[Dict[str, Any]] = None) -> float: + """Return provider cooldown seconds from structured error context. + + ``extract_api_error_context`` normalizes retry-after / reset timestamps + into ``reset_at``. Use that provider-supplied window when available so + repeated turns do not restore an exhausted primary after the old fixed + 60-second grace. Clamp absurdly-low/absurdly-high values to keep the + fallback mechanism useful and bounded. + """ + default = 60.0 + if not isinstance(error_context, dict): + return default + reset_at = error_context.get("reset_at") + if reset_at is None or reset_at == "": + return default + try: + reset_at_float = float(reset_at) + except (TypeError, ValueError): + return default + # ``reset_at`` is stored as epoch seconds by extract_api_error_context; + # a few provider bodies may carry small retry-after-like numbers instead. + seconds = reset_at_float - time.time() if reset_at_float > 1_000_000_000 else reset_at_float + if seconds <= 0: + return default + return max(default, min(seconds, 6 * 60 * 60)) + + +def _format_context_window(tokens: "int | None") -> str: + """Compact human label for a context window: 1M, 272K, 128K, etc.""" + if not tokens or tokens <= 0: + return "" + if tokens >= 1_000_000: + whole = tokens / 1_000_000 + return f"{whole:.0f}M" if whole == int(whole) else f"{whole:.1f}M" + if tokens >= 1_000: + return f"{tokens // 1000}K" + return str(tokens) + + +def _emit_fallback_announce( + agent, + old_model: str, + new_model: str, + new_provider: str, + *, + old_provider: "str | None" = None, + old_window: "int | None" = None, + new_window: "int | None" = None, +) -> None: + """Emit a single, always-visible chat status line when a model fallback + activates successfully. + + Unlike ``_buffer_status`` (suppressed on successful recovery, flushed only on + a terminal turn failure), this routes through ``_emit_status`` so it reaches + the gateway ``status_callback`` (Discord/Telegram) AND the CLI every time — + closing the 2026-06-19 gap where an opus->gpt-5.5 fallback that *succeeded* + was invisible to the user. + + Both sides are rendered ``provider/model`` (e.g. + ``claude-app/claude-opus-4-8 → openai-codex/gpt-5.5``) so the *route* is + unambiguous — the same model slug can be served by different providers, and + the provider is the thing that explains a window/behavior change. When the + old provider is unknown the source side degrades to the bare model slug. + + De-duplicated on the ``(old_model, new_model)`` pair so a re-entrant fallback + chain that bounces to the same destination within a turn announces once + (Invariant I5). A no-op transition (``old_model == new_model``) is silent. + """ + if not new_model or old_model == new_model: + return + transition = (old_model, new_model) + if getattr(agent, "_last_fallback_announced", None) == transition: + return + agent._last_fallback_announced = transition + + # Record a structured fallback event so a compaction shortly afterward, in + # the SAME logical turn, can note it was "after model fallback" (turn-scoped + # causality, not wall-clock proximity). See conversation_compression. + try: + agent._last_fallback_event = { + "old_model": old_model, + "new_model": new_model, + "old_provider": old_provider, + "new_provider": new_provider, + "old_window": old_window, + "new_window": new_window, + "turn_id": getattr(agent, "_current_turn_id", None), + "monotonic_time": time.monotonic(), + } + except Exception: + pass + + old_label = f"{old_provider}/{old_model}" if old_provider else old_model + new_label = f"{new_provider}/{new_model}" if new_provider else new_model + msg = f"🔄 Model fallback: {old_label} → {new_label}" + old_lbl = _format_context_window(old_window) + new_lbl = _format_context_window(new_window) + if old_lbl and new_lbl and old_lbl != new_lbl: + msg += f" · context window {old_lbl}→{new_lbl}" + emit = getattr(agent, "_emit_status", None) + if callable(emit): + emit(msg) + + +def try_activate_fallback( + agent, + reason: "FailoverReason | None" = None, + *, + error_context: Optional[Dict[str, Any]] = None, +) -> bool: """Switch to the next fallback model/provider in the chain. Called when the current model is failing after retries. Swaps the @@ -1062,7 +1343,7 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool current_provider = (getattr(agent, "provider", "") or "").strip().lower() primary_provider = ((agent._primary_runtime or {}).get("provider") or "").strip().lower() if (not fallback_already_active) or (primary_provider and current_provider == primary_provider): - agent._rate_limited_until = time.monotonic() + 60 + agent._rate_limited_until = time.monotonic() + _primary_cooldown_seconds(error_context) if agent._fallback_index >= len(agent._fallback_chain): return False @@ -1168,6 +1449,7 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool fb_api_mode = "bedrock_converse" old_model = agent.model + old_provider = agent.provider # Clear the per-config context_length override so the fallback # model's actual context window is resolved instead of inheriting @@ -1181,6 +1463,67 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool agent._transport_cache.clear() agent._fallback_activated = True + # Per-entry reasoning effort override (#21256). A fallback_model + # entry may carry ``reasoning_effort`` (none/minimal/low/medium/high/ + # xhigh) to run that tier at a different thinking depth than the + # global ``agent.reasoning_effort`` — e.g. a last-resort Codex tier at + # ``xhigh``. The primary's reasoning_config is snapshotted in + # ``_primary_runtime`` and restored on primary recovery, so this only + # affects the fallback turn(s). Absent/blank → keep the current + # (primary/global) reasoning effort unchanged. + _fb_reasoning_effort = str(fb.get("reasoning_effort") or "").strip() + if _fb_reasoning_effort: + try: + from hermes_constants import parse_reasoning_effort + _fb_reasoning_config = parse_reasoning_effort(_fb_reasoning_effort) + # parse_reasoning_effort returns None for an unrecognized + # level; only override when it parsed to a real config so a + # typo doesn't silently disable reasoning. + if _fb_reasoning_config is not None: + agent.reasoning_config = _fb_reasoning_config + logger.info( + "Fallback %s/%s: reasoning effort override → %s", + fb_provider, fb_model, _fb_reasoning_effort, + ) + else: + logger.warning( + "Fallback %s/%s: unknown reasoning_effort '%s' — " + "keeping current effort", + fb_provider, fb_model, _fb_reasoning_effort, + ) + except Exception as _re_exc: # noqa: BLE001 + logger.warning( + "Fallback %s/%s: failed to apply reasoning_effort '%s': %s", + fb_provider, fb_model, _fb_reasoning_effort, _re_exc, + ) + + # Re-sync the auxiliary-routing runtime globals to the fallback tuple. + # ``turn_context`` set these to the PRIMARY (provider+model) at turn + # start; without updating them here, auxiliary tasks (context + # compression, web extract, etc.) whose explicit model is dropped in + # ``_resolve_auto`` fall back to ``_read_main_model()`` -> the stale + # PRIMARY model name, while the provider resolves to the fallback. + # That mismatch sent a Claude model name (``claude-opus-4-8``) to the + # Codex/ChatGPT endpoint -> ``400 ... not supported when using Codex + # with a ChatGPT account``. Keeping the globals in lockstep with the + # live runtime is exactly what they are documented to track. + try: + from agent.auxiliary_client import set_runtime_main + # NOTE: the fallback client's own api_key is not swapped onto the + # agent until further below, so pass an empty key here — aux + # resolution then uses the fallback PROVIDER's own credentials + # (e.g. the Codex OAuth token) instead of pinning the stale + # primary key. provider+model+base_url+api_mode are the load- + # bearing fields for correct routing and are all final here. + set_runtime_main( + fb_provider, fb_model, + base_url=fb_base_url, + api_key="", + api_mode=fb_api_mode, + ) + except Exception: # noqa: BLE001 — never let aux-routing sync break failover + pass + # Clear the credential pool when the fallback provider doesn't match # the pool's provider. The pool was seeded for the primary provider; # leaving it attached means downstream recovery (rate_limit / billing / @@ -1278,6 +1621,9 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool config_context_length=getattr(agent, "_config_context_length", None), custom_providers=getattr(agent, "_custom_providers", None), ) + _old_ctx_window = getattr( + getattr(agent, "context_compressor", None), "context_length", None + ) agent.context_compressor.update_model( model=agent.model, context_length=fb_context_length, @@ -1295,6 +1641,25 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool "Fallback activated: %s → %s (%s)", old_model, fb_model, fb_provider, ) + # Always-emitted, deduped fallback ANNOUNCE (separate from the + # suppress-on-recovery _buffer_status above, which only flushes on a + # TERMINAL turn failure — so a fallback that SUCCEEDS was invisible, + # the 2026-06-19 opus->gpt-5.5 case). This reaches the gateway + # status_callback (Discord/Telegram), not just the CLI, via _emit_status. + # Deduped on the (old->new) model pair so a re-entrant fallback chain + # that bounces to the same destination in one turn announces once (I5). + try: + _new_ctx_window = getattr( + getattr(agent, "context_compressor", None), "context_length", None + ) + _emit_fallback_announce( + agent, old_model, fb_model, fb_provider, + old_provider=old_provider, + old_window=locals().get("_old_ctx_window"), + new_window=_new_ctx_window, + ) + except Exception: + logger.debug("fallback announce failed", exc_info=True) return True except Exception as e: logger.error("Failed to activate fallback %s: %s", fb_model, e) @@ -1844,6 +2209,8 @@ def _call_chat_completions(): agent._check_openrouter_cache_status(getattr(stream, "response", None)) content_parts: list = [] + content_splicer = _SurrogateSplicer() + reasoning_splicer = _SurrogateSplicer() tool_calls_acc: dict = {} tool_gen_notified: set = set() # Ollama-compatible endpoints reuse index 0 for every tool call @@ -1896,6 +2263,8 @@ def _call_chat_completions(): # Accumulate reasoning content reasoning_text = getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None) + if reasoning_text: + reasoning_text = reasoning_splicer.feed(reasoning_text) if reasoning_text: reasoning_parts.append(reasoning_text) _fire_first_delta() @@ -1903,28 +2272,30 @@ def _call_chat_completions(): # Accumulate text content — fire callback only when no tool calls if delta and delta.content: - content_parts.append(delta.content) - if not tool_calls_acc: - _fire_first_delta() - agent._fire_stream_delta(delta.content) - deltas_were_sent["yes"] = True - # Tool calls suppress regular content streaming (avoids - # displaying chatty "I'll use the tool..." text alongside - # tool calls). But reasoning tags embedded in suppressed - # content should still reach the display — otherwise the - # reasoning box only appears as a post-response fallback, - # rendering it confusingly after the already-streamed - # response. Route suppressed content through the stream - # delta callback so its tag extraction can fire the - # reasoning display. Non-reasoning text is harmlessly - # suppressed by the CLI's _stream_delta when the stream - # box is already closed (tool boundary flush). - elif agent.stream_delta_callback: - try: - agent.stream_delta_callback(delta.content) - agent._record_streamed_assistant_text(delta.content) - except Exception: - pass + content_delta = content_splicer.feed(delta.content) + if content_delta: + content_parts.append(content_delta) + if not tool_calls_acc: + _fire_first_delta() + agent._fire_stream_delta(content_delta) + deltas_were_sent["yes"] = True + # Tool calls suppress regular content streaming (avoids + # displaying chatty "I'll use the tool..." text alongside + # tool calls). But reasoning tags embedded in suppressed + # content should still reach the display — otherwise the + # reasoning box only appears as a post-response fallback, + # rendering it confusingly after the already-streamed + # response. Route suppressed content through the stream + # delta callback so its tag extraction can fire the + # reasoning display. Non-reasoning text is harmlessly + # suppressed by the CLI's _stream_delta when the stream + # box is already closed (tool boundary flush). + elif agent.stream_delta_callback: + try: + agent.stream_delta_callback(content_delta) + agent._record_streamed_assistant_text(content_delta) + except Exception: + pass # Accumulate tool call deltas — notify display on first name if delta and delta.tool_calls: @@ -1999,6 +2370,26 @@ def _call_chat_completions(): if hasattr(chunk, "usage") and chunk.usage: usage_obj = chunk.usage + reasoning_tail = reasoning_splicer.flush() + if reasoning_tail: + reasoning_parts.append(reasoning_tail) + _fire_first_delta() + agent._fire_reasoning_delta(reasoning_tail) + + content_tail = content_splicer.flush() + if content_tail: + content_parts.append(content_tail) + if not tool_calls_acc: + _fire_first_delta() + agent._fire_stream_delta(content_tail) + deltas_were_sent["yes"] = True + elif agent.stream_delta_callback: + try: + agent.stream_delta_callback(content_tail) + agent._record_streamed_assistant_text(content_tail) + except Exception: + pass + # Build mock response matching non-streaming shape full_content = "".join(content_parts) or None mock_tool_calls = None @@ -2135,6 +2526,8 @@ def _call_anthropic(): works unchanged. """ has_tool_use = False + text_splicer = _SurrogateSplicer() + thinking_splicer = _SurrogateSplicer() # Reset stale-stream timer for this attempt last_chunk_time["t"] = time.time() @@ -2203,18 +2596,33 @@ def _call_anthropic(): delta_type = getattr(delta, "type", None) if delta_type == "text_delta": text = getattr(delta, "text", "") + if text: + text = text_splicer.feed(text) if text and not has_tool_use: _fire_first_delta() agent._fire_stream_delta(text) deltas_were_sent["yes"] = True elif delta_type == "thinking_delta": thinking_text = getattr(delta, "thinking", "") + if thinking_text: + thinking_text = thinking_splicer.feed(thinking_text) if thinking_text: _fire_first_delta() agent._fire_reasoning_delta(thinking_text) + text_tail = text_splicer.flush() + if text_tail and not has_tool_use: + _fire_first_delta() + agent._fire_stream_delta(text_tail) + deltas_were_sent["yes"] = True + + thinking_tail = thinking_splicer.flush() + if thinking_tail: + _fire_first_delta() + agent._fire_reasoning_delta(thinking_tail) + # Return the native Anthropic Message for downstream processing - return stream.get_final_message() + return _repair_anthropic_message_surrogates(stream.get_final_message()) def _call(): import httpx as _httpx diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 7f175fff97fa1..b4ecc3ae3587e 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -595,7 +595,31 @@ def _on_reasoning_delta(text: str) -> None: def _on_event(event: Any) -> None: # TTFB watchdog and activity touch — runs once per SSE event. - agent._codex_stream_last_event_ts = time.time() + # ``_codex_stream_last_event_ts`` is refreshed by ANY event, including + # content-free keepalive / ``response.in_progress`` frames — that's the + # signal the no-byte and stream-idle watchdogs key on. + now = time.time() + agent._codex_stream_last_event_ts = now + # ``_codex_stream_last_progress_ts`` tracks REAL forward progress only: + # text/reasoning deltas, function-call frames, output-item completions, + # or a terminal event. A backend that emits only periodic keepalives + # (observed on chatgpt.com/backend-api/codex: the socket stays "alive" + # with in_progress frames but never produces a delta or completes) keeps + # _last_event_ts fresh forever, so the idle watchdog never trips and the + # call burns the full blunt stale timeout. Stamping progress separately + # lets a progress-stall watchdog reconnect at the fast cutoff instead. + _etype = _event_field(event, "type", "") + if not isinstance(_etype, str): + _etype = "" + _is_progress = ( + "delta" in _etype + or "function_call" in _etype + or _etype == "response.output_item.done" + or _etype == "error" + or _etype in _TERMINAL_EVENT_TYPES + ) + if _is_progress: + agent._codex_stream_last_progress_ts = now agent._touch_activity("receiving stream response") def _interrupt_check() -> bool: diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 16db1bedc30f3..652a29a9d7447 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -215,6 +215,39 @@ def _collect_path_mentions(text: str, relevant_files: list[str], *, limit: int = _dedupe_append(relevant_files, match.rstrip(".,:;"), limit=limit) +_PROVIDER_ERROR_SUMMARY_PATTERNS = ( + # Generic OpenAI-compatible provider "model not found" responses (e.g. the + # claude-bridge / Codex-via-bridge case where literal 'auto' propagated to + # the wire and the bridge politely refused). + "there's an issue with the selected model", + "there is an issue with the selected model", + "run --model to pick a different model", + "may not exist or you may not have access", + # OpenAI / Anthropic / OpenRouter refusal-by-content patterns. Conservative + # — these phrases would not occur naturally in a faithful conversation + # summary; if they appear in summary output it means the auxiliary LLM + # refused the prompt rather than producing one. + "i can't help with that", + "i cannot help with that", + "i'm sorry, but i can't", + "i'm sorry, but i cannot", + "i am sorry, but i can't", + "i am sorry, but i cannot", +) + + +def _looks_like_provider_error_summary(summary: str) -> bool: + """True when the auxiliary LLM returned an error / refusal string instead of a summary. + + Used to reject 200-OK responses whose body is actually the provider's + control plane talking. See ``_PROVIDER_ERROR_SUMMARY_PATTERNS``. + """ + if not summary or not isinstance(summary, str): + return False + needle = summary.lower() + return any(p in needle for p in _PROVIDER_ERROR_SUMMARY_PATTERNS) + + def _content_length_for_budget(raw_content: Any) -> int: """Return the effective char-length of a message's content for token budgeting. @@ -656,6 +689,28 @@ def update_model( self.provider = provider self.api_mode = api_mode self.context_length = context_length + # Re-resolve the threshold for the DESTINATION model when the + # compression config was threaded at construction (agent_init). A + # mid-session fallback (e.g. opus 1M -> gpt-5.5 272K) must honor the new + # model's compression.per_model_threshold instead of re-applying the OLD + # model's stored threshold_percent to the new, smaller window — the + # 2026-06-19 compaction-thrash root cause. Shared resolver = same + # precedence as init (per_model_threshold -> built-in family -> global). + # When the config was NOT threaded (older/direct callers), keep the + # legacy behavior of re-applying the stored percent. + if self._per_model_threshold_cfg is not None or self._global_threshold_percent is not None: + from agent.auxiliary_client import resolve_compression_threshold + self.threshold_percent = resolve_compression_threshold( + self._per_model_threshold_cfg, + model, + provider, + global_threshold=( + self._global_threshold_percent + if self._global_threshold_percent is not None + else self.threshold_percent + ), + allow_codex_gpt55_autoraise=self._codex_gpt55_autoraise, + ) self.threshold_tokens = max( int(context_length * self.threshold_percent), MINIMUM_CONTEXT_LENGTH, @@ -683,7 +738,19 @@ def __init__( provider: str = "", api_mode: str = "", abort_on_summary_failure: bool = False, + per_model_threshold: "dict | None" = None, + global_threshold_percent: "float | None" = None, + codex_gpt55_autoraise: bool = True, ): + # Compression-threshold config for re-resolving the destination model's + # threshold on a model switch / fallback (update_model). When threaded + # by agent_init these let update_model honor compression.per_model_threshold + # for the NEW model instead of re-applying the OLD model's stored percent + # (the 2026-06-19 compaction-thrash bug). When None (older/direct callers) + # update_model keeps the legacy "re-apply stored percent" behavior. + self._per_model_threshold_cfg = per_model_threshold + self._global_threshold_percent = global_threshold_percent + self._codex_gpt55_autoraise = codex_gpt55_autoraise self.model = model self.base_url = base_url self.api_key = api_key @@ -834,6 +901,45 @@ def should_compress(self, prompt_tokens: int = None) -> bool: return False return True + def record_compaction_effectiveness( + self, pre_request_tokens: int, post_request_tokens: int, + ) -> None: + """Single authoritative verdict on whether a compaction was effective, + measured at the REQUEST level (system prompt + messages + tool schemas), + the same level that ``should_compress`` uses to decide re-trigger. + + This corrects the 2026-06-19 compaction-thrash root cause: ``compress()`` + compared a request-level PRE value (``display_tokens``) against a + MESSAGES-ONLY post estimate, so the ~30K of tool-schema overhead made + every pass look like a >=10% saving and reset the anti-thrash counter, + even when the real request stayed over the trigger and immediately + re-fired (the ``205,072 -> 297,723`` "tokens went UP" case). + + A pass is INEFFECTIVE when the post-compaction request estimate did not + get below the trigger AND it did not shed at least 10% of the request. + Both arms matter: a huge transcript that drops 60% but is still over + threshold is making real progress (effective); a pass that "saved" a few + percent and is still over threshold is thrash (ineffective). + + This is the SINGLE owner of the counter's increment/reset for the normal + compaction path — ``compress()`` no longer independently mutates it from + its messages-only verdict (it still increments for the structural + "nothing to compress" no-window case, which is request-independent). + Called exactly once per completed compaction by the done-site, so a + single ineffective pass increments by exactly 1. + """ + if pre_request_tokens > 0: + saved_pct = (pre_request_tokens - post_request_tokens) / pre_request_tokens * 100 + else: + saved_pct = 0.0 + self._last_compression_savings_pct = saved_pct + still_over_threshold = post_request_tokens >= self.threshold_tokens + if still_over_threshold and saved_pct < 10: + self._ineffective_compression_count += 1 + else: + self._ineffective_compression_count = 0 + + # ------------------------------------------------------------------ # Tool output pruning (cheap pre-pass, no LLM call) # ------------------------------------------------------------------ @@ -2405,17 +2511,20 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f new_estimate = estimate_messages_tokens_rough(compressed) saved_estimate = display_tokens - new_estimate - # Anti-thrashing: track compression effectiveness + # NOTE: the anti-thrash effectiveness verdict is NOT decided here. + # ``display_tokens`` is a REQUEST-level pre value but ``new_estimate`` is + # MESSAGES-ONLY, so this comparison is inflated by the ~30K of tool-schema + # overhead and historically reset the counter on every pass (the + # 2026-06-19 compaction-thrash root cause). The authoritative, + # apples-to-apples REQUEST-level verdict is recorded by the done-site via + # ``record_compaction_effectiveness`` (see conversation_compression). + # ``saved_estimate`` / ``savings_pct`` below remain for human-readable + # logging only. savings_pct = (saved_estimate / display_tokens * 100) if display_tokens > 0 else 0 - self._last_compression_savings_pct = savings_pct - if savings_pct < 10: - self._ineffective_compression_count += 1 - else: - self._ineffective_compression_count = 0 if not self.quiet_mode: logger.info( - "Compressed: %d -> %d messages (~%d tokens saved, %.0f%%)", + "Compressed: %d -> %d messages (~%d tokens saved, %.0f%% messages-only)", n_messages, len(compressed), saved_estimate, diff --git a/agent/context_engine.py b/agent/context_engine.py index 79c31fb48e6cd..b595739ccf01f 100644 --- a/agent/context_engine.py +++ b/agent/context_engine.py @@ -170,8 +170,10 @@ def on_session_reset(self) -> None: def get_tool_schemas(self) -> List[Dict[str, Any]]: """Return tool schemas this engine provides to the agent. - Default returns empty list (no tools). LCM would return schemas - for lcm_grep, lcm_describe, lcm_expand here. + Engines may return bare OpenAI function schemas or full + {"type": "function", "function": ...} tool definitions; the host + normalizes both. LCM returns schemas for lcm_grep, lcm_describe, + lcm_expand here. """ return [] diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index d5469a1b344f2..a80db9a687374 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -31,6 +31,7 @@ import logging import os import tempfile +import time import uuid from datetime import datetime from pathlib import Path @@ -50,6 +51,260 @@ f"🗜️ {COMPACTION_STATUS_MARKER} — summarizing earlier conversation so I can continue..." ) +# ── Compaction completion announce (engine-aware) ────────────────────────── +# Spec: ~/.hermes/plans/2026-06-20_compaction-announce-with-context-reference.md +# A persistent, in-chat marker emitted when context is actually compacted, for +# BOTH the built-in ContextCompressor (lossy, session-rotating) and the LCM/DAG +# engine (lossless raw store + lcm_grep/lcm_expand recovery). Additive to the +# fallback announce (never a replacement). Emitted out-of-band via _emit_status, +# never injected into model history. + +# Markers stripped from a summary snippet before display. +_COMPACTION_SUMMARY_MARKERS = ( + "[CONTEXT COMPACTION — REFERENCE ONLY]", + "[CONTEXT COMPACTION - REFERENCE ONLY]", + "[CONTEXT COMPACTION]", +) + +# Allow-list gating (Invariant I8 / §5.5). Statuses that ALWAYS represent a real +# context reduction → announce unconditionally. +_ANNOUNCE_STATUS_UNCONDITIONAL = frozenset( + {"compacted", "overflow_recovery", "degraded_fallback_compressed"} +) +# Statuses that announce ONLY when context tokens actually dropped (token +# reduction is monotonic across LCM node reassignment; message count is not). +_ANNOUNCE_STATUS_CONDITIONAL = frozenset({"degraded_fail_open", "sanitized"}) +# Statuses carrying a degraded-summary caveat. +_DEGRADED_STATUSES = frozenset({"degraded_fallback_compressed", "degraded_fail_open"}) + + +def _compaction_window_label(tokens: "int | None") -> str: + """Compact human label for a context window: 1M, 272K, 128K (mirror of + chat_completion_helpers._format_context_window; kept local to avoid a + cross-module import cycle).""" + if not tokens or tokens <= 0: + return "" + if tokens >= 1_000_000: + whole = tokens / 1_000_000 + return f"{whole:.0f}M" if whole == int(whole) else f"{whole:.1f}M" + if tokens >= 1_000: + return f"{tokens // 1000}K" + return str(tokens) + + +def _abbrev_tokens(tokens: "int | None") -> str: + """~323K / ~15K / ~1.2M style for the token-delta display.""" + if tokens is None or tokens <= 0: + return "?" + if tokens >= 1_000_000: + whole = tokens / 1_000_000 + return f"~{whole:.0f}M" if whole == int(whole) else f"~{whole:.1f}M" + if tokens >= 1_000: + return f"~{tokens // 1000}K" + return f"~{tokens}" + + +def _msg_text(content: Any) -> str: + """Flatten a message ``content`` (str or list-of-blocks) to plain text.""" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for block in content: + if isinstance(block, dict): + t = block.get("text") or block.get("content") + if isinstance(t, str): + parts.append(t) + elif isinstance(block, str): + parts.append(block) + return "\n".join(parts) + return "" + + +def _extract_compaction_summary_snippet( + compressed_messages: list, *, max_chars: int = 160 +) -> "str | None": + """Deterministically pull a one-line snippet of what was summarised. + + Scans for the first message carrying a compaction summary marker, strips the + marker boilerplate, collapses whitespace, and truncates at a word boundary. + Returns ``None`` when no usable summary text exists (e.g. a placeholder-only + marker). Not an LLM call. + """ + if not compressed_messages: + return None + for msg in compressed_messages: + if not isinstance(msg, dict): + continue + text = _msg_text(msg.get("content")) + if not text: + continue + if not any(m in text for m in _COMPACTION_SUMMARY_MARKERS): + continue + for m in _COMPACTION_SUMMARY_MARKERS: + text = text.replace(m, " ") + # collapse all whitespace runs to single spaces + cleaned = " ".join(text.split()) + if not cleaned: + return None + if len(cleaned) <= max_chars: + return cleaned + # truncate at a word boundary, then append an ellipsis + cut = cleaned[:max_chars].rsplit(" ", 1)[0].rstrip() + if not cut: + cut = cleaned[:max_chars].rstrip() + return cut + "…" + return None + + +def _format_compaction_announce( + engine_name: "str | None", + status: "str | None", + *, + old_session_id: "str | None", + new_session_id: "str | None", + old_messages: int, + new_messages: int, + pre_tokens: "int | None", + post_tokens: "int | None", + model: "str | None", + provider: "str | None", + window_from: "int | None" = None, + window_to: "int | None" = None, + summary_snippet: "str | None" = None, + raw_store_count: "int | None" = None, + after_fallback: bool = False, +) -> "str | None": + """Build the engine-aware announce line, or ``None`` if gating says skip. + + Gating (§5.5, Invariant I8): + - LCM (``engine_name == "lcm"``): allow-list on ``status`` — unconditional + set always announces; conditional set announces only on a real token drop; + everything else (incl. unknown/future) is silent. + - Built-in (no engine name / no status): announce only when a real rotation + happened (``old_session_id != new_session_id``). + """ + is_lcm = engine_name == "lcm" + + if is_lcm: + if status in _ANNOUNCE_STATUS_UNCONDITIONAL: + pass + elif status in _ANNOUNCE_STATUS_CONDITIONAL: + if not (pre_tokens and post_tokens and post_tokens < pre_tokens): + return None + else: + return None # default-deny: noop/idle/running/bypassed/unknown + else: + # built-in compressor: a real compaction rotates the session id + if not old_session_id or not new_session_id or old_session_id == new_session_id: + return None + + degraded = status in _DEGRADED_STATUSES + head = "🗜️ Context compacted" + if after_fallback: + head += " after model fallback" + if degraded: + head += " (degraded)" + + parts = [f"{head}: {old_messages}→{new_messages} messages"] + parts.append(f"{_abbrev_tokens(pre_tokens)}→{_abbrev_tokens(post_tokens)} tokens") + + if model: + parts.append(f"{provider}/{model}" if provider else str(model)) + + if after_fallback: + wf, wt = _compaction_window_label(window_from), _compaction_window_label(window_to) + if wf and wt and wf != wt: + parts.append(f"window {wf}→{wt}") + + if is_lcm: + parts.append("engine: lcm") + + line = " · ".join(parts) + + # Recovery reference (engine-correct). + if is_lcm: + if raw_store_count and raw_store_count > 0: + ref = ( + f"↩ nothing lost — {raw_store_count:,} raw turns from this session " + "preserved in lcm.db · recover with lcm_grep / lcm_expand" + ) + else: + ref = ( + "↩ nothing lost — raw turns preserved in lcm.db · " + "recover with lcm_grep / lcm_expand" + ) + else: + ref = f"↩ previous: {old_session_id} → current: {new_session_id}" + line += "\n" + ref + + if summary_snippet: + line += f"\nSummary: {summary_snippet}" + elif degraded: + line += "\nSummary: unavailable — summarizer degraded this pass; raw store is intact." + + return line + + +def _emit_compaction_announce(agent: Any, *, dedupe_key, **fmt_kwargs) -> None: + """Emit the compaction announce once per real compaction boundary. + + Dedupe: ``agent._last_compaction_announced`` holds an engine-namespaced + ``dedupe_key``; a repeat key is skipped. The key is set ONLY after a + successful emit (D7), so a swallowed emit failure does not suppress the next + real compaction's announce. The caller holds the compression lock, so the + read-then-write is serialized per session. + """ + line = _format_compaction_announce(**fmt_kwargs) + if line is None: + return # gating skip — do not advance the key + if getattr(agent, "_last_compaction_announced", None) == dedupe_key: + return # already announced this boundary + emit = getattr(agent, "_emit_status", None) + if not callable(emit): + return + try: + emit(line) + except Exception: + logger.debug("compaction announce emit failed", exc_info=True) + return # key NOT advanced — next real compaction can still announce + agent._last_compaction_announced = dedupe_key + + +# Tight wall-clock window (seconds) used ONLY when no turn id is available to +# link a fallback to a following compaction. Real chained fallback→compaction +# happens within one turn (seconds), not minutes — keep this tight. +_POST_FALLBACK_WALLCLOCK_SECS = 75.0 + + +def _compaction_after_fallback( + agent: Any, *, now_monotonic: float, current_turn_id: "str | None" +) -> "Tuple[bool, Optional[int], Optional[int]]": + """Decide whether this compaction follows a model fallback, turn-scoped. + + Returns ``(after_fallback, window_from, window_to)``. The causal signal is + *same logical turn AND fallback-before-compaction* — NOT wall-clock + proximity (the §0 incident had the fallback AFTER the compaction, which must + NOT be labeled). Only when no turn id exists on either side does it fall back + to a tight wall-clock window, still requiring fallback-before-compaction. + """ + ev = getattr(agent, "_last_fallback_event", None) + if not isinstance(ev, dict): + return (False, None, None) + fb_mono = ev.get("monotonic_time") + if fb_mono is None or fb_mono > now_monotonic: + # fallback happened AFTER this compaction (or unknown time) → not causal + return (False, None, None) + fb_turn = ev.get("turn_id") + if fb_turn is not None and current_turn_id is not None: + if fb_turn != current_turn_id: + return (False, None, None) + else: + # no turn linkage available → tight wall-clock fallback + if (now_monotonic - fb_mono) > _POST_FALLBACK_WALLCLOCK_SECS: + return (False, None, None) + return (True, ev.get("old_window"), ev.get("new_window")) + def _compression_lock_holder(agent: Any) -> str: """Build a unique holder id for the lock: pid:tid:agent-instance:uuid. @@ -611,6 +866,30 @@ def _release_lock() -> None: system_prompt=new_system_prompt or "", tools=agent.tools or None, ) + + # Record the anti-thrash effectiveness verdict at the REQUEST level (the + # same level should_compress uses), apples-to-apples: pre = the request-level + # estimate of the messages that triggered this compaction, post = + # _compressed_est. This is the SINGLE owner of the counter for the normal + # compaction path and fixes the 2026-06-19 thrash where compress()'s + # messages-only verdict reset the counter on every pass (the + # 205,072 -> 297,723 "tokens went UP" case). Use the pre-compaction request + # estimate of the ORIGINAL messages so a failed/placeholder summary that + # leaves the request over threshold is correctly counted as ineffective. + try: + _pre_request_est = estimate_request_tokens_rough( + messages, + system_prompt=system_message or "", + tools=agent.tools or None, + ) + if hasattr(agent.context_compressor, "record_compaction_effectiveness"): + agent.context_compressor.record_compaction_effectiveness( + pre_request_tokens=_pre_request_est, + post_request_tokens=_compressed_est, + ) + except Exception as _eff_err: + logger.debug("record_compaction_effectiveness failed: %s", _eff_err) + agent.context_compressor.last_compression_rough_tokens = _compressed_est agent.context_compressor.last_prompt_tokens = -1 agent.context_compressor.last_completion_tokens = 0 @@ -630,6 +909,51 @@ def _release_lock() -> None: agent.session_id or "none", _pre_msg_count, len(compressed), f"{_compressed_est:,}", ) + + # ── In-chat compaction announce (engine-aware; additive to fallback) ────── + # Emitted out-of-band via _emit_status (persistent chat line, never injected + # into model history). Engine-correct recovery reference: built-in → session + # pointer; LCM → lossless-store + lcm_grep/lcm_expand guidance. Gating is the + # allow-list in _format_compaction_announce. Runs INSIDE the lock hold (the + # _release_lock() below) so the dedupe read-then-write is serialized. + try: + _cc = agent.context_compressor + _engine_name = getattr(_cc, "name", None) + _status = getattr(_cc, "_last_compression_status", None) + _old_sid = locals().get("old_session_id") + _new_sid = agent.session_id + if _engine_name == "lcm": + _dedupe_key = ("lcm", getattr(_cc, "compression_count", None)) + else: + _dedupe_key = ("builtin", (_old_sid, _new_sid)) + _now_mono = time.monotonic() + _after_fb, _win_from, _win_to = _compaction_after_fallback( + agent, + now_monotonic=_now_mono, + current_turn_id=getattr(agent, "_current_turn_id", None), + ) + _emit_compaction_announce( + agent, + dedupe_key=_dedupe_key, + engine_name=_engine_name, + status=_status, + old_session_id=_old_sid, + new_session_id=_new_sid, + old_messages=_pre_msg_count, + new_messages=len(compressed), + pre_tokens=locals().get("_pre_request_est"), + post_tokens=_compressed_est, + model=getattr(agent, "model", None), + provider=getattr(agent, "provider", None), + window_from=_win_from, + window_to=_win_to, + summary_snippet=_extract_compaction_summary_snippet(compressed), + raw_store_count=None, # session-scoped count not cheap here; omit (N-NEW-3) + after_fallback=_after_fb, + ) + except Exception: + logger.debug("compaction announce skipped (non-fatal)", exc_info=True) + # Release the lock on the OLD session_id only AFTER rotation completed # and all post-rotation bookkeeping (memory manager, context engine, # file dedup) ran. A concurrent path that wakes up the moment we diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 379a038a9e09b..e7a700657ed4a 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -47,6 +47,7 @@ ) from agent.model_metadata import ( MINIMUM_CONTEXT_LENGTH, + compose_request_breakdown, estimate_messages_tokens_rough, estimate_request_tokens_rough, get_context_length_from_provider_error, @@ -251,6 +252,48 @@ def _try_refresh_nous_paid_entitlement_credentials(agent) -> bool: return False +def _resolve_skills_prompt_text(agent) -> str: + """Return the skill-index substring embedded in the system prompt. + + Used only by the request-composition telemetry (compose_request_breakdown) + to split the System-prompt bucket into identity/rules vs skill catalog. + Display-only — never affects the bytes actually sent. + + Prefers the stash set during a from-scratch prompt build + (``agent._skills_prompt_text``). On the gateway path a fresh AIAgent is + constructed per turn and the system prompt is usually RESTORED verbatim + from the session DB (see _restore_or_build_system_prompt), so the stash is + never set and stays "". In that case recompute it here via the cached + ``build_skills_system_prompt`` (in-process LRU + disk snapshot, so this is + cheap), mirroring the exact gating used in build_system_prompt_parts: + only when the skills tools are present. Best-effort: any failure returns + the (possibly empty) stash so telemetry never breaks the loop. + """ + stashed = getattr(agent, "_skills_prompt_text", "") or "" + if stashed: + return stashed + try: + valid = getattr(agent, "valid_tool_names", None) or set() + if not any(n in valid for n in ("skills_list", "skill_view", "skill_manage")): + return "" + _r = _ra() + avail_toolsets = { + ts for ts in (_r.get_toolset_for_tool(t) for t in valid) if ts + } + text = _r.build_skills_system_prompt( + available_tools=valid, + available_toolsets=avail_toolsets, + ) or "" + # Memoize so subsequent calls this turn skip the recompute. + try: + agent._skills_prompt_text = text + except Exception: + pass + return text + except Exception: + return stashed + + def _restore_or_build_system_prompt(agent, system_message, conversation_history): """Restore the cached system prompt from the session DB or build it fresh. @@ -499,6 +542,11 @@ def run_conversation( # Main conversation loop counters (pure locals consumed by the loop below). api_call_count = 0 + # Blackbox: per-turn token/cost/latency accumulator. LOCAL to this + # run_conversation frame (re-entrant-safe across subagents). Each + # successful provider call appends a dict at the usage-commit site below; + # folded into the on_session_end `turn_usage` kwarg at the end of the turn. + _turn_calls: List[Dict[str, Any]] = [] final_response = None interrupted = False failed = False @@ -822,6 +870,21 @@ def run_conversation( approx_request_tokens = estimate_request_tokens_rough( api_messages, tools=agent.tools or None ) + # Real per-call request composition (char/4 over the EXACT payload being + # sent this call): fixed (system+tool schemas) vs non-fixed (history, + # tool results, tool-call args). Captured here, attached to the per-call + # _turn_calls entry below; the FINAL call's composition becomes the + # turn's recorded breakdown. Telemetry must never break the loop. + try: + _skills_text = _resolve_skills_prompt_text(agent) + _call_composition = compose_request_breakdown( + api_messages, + system_prompt=effective_system or "", + skills_prompt=_skills_text, + tools=agent.tools or None, + ) + except Exception: + _call_composition = None _runtime_context_error = _ollama_context_limit_error( agent, approx_request_tokens @@ -1781,6 +1844,45 @@ def _perform_api_call(next_api_kwargs): 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 + # Keep the final successful provider-call usage available for + # `/context` / `/usage` style surfaces. The session_* counters + # above are cumulative; this snapshot preserves the last turn's + # cache split without provider-specific payload parsing later. + agent.last_turn_usage = { + "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, + "last_composition": _call_composition, + "reasoning_tokens": canonical_usage.reasoning_tokens, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": total_tokens, + } + # Blackbox per-TURN accumulator (separate from the per-CALL + # snapshot above). INVARIANT: this append lives INSIDE the + # successful-usage commit block — the same place that mutates + # session_*_tokens — and MUST NOT be moved into any retry or + # exception branch, or a retried 5xx would double-count. + # `_turn_calls` is a LOCAL list (initialized at the top of + # run_conversation), never an agent attribute, so concurrent + # subagents running their own run_conversation frame cannot + # stomp each other's accumulator. + try: + _turn_calls.append({ + "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, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": total_tokens, + "latency_s": api_duration, + "composition": _call_composition, + }) + except Exception: + pass # telemetry must never break the conversation loop # Log API call details for debugging/observability _cache_pct = "" @@ -1839,6 +1941,11 @@ def _perform_api_call(next_api_kwargs): if cost_result.status == "included" else None, model=agent.model, api_call_count=1, + last_turn_input_tokens=canonical_usage.input_tokens, + last_turn_output_tokens=canonical_usage.output_tokens, + last_turn_cache_read_tokens=canonical_usage.cache_read_tokens, + last_turn_cache_write_tokens=canonical_usage.cache_write_tokens, + last_turn_reasoning_tokens=canonical_usage.reasoning_tokens, ) except Exception as e: # Log token persistence failures so they're @@ -2745,7 +2852,10 @@ def _perform_api_call(next_api_kwargs): ) else: agent._buffer_status("⚠️ Rate limited — switching to fallback provider...") - if agent._try_activate_fallback(reason=classified.reason): + if agent._try_activate_fallback( + reason=classified.reason, + error_context=error_context, + ): retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False @@ -4414,6 +4524,7 @@ def _perform_api_call(next_api_kwargs): original_user_message=original_user_message, _should_review_memory=_should_review_memory, _turn_exit_reason=_turn_exit_reason, + _turn_calls=_turn_calls, ) diff --git a/agent/message_sanitization.py b/agent/message_sanitization.py index ff53d247a84a9..fe4160e20cf3d 100644 --- a/agent/message_sanitization.py +++ b/agent/message_sanitization.py @@ -39,6 +39,86 @@ def _sanitize_surrogates(text: str) -> str: return text +_HIGH_SURROGATE_MIN = 0xD800 +_HIGH_SURROGATE_MAX = 0xDBFF +_LOW_SURROGATE_MIN = 0xDC00 +_LOW_SURROGATE_MAX = 0xDFFF + + +def _is_high_surrogate(ch: str) -> bool: + return len(ch) == 1 and _HIGH_SURROGATE_MIN <= ord(ch) <= _HIGH_SURROGATE_MAX + + +def _is_low_surrogate(ch: str) -> bool: + return len(ch) == 1 and _LOW_SURROGATE_MIN <= ord(ch) <= _LOW_SURROGATE_MAX + + +def _combine_surrogate_pair(high: str, low: str) -> str: + """Actively recombine a UTF-16 surrogate pair into one Unicode scalar.""" + return (high + low).encode("utf-16-le", "surrogatepass").decode("utf-16-le") + + +class _SurrogateSplicer: + """Recombine UTF-16 surrogate pairs split across streaming deltas. + + Some providers JSON-decode ``"\\ud83d"`` and ``"\\ude00"`` in separate + stream deltas. Python stores those as two invalid surrogate code points; + concatenation alone does not turn them into 😀, and a later UTF-8 encode + crashes. This stateful helper carries one trailing high surrogate across + deltas, actively recombines valid high+low pairs, and floors orphaned + surrogates to U+FFFD. + """ + + def __init__(self) -> None: + self._pending_high: str = "" + + def feed(self, text: str) -> str: + if not isinstance(text, str) or not text: + return "" + + if self._pending_high: + text = self._pending_high + text + self._pending_high = "" + + out: list[str] = [] + i = 0 + n = len(text) + while i < n: + ch = text[i] + if _is_high_surrogate(ch): + if i + 1 >= n: + self._pending_high = ch + i += 1 + continue + nxt = text[i + 1] + if _is_low_surrogate(nxt): + out.append(_combine_surrogate_pair(ch, nxt)) + i += 2 + continue + out.append("\ufffd") + i += 1 + continue + if _is_low_surrogate(ch): + out.append("\ufffd") + i += 1 + continue + out.append(ch) + i += 1 + return "".join(out) + + def flush(self) -> str: + if self._pending_high: + self._pending_high = "" + return "\ufffd" + return "" + + +def _splice_surrogates(text: str) -> str: + """Recombine adjacent surrogate pairs in one string and floor orphans.""" + splicer = _SurrogateSplicer() + return splicer.feed(text) + splicer.flush() + + def _sanitize_structure_surrogates(payload: Any) -> bool: """Replace surrogate code points in nested dict/list payloads in-place. @@ -432,6 +512,8 @@ def _walk(node): __all__ = [ "_SURROGATE_RE", "_sanitize_surrogates", + "_SurrogateSplicer", + "_splice_surrogates", "_sanitize_structure_surrogates", "_sanitize_messages_surrogates", "_escape_invalid_chars_in_json_strings", diff --git a/agent/model_metadata.py b/agent/model_metadata.py index e31fcdea48db4..2954a848b4713 100644 --- a/agent/model_metadata.py +++ b/agent/model_metadata.py @@ -7,6 +7,7 @@ import ipaddress import json import logging +import math import os import re import time @@ -107,7 +108,7 @@ def _strip_provider_prefix(model: str) -> str: _model_metadata_cache_time: float = 0 _novita_metadata_cache: Dict[str, Dict[str, Any]] = {} _novita_metadata_cache_time: float = 0 -_MODEL_CACHE_TTL = 3600 +_MODEL_CACHE_TTL = 259200 # 3 days — model pricing/context rarely change; long cache keeps notional cost cheap & resilient to OpenRouter blips _endpoint_model_metadata_cache: Dict[str, Dict[str, Dict[str, Any]]] = {} _endpoint_model_metadata_cache_time: Dict[str, float] = {} _ENDPOINT_MODEL_CACHE_TTL = 300 @@ -184,6 +185,96 @@ def _save_model_metadata_disk_cache(data: Dict[str, Dict[str, Any]]) -> None: # Sessions, model switches, and cron jobs should reject models below this. MINIMUM_CONTEXT_LENGTH = 64_000 +# Chars-per-token divisor for the /context /usage /compress composition +# estimate (compose_request_breakdown). The classic rule of thumb is ~4 +# chars/token, but that is tuned for English prose; real Hermes requests are +# dominated by JSON/code-dense tool results which tokenize closer to ~3.5 +# chars/token, so a flat /4 systematically UNDER-counts (empirically ~-8% at +# 170k context, ~-15% at 280k on tool-heavy turns). 3.5 tightened the live +# delta-vs-measured band to roughly +-5% across 170k-280k turns. Tunable via +# env HERMES_COMPOSITION_CHARS_PER_TOKEN without a code change; clamped to a +# sane 2.0-8.0 range so a typo can't make the estimate absurd. This only +# affects the displayed composition estimate -- never billing or the provider's +# real prompt_tokens (Measured occupancy). As of the divisor-unification work +# every char-based estimator (display, rough/compression, 413/stale-call) reads +# this same constant, so 3.5 is the single tunable knob across all of them. +def _composition_chars_per_token() -> float: + raw = os.environ.get("HERMES_COMPOSITION_CHARS_PER_TOKEN", "").strip() + if raw: + try: + val = float(raw) + if 2.0 <= val <= 8.0: + return val + except (TypeError, ValueError): + pass + return 3.5 + + +COMPOSITION_CHARS_PER_TOKEN = _composition_chars_per_token() + + +# Fixed-bucket char->token divisor. The shared 3.5 above is tuned for NON-FIXED +# content (history/tool-results/tool-args) which packs DENSER than 4 chars/token +# (see the under-count note above: /4 under-counts at 170k-280k history-heavy +# turns). The FIXED prefix -- system prompt + tool schemas + skills index -- is +# different content and was MEASURED to pack LOOSER. Pooled live measurement +# (n=60 claude-opus full-tool near-zero-history turns across two turns.db's) put +# the REAL fixed ratio at ~4.38 chars/tok (median 4.45); o200k/gpt-5.x measured +# ~4.09. So /3.5 over-counts the fixed prefix by ~20-25%. 4.5 is the chosen +# divisor (one global number for ALL gateways, by Ace 2026-06-15 -- consistency +# over per-host tuning). Post-deploy LIVE measurement of real fixed density +# (back-out context_used - non-fixed on post-reload full-tool rows) put +# Apollo/default at ~4.36 chars/tok and Aegis at ~4.83 (smaller SOUL + fewer +# tools tokenize denser); o200k/gpt-5.x ~4.09. 4.5 is the first single divisor +# that lands BOTH high-traffic gateways inside the [0.90,1.05] acceptance band: +# Apollo 4.5/4.36 = ~1.03 (3% under, safe), Aegis 4.5/4.83 = ~0.93 (7% over, +# safe). 4.2 left Aegis at ~0.87 (just under band); 4.5 fixes that without +# tipping Apollo into a dangerous under-count. Applied to ALL families by fiat +# (the fixed bytes are identical across models, only tokenizer density varies, +# and it clustered 4.09-4.83); override via +# HERMES_COMPOSITION_CHARS_PER_TOKEN_FIXED if a family proves materially +# different. Same 2.0-8.0 clamp; display-only, never billing. Two divisors on +# purpose: accurate-but-slightly-conservative on the stable fixed prefix, more +# conservative on the volatile non-fixed tail so the context-pressure warning +# fires early rather than late. +def _composition_chars_per_token_fixed() -> float: + raw = os.environ.get("HERMES_COMPOSITION_CHARS_PER_TOKEN_FIXED", "").strip() + if raw: + try: + val = float(raw) + if 2.0 <= val <= 8.0: + return val + except (TypeError, ValueError): + pass + return 4.5 + + +COMPOSITION_CHARS_PER_TOKEN_FIXED = _composition_chars_per_token_fixed() + + +# Per-message wire-framing overhead, in tokens. Every chat message carries +# structural scaffolding the char-based estimators don't see: role tokens, +# message delimiters, and (for tool turns) tool-call id wrappers. The content +# char-count misses all of it, which is a large part of why the displayed +# Estimated request size runs under the provider's Measured occupancy on +# many-message sessions. A flat per-message constant absorbs most of that gap +# without tokenizing. ~4 tok/message is a conservative empirical fit across +# Anthropic + OpenAI-wire requests. Tunable via env (clamped 0-20; 0 disables). +# Display-only: never affects billing, Measured occupancy, or compression. +def _per_message_framing_tokens() -> int: + raw = os.environ.get("HERMES_PER_MESSAGE_FRAMING_TOKENS", "").strip() + if raw: + try: + val = int(raw) + if 0 <= val <= 20: + return val + except (TypeError, ValueError): + pass + return 4 + + +PER_MESSAGE_FRAMING_TOKENS = _per_message_framing_tokens() + # Thin fallback defaults — only broad model family patterns. # These fire only when provider is unknown AND models.dev/OpenRouter/Anthropic # all miss. Replaced the previous 80+ entry dict. @@ -203,6 +294,9 @@ def _save_model_metadata_disk_cache(data: Dict[str, Dict[str, Any]]) -> None: "claude-sonnet-4-6": 1000000, "claude-opus-4.6": 1000000, "claude-sonnet-4.6": 1000000, + # Claude Fable 5 / Mythos 5 (1M context, 128k max output) — released 2026-06-09. + "claude-fable-5": 1000000, + "claude-mythos-5": 1000000, # Catch-all for older Claude models (must sort after specific entries) "claude": 200000, # OpenAI — GPT-5 family (most have 400k; specific overrides first) @@ -679,7 +773,7 @@ def _add_model_aliases(cache: Dict[str, Dict[str, Any]], model_id: str, entry: D def fetch_model_metadata(force_refresh: bool = False) -> Dict[str, Dict[str, Any]]: - """Fetch model metadata from OpenRouter (cached for 1 hour).""" + """Fetch model metadata from OpenRouter (cached for _MODEL_CACHE_TTL seconds).""" global _model_metadata_cache, _model_metadata_cache_time if not force_refresh and _model_metadata_cache and (time.time() - _model_metadata_cache_time) < _MODEL_CACHE_TTL: @@ -1973,16 +2067,50 @@ def get_model_context_length( return DEFAULT_FALLBACK_CONTEXT +def _ceil_chars_to_tokens(chars: int) -> int: + """Ceiling-divide a char count into tokens by COMPOSITION_CHARS_PER_TOKEN. + + Single shared char->token rule for ALL of Hermes' rough estimators + (estimate_tokens_rough / estimate_messages_tokens_rough / + estimate_request_tokens_rough) AND the /context composition breakdown. + Default divisor 3.5 (tunable via HERMES_COMPOSITION_CHARS_PER_TOKEN); see + the COMPOSITION_CHARS_PER_TOKEN comment. Ceiling so short texts never round + to 0 tokens (which would let many small tool results vanish from an estimate + and undercount). char/3.5 estimates HIGHER than the old char/4, so + compression / pre-flight 413 checks fire *sooner*, never later -- strictly + safer for context-overrun protection. + """ + if chars <= 0: + return 0 + return int(math.ceil(chars / COMPOSITION_CHARS_PER_TOKEN)) + + +def _ceil_chars_to_tokens_fixed(chars: int) -> int: + """Ceiling-divide a char count into tokens by COMPOSITION_CHARS_PER_TOKEN_FIXED. + + Sibling of _ceil_chars_to_tokens, but for the FIXED request prefix (system + prompt + tool schemas + skills index). That content was measured to pack + looser than the non-fixed tail (~4.0-4.35 chars/tok vs ~3.5), so it uses its + own divisor (default 4.0, tunable via HERMES_COMPOSITION_CHARS_PER_TOKEN_FIXED) + -- see the COMPOSITION_CHARS_PER_TOKEN_FIXED comment. Same ceiling/min-0 + behavior. Display-only; never billing. + """ + if chars <= 0: + return 0 + return int(math.ceil(chars / COMPOSITION_CHARS_PER_TOKEN_FIXED)) + + def estimate_tokens_rough(text: str) -> int: - """Rough token estimate (~4 chars/token) for pre-flight checks. + """Rough token estimate (~3.5 chars/token) for pre-flight checks. Uses ceiling division so short texts (1-3 chars) never estimate as 0 tokens, which would cause the compressor and pre-flight checks to systematically undercount when many short tool results are present. + Divisor is the shared, tunable COMPOSITION_CHARS_PER_TOKEN (default 3.5). """ if not text: return 0 - return (len(text) + 3) // 4 + return _ceil_chars_to_tokens(len(text)) def estimate_messages_tokens_rough(messages: List[Dict[str, Any]]) -> int: @@ -1999,7 +2127,7 @@ def estimate_messages_tokens_rough(messages: List[Dict[str, Any]]) -> int: for msg in messages: total_chars += _estimate_message_chars(msg) image_tokens += _count_image_tokens(msg, _IMAGE_TOKEN_COST) - return ((total_chars + 3) // 4) + image_tokens + return _ceil_chars_to_tokens(total_chars) + image_tokens def _count_image_tokens(msg: Dict[str, Any], cost_per_image: int) -> int: @@ -2077,9 +2205,153 @@ def estimate_request_tokens_rough( """ total = 0 if system_prompt: - total += (len(system_prompt) + 3) // 4 + total += _ceil_chars_to_tokens_fixed(len(system_prompt)) if messages: total += estimate_messages_tokens_rough(messages) if tools: - total += (len(str(tools)) + 3) // 4 + total += _ceil_chars_to_tokens_fixed(len(str(tools))) return total + + +def compose_request_breakdown( + messages: List[Dict[str, Any]], + *, + system_prompt: str = "", + skills_prompt: str = "", + tools: Optional[List[Dict[str, Any]]] = None, +) -> Dict[str, int]: + """Decompose an outgoing request into fixed vs non-fixed token buckets. + + Single source of truth for "what did this request actually contain" -- + measured at the call site over the EXACT payload being sent (final + api_messages + effective_system + agent.tools), so the buckets describe + reality, not a re-derived proxy over the stored transcript. + + Buckets (char/N estimates with a TWO-TIER divisor N): + Fixed (stable cacheable prefix) use COMPOSITION_CHARS_PER_TOKEN_FIXED + (default 4.0; env HERMES_COMPOSITION_CHARS_PER_TOKEN_FIXED) -- this + content was measured to pack ~4.0-4.35 chars/tok: + sys_tokens, tool_schema_tokens + - sys_tokens splits into identity_tokens + skills_tokens, where + skills_tokens is the skill-index catalog embedded in the system + prompt (passed separately as skills_prompt; it is a substring of + system_prompt, so identity_tokens = sys_tokens - skills_tokens). + Non-fixed (grows with the conversation) use COMPOSITION_CHARS_PER_TOKEN + (default 3.5; env HERMES_COMPOSITION_CHARS_PER_TOKEN) -- this content + packs denser (~3.5 chars/tok), and a slight over-estimate here is + intentional (context-pressure warning fires early): + history_tokens, tool_result_tokens, tool_arg_tokens, framing_tokens + - framing_tokens is a flat per-message wire-overhead estimate + (PER_MESSAGE_FRAMING_TOKENS * message_count): role tokens, message + delimiters, tool-call id wrappers that the char walk never sees. + It scales with message count, so it lives in non-fixed. + + Image parts count at the flat per-image cost (matching + estimate_messages_tokens_rough). The system message inside messages is + skipped (accounted via system_prompt) to avoid double-counting the prefix. + + total == fixed + nonfixed exactly (no scaling). Distinct from the + provider's prompt_tokens (true tokenizer count); surfaces show both. + """ + IMG_COST = 1500 + sys_chars = len(system_prompt or "") + skills_chars = len(skills_prompt or "") + # The skills catalog is a substring of the system prompt; never let it + # exceed the whole prompt (defensive against a stale/mismatched stash). + skills_chars = min(skills_chars, sys_chars) + # Count the skills listed in the index. Each skill is rendered as a + # bullet line " - {name}[: {desc}]" (4-space indent) by + # build_skills_system_prompt; category headers use a 2-space indent, so + # the 4-space " - " prefix uniquely identifies a skill entry. + skills_count = 0 + if skills_prompt: + for _line in skills_prompt.splitlines(): + if _line.startswith(" - "): + skills_count += 1 + tool_schema_chars = len(str(tools)) if tools else 0 + history_chars = 0 + tool_result_chars = 0 + tool_arg_chars = 0 + image_tokens = 0 + tool_result_count = 0 + history_message_count = 0 + for msg in messages or []: + if not isinstance(msg, dict): + history_chars += len(str(msg)) + history_message_count += 1 + continue + role = msg.get("role") + if role == "system": + continue + image_tokens += _count_image_tokens(msg, IMG_COST) + if role == "tool": + tool_result_count += 1 + tool_result_chars += _estimate_message_chars(msg) + else: + history_message_count += 1 + # History = visible text content only. tool_calls JSON is counted + # separately in tool_arg_chars below; counting the whole message + # dict here would double-count the args (it serializes tool_calls). + tcs = msg.get("tool_calls") + if tcs: + stripped = {k: v for k, v in msg.items() if k != "tool_calls"} + history_chars += _estimate_message_chars(stripped) + try: + tool_arg_chars += len( + tcs if isinstance(tcs, str) else json.dumps(tcs) + ) + except Exception: + tool_arg_chars += len(str(tcs)) + else: + history_chars += _estimate_message_chars(msg) + + def _t(chars: int) -> int: + # NON-FIXED char->token rule (COMPOSITION_CHARS_PER_TOKEN, default 3.5), + # ceiling so short tool results never round to 0 and vanish. Used for the + # volatile buckets (history/tool-results/tool-args) which pack denser. + return _ceil_chars_to_tokens(chars) + + def _t_fixed(chars: int) -> int: + # FIXED char->token rule (COMPOSITION_CHARS_PER_TOKEN_FIXED, default 4.0). + # The stable request prefix (system prompt + tool schemas + skills index) + # was measured to pack looser (~4.0-4.35 chars/tok) than the non-fixed + # tail; using /4.0 here removes the ~17-24% over-count /3.5 caused on it. + return _ceil_chars_to_tokens_fixed(chars) + + sys_tokens = _t_fixed(sys_chars) + # Split the fixed system prompt into the skill-index catalog vs everything + # else (identity/rules/guidance). skills_tokens is derived from the same + # FIXED char->token rule; identity is the remainder so the two always sum to + # sys_tokens exactly (no double-count, invariant preserved). + skills_tokens = _t_fixed(skills_chars) + skills_tokens = min(skills_tokens, sys_tokens) + identity_tokens = sys_tokens - skills_tokens + tool_schema_tokens = _t_fixed(tool_schema_chars) + # Per-message wire framing: flat constant * message count. Counts every + # message on the wire (including the system message) since each carries + # role/delimiter overhead. Scales with the conversation -> non-fixed. + framing_tokens = PER_MESSAGE_FRAMING_TOKENS * len(messages or []) + history_tokens = _t(history_chars) + image_tokens + tool_result_tokens = _t(tool_result_chars) + tool_arg_tokens = _t(tool_arg_chars) + fixed_tokens = sys_tokens + tool_schema_tokens + nonfixed_tokens = ( + history_tokens + tool_result_tokens + tool_arg_tokens + framing_tokens + ) + + return { + "sys_tokens": sys_tokens, + "identity_tokens": identity_tokens, + "skills_tokens": skills_tokens, + "skills_count": skills_count, + "tool_schema_tokens": tool_schema_tokens, + "history_tokens": history_tokens, + "history_message_count": history_message_count, + "tool_result_tokens": tool_result_tokens, + "tool_arg_tokens": tool_arg_tokens, + "tool_result_count": tool_result_count, + "framing_tokens": framing_tokens, + "fixed_tokens": fixed_tokens, + "nonfixed_tokens": nonfixed_tokens, + "total_tokens": fixed_tokens + nonfixed_tokens, + } diff --git a/agent/redact.py b/agent/redact.py index de247ec0ad2d4..f622ce61f6413 100644 --- a/agent/redact.py +++ b/agent/redact.py @@ -109,8 +109,13 @@ # ENV assignment patterns: KEY=value where KEY contains a secret-like name _SECRET_ENV_NAMES = r"(?:API_?KEY|TOKEN|SECRET|PASSWORD|PASSWD|CREDENTIAL|AUTH)" +# Git identity vars (GIT_AUTHOR_*/GIT_COMMITTER_*) contain the substring +# "AUTH" (in "AUTHOR") and so falsely match the secret-name pattern below. +# They are never secrets -- exempt them with a leading word-boundary + +# negative lookahead so commit-authoring commands aren't mangled in output. +_GIT_IDENTITY_ALLOWLIST = r"\b(?!GIT_AUTHOR_NAME=)(?!GIT_AUTHOR_EMAIL=)(?!GIT_COMMITTER_NAME=)(?!GIT_COMMITTER_EMAIL=)" _ENV_ASSIGN_RE = re.compile( - rf"([A-Z0-9_]{{0,50}}{_SECRET_ENV_NAMES}[A-Z0-9_]{{0,50}})\s*=\s*(['\"]?)(\S+)\2", + rf"{_GIT_IDENTITY_ALLOWLIST}([A-Z0-9_]{{0,50}}{_SECRET_ENV_NAMES}[A-Z0-9_]{{0,50}})\s*=\s*(['\"]?)(\S+)\2", ) # JSON field patterns: "apiKey": "value", "token": "value", etc. diff --git a/agent/system_prompt.py b/agent/system_prompt.py index 76f57dfcdbc00..3e9c08a38eaf4 100644 --- a/agent/system_prompt.py +++ b/agent/system_prompt.py @@ -213,6 +213,14 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None) skills_prompt = "" if skills_prompt: stable_parts.append(skills_prompt) + # Stash the skills-index substring on the agent so the request-composition + # telemetry (compose_request_breakdown) can split System prompt into + # identity/rules vs skill catalog. Display-only; never affects the prompt + # bytes actually sent. Cleared to "" when no skills block is present. + try: + agent._skills_prompt_text = skills_prompt or "" + except Exception: + pass # Alibaba Coding Plan API always returns "glm-4.7" as model name regardless # of the requested model. Inject explicit model identity into the system prompt diff --git a/agent/transports/chat_completions.py b/agent/transports/chat_completions.py index c0b2a13d250f8..b8fad1439f418 100644 --- a/agent/transports/chat_completions.py +++ b/agent/transports/chat_completions.py @@ -534,6 +534,7 @@ def _build_kwargs_from_profile(self, profile, model, sanitized, tools, params): base_url=params.get("base_url"), ollama_num_ctx=params.get("ollama_num_ctx"), session_id=params.get("session_id"), + bridge_route_suffix=params.get("bridge_route_suffix"), ) ) api_kwargs.update(top_level_from_profile) diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index 20db3fcef9f61..f17cd3b51076f 100644 --- a/agent/turn_finalizer.py +++ b/agent/turn_finalizer.py @@ -42,6 +42,7 @@ def finalize_turn( original_user_message, _should_review_memory, _turn_exit_reason, + _turn_calls=None, ): """Run the post-loop finalization and return the turn ``result`` dict. @@ -167,13 +168,13 @@ def finalize_turn( _diag_msg = ( "Turn ended: reason=%s model=%s api_calls=%d/%d budget=%d/%d " - "tool_turns=%d last_msg_role=%s response_len=%d session=%s" + "tool_turns=%d last_msg_role=%s response_len=%d session=%s task=%s" ) _diag_args = ( _turn_exit_reason, agent.model, api_call_count, agent.max_iterations, _budget_used, _budget_max, _turn_tool_count, _last_msg_role, _resp_len, - agent.session_id or "none", + agent.session_id or "none", effective_task_id or "none", ) if _last_msg_role == "tool" and not interrupted: @@ -408,10 +409,80 @@ def finalize_turn( # _reset_session). # Plugin hook: on_session_end - # Fired at the very end of every run_conversation call. - # Plugins can use this for cleanup, flushing buffers, etc. + # Fired at the very end of every run_conversation call (i.e. once per TURN, + # despite the name). Plugins can use this for cleanup, flushing buffers, + # and per-turn telemetry. The `turn_usage` kwarg is an ADDITIVE, optional + # payload (Blackbox plugin) — existing consumers take **kwargs and ignore + # it; invoke_hook wraps each callback in try/except so a strict-signature + # callback cannot break the loop. try: from hermes_cli.plugins import invoke_hook as _invoke_hook + # Fold the per-turn accumulator into a compact summary. Telemetry must + # never break the turn, so guard the fold. + _turn_calls = _turn_calls or [] + _turn_usage = None + try: + if _turn_calls: + # Last-call cache split — the FINAL provider call's own + # cache_read/cache_write/uncached. Distinct from the summed + # cache_* totals above (which are whole-turn billing and + # double-count re-sent context). These three sum to the final + # call's prompt_tokens == context_used, so they decompose the + # context WINDOW (occupancy) rather than the turn's SPEND. The + # /context "Context window" line renders this split so the + # window numbers are visibly different from the billed sums. + _last_call = _turn_calls[-1] + _last_cache_read = int(_last_call.get("cache_read_tokens", 0) or 0) + _last_cache_write = int(_last_call.get("cache_write_tokens", 0) or 0) + _last_uncached = int(_last_call.get("input_tokens", 0) or 0) + # Request composition of the FINAL call — the char/4 fixed vs + # non-fixed breakdown of the exact payload that produced the + # window occupancy (context_used). This is the authoritative + # "what was in the last request" record /context + /usage read. + _last_composition = _last_call.get("composition") or None + # Per-call composition history (small ints, ~60B/call) so + # /cost can show how the request grew call-by-call. + _comp_calls = [ + { + "composition": c.get("composition"), + "output_tokens": int(c.get("output_tokens", 0) or 0), + "reasoning_tokens": int(c.get("reasoning_tokens", 0) or 0), + } + for c in _turn_calls + ] + _turn_usage = { + "api_calls": len(_turn_calls), + "input_tokens": sum(c["input_tokens"] for c in _turn_calls), + "output_tokens": sum(c["output_tokens"] for c in _turn_calls), + "cache_read_tokens": sum(c["cache_read_tokens"] for c in _turn_calls), + "cache_write_tokens": sum(c["cache_write_tokens"] for c in _turn_calls), + "reasoning_tokens": sum(c["reasoning_tokens"] for c in _turn_calls), + "total_tokens": sum(c["total_tokens"] for c in _turn_calls), + "latency_s": sum(c.get("latency_s", 0.0) for c in _turn_calls), + # Per-call breakdown so the plugin can price each call + # against tiered pricing and reconcile cost_status worst-of. + "calls": list(_turn_calls), + "context_used": getattr(agent.context_compressor, "last_prompt_tokens", 0) + if getattr(agent, "context_compressor", None) else 0, + "context_length": getattr(agent.context_compressor, "context_length", 0) + if getattr(agent, "context_compressor", None) else 0, + # Last-call cache split (decomposes the context window). + "last_cache_read_tokens": _last_cache_read, + "last_cache_write_tokens": _last_cache_write, + "last_uncached_tokens": _last_uncached, + # Real request composition (fixed vs non-fixed, char/4) of + # the final call + per-call history. See compose_request_breakdown. + "last_composition": _last_composition, + "composition_calls": _comp_calls, + # Subagent attribution (set by delegate_tool before run; absent at top level). + "parent_turn_id": getattr(agent, "_blackbox_parent_turn_id", None), + "parent_platform": getattr(agent, "_blackbox_parent_platform", None), + "parent_chat_id": getattr(agent, "_blackbox_parent_chat_id", None), + "parent_chat_name": getattr(agent, "_blackbox_parent_chat_name", None), + "is_subagent": bool(getattr(agent, "_blackbox_is_subagent", False)), + } + except Exception: + _turn_usage = None _invoke_hook( "on_session_end", session_id=agent.session_id, @@ -421,6 +492,12 @@ def finalize_turn( interrupted=interrupted, model=agent.model, platform=getattr(agent, "platform", None) or "", + provider=getattr(agent, "provider", None) or "", + chat_id=getattr(agent, "_chat_id", None) or "", + chat_name=getattr(agent, "_chat_name", None) or "", + user_message=original_user_message, + final_response=final_response, + turn_usage=_turn_usage, ) except Exception as exc: logger.warning("on_session_end hook failed: %s", exc) diff --git a/agent/usage_pricing.py b/agent/usage_pricing.py index 95bb11df521e4..7a14620d3b245 100644 --- a/agent/usage_pricing.py +++ b/agent/usage_pricing.py @@ -27,6 +27,59 @@ ] +# Local subscription proxies / bridges / pools that front the Anthropic API. +# Their marginal cash cost is $0 (flat Claude subscription / tailnet failover / +# local relay), but we price them at official Anthropic API rates for fleet cost +# *visibility* and label the result "estimated". Provider names match the +# `provider:` keys used in ~/.hermes/config.yaml + plugins/model-providers/* +# across the fleet. +# +# This is the set of EXACT BASE names. The numbered failover family +# (claude-api-proxy-fN / claude-bridge-fN, any integer N) is matched by PATTERN +# in is_notional_anthropic_provider() below — so a NEW failover lane (-f6, -f7, +# …) can never again silently price as $0 the way -f2/-f3/-f4/-f5 did. Use the +# predicate, not bare `in` membership, at every call site. +NOTIONAL_ANTHROPIC_PROVIDERS = frozenset({ + "claude-api-proxy", + "claude-bridge", + "claude-pool", +}) + +# claude-api-proxy-f1, claude-bridge-f2, … -f. Anchored + +# integer-only so it matches ONLY the disciplined failover naming and never an +# unrelated "claude-bridge-frobnicate". claude-pool has no -fN family (failover +# happens inside the relay), so it stays an exact base member above. +_NOTIONAL_ANTHROPIC_FN_RE = re.compile(r"^claude-(?:api-proxy|bridge)-f\d+$") + + +def is_notional_anthropic_provider(provider_name: Optional[str]) -> bool: + """True if a provider key should be priced at notional Anthropic rates. + + Covers the exact base relays/proxies/pool AND the numbered -fN failover + family by pattern, so adding a new -fN lane (or a config that references + one) never needs a code change. Normalizes (strip/lower) defensively for + direct callers; resolve_billing_route already normalizes before calling. + """ + p = (provider_name or "").strip().lower() + if not p: + return False + return p in NOTIONAL_ANTHROPIC_PROVIDERS or bool( + _NOTIONAL_ANTHROPIC_FN_RE.match(p) + ) + +# Notional pricing for ChatGPT-subscription Codex providers (openai-codex). +# Marginal cash cost is $0 (covered by a flat ChatGPT subscription), but for +# fleet cost *visibility* we price these at OpenRouter's live catalog rates for +# the underlying OpenAI model and label the result "estimated". Unlike the +# Anthropic proxies (which fall back to a static docs snapshot), OpenAI models +# are priced purely from the live OpenRouter models API — the same dynamic +# source that already powers `provider: openrouter` routes fleet-wide. See +# resolve_billing_route() and _openrouter_pricing_entry(). +NOTIONAL_OPENROUTER_PROVIDERS = frozenset({ + "openai-codex", +}) + + @dataclass(frozen=True) class CanonicalUsage: input_tokens: int = 0 @@ -567,6 +620,34 @@ def resolve_billing_route( provider_name = inferred_provider model = bare_model + # Notional pricing for local subscription proxies/bridges that front the + # Anthropic API (Claude Code OAuth billing, tailnet failovers, etc.). The + # marginal cash cost is $0 (covered by a flat subscription), but for fleet + # cost *visibility* we price these at official Anthropic API rates and label + # the result "estimated" so /cost cards, top, and session rollups carry + # meaningful numbers. See NOTIONAL_ANTHROPIC_PROVIDERS. + if is_notional_anthropic_provider(provider_name): + return BillingRoute( + provider="anthropic", + model=model.split("/")[-1], + base_url=base_url or "", + billing_mode="official_docs_snapshot", + ) + + # Notional pricing for ChatGPT-subscription Codex routes. Marginal cash cost + # is $0, but for cost *visibility* we resolve to the underlying OpenAI model + # and price it from the live OpenRouter catalog (status "estimated"). The + # OpenRouter catalog lists base + dated families (gpt-5.5, gpt-5.3-codex, + # ...) but not every "-codex" variant (e.g. gpt-5.5-codex is absent while + # gpt-5.5 is present), so _normalize_codex_model_name() strips a trailing + # "-codex" as a fallback when the exact id is missing. + if provider_name in NOTIONAL_OPENROUTER_PROVIDERS: + return BillingRoute( + provider="openrouter", + model=model.split("/")[-1], + base_url=base_url or "", + billing_mode="official_models_api", + ) if provider_name == "openai-codex": return BillingRoute(provider="openai-codex", model=model, base_url=base_url or "", billing_mode="subscription_included") if provider_name == "openrouter" or base_url_host_matches(base_url or "", "openrouter.ai"): @@ -601,7 +682,38 @@ def _normalize_anthropic_model_name(model: str) -> str: return name +# Trailing 8-digit release-date suffix on a NEW-scheme Anthropic model id, e.g. +# claude-haiku-4-5-20251001 → the dated alias of claude-haiku-4-5. Anchored to +# the end and require a preceding "-N" version segment so we only strip a date +# that was APPENDED to an already-versioned name — never the canonical OLD-scheme +# ids whose date IS part of the name (claude-3-5-haiku-20241022, where stripping +# would land on the non-existent claude-3-5-haiku). Used ONLY as a last-resort +# snapshot fallback, after direct + dot-normalized lookups, so a real dated entry +# always wins. +_ANTHROPIC_DATED_SUFFIX_RE = re.compile(r"^(claude-.*-\d+-\d+)-\d{8}$") + + +def _strip_anthropic_release_date(name: str) -> Optional[str]: + """Strip a trailing -YYYYMMDD appended to a versioned new-scheme name. + + claude-haiku-4-5-20251001 → claude-haiku-4-5 ; claude-opus-4-8-20260115 → + claude-opus-4-8. Returns None when there is no such suffix to strip (incl. + the old-scheme claude-3-5-haiku-20241022, which lacks the -N-N version tail + before the date and so is left intact for its own direct entry). + """ + m = _ANTHROPIC_DATED_SUFFIX_RE.match(name) + return m.group(1) if m else None + + def _lookup_official_docs_pricing(route: BillingRoute) -> Optional[PricingEntry]: + """Resolve a route to a static official-docs pricing entry, most-specific first. + + Lookup precedence (first hit wins, so a more-specific entry never loses to a + fallback): exact ``(provider, model)`` → for Anthropic, the dot-normalized + name (e.g. ``opus-4.7`` → ``opus-4-7``) → then the date-stripped base of a + versioned new-scheme id (``claude-haiku-4-5-20251001`` → ``claude-haiku-4-5``). + Returns None when no entry matches at any tier. + """ model = route.model.lower() # Direct lookup first entry = _OFFICIAL_DOCS_PRICING.get((route.provider, model)) @@ -614,16 +726,53 @@ def _lookup_official_docs_pricing(route: BillingRoute) -> Optional[PricingEntry] entry = _OFFICIAL_DOCS_PRICING.get((route.provider, normalized)) if entry: return entry + # Last-resort: strip a trailing -YYYYMMDD release date appended to a + # versioned new-scheme id and retry on the base (claude-haiku-4-5- + # 20251001 → claude-haiku-4-5). Runs AFTER direct + dot-normalized so a + # real dated entry (e.g. claude-3-5-haiku-20241022) always wins on its + # own key. Fixes the dated-Haiku unpriced gap (audit 2026-06-17). + base = _strip_anthropic_release_date(normalized) + if base and base != normalized: + entry = _OFFICIAL_DOCS_PRICING.get((route.provider, base)) + if entry: + return entry + return None + + +def _normalize_codex_model_name(model: str) -> Optional[str]: + """Map a Codex model id to its underlying OpenAI catalog id when they differ. + + OpenRouter lists base/dated OpenAI families (gpt-5.5, gpt-5.3-codex, ...) but + not every "-codex" variant. When an exact id is missing we strip a trailing + "-codex" segment so e.g. gpt-5.5-codex falls back to gpt-5.5. Returns None if + no normalization applies (i.e. the name has no "-codex" suffix to strip). + """ + name = model.lower().strip() + if name.endswith("-codex"): + return name[: -len("-codex")] return None def _openrouter_pricing_entry(route: BillingRoute) -> Optional[PricingEntry]: - return _pricing_entry_from_metadata( - fetch_model_metadata(), + metadata = fetch_model_metadata() + entry = _pricing_entry_from_metadata( + metadata, route.model, source_url="https://openrouter.ai/docs/api/api-reference/models/get-models", pricing_version="openrouter-models-api", ) + if entry is not None: + return entry + # Fallback: a "-codex" variant absent from the catalog → price the base model. + fallback = _normalize_codex_model_name(route.model) + if fallback: + return _pricing_entry_from_metadata( + metadata, + fallback, + source_url="https://openrouter.ai/docs/api/api-reference/models/get-models", + pricing_version="openrouter-models-api", + ) + return None def _pricing_entry_from_metadata( @@ -813,12 +962,24 @@ def estimate_usage_cost( ) if usage.cache_write_tokens: if entry.cache_write_cost_per_million is None: - return CostResult( - amount_usd=None, - status="unknown", - source=entry.source, - label="n/a", - notes=("cache-write pricing unavailable for route",), + # No published cache-write rate. For OpenAI-family routes (and any + # provider that doesn't charge a separate cache-write premium) the + # live models API omits this field by design — cache-write tokens + # are just input tokens that were also written to cache and are + # billed at the regular input rate. So if we DO know the input + # rate, price cache-write at the input rate rather than dropping + # the entire turn as unpriced (which silently loses real spend). + # Only bail when input pricing is ALSO missing (truly unpriceable). + if entry.input_cost_per_million is None: + return CostResult( + amount_usd=None, + status="unknown", + source=entry.source, + label="n/a", + notes=("cache-write pricing unavailable for route",), + ) + notes.append( + "cache-write priced at input rate (no separate cache-write rate published)" ) if entry.input_cost_per_million is not None: @@ -829,6 +990,10 @@ def estimate_usage_cost( amount += Decimal(usage.cache_read_tokens) * entry.cache_read_cost_per_million / _ONE_MILLION if entry.cache_write_cost_per_million is not None: amount += Decimal(usage.cache_write_tokens) * entry.cache_write_cost_per_million / _ONE_MILLION + elif usage.cache_write_tokens and entry.input_cost_per_million is not None: + # Fallback: no published cache-write rate → bill at the input rate + # (see the cache-write guard above). Correct for OpenAI-family routes. + amount += Decimal(usage.cache_write_tokens) * entry.input_cost_per_million / _ONE_MILLION if entry.request_cost is not None and usage.request_count: amount += Decimal(usage.request_count) * entry.request_cost diff --git a/cli.py b/cli.py index bc4f4a76befb4..604f1323f4cd5 100644 --- a/cli.py +++ b/cli.py @@ -6157,89 +6157,76 @@ def retry_last(self): return last_message def undo_last(self, n: int = 1, prefill: bool = True): - """Back up N user turns: truncate history, soft-delete on disk, prefill. - - Walks backwards N user messages and discards everything from the - Nth-from-last user message onward (its assistant response, tool - calls, etc.). ``n`` defaults to 1 (the last exchange); ``/undo 3`` - backs up three user turns. If ``n`` exceeds the number of user - turns, it backs up to the oldest one. - - Beyond the in-memory ``conversation_history`` slice, this also: - • soft-deletes the truncated rows in SessionDB (``active=0``) so - they're hidden from re-prompts and search but kept for audit; - • notifies memory providers via ``on_session_switch(rewound=True)``; - • mirrors /branch's agent surgery (system-prompt invalidation + - flush-index reset); - • when ``prefill`` is set and an input buffer is available, - pre-fills the composer with the backed-up message text so it - can be edited and resubmitted. - - ``prefill=False`` is used by callers that drive the undo - programmatically (e.g. checkpoint rollback) and don't want to - touch the user's input buffer. - """ - if not self.conversation_history: - print("(._.) No messages to undo.") + """Undo N half-turns via the shared undo core and render its prefill.""" + if not self.session_id: + print("(._.) No active session to undo.") return + try: + import hermes_undo - if n < 1: - n = 1 + if self._session_db is not None: + hermes_undo._session_db = self._session_db + result = hermes_undo.undo(self.session_id, n) + except Exception as e: + logger.debug("undo: failed: %s", e) + print(f"(._.) Undo failed: {e}") + return - # Walk backwards collecting the indices of the last N user messages. - user_indices = [] - for i in range(len(self.conversation_history) - 1, -1, -1): - if self.conversation_history[i].get("role") == "user": - user_indices.append(i) - if len(user_indices) >= n: - break + rewound_ids = list(result.get("rewound_ids") or []) + if not rewound_ids: + print("(._.) Nothing to undo.") + return - if not user_indices: - print("(._.) No user message found to undo.") + self._reload_active_history_after_rewind(rewound=True) + count = len(rewound_ids) + prefill_text = result.get("prefill_text") + print(f"(^_^)b Undid {n} half-turn(s) ({count} message(s)).") + print(f" {len(self.conversation_history)} message(s) remaining in history.") + if prefill and isinstance(prefill_text, str): + self._prefill_input_buffer(prefill_text) + + def redo_last(self, n: int = 1): + """Redo N undo operations via the shared undo core.""" + if not self.session_id: + print("(._.) No active session to redo.") return + try: + import hermes_undo - # The oldest of the collected user messages is our truncation point. - cut_idx = user_indices[-1] - turns_undone = len(user_indices) + if self._session_db is not None: + hermes_undo._session_db = self._session_db + result = hermes_undo.redo(self.session_id, n) + except Exception as e: + logger.debug("redo: failed: %s", e) + print(f"(._.) Redo failed: {e}") + return - removed_count = len(self.conversation_history) - cut_idx - removed_msg = self.conversation_history[cut_idx].get("content", "") - removed_text = self._undo_content_to_text(removed_msg) + reactivated = int(result.get("reactivated_count") or 0) + if reactivated <= 0: + print(f"(._.) {result.get('message') or 'Nothing to redo.'}") + return - # Truncate the in-memory history to before that user message. - self.conversation_history = self.conversation_history[:cut_idx] + self._reload_active_history_after_rewind(rewound=True) + print(f"(^_^)b Redid {n} undo operation(s) ({reactivated} message(s) restored).") + tail = self.conversation_history[-1] if self.conversation_history else None + if tail: + role = tail.get("role", "message") + content = tail.get("content") + if isinstance(content, str) and content: + preview = content[:60] + ("..." if len(content) > 60 else "") + print(f" Restored tail ({role}): \"{preview}\"") + else: + print(f" Restored tail: {role} turn.") - # Soft-delete the truncated rows on disk so re-prompts and search - # see the clean transcript while the rows survive for audit. - rewound_rows = 0 + def _reload_active_history_after_rewind(self, *, rewound: bool = False) -> None: if self._session_db is not None and self.session_id: try: - recents = self._session_db.list_recent_user_messages( - self.session_id, limit=max(turns_undone, 10) + self.conversation_history = self._session_db.get_messages_as_conversation( + self.session_id ) - if recents: - target_idx = min(turns_undone - 1, len(recents) - 1) - target_id = recents[target_idx]["id"] - result = self._session_db.rewind_to_message( - self.session_id, target_id - ) - rewound_rows = result.get("rewound_count", 0) - # Prefer the DB's decoded target text for the prefill — - # it's the canonical persisted copy. - db_text = self._undo_content_to_text( - (result.get("target_message") or {}).get("content") - ) - if db_text: - removed_text = db_text - except ValueError as e: - # Non-user target / cross-session — keep the in-memory undo - # but skip the soft-delete; surface a debug-level note. - logger.debug("undo: soft-delete skipped: %s", e) except Exception as e: - logger.debug("undo: soft-delete failed: %s", e) + logger.debug("rewind: active history reload failed: %s", e) - # Agent surgery: invalidate the system-prompt cache and reset the - # flush index so the next turn re-flushes from the truncated head. if self.agent is not None: if hasattr(self.agent, "_invalidate_system_prompt"): try: @@ -6251,8 +6238,6 @@ def undo_last(self, n: int = 1, prefill: bool = True): self.agent._last_flushed_db_idx = len(self.conversation_history) except Exception: pass - # Notify memory providers — same hook /branch fires, with the - # rewound flag so per-turn document caches invalidate (#6672, #21910). try: _mm = getattr(self.agent, "_memory_manager", None) if _mm is not None and self.session_id: @@ -6260,25 +6245,11 @@ def undo_last(self, n: int = 1, prefill: bool = True): self.session_id, parent_session_id="", reset=False, - rewound=True, + rewound=rewound, ) except Exception: pass - turn_word = "turn" if turns_undone == 1 else "turns" - msg_count = rewound_rows or removed_count - print( - f"(^_^)b Undid {turns_undone} {turn_word} ({msg_count} message(s)). " - f"Backed up to: \"{removed_text[:60]}{'...' if len(removed_text) > 60 else ''}\"" - ) - remaining = len(self.conversation_history) - print(f" {remaining} message(s) remaining in history.") - - # Pre-fill the composer with the backed-up message so the user can - # edit and resubmit (Claude-Code-style). Editable, not auto-sent. - if prefill and removed_text: - self._prefill_input_buffer(removed_text) - @staticmethod def _undo_content_to_text(content) -> str: """Flatten message content (str or content-part list) to plain text.""" @@ -7440,7 +7411,7 @@ def process_command(self, command: str) -> bool: # Re-queue the message so process_loop sends it to the agent self._pending_input.put(retry_msg) elif canonical == "undo": - # Parse optional turn count: "/undo" → 1, "/undo 3" → 3. + # Parse optional half-turn count: "/undo" → 1, "/undo 3" → 3. _undo_n = 1 _undo_parts = cmd_original.split() if len(_undo_parts) > 1: @@ -7452,9 +7423,9 @@ def process_command(self, command: str) -> bool: if _undo_n < 1: _undo_n = 1 _undo_desc = ( - "This removes the last user/assistant exchange from history." + "This backs up the last half-turn in history." if _undo_n == 1 - else f"This removes the last {_undo_n} user turns from history." + else f"This backs up the last {_undo_n} half-turns from history." ) if self._confirm_destructive_slash( "undo", @@ -7463,6 +7434,20 @@ def process_command(self, command: str) -> bool: ) is None: return True # confirmation cancelled — command handled, keep REPL alive self.undo_last(_undo_n) + elif canonical == "redo": + _redo_n = 1 + _redo_parts = cmd_original.split() + if len(_redo_parts) > 1: + try: + _redo_n = int(_redo_parts[1]) + except ValueError: + print(f"(._.) Invalid count {_redo_parts[1]!r} — use /redo or /redo N.") + return + if _redo_n < 1: + # Match /undo: a non-positive count clamps to 1 rather than + # falling through to a misleading "nothing to redo". + _redo_n = 1 + self.redo_last(_redo_n) elif canonical == "branch": self._handle_branch_command(cmd_original) elif canonical == "save": @@ -7765,15 +7750,30 @@ def process_command(self, command: str) -> bool: if len(matches) > 1: # Prefer an exact match (typed the full command name) exact = [c for c in matches if c == typed_base] + builtin_matches = [c for c in matches if c in COMMANDS] + skill_matches = [c for c in matches if c not in COMMANDS] if len(exact) == 1: matches = exact - else: - # Prefer the unique shortest match: - # /qui → /quit (5) wins over /quint-pipeline (15) - min_len = min(len(c) for c in matches) - shortest = [c for c in matches if len(c) == min_len] - if len(shortest) == 1: - matches = shortest + elif len(builtin_matches) == 1: + # A single built-in command is uniquely identified — prefer it + # over any longer skill/bundle names that share the prefix: + # /qui → /quit wins over the /quint-pipeline skill; + # /con → /config. + matches = builtin_matches + elif len(builtin_matches) > 1: + # Multiple built-ins share the prefix. Resolve to the shortest + # ONLY when it is itself a prefix of every other match — i.e. + # the others are extensions of one base command + # (/sta → /status, with /statusbar an extension). When the + # matches are unrelated siblings (/re → /redo, /reset, /retry) + # there is no such base, so we stay ambiguous instead of + # silently picking the shortest by length alone. + shortest = min(builtin_matches, key=len) + if all(c.startswith(shortest) for c in builtin_matches): + matches = [shortest] + elif not builtin_matches and len(skill_matches) == 1: + # No built-in matched but exactly one skill/bundle did. + matches = skill_matches if len(matches) == 1: # Expand the prefix to the full command name, preserving arguments. # Guard against redispatching the same token to avoid infinite @@ -10109,6 +10109,13 @@ def chat(self, message, images: list = None) -> Optional[str]: # Add user message to history self.conversation_history.append({"role": "user", "content": message}) + if self.session_id: + try: + from hermes_undo import on_user_message_appended + + on_user_message_appended(self.session_id) + except Exception as e: + logger.debug("redo clear on user append failed: %s", e) ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]") print(flush=True) diff --git a/cron/scheduler.py b/cron/scheduler.py index 3590699661950..28a46b95edb0a 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -111,6 +111,77 @@ def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None: ) return None +def _filter_fallback_chain_for_pinned_job( + job: dict, fallback_model, job_id: str +): + """Restrict a cron job's fallback chain to its pinned provider. + + Root cause of the recurring "Model fallback detected" alerts (morning-digest + et al.): a cron job pins ``provider=openai-codex`` and STARTS on Codex + correctly, but the GLOBAL ``config.yaml`` fallback chain — whose first entry + is Opus — gets handed to every job. When Codex hiccups mid-run (rate-limit, + empty-TTFB watchdog) the agent walks that chain and silently finishes the + "codex-only" job on Opus. + + Fix: if a job pins a provider, drop every fallback entry whose provider + doesn't match. A Codex-pinned job then either runs on Codex or FAILS LOUDLY + (→ the job error surfaces as an alert) — it can never silently become Opus. + + Jobs that don't pin a provider are unchanged (full global chain preserved). + Honoured by both the init-time fallback (agent_init) and the mid-run + fallback chain, since both consume this same list. + + Override: set ``HERMES_CRON_ALLOW_CROSS_PROVIDER_FALLBACK=1`` to restore the + old permissive behavior (revert switch Ace asked for). + """ + if os.getenv("HERMES_CRON_ALLOW_CROSS_PROVIDER_FALLBACK", "").strip() in ("1", "true", "yes"): + return fallback_model + + pinned = str(job.get("provider") or "").strip().lower() + if not pinned or not fallback_model: + return fallback_model + + chain = fallback_model if isinstance(fallback_model, list) else [fallback_model] + filtered = [ + f for f in chain + if isinstance(f, dict) + and str(f.get("provider") or "").strip().lower() == pinned + ] + dropped = len(chain) - len(filtered) + if dropped: + logger.warning( + "Job '%s': pinned provider '%s' — dropped %d cross-provider fallback " + "entr%s to prevent silent off-provider fallback (e.g. codex→opus). " + "Set HERMES_CRON_ALLOW_CROSS_PROVIDER_FALLBACK=1 to revert.", + job_id, pinned, dropped, "y" if dropped == 1 else "ies", + ) + return filtered or None + + +def _resolve_job_fallback_chain(job: dict, global_chain, job_id: str): + """Resolve the effective fallback chain for a cron job. + + Precedence: + 1. If the job declares its OWN ``fallback`` (list or single dict), use that + as the base chain instead of the global config.yaml chain. This lets a + job add a *same-provider* fallback (e.g. codex/gpt-5.5 → codex/gpt-5.4) + for resilience against a single-model backend hiccup, which the global + chain can't provide (its only same-provider entry is the model the job + already runs). + 2. Otherwise fall back to the global chain. + + Either way the result passes through ``_filter_fallback_chain_for_pinned_job`` + so a provider-pinned job can NEVER cross providers — a per-job chain cannot be + used as a codex→opus escape hatch. The pin-filter's revert env var still + applies. + """ + job_fb = job.get("fallback") + if isinstance(job_fb, dict): + job_fb = [job_fb] + base = job_fb if job_fb else global_chain + return _filter_fallback_chain_for_pinned_job(job, base, job_id) + + # Valid delivery platforms — used to validate user-supplied platform names # in cron delivery targets, preventing env var enumeration via crafted names. _KNOWN_DELIVERY_PLATFORMS = frozenset({ @@ -149,7 +220,7 @@ def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None: "QQBOT_HOME_CHANNEL": "QQ_HOME_CHANNEL", } -from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run +from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run, get_job # Sentinel: when a cron agent has nothing new to report, it can start its # response with this marker to suppress delivery. Output is still saved @@ -656,7 +727,7 @@ def _send_media_via_adapter( logger.warning("Job '%s': failed to send media %s: %s", job.get("id", "?"), media_path, e) -def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]: +def _deliver_result(job: dict, content: str, success: bool = True, adapters=None, loop=None) -> Optional[str]: """ Deliver job output to the configured target(s) (origin chat, specific platform, etc.). @@ -665,6 +736,9 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option the standalone HTTP path cannot encrypt. Falls back to standalone send if the adapter path fails or is unavailable. + ``success`` selects the framing of the wrapped delivery (clean ✅ header for + successful runs, ⚠️ failure header carrying the error as the body). + Returns None on success, or an error string on failure. """ targets = _resolve_delivery_targets(job) @@ -691,13 +765,28 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option if wrap_response: task_name = job.get("name", job["id"]) job_id = job.get("id", "") - delivery_content = ( - f"Cronjob Response: {task_name}\n" - f"(job_id: {job_id})\n" - f"-------------\n\n" - f"{content}\n\n" - f"To stop or manage this job, send me a new message (e.g. \"stop reminder {task_name}\")." + manage_hint = ( + f'To stop or manage this job, send me a new message ' + f'(e.g. "stop reminder {task_name}").' ) + if success: + delivery_content = ( + f"✅ Cronjob Response: {task_name}\n" + f"🪪 job_id: {job_id}\n" + f"-------------\n\n" + f"{content}\n\n" + f"{manage_hint}" + ) + else: + # Failure framing: a single ⚠️ header carrying the error as the body, + # instead of nesting a "Cron job failed" line inside the success header. + delivery_content = ( + f"⚠️ Cronjob Failed: {task_name}\n" + f"🪪 job_id: {job_id}\n" + f"-------------\n\n" + f"{content}\n\n" + f"{manage_hint}" + ) else: delivery_content = content @@ -1685,6 +1774,13 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: raise RuntimeError(message) from exc fallback_model = _cfg.get("fallback_providers") or _cfg.get("fallback_model") or None + # Hard rule: a cron job that PINS a provider must never silently fall back + # to a different provider (the recurring codex→opus alert). A job may also + # declare its own same-provider fallback (e.g. codex/gpt-5.5 → codex/gpt-5.4) + # for single-model-outage resilience; either way the chain is pin-filtered + # so a pinned job runs on its provider or fails loudly. Revert with + # HERMES_CRON_ALLOW_CROSS_PROVIDER_FALLBACK=1. + fallback_model = _resolve_job_fallback_chain(job, fallback_model, job_id) credential_pool = None runtime_provider = str(runtime.get("provider") or "").strip().lower() if runtime_provider: @@ -1755,7 +1851,31 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: session_id=_cron_session_id, session_db=_session_db, ) - + + # Per-job API retry override. By default the agent inherits the global + # agent.api_max_retries (config.yaml, default 3). Jobs that pin a flaky + # subscription backend (e.g. the openai-codex morning digest) can set a + # higher per-job count so transient backend hangs are retried more times + # on the requested model before the fallback chain swaps to another + # model. Combined with the fast-reconnect Codex TTFB watchdog, extra + # attempts are cheap (~40s each) and keep the job on its requested model + # in almost all cases. Only ever raises the floor; never lowers below 1. + _job_max_retries = job.get("api_max_retries") + if _job_max_retries is not None: + try: + _job_retries = max(int(_job_max_retries), 1) + _prev_retries = getattr(agent, "_api_max_retries", 3) + agent._api_max_retries = _job_retries # type: ignore[attr-defined] + logger.info( + "Job '%s': per-job api_max_retries override = %d (was %d)", + job_id, _job_retries, _prev_retries, + ) + except (TypeError, ValueError): + logger.warning( + "Job '%s': invalid api_max_retries=%r; using agent default", + job_id, _job_max_retries, + ) + # Run the agent with an *inactivity*-based timeout: the job can run # for hours if it's actively calling tools / receiving stream tokens, # but a hung API call or stuck tool with no activity for the configured @@ -1967,6 +2087,106 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: logger.debug("Job '%s': failed to reap stale auxiliary clients: %s", job_id, e) +def _process_one_job(job: dict, *, verbose: bool = True, adapters=None, loop=None) -> dict: + """Run one job end-to-end: execute → save → deliver → mark. + + Shared by the scheduler tick (parallel/sequential passes) and by + ``run_job_now`` (synchronous single-job execution for ``cron run --wait``). + Returns a structured result dict; callers that only need a truthiness + signal can read ``result["processed"]``. + """ + try: + success, output, final_response, error = run_job(job) + + output_file = save_job_output(job["id"], output) + if verbose: + logger.info("Output saved to: %s", output_file) + + # Deliver the final response to the origin/target chat. + # If the agent responded with [SILENT], skip delivery (but + # output is already saved above). Failed jobs always deliver. + # On failure, hand the raw error to _deliver_result and let the wrapper + # own the status framing (avoids the old double-wrap: a "Cron job failed" + # body nested inside a separate "Cronjob Response" header). + deliver_content = final_response if success else (error or "unknown error") + # Treat whitespace-only final responses the same as empty + # responses: do not deliver a blank message, and let the + # empty-response guard below mark the run as a soft failure. + should_deliver = bool(deliver_content.strip()) + if should_deliver and success and SILENT_MARKER in deliver_content.strip().upper(): + logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER) + should_deliver = False + + delivery_error = None + if should_deliver: + try: + delivery_error = _deliver_result(job, deliver_content, success=success, adapters=adapters, loop=loop) + except Exception as de: + delivery_error = str(de) + logger.error("Delivery failed for job %s: %s", job["id"], de) + + # Treat empty final_response as a soft failure so last_status + # is not "ok" — the agent ran but produced nothing useful. + # (issue #8585) + if success and not final_response.strip(): + success = False + error = "Agent completed but produced empty response (model error, timeout, or misconfiguration)" + + mark_job_run(job["id"], success, error, delivery_error=delivery_error) + return { + "processed": True, + "success": success, + "job_id": job["id"], + "final_response": final_response, + "error": error, + "delivery_error": delivery_error, + "output_file": output_file, + } + + except Exception as e: + logger.error("Error processing job %s: %s", job['id'], e) + mark_job_run(job["id"], False, str(e)) + return { + "processed": False, + "success": False, + "job_id": job["id"], + "final_response": "", + "error": str(e), + "delivery_error": None, + "output_file": None, + } + + +def run_job_now(job_id: str, *, verbose: bool = True, adapters=None, loop=None) -> dict: + """Run a single job synchronously to completion, in the calling thread. + + Unlike ``tick()``, this does NOT consult the due-list or advance + ``next_run_at`` — it runs the named job right now regardless of schedule, + using the exact same execute → save → deliver → mark pipeline. Backs the + ``hermes cron run --wait`` CLI flag so slow jobs can be verified to + completion in the foreground. + + Returns the structured result dict from ``_process_one_job``, or a + ``{"success": False, "error": "... not found"}`` dict if the id is unknown. + """ + job = get_job(job_id) + if not job: + return { + "processed": False, + "success": False, + "job_id": job_id, + "final_response": "", + "error": f"Job not found: {job_id}", + "delivery_error": None, + "output_file": None, + } + if verbose: + logger.info("Running job '%s' (%s) synchronously now", job.get("name", job_id), job_id) + # Preserve scheduler-scoped ContextVar state, mirroring the tick passes. + _ctx = contextvars.copy_context() + return _ctx.run(_process_one_job, job, verbose=verbose, adapters=adapters, loop=loop) + + def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> int: """ Check and run all due jobs. @@ -2046,47 +2266,9 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i def _process_job(job: dict) -> bool: """Run one due job end-to-end: execute, save, deliver, mark.""" - try: - success, output, final_response, error = run_job(job) - - output_file = save_job_output(job["id"], output) - if verbose: - logger.info("Output saved to: %s", output_file) - - # Deliver the final response to the origin/target chat. - # If the agent responded with [SILENT], skip delivery (but - # output is already saved above). Failed jobs always deliver. - deliver_content = final_response if success else f"⚠️ Cron job '{job.get('name', job['id'])}' failed:\n{error}" - # Treat whitespace-only final responses the same as empty - # responses: do not deliver a blank message, and let the - # empty-response guard below mark the run as a soft failure. - should_deliver = bool(deliver_content.strip()) - if should_deliver and success and SILENT_MARKER in deliver_content.strip().upper(): - logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER) - should_deliver = False - - delivery_error = None - if should_deliver: - try: - delivery_error = _deliver_result(job, deliver_content, adapters=adapters, loop=loop) - except Exception as de: - delivery_error = str(de) - logger.error("Delivery failed for job %s: %s", job["id"], de) - - # Treat empty final_response as a soft failure so last_status - # is not "ok" — the agent ran but produced nothing useful. - # (issue #8585) - if success and not final_response.strip(): - success = False - error = "Agent completed but produced empty response (model error, timeout, or misconfiguration)" - - mark_job_run(job["id"], success, error, delivery_error=delivery_error) - return True - - except Exception as e: - logger.error("Error processing job %s: %s", job['id'], e) - mark_job_run(job["id"], False, str(e)) - return False + return _process_one_job( + job, verbose=verbose, adapters=adapters, loop=loop, + )["processed"] # Partition due jobs: those with a per-job workdir mutate # os.environ["TERMINAL_CWD"] inside run_job, which is process-global — diff --git a/docs/PRD-8-aegis-lcm-store-reset.md b/docs/PRD-8-aegis-lcm-store-reset.md new file mode 100644 index 0000000000000..a1341c9d975b0 --- /dev/null +++ b/docs/PRD-8-aegis-lcm-store-reset.md @@ -0,0 +1,172 @@ +# PRD-8 — Aegis LCM Store Snapshot + Fresh-Reset for a Clean Phase-2 Benchmark + +**Status:** DRAFT (pre-review) +**Owner:** Apollo +**Blast radius:** Aegis (break-glass agent) gateway + its LCM store. NOT Apollo, NOT any other profile. +**Privilege:** Bouncing the Aegis gateway is a privileged action (SOUL §7). Requires explicit go. + +## 1. Problem + +The Phase-2 LCM benchmark (PRD-7, Arm A raw-store + Arm B DAG node-served) drives the **live +Aegis** profile and writes every planted sentinel into Aegis's real LCM store +`~/.hermes/profiles/aegis/lcm.db`. After ~dozens of dev/validation runs that store now holds: + +- **17,485 messages**, **296 MB**, **44 summary nodes** +- **1,421 sentinel-bearing rows** from PRIOR runs (`LCM-LIVE-RECOVERY-*`, `LCM-ARMB-*`, `recover-*`) + +A confirmed consequence (smoke `arm-a-smoke-n3-tight`, trial `exact-1729-000`): the model called +`lcm_grep`, but the polluted store returned **other runs' sentinels** and it answered with the wrong +one. An N=180 run against this store measures "find the right needle among 1,421 stale needles," not +"does LCM recover the planted fact." The benchmark result would be **contaminated and pessimistic** — +unacceptable for a gate that decides the privileged Apollo cutover. + +## 2. Goal + +Give the Phase-2 campaign a **clean LCM store** so each trial recovers only its own planted fact, +while **losing nothing** (full reversibility) and **not disturbing any non-Aegis profile**. + +Non-goal: changing the harness, the gate math, or the engine. This is store hygiene only. + +## 3. Approach (Option 1 — snapshot + fresh store) + +1. **Quiesce** the Aegis gateway (it holds `lcm.db` open with a 7.3 MB WAL). +2. **Checkpoint + snapshot** the current store to a timestamped backup (verifiable copy, incl. WAL). +3. **Move** the polluted db aside (do not delete) so Aegis recreates a fresh, empty store on start. +4. **Restart** Aegis; confirm a fresh `lcm.db` exists and is empty (0 messages / 0 nodes). +5. Campaign then runs against the clean store. + +Rejected alternatives (documented): (2) per-run `--lcm-db` throwaway — more code, deferred; (3) run +dirty + post-filter — leaves `lcm_grep` noise, gives a contaminated number. Both rejected for THIS +gate; (2) is a reasonable future enhancement. + +## 4. Detailed steps & commands + +All paths under `~/.hermes/profiles/aegis/`. `$TS` = `date +%Y%m%d_%H%M%S`. +`DB=~/.hermes/profiles/aegis/lcm.db`. + +### 4.0 Resolve + PIN the launchd label AND plist (config-drift defense) +- Resolve the real Aegis gateway label once: `launchctl list | grep -i 'hermes.*aegis'` → + capture the exact label into `$LABEL`. **Echo `$LABEL` into the report** — it is evidence, never + re-guessed on the next run. Abort if zero or >1 match (ambiguous → human). +- Resolve and PIN the **plist path** too: `launchctl print gui/$UID/$LABEL | grep -i 'path ='` (or + the known `~/Library/LaunchAgents/