cli: worktree lock + dirty-tree preservation — stop pruning uncommitted work

Three behavior changes to the hermes -w worktree lifecycle:

1. Git-native locks. _setup_worktree now locks its worktree
   (git worktree lock --reason "hermes session pid=<pid>"), and
   _prune_stale_worktrees skips locked worktrees at ANY age — a lock
   from a live or crashed session means "do not touch". New helpers
   _lock_worktree / _unlock_worktree / _worktree_is_locked (fail-safe:
   any error reads as locked) / _worktree_is_dirty (fail-safe: any
   error reads as dirty).

2. Dirty trees are preserved. _cleanup_worktree previously destroyed
   worktrees with uncommitted changes if there were no unpushed
   commits; it now keeps the worktree, branch, and lock when the tree
   is dirty OR has unpushed commits, and prints manual cleanup hints
   (git worktree unlock + remove --force). The >72h "force remove
   regardless" prune tier is removed: pruning may only ever delete
   clean, unlocked, fully-pushed worktrees.

3. Branch deletion is gated on removal success. Both cleanup and
   prune previously deleted the branch without checking the
   git worktree remove returncode, dropping easy reachability of the
   commits even when removal failed; the branch is now only deleted
   after a successful remove.

agent/account_usage.py

@@ -243,17 +243,6 @@ def nous_credits_lines(*, markdown: bool = False, timeout: float = 10.0) -> list
     renders from that fixture instead of the real portal (so the block + gauge are
     testable without a live account). Throwaway scaffolding.
     """
-    snapshot = _fetch_nous_credits_snapshot(timeout=timeout)
-    return render_account_usage_lines(snapshot, markdown=markdown)
-
-
-def _fetch_nous_credits_snapshot(timeout: float = 10.0) -> Optional[AccountUsageSnapshot]:
-    """Auth-gate + portal fetch + snapshot build for the Nous credits block.
-
-    Shared by ``nous_credits_lines`` (full block) and
-    ``nous_credits_compact_line`` (one-liner). Honors the
-    HERMES_DEV_CREDITS_FIXTURE dev override. Fail-open → None.
-    """
     # Dev fixture short-circuit — render /usage from the injected state, no portal.
     try:
         from agent.credits_tracker import dev_fixture_credits_state
@@ -262,16 +251,17 @@ def _fetch_nous_credits_snapshot(timeout: float = 10.0) -> Optional[AccountUsage
     except Exception:
         fixture = None
     if fixture is not None:
-        return _snapshot_from_credits_state(fixture)
+        snapshot = _snapshot_from_credits_state(fixture)
+        return render_account_usage_lines(snapshot, markdown=markdown)
 
     try:
         from hermes_cli.auth import get_provider_auth_state
 
         tok = (get_provider_auth_state("nous") or {}).get("access_token")
         if not (isinstance(tok, str) and tok.strip()):
-            return None
+            return []
     except Exception:
-        return None
+        return []
     try:
         import concurrent.futures
 
@@ -281,36 +271,13 @@ def _fetch_nous_credits_snapshot(timeout: float = 10.0) -> Optional[AccountUsage
             account = pool.submit(
                 get_nous_portal_account_info, force_fresh=True
             ).result(timeout=timeout)
-        return build_nous_credits_snapshot(account)
+        snapshot = build_nous_credits_snapshot(account)
+        return render_account_usage_lines(snapshot, markdown=markdown)
     except Exception:
         # Fail-open (caller shows nothing), but leave a breadcrumb so a dead
         # /usage credits block is diagnosable in agent.log without a dev flag.
         logger.debug("credits ▸ /usage portal fetch/render failed (fail-open)", exc_info=True)
-        return None
-
-
-def nous_credits_compact_line(*, timeout: float = 10.0) -> Optional[str]:
-    """One-line Nous credits summary for the compact /usage view, or None.
-
-    Condenses the snapshot's own detail strings (stable, locally-built
-    formats) into ``Nous credits (Plan): Total usable: $X · Renews: …``.
-    Same gating/fail-open semantics as ``nous_credits_lines``.
-    """
-    snap = _fetch_nous_credits_snapshot(timeout=timeout)
-    if snap is None or not snap.available:
-        return None
-    picked = [
-        d for d in snap.details
-        if d.startswith(("Total usable:", "Renews:", "Status:"))
-    ]
-    if not picked:
-        picked = [d for d in snap.details if not d.startswith("Manage / top up:")][:2]
-    if not picked:
-        return None
-    title = snap.title
-    if snap.plan:
-        title += f" ({snap.plan})"
-    return f"{title}: " + " · ".join(picked)
+        return []
 
 
 def _snapshot_from_credits_state(state) -> Optional[AccountUsageSnapshot]:

agent/agent_init.py

@@ -1598,12 +1598,6 @@ def init_agent(
     agent.session_cache_write_tokens = 0
     agent.session_reasoning_tokens = 0
     agent.session_estimated_cost_usd = 0.0
-    # Provider-REPORTED cost only (e.g. OpenRouter usage.cost). None means
-    # "nothing reported" — distinct from a real $0.00.
-    agent.session_actual_cost_usd = None
-    # Per-model session usage rows for /usage: {model: {calls, input, output,
-    # cache_read, cache_write, cost_usd|None}}.
-    agent.session_model_usage = {}
     agent.session_cost_status = "unknown"
     agent.session_cost_source = "none"
     

agent/auxiliary_client.py

@@ -3079,23 +3079,20 @@ def _try_configured_fallback_chain(
         if not fb_provider or fb_provider.lower() == skip:
             continue
         fb_model = str(entry.get("model", "")).strip() or None
-        fb_base_url = str(entry.get("base_url", "")).strip() or None
-        fb_api_key = str(entry.get("api_key", "")).strip() or None
 
         label = f"fallback_chain[{i}]({fb_provider})"
 
         try:
-            fb_client = _resolve_single_provider(
-                fb_provider, fb_model, fb_base_url, fb_api_key)
+            fb_client, resolved_model = _resolve_fallback_entry(entry)
         except Exception:
-            fb_client = None
+            fb_client, resolved_model = None, None
 
         if fb_client is not None:
             logger.info(
                 "Auxiliary %s: %s on %s — configured fallback to %s (%s)",
-                task, reason, failed_provider, label, fb_model or "default",
+                task, reason, failed_provider, label, resolved_model or fb_model or "default",
             )
-            return fb_client, fb_model, label
+            return fb_client, resolved_model or fb_model, label
         tried.append(label)
 
     if tried:
@@ -3106,6 +3103,103 @@ def _try_configured_fallback_chain(
     return None, None, ""
 
 
+def _fallback_entry_api_key(entry: Dict[str, Any]) -> Optional[str]:
+    """Resolve inline or env-backed API key from a fallback-chain entry."""
+    explicit = str(entry.get("api_key") or "").strip()
+    if explicit:
+        return explicit
+    key_env = str(entry.get("key_env") or entry.get("api_key_env") or "").strip()
+    if key_env:
+        return os.getenv(key_env, "").strip() or None
+    return None
+
+
+def _resolve_fallback_entry(entry: Dict[str, Any]) -> Tuple[Optional[Any], Optional[str]]:
+    """Resolve one fallback entry through the central provider router."""
+    provider = str(entry.get("provider") or "").strip()
+    model = str(entry.get("model") or "").strip() or None
+    if not provider or not model:
+        return None, None
+    base_url = str(entry.get("base_url") or "").strip() or None
+    api_key = _fallback_entry_api_key(entry)
+    api_mode = str(entry.get("api_mode") or entry.get("transport") or "").strip() or None
+    return resolve_provider_client(
+        provider,
+        model=model,
+        explicit_base_url=base_url,
+        explicit_api_key=api_key,
+        api_mode=api_mode,
+    )
+
+
+def _try_main_fallback_chain(
+    task: Optional[str],
+    failed_provider: str = "",
+    reason: str = "error",
+) -> Tuple[Optional[Any], Optional[str], str]:
+    """Try the top-level main-agent fallback chain for an auxiliary call.
+
+    ``provider: auto`` auxiliary tasks should respect the user's declared
+    main fallback policy before dropping into Hermes' built-in discovery
+    chain. The top-level chain is read through ``get_fallback_chain`` so
+    both modern ``fallback_providers`` and legacy ``fallback_model`` entries
+    participate in the same order as the main agent.
+    """
+    try:
+        from hermes_cli.config import load_config
+        from hermes_cli.fallback_config import get_fallback_chain
+
+        chain = get_fallback_chain(load_config())
+    except Exception as exc:
+        logger.debug("Auxiliary %s: could not load main fallback chain: %s", task or "call", exc)
+        return None, None, ""
+
+    if not chain:
+        return None, None, ""
+
+    failed_norm = (failed_provider or "").strip().lower()
+    main_norm = (_read_main_provider() or "").strip().lower()
+    skip = {p for p in (failed_norm, main_norm, "auto") if p}
+    tried: List[str] = []
+
+    for i, entry in enumerate(chain):
+        if not isinstance(entry, dict):
+            continue
+        fb_provider = str(entry.get("provider") or "").strip()
+        fb_model = str(entry.get("model") or "").strip()
+        if not fb_provider or not fb_model:
+            continue
+        fb_norm = fb_provider.lower()
+        label = f"fallback_providers[{i}]({fb_provider})"
+        if fb_norm in skip:
+            tried.append(f"{label} (skipped)")
+            continue
+        if _is_provider_unhealthy(fb_norm):
+            _log_skip_unhealthy(fb_norm, task)
+            tried.append(f"{label} (unhealthy)")
+            continue
+        try:
+            fb_client, resolved_model = _resolve_fallback_entry(entry)
+        except Exception as exc:
+            logger.debug("Auxiliary %s: main fallback %s failed to resolve: %s", task or "call", label, exc)
+            fb_client, resolved_model = None, None
+        if fb_client is not None:
+            logger.info(
+                "Auxiliary %s: %s on %s — main fallback chain to %s (%s)",
+                task or "call", reason, failed_provider or "auto", label,
+                resolved_model or fb_model,
+            )
+            return fb_client, resolved_model or fb_model, fb_provider
+        tried.append(label)
+
+    if tried:
+        logger.debug(
+            "Auxiliary %s: main fallback chain exhausted (tried: %s)",
+            task or "call", ", ".join(tried),
+        )
+    return None, None, ""
+
+
 def _resolve_single_provider(
     provider: str,
     model: Optional[str] = None,
@@ -3116,16 +3210,19 @@ def _resolve_single_provider(
 
     Uses the existing provider resolution infrastructure where possible.
     """
-    # Reuse resolve_provider_client which handles provider→client mapping
+    # Reuse resolve_provider_client which handles provider→client mapping.
     client, resolved_model = resolve_provider_client(
         provider=provider,
         model=model,
-        base_url=base_url,
-        api_key=api_key,
+        explicit_base_url=base_url,
+        explicit_api_key=api_key,
     )
     return client
 
-def _resolve_auto(main_runtime: Optional[Dict[str, Any]] = None) -> Tuple[Optional[OpenAI], Optional[str]]:
+def _resolve_auto(
+    main_runtime: Optional[Dict[str, Any]] = None,
+    task: Optional[str] = None,
+) -> Tuple[Optional[OpenAI], Optional[str]]:
     """Full auto-detection chain.
 
     Priority:
@@ -3223,7 +3320,22 @@ def _resolve_auto(main_runtime: Optional[Dict[str, Any]] = None) -> Tuple[Option
                             main_provider, resolved or main_model)
                 return client, resolved or main_model
 
-    # ── Step 2: aggregator / fallback chain ──────────────────────────────
+    # ── Step 2: user-configured fallback policy ─────────────────────────
+    # In auto mode, respect the task-specific fallback chain first, then the
+    # main agent's top-level fallback_providers/fallback_model chain. The
+    # hardcoded provider discovery chain below is only the convenience default
+    # for users who have not declared a fallback policy.
+    if task:
+        fb_client, fb_model, _fb_label = _try_configured_fallback_chain(
+            task, main_provider or "auto", reason="main provider unavailable")
+        if fb_client is not None:
+            return fb_client, fb_model
+    fb_client, fb_model, _fb_label = _try_main_fallback_chain(
+        task, main_provider or "auto", reason="main provider unavailable")
+    if fb_client is not None:
+        return fb_client, fb_model
+
+    # ── Step 3: aggregator / fallback chain ──────────────────────────────
     tried = []
     for label, try_fn in _get_provider_chain():
         if _is_provider_unhealthy(label):
@@ -3344,6 +3456,7 @@ def resolve_provider_client(
     api_mode: str = None,
     main_runtime: Optional[Dict[str, Any]] = None,
     is_vision: bool = False,
+    task: Optional[str] = None,
 ) -> Tuple[Optional[Any], Optional[str]]:
     """Central router: given a provider name and optional model, return a
     configured client with the correct auth, base URL, and API format.
@@ -3464,7 +3577,7 @@ def resolve_provider_client(
 
     # ── Auto: try all providers in priority order ────────────────────
     if provider == "auto":
-        client, resolved = _resolve_auto(main_runtime=main_runtime)
+        client, resolved = _resolve_auto(main_runtime=main_runtime, task=task)
         if client is None:
             return None, None
         # When auto-detection lands on a non-OpenRouter provider (e.g. a
@@ -4357,11 +4470,16 @@ def _client_cache_key(
     api_mode: Optional[str] = None,
     main_runtime: Optional[Dict[str, Any]] = None,
     is_vision: bool = False,
+    task: Optional[str] = None,
 ) -> tuple:
     runtime = _normalize_main_runtime(main_runtime)
     runtime_key = tuple(runtime.get(field, "") for field in _MAIN_RUNTIME_FIELDS) if provider == "auto" else ()
+    # `auto` can now resolve through task-specific or main fallback policy,
+    # so the task participates in the cache key. Non-auto providers keep the
+    # old cache shape because the explicit provider/model tuple is sufficient.
+    task_key = (task or "") if provider == "auto" else ""
     pool_hint = _pool_cache_hint(provider, main_runtime=main_runtime)
-    return (provider, async_mode, base_url or "", api_key or "", api_mode or "", runtime_key, is_vision, pool_hint)
+    return (provider, async_mode, base_url or "", api_key or "", api_mode or "", runtime_key, is_vision, task_key, pool_hint)
 
 
 def _store_cached_client(cache_key: tuple, client: Any, default_model: Optional[str], *, bound_loop: Any = None) -> None:
@@ -4554,6 +4672,7 @@ def _get_cached_client(
     api_mode: str = None,
     main_runtime: Optional[Dict[str, Any]] = None,
     is_vision: bool = False,
+    task: Optional[str] = None,
 ) -> Tuple[Optional[Any], Optional[str]]:
     """Get or create a cached client for the given provider.
 
@@ -4591,6 +4710,7 @@ def _get_cached_client(
         api_mode=api_mode,
         main_runtime=main_runtime,
         is_vision=is_vision,
+        task=task,
     )
     with _client_cache_lock:
         if cache_key in _client_cache:
@@ -4635,6 +4755,7 @@ def _get_cached_client(
         api_mode=api_mode,
         main_runtime=runtime,
         is_vision=is_vision,
+        task=task,
     )
     if client is not None:
         # For async clients, remember which loop they were created on so we
@@ -5140,7 +5261,7 @@ def call_llm(
             if not resolved_base_url:
                 logger.info("Auxiliary %s: provider %s unavailable, trying auto-detection chain",
                             task or "call", resolved_provider)
-                client, final_model = _get_cached_client("auto", main_runtime=main_runtime)
+                client, final_model = _get_cached_client("auto", main_runtime=main_runtime, task=task)
         if client is None:
             raise RuntimeError(
                 f"No LLM provider configured for task={task} provider={resolved_provider}. "
@@ -5466,14 +5587,19 @@ def call_llm(
 
             # Fallback order (#26882, #26803):
             #   1. User-configured fallback_chain (per-task) if set
-            #   2. Main agent model (last-resort safety net)
-            # For auto users (no explicit aux provider), use the full
-            # auto-detection chain instead — its Step 1 IS the main agent
-            # model, so users on `auto` already get main-model fallback.
+            #   2. For auto: top-level main fallback_providers/fallback_model
+            #   3. For auto: built-in auxiliary discovery chain
+            #   4. For explicit aux providers: main agent model safety net
             fb_client, fb_model, fb_label = (None, None, "")
             if is_auto:
-                fb_client, fb_model, fb_label = _try_payment_fallback(
-                    resolved_provider, task, reason=reason)
+                fb_client, fb_model, fb_label = _try_configured_fallback_chain(
+                    task, resolved_provider or "auto", reason=reason)
+                if fb_client is None:
+                    fb_client, fb_model, fb_label = _try_main_fallback_chain(
+                        task, resolved_provider or "auto", reason=reason)
+                if fb_client is None:
+                    fb_client, fb_model, fb_label = _try_payment_fallback(
+                        resolved_provider, task, reason=reason)
             else:
                 fb_client, fb_model, fb_label = _try_configured_fallback_chain(
                     task, resolved_provider or "auto", reason=reason)
@@ -5636,7 +5762,7 @@ async def async_call_llm(
             if not resolved_base_url:
                 logger.info("Auxiliary %s: provider %s unavailable, trying auto-detection chain",
                             task or "call", resolved_provider)
-                client, final_model = _get_cached_client("auto", async_mode=True)
+                client, final_model = _get_cached_client("auto", async_mode=True, main_runtime=main_runtime, task=task)
         if client is None:
             raise RuntimeError(
                 f"No LLM provider configured for task={task} provider={resolved_provider}. "
@@ -5904,13 +6030,19 @@ async def async_call_llm(
 
             # Fallback order (#26882, #26803):
             #   1. User-configured fallback_chain (per-task) if set
-            #   2. Main agent model (last-resort safety net)
-            # Auto users get the full auto-detection chain instead — its
-            # Step 1 IS the main agent model.
+            #   2. For auto: top-level main fallback_providers/fallback_model
+            #   3. For auto: built-in auxiliary discovery chain
+            #   4. For explicit aux providers: main agent model safety net
             fb_client, fb_model, fb_label = (None, None, "")
             if is_auto:
-                fb_client, fb_model, fb_label = _try_payment_fallback(
-                    resolved_provider, task, reason=reason)
+                fb_client, fb_model, fb_label = _try_configured_fallback_chain(
+                    task, resolved_provider or "auto", reason=reason)
+                if fb_client is None:
+                    fb_client, fb_model, fb_label = _try_main_fallback_chain(
+                        task, resolved_provider or "auto", reason=reason)
+                if fb_client is None:
+                    fb_client, fb_model, fb_label = _try_payment_fallback(
+                        resolved_provider, task, reason=reason)
             else:
                 fb_client, fb_model, fb_label = _try_configured_fallback_chain(
                     task, resolved_provider or "auto", reason=reason)

agent/conversation_loop.py

@@ -57,11 +57,7 @@ from agent.process_bootstrap import _install_safe_stdio
 from agent.prompt_caching import apply_anthropic_cache_control
 from agent.retry_utils import jittered_backoff
 from agent.trajectory import has_incomplete_scratchpad
-from agent.usage_pricing import (
-    estimate_usage_cost,
-    extract_provider_cost_usd,
-    normalize_usage,
-)
+from agent.usage_pricing import estimate_usage_cost, normalize_usage
 from hermes_constants import PARTIAL_STREAM_STUB_ID
 from hermes_logging import set_session_context
 from tools.skill_provenance import set_current_write_origin
@@ -1809,37 +1805,6 @@ def run_conversation(
                     agent.session_cost_status = cost_result.status
                     agent.session_cost_source = cost_result.source
 
-                    # ── Real provider-REPORTED cost (never estimated) ──
-                    # OpenRouter usage accounting returns ``usage.cost`` on the
-                    # response when the request carries usage:{include:true}
-                    # (added on OpenRouter routes). When the provider reports
-                    # nothing, this stays None — absent, NOT zero — so cost
-                    # displays hide instead of showing a fabricated $0.00.
-                    reported_cost_usd = extract_provider_cost_usd(response.usage)
-                    if reported_cost_usd is not None:
-                        _prev_actual = getattr(agent, "session_actual_cost_usd", None)
-                        agent.session_actual_cost_usd = (_prev_actual or 0.0) + reported_cost_usd
-                        agent.session_cost_status = "actual"
-                        agent.session_cost_source = "provider_cost_api"
-
-                    # Per-model session breakdown for /usage — counts are always
-                    # real; cost_usd only accumulates provider-reported values
-                    # and stays None when the provider reports nothing.
-                    _model_usage = getattr(agent, "session_model_usage", None)
-                    if _model_usage is None:
-                        _model_usage = agent.session_model_usage = {}
-                    _mrow = _model_usage.setdefault(agent.model, {
-                        "calls": 0, "input": 0, "output": 0,
-                        "cache_read": 0, "cache_write": 0, "cost_usd": None,
-                    })
-                    _mrow["calls"] += 1
-                    _mrow["input"] += canonical_usage.input_tokens
-                    _mrow["output"] += canonical_usage.output_tokens
-                    _mrow["cache_read"] += canonical_usage.cache_read_tokens
-                    _mrow["cache_write"] += canonical_usage.cache_write_tokens
-                    if reported_cost_usd is not None:
-                        _mrow["cost_usd"] = (_mrow["cost_usd"] or 0.0) + reported_cost_usd
-
                     # Persist token counts to session DB for /insights.
                     # Do this for every platform with a session_id so non-CLI
                     # sessions (gateway, cron, delegated runs) cannot lose
@@ -1866,14 +1831,8 @@ def run_conversation(
                                 reasoning_tokens=canonical_usage.reasoning_tokens,
                                 estimated_cost_usd=float(cost_result.amount_usd)
                                 if cost_result.amount_usd is not None else None,
-                                # Provider-reported per-call cost delta. NULL
-                                # (not 0) when the provider reported nothing —
-                                # the SQL CASE keeps actual_cost_usd untouched.
-                                actual_cost_usd=reported_cost_usd,
-                                cost_status="actual"
-                                if reported_cost_usd is not None else cost_result.status,
-                                cost_source="provider_cost_api"
-                                if reported_cost_usd is not None else cost_result.source,
+                                cost_status=cost_result.status,
+                                cost_source=cost_result.source,
                                 billing_provider=agent.provider,
                                 billing_base_url=agent.base_url,
                                 billing_mode="subscription_included"

agent/transports/chat_completions.py

@@ -388,13 +388,6 @@ class ChatCompletionsTransport(ProviderTransport):
         if provider_prefs and is_openrouter:
             extra_body["provider"] = provider_prefs
 
-        # OpenRouter usage accounting — response `usage.cost` carries the REAL
-        # charged cost (credits are 1:1 USD). Parity with the profile path in
-        # plugins/model-providers/openrouter/__init__.py; this branch only runs
-        # when the OpenRouter profile isn't loaded.
-        if is_openrouter:
-            extra_body["usage"] = {"include": True}
-
         # Pareto Code router plugin — model-gated. Same shape as the
         # profile path in plugins/model-providers/openrouter/__init__.py;
         # this branch only runs when the OpenRouter profile isn't loaded.

agent/usage_pricing.py

@@ -852,100 +852,6 @@ def estimate_usage_cost(
     )
 
 
-def _finite_nonneg_number(value: Any) -> Optional[float]:
-    """Return ``value`` as a float when it is a real, finite, non-negative
-    number (int/float, not bool); otherwise None."""
-    if isinstance(value, bool) or not isinstance(value, (int, float)):
-        return None
-    try:
-        f = float(value)
-    except (TypeError, ValueError):
-        return None
-    if f != f or f in (float("inf"), float("-inf")) or f < 0:
-        return None
-    return f
-
-
-def extract_provider_cost_usd(response_usage: Any) -> Optional[float]:
-    """Provider-REPORTED cost (USD) from a response ``usage`` object, or None.
-
-    Reads the ``usage.cost`` field that OpenRouter's usage accounting returns
-    (``usage: {"include": true}`` request param; OpenRouter credits are 1:1
-    USD). OpenRouter-compatible aggregators use the same field. This NEVER
-    estimates: when the provider reports nothing, the result is None — callers
-    must treat None as "no cost data", not zero. A reported ``0`` is a real
-    zero (e.g. free-tier models) and is returned as ``0.0``.
-    """
-    if response_usage is None:
-        return None
-    cost = getattr(response_usage, "cost", None)
-    if cost is None and isinstance(response_usage, dict):
-        cost = response_usage.get("cost")
-    return _finite_nonneg_number(cost)
-
-
-def real_session_cost_usd(agent: Any) -> Optional[float]:
-    """Session-cumulative provider-REPORTED cost in USD, or None.
-
-    Combines the two real sources Hermes has — no estimation, ever:
-      - ``agent.session_actual_cost_usd``: per-response ``usage.cost``
-        accumulator (OpenRouter usage accounting).
-      - Nous ``x-nous-credits-*`` header delta via
-        ``agent.get_credits_spent_micros()`` (account-level spend since the
-        session first saw a header; clamped at 0 so a mid-session top-up
-        doesn't render a negative cost).
-
-    Returns None when neither source has reported anything — callers must
-    hide their cost display in that case rather than showing $0.00.
-    """
-    total: Optional[float] = None
-
-    actual = _finite_nonneg_number(getattr(agent, "session_actual_cost_usd", None))
-    if actual is not None:
-        total = actual
-
-    try:
-        spent_micros = agent.get_credits_spent_micros()
-    except Exception:
-        spent_micros = None
-    if spent_micros is not None:
-        try:
-            spent_usd = max(0, int(spent_micros)) / 1_000_000
-        except (TypeError, ValueError):
-            spent_usd = None
-        if spent_usd is not None:
-            total = (total or 0.0) + spent_usd
-
-    return total
-
-
-def nous_header_cost_usd(agent: Any) -> Optional[float]:
-    """Session-cumulative cost in USD derived ONLY from the Nous portal
-    ``x-nous-credits-*`` header delta, or None.
-
-    This is the STATUS-BAR cost source (glitch 2026-06-13, F3): the TUI chrome
-    must show cost ONLY when the session runs against the Nous portal, because
-    the header delta is the one figure we can trust without re-deriving per-model
-    cache/input/output pricing (which is unreliable across the model long tail).
-    Unlike :func:`real_session_cost_usd`, this DELIBERATELY ignores the
-    OpenRouter ``usage.cost`` accumulator — a non-Nous route reports no header,
-    so the chrome hides its cost segment entirely.
-
-    The ``/usage`` accounting page keeps using ``real_session_cost_usd`` (both
-    provider-reported sources); only the chrome bar narrows to header-only.
-    """
-    try:
-        spent_micros = agent.get_credits_spent_micros()
-    except Exception:
-        return None
-    if spent_micros is None:
-        return None
-    try:
-        return max(0, int(spent_micros)) / 1_000_000
-    except (TypeError, ValueError):
-        return None
-
-
 def has_known_pricing(
     model_name: str,
     provider: Optional[str] = None,

apps/desktop/src/app/session/hooks/use-prompt-actions.ts

@@ -58,6 +58,7 @@ import { clearSessionTodos } from '@/store/todos'
 
 import type {
   ClientSessionState,
+  BrowserManageResponse,
   FileAttachResponse,
   HandoffFailResponse,
   HandoffRequestResponse,
@@ -1141,6 +1142,81 @@ export function usePromptActions({
           } catch (err) {
             renderSlashOutput(`error: ${err instanceof Error ? err.message : String(err)}`)
           }
+        },
+        // /browser connect|disconnect|status manages the live CDP connection on
+        // the gateway host, mirroring the TUI's browser.manage RPC. It mutates
+        // BROWSER_CDP_URL (and may launch Chrome) in the gateway process — only
+        // meaningful when that process runs on this machine, so it's gated to
+        // local connections. A remote gateway would act on the wrong host.
+        browser: async ctx => {
+          const resolved = await withSlashOutput(ctx)
+
+          if (!resolved) {
+            return
+          }
+
+          const { render: renderSlashOutput, sessionId } = resolved
+
+          if ($connection.get()?.mode === 'remote') {
+            renderSlashOutput(
+              '/browser manages a Chromium-family browser on the gateway host — only available when connected to a local gateway.'
+            )
+
+            return
+          }
+
+          const [rawAction = 'status', ...rest] = ctx.arg.trim().split(/\s+/).filter(Boolean)
+          const cmdAction = rawAction.toLowerCase()
+
+          if (!['connect', 'disconnect', 'status'].includes(cmdAction)) {
+            renderSlashOutput(
+              'usage: /browser [connect|disconnect|status] [url] · persistent: set browser.cdp_url in config.yaml'
+            )
+
+            return
+          }
+
+          const url = cmdAction === 'connect' ? rest.join(' ').trim() || 'http://127.0.0.1:9222' : undefined
+
+          if (url) {
+            renderSlashOutput(`checking Chromium-family browser remote debugging at ${url}...`)
+          }
+
+          try {
+            const result = await requestGateway<BrowserManageResponse>('browser.manage', {
+              action: cmdAction,
+              session_id: sessionId,
+              ...(url && { url })
+            })
+
+            // Without a streamed session subscription, the gateway bundles its
+            // progress lines into `messages` — flush them inline.
+            result?.messages?.forEach(message => renderSlashOutput(message))
+
+            if (cmdAction === 'status') {
+              renderSlashOutput(
+                result?.connected
+                  ? `browser connected: ${result.url || '(url unavailable)'}`
+                  : 'browser not connected (try /browser connect <url> or set browser.cdp_url in config.yaml)'
+              )
+
+              return
+            }
+
+            if (cmdAction === 'disconnect') {
+              renderSlashOutput('browser disconnected')
+
+              return
+            }
+
+            if (result?.connected) {
+              renderSlashOutput('Browser connected to live Chromium-family browser via CDP')
+              renderSlashOutput(`Endpoint: ${result.url || '(url unavailable)'}`)
+              renderSlashOutput('next browser tool call will use this CDP endpoint')
+            }
+          } catch (err) {
+            renderSlashOutput(`error: ${err instanceof Error ? err.message : String(err)}`)
+          }
         }
       }
 

apps/desktop/src/app/types.ts

@@ -46,6 +46,12 @@ export interface SlashExecResponse {
   warning?: string
 }
 
+export interface BrowserManageResponse {
+  connected?: boolean
+  url?: string
+  messages?: string[]
+}
+
 export interface SessionSteerResponse {
   // 'queued' == accepted into the live turn's steer slot (injected at the next
   // tool-result boundary); 'rejected' == no live tool window, caller queues.

apps/desktop/src/lib/desktop-slash-commands.test.ts

@@ -52,6 +52,17 @@ describe('desktop slash command curation', () => {
     expect(desktopSlashUnavailableMessage('/personality')).toBeNull()
   })
 
+  it('treats /browser as an executable action command (local-gateway connect)', () => {
+    // /browser used to be terminal-only; it now resolves to a desktop action
+    // handler that routes browser.manage RPC when the gateway is local.
+    expect(isDesktopSlashCommand('/browser')).toBe(true)
+    expect(isDesktopSlashSuggestion('/browser')).toBe(true)
+    expect(desktopSlashUnavailableMessage('/browser')).toBeNull()
+    expect(resolveDesktopCommand('/browser')?.surface).toEqual({ kind: 'action', action: 'browser' })
+    // Bare /browser expands to its sub-action options in the popover.
+    expect(resolveDesktopCommand('/browser')?.args).toBe(true)
+  })
+
   it('allows aliases to execute without cluttering the popover', () => {
     expect(isDesktopSlashSuggestion('/reset')).toBe(false)
     expect(isDesktopSlashCommand('/reset')).toBe(true)

apps/desktop/src/lib/desktop-slash-commands.ts

@@ -30,6 +30,7 @@ export interface DesktopThemeCommandOption {
  */
 export type DesktopActionId =
   | 'branch'
+  | 'browser'
   | 'handoff'
   | 'help'
   | 'new'
@@ -103,6 +104,12 @@ const DESKTOP_COMMAND_SPECS: readonly DesktopCommandSpec[] = [
   { name: '/skin', description: 'Switch desktop theme or cycle to the next one', surface: action('skin'), args: true },
   { name: '/title', description: 'Rename the current session', surface: action('title') },
   { name: '/help', description: 'Show desktop slash commands', aliases: ['/commands'], surface: action('help') },
+  {
+    name: '/browser',
+    description: 'Manage browser CDP connection [connect|disconnect|status] (local gateway only)',
+    surface: action('browser'),
+    args: true
+  },
 
   // Overlay pickers
   { name: '/model', description: 'Switch the model for this session', surface: picker('model'), hidden: true },
@@ -142,7 +149,7 @@ const DESKTOP_COMMAND_SPECS: readonly DesktopCommandSpec[] = [
 // per reason beats 40 identical object literals.
 const NO_DESKTOP_SURFACE: Record<DesktopUnavailableReason, readonly string[]> = {
   terminal: [
-    '/browser', '/busy', '/clear', '/compact', '/config', '/copy', '/cron', '/details',
+    '/busy', '/clear', '/compact', '/config', '/copy', '/cron', '/details',
     '/exit', '/footer', '/gateway', '/gquota', '/history', '/image', '/indicator', '/logs',
     '/mouse', '/paste', '/platforms', '/plugins', '/quit', '/redraw', '/reload', '/restart',
     '/sb', '/set-home', '/sethome', '/snap', '/snapshot', '/statusbar', '/toolsets', '/update', '/verbose'

cli.py

@@ -1273,6 +1273,11 @@ def _setup_worktree(repo_root: str = None) -> Optional[Dict[str, str]]:
         print(f"\033[31m✗ Failed to create worktree: {e}\033[0m")
         return None
 
+    # Lock the worktree so concurrent/later hermes processes' pruning
+    # leaves this session's work alone (locks survive crashes too).
+    # Lock failure is non-fatal — _lock_worktree logs at debug level.
+    _lock_worktree(repo_root, str(wt_path))
+
     # Copy files listed in .worktreeinclude (gitignored files the agent needs)
     include_file = Path(repo_root) / ".worktreeinclude"
     if include_file.exists():
@@ -1383,13 +1388,109 @@ def _worktree_has_unpushed_commits(worktree_path: str, timeout: int = 10) -> boo
         return True
 
 
-def _cleanup_worktree(info: Dict[str, str] = None) -> None:
-    """Remove a worktree and its branch on exit.
+def _lock_worktree(repo_root: str, wt_path: str, timeout: int = 10) -> bool:
+    """Lock a worktree using git's native lock mechanism.
 
-    Preserves the worktree only if it has unpushed commits (real work
-    that hasn't been pushed to any remote).  Uncommitted changes alone
-    (untracked files, test artifacts) are not enough to keep it — agent
-    work lives in commits/PRs, not the working tree.
+    The lock marks the worktree as in-use by a live (or crashed) hermes
+    session so that other hermes processes' pruning leaves it alone.
+    Never raises; returns whether the lock was taken.
+    """
+    import subprocess
+
+    try:
+        result = subprocess.run(
+            ["git", "worktree", "lock",
+             "--reason", f"hermes session pid={os.getpid()}", str(wt_path)],
+            capture_output=True, text=True, timeout=timeout, cwd=repo_root,
+        )
+        if result.returncode != 0:
+            logger.debug(
+                "Failed to lock worktree %s: %s", wt_path, result.stderr.strip()
+            )
+            return False
+        return True
+    except Exception as e:
+        logger.debug("Failed to lock worktree %s: %s", wt_path, e)
+        return False
+
+
+def _unlock_worktree(repo_root: str, wt_path: str, timeout: int = 10) -> bool:
+    """Release a git worktree lock. Never raises."""
+    import subprocess
+
+    try:
+        result = subprocess.run(
+            ["git", "worktree", "unlock", str(wt_path)],
+            capture_output=True, text=True, timeout=timeout, cwd=repo_root,
+        )
+        if result.returncode != 0:
+            logger.debug(
+                "Failed to unlock worktree %s: %s", wt_path, result.stderr.strip()
+            )
+            return False
+        return True
+    except Exception as e:
+        logger.debug("Failed to unlock worktree %s: %s", wt_path, e)
+        return False
+
+
+def _worktree_is_locked(repo_root: str, wt_path: str, timeout: int = 10) -> bool:
+    """Return whether a worktree is locked (per ``git worktree list --porcelain``).
+
+    Fails SAFE: on any error (bad repo_root, git failure, timeout) returns
+    True so callers treat the worktree as in-use and do not delete it.
+    """
+    import subprocess
+
+    try:
+        result = subprocess.run(
+            ["git", "worktree", "list", "--porcelain"],
+            capture_output=True, text=True, timeout=timeout, cwd=repo_root,
+        )
+        if result.returncode != 0:
+            return True
+        target = Path(wt_path).resolve()
+        current_path: Optional[Path] = None
+        for line in result.stdout.splitlines():
+            if line.startswith("worktree "):
+                current_path = Path(line[len("worktree "):].strip()).resolve()
+            elif line == "locked" or line.startswith("locked "):
+                if current_path == target:
+                    return True
+        return False
+    except Exception:
+        return True
+
+
+def _worktree_is_dirty(wt_path: str, timeout: int = 10) -> bool:
+    """Return whether a worktree has uncommitted changes (staged, unstaged,
+    or untracked).
+
+    Fails SAFE: on any error returns True so callers do not delete a
+    worktree whose state they cannot determine.
+    """
+    import subprocess
+
+    try:
+        result = subprocess.run(
+            ["git", "status", "--porcelain"],
+            capture_output=True, text=True, timeout=timeout, cwd=wt_path,
+        )
+        if result.returncode != 0:
+            return True
+        return bool(result.stdout.strip())
+    except Exception:
+        return True
+
+
+def _cleanup_worktree(info: Dict[str, str] = None) -> None:
+    """Remove a worktree and its branch on graceful exit.
+
+    Preserves the worktree (along with its branch and lock) if it has
+    unpushed commits OR uncommitted changes — either may be work the user
+    has not retrieved yet.  Only clean, fully-pushed worktrees are
+    removed, and the branch is only deleted after ``git worktree remove``
+    actually succeeded.
     """
     global _active_worktree
     info = info or _active_worktree
@@ -1406,24 +1507,41 @@ def _cleanup_worktree(info: Dict[str, str] = None) -> None:
         return
 
     has_unpushed = _worktree_has_unpushed_commits(wt_path, timeout=10)
+    is_dirty = _worktree_is_dirty(wt_path)
 
-    if has_unpushed:
-        print(f"\n\033[33m⚠ Worktree has unpushed commits, keeping: {wt_path}\033[0m")
-        print(f"  To clean up manually: git worktree remove --force {wt_path}")
+    if has_unpushed or is_dirty:
+        reason = "unpushed commits" if has_unpushed else "uncommitted changes"
+        print(f"\n\033[33m⚠ Worktree has {reason}, keeping: {wt_path}\033[0m")
+        print(f"  To clean up manually: git worktree unlock {wt_path}")
+        print(f"  then: git worktree remove --force {wt_path}")
         _active_worktree = None
         return
 
-    # Remove worktree (even if working tree is dirty — uncommitted
-    # changes without unpushed commits are just artifacts)
+    # Clean and fully pushed — release our lock, then remove.
+    _unlock_worktree(repo_root, wt_path)
+
+    removed = False
     try:
-        subprocess.run(
+        result = subprocess.run(
             ["git", "worktree", "remove", wt_path, "--force"],
             capture_output=True, text=True, timeout=15, cwd=repo_root,
         )
+        removed = result.returncode == 0
+        if not removed:
+            logger.debug(
+                "Failed to remove worktree %s: %s", wt_path, result.stderr.strip()
+            )
     except Exception as e:
         logger.debug("Failed to remove worktree: %s", e)
 
-    # Delete the branch
+    if not removed:
+        # Removal failed — keep the branch so the commits stay reachable.
+        print(f"\033[33m⚠ Could not remove worktree, keeping it (and branch "
+              f"{branch}): {wt_path}\033[0m")
+        _active_worktree = None
+        return
+
+    # Delete the branch only now that the worktree is actually gone.
     try:
         subprocess.run(
             ["git", "branch", "-D", branch],
@@ -1517,10 +1635,14 @@ def _run_checkpoint_auto_maintenance() -> None:
 def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
     """Remove stale worktrees and orphaned branches on startup.
 
-    Age-based tiers:
+    Pruning may only ever delete clean, unlocked, fully-pushed worktrees:
     - Under max_age_hours (24h): skip — session may still be active.
-    - 24h–72h: remove if no unpushed commits.
-    - Over 72h: force remove regardless (nothing should sit this long).
+    - Locked (a live or crashed hermes session): skip at ANY age.
+    - Dirty working tree (uncommitted changes): skip at ANY age.
+    - Unpushed commits: skip at ANY age.
+
+    The branch is only deleted after ``git worktree remove`` actually
+    succeeded, so commits never lose their easy reachability.
 
     Also prunes orphaned ``hermes/*`` and ``pr-*`` local branches that
     have no corresponding worktree.
@@ -1535,7 +1657,6 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
 
     now = time.time()
     soft_cutoff = now - (max_age_hours * 3600)       # 24h default
-    hard_cutoff = now - (max_age_hours * 3 * 3600)   # 72h default
 
     for entry in worktrees_dir.iterdir():
         if not entry.is_dir() or not entry.name.startswith("hermes-"):
@@ -1549,14 +1670,22 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
         except Exception:
             continue
 
-        force = mtime <= hard_cutoff  # Over 72h — force remove
+        # A lock means a session (live, or crashed mid-work) owns this
+        # worktree — never touch it, regardless of age.
+        if _worktree_is_locked(repo_root, str(entry)):
+            logger.debug("Skipping locked worktree: %s", entry.name)
+            continue
 
-        if not force:
-            # 24h–72h tier: only remove if no unpushed commits
-            if _worktree_has_unpushed_commits(str(entry), timeout=5):
-                continue  # Has unpushed commits or can't check — skip
+        # Uncommitted changes may be work the user hasn't retrieved.
+        if _worktree_is_dirty(str(entry)):
+            logger.debug("Skipping dirty worktree: %s", entry.name)
+            continue
 
-        # Safe to remove
+        # Unpushed commits are definitely work — keep at any age.
+        if _worktree_has_unpushed_commits(str(entry), timeout=5):
+            continue
+
+        # Safe to remove: clean, unlocked, fully pushed.
         try:
             branch_result = subprocess.run(
                 ["git", "branch", "--show-current"],
@@ -1564,16 +1693,24 @@ def _prune_stale_worktrees(repo_root: str, max_age_hours: int = 24) -> None:
             )
             branch = branch_result.stdout.strip()
 
-            subprocess.run(
+            remove_result = subprocess.run(
                 ["git", "worktree", "remove", str(entry), "--force"],
                 capture_output=True, text=True, timeout=15, cwd=repo_root,
             )
+            if remove_result.returncode != 0:
+                # Removal failed — keep the branch so the commits stay
+                # reachable.
+                logger.debug(
+                    "Failed to remove worktree %s: %s",
+                    entry.name, remove_result.stderr.strip(),
+                )
+                continue
             if branch:
                 subprocess.run(
                     ["git", "branch", "-D", branch],
                     capture_output=True, text=True, timeout=10, cwd=repo_root,
                 )
-            logger.debug("Pruned stale worktree: %s (force=%s)", entry.name, force)
+            logger.debug("Pruned stale worktree: %s", entry.name)
         except Exception as e:
             logger.debug("Failed to prune worktree %s: %s", entry.name, e)
 
@@ -8301,13 +8438,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         compressions = compressor.compression_count
 
         msg_count = len(self.conversation_history)
-        # Cost — provider-REPORTED only (OpenRouter usage.cost accumulator
-        # and/or Nous credits-header delta). No estimation: an unreported
-        # cost shows as "not reported", never a fabricated dollar figure.
-        from agent.usage_pricing import real_session_cost_usd, resolve_billing_route
-        real_cost_usd = real_session_cost_usd(agent)
-        _billing_route = resolve_billing_route(
+        cost_result = estimate_usage_cost(
             agent.model,
+            CanonicalUsage(
+                input_tokens=input_tokens,
+                output_tokens=output_tokens,
+                cache_read_tokens=cache_read_tokens,
+                cache_write_tokens=cache_write_tokens,
+            ),
             provider=getattr(agent, "provider", None),
             base_url=getattr(agent, "base_url", None),
         )
@@ -8327,16 +8465,21 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         print(f"  Total tokens:              {total:>10,}")
         print(f"  API calls:                 {calls:>10,}")
         print(f"  Session duration:          {elapsed:>10}")
-        if real_cost_usd is not None:
-            print(f"  Cost (provider-reported): ${real_cost_usd:>9.4f}")
-        elif _billing_route.billing_mode == "subscription_included":
-            print(f"  Cost:                    {'included':>11}")
+        print(f"  Cost status:              {cost_result.status:>10}")
+        print(f"  Cost source:              {cost_result.source:>10}")
+        if cost_result.amount_usd is not None:
+            prefix = "~" if cost_result.status == "estimated" else ""
+            print(f"  Total cost:              {prefix}${float(cost_result.amount_usd):>10.4f}")
+        elif cost_result.status == "included":
+            print(f"  Total cost:              {'included':>10}")
         else:
-            print(f"  Cost:        {'not reported by provider':>23}")
+            print(f"  Total cost:              {'n/a':>10}")
         print(f"  {'─' * 40}")
         print(f"  Current context:  {last_prompt:,} / {ctx_len:,} ({pct:.0f}%)")
         print(f"  Messages:         {msg_count}")
         print(f"  Compressions:     {compressions}")
+        if cost_result.status == "unknown":
+            print(f"  Note:             Pricing unknown for {agent.model}")
 
         # Account limits -- fetched off-thread with a hard timeout so slow
         # provider APIs don't hang the prompt.

gateway/platforms/base.py

@@ -77,6 +77,13 @@ def _thread_metadata_for_source(source, reply_to_message_id: str | None = None)
     return metadata
 
 
+def _mark_notify_metadata(metadata: dict | None) -> dict:
+    """Clone metadata and mark a user-visible reply as notify-worthy."""
+    notify_metadata = dict(metadata) if metadata else {}
+    notify_metadata["notify"] = True
+    return notify_metadata
+
+
 def _reply_anchor_for_event(event) -> str | None:
     """Return reply_to id for platforms that need reply semantics.
 
@@ -3889,7 +3896,7 @@ class BasePlatformAdapter(ABC):
                     chat_id=event.source.chat_id,
                     content=_text,
                     reply_to=_reply_anchor_for_event(event),
-                    metadata=thread_meta,
+                    metadata=_mark_notify_metadata(thread_meta),
                 )
                 if _eph_ttl > 0 and _r.success and _r.message_id:
                     self._schedule_ephemeral_delete(
@@ -3995,7 +4002,7 @@ class BasePlatformAdapter(ABC):
                             chat_id=event.source.chat_id,
                             content=_text,
                             reply_to=_reply_anchor_for_event(event),
-                            metadata=_thread_meta,
+                            metadata=_mark_notify_metadata(_thread_meta),
                         )
                         if _eph_ttl > 0 and _r.success and _r.message_id:
                             self._schedule_ephemeral_delete(
@@ -4045,7 +4052,7 @@ class BasePlatformAdapter(ABC):
                                 chat_id=event.source.chat_id,
                                 content=_text,
                                 reply_to=_reply_anchor_for_event(event),
-                                metadata=_thread_meta,
+                                metadata=_mark_notify_metadata(_thread_meta),
                             )
                             if _eph_ttl > 0 and _r.success and _r.message_id:
                                 self._schedule_ephemeral_delete(
@@ -4268,6 +4275,12 @@ class BasePlatformAdapter(ABC):
                         )
                         text_content = _recovered
 
+                # Final user-visible content (text, TTS, media, files) gets
+                # the existing notify=True marker. Clone once so typing/status
+                # metadata stays unmarked and progress bubbles remain
+                # thread-strict.
+                _final_thread_metadata = _mark_notify_metadata(_thread_metadata)
+
                 # Auto-TTS: if voice message, generate audio FIRST (before sending text)
                 # Gated via ``_should_auto_tts_for_chat``: fires when the chat has
                 # an explicit ``/voice on|tts`` opt-in OR when ``voice.auto_tts`` is
@@ -4307,7 +4320,7 @@ class BasePlatformAdapter(ABC):
                             chat_id=event.source.chat_id,
                             audio_path=_tts_path,
                             caption=telegram_tts_caption,
-                            metadata=_thread_metadata,
+                            metadata=_final_thread_metadata,
                         )
                         _tts_caption_delivered = bool(
                             telegram_tts_caption and getattr(tts_result, "success", False)
@@ -4322,23 +4335,11 @@ class BasePlatformAdapter(ABC):
                 if text_content and not _tts_caption_delivered:
                     logger.info("[%s] Sending response (%d chars) to %s", self.name, len(text_content), event.source.chat_id)
                     _reply_anchor = _reply_anchor_for_event(event)
-                    # Mark final response messages for notification delivery.
-                    # Platform adapters that support per-message notification
-                    # control (e.g. Telegram's disable_notification) use this
-                    # flag to override silent-mode and ensure the final
-                    # response triggers a push notification.
-                    # Clone to avoid mutating the metadata shared with the
-                    # typing-indicator task (which must remain unmarked).
-                    if _thread_metadata is not None:
-                        _thread_metadata = dict(_thread_metadata)
-                        _thread_metadata["notify"] = True
-                    else:
-                        _thread_metadata = {"notify": True}
                     result = await self._send_with_retry(
                         chat_id=event.source.chat_id,
                         content=text_content,
                         reply_to=_reply_anchor,
-                        metadata=_thread_metadata,
+                        metadata=_final_thread_metadata,
                     )
                     _record_delivery(result)
 
@@ -4367,7 +4368,7 @@ class BasePlatformAdapter(ABC):
                         await self.send_multiple_images(
                             chat_id=event.source.chat_id,
                             images=images,
-                            metadata=_thread_metadata,
+                            metadata=_final_thread_metadata,
                             human_delay=human_delay,
                         )
                     except Exception as batch_err:
@@ -4409,7 +4410,7 @@ class BasePlatformAdapter(ABC):
                         await self.send_multiple_images(
                             chat_id=event.source.chat_id,
                             images=_batch,
-                            metadata=_thread_metadata,
+                            metadata=_final_thread_metadata,
                             human_delay=human_delay,
                         )
                     except Exception as batch_err:
@@ -4424,19 +4425,19 @@ class BasePlatformAdapter(ABC):
                             media_result = await self.send_voice(
                                 chat_id=event.source.chat_id,
                                 audio_path=media_path,
-                                metadata=_thread_metadata,
+                                metadata=_final_thread_metadata,
                             )
                         elif ext in _VIDEO_EXTS:
                             media_result = await self.send_video(
                                 chat_id=event.source.chat_id,
                                 video_path=media_path,
-                                metadata=_thread_metadata,
+                                metadata=_final_thread_metadata,
                             )
                         else:
                             media_result = await self.send_document(
                                 chat_id=event.source.chat_id,
                                 file_path=media_path,
-                                metadata=_thread_metadata,
+                                metadata=_final_thread_metadata,
                             )
 
                         if not media_result.success:
@@ -4454,13 +4455,13 @@ class BasePlatformAdapter(ABC):
                             await self.send_video(
                                 chat_id=event.source.chat_id,
                                 video_path=file_path,
-                                metadata=_thread_metadata,
+                                metadata=_final_thread_metadata,
                             )
                         else:
                             await self.send_document(
                                 chat_id=event.source.chat_id,
                                 file_path=file_path,
-                                metadata=_thread_metadata,
+                                metadata=_final_thread_metadata,
                             )
                     except Exception as file_err:
                         logger.error("[%s] Error sending local file %s: %s", self.name, file_path, file_err)

gateway/platforms/email.py

@@ -678,8 +678,13 @@ class EmailAdapter(BasePlatformAdapter):
         image_url: str,
         caption: Optional[str] = None,
         reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
     ) -> SendResult:
-        """Send an image URL as part of an email body."""
+        """Send an image URL as part of an email body.
+
+        ``metadata`` is accepted to honor the base-class contract; the
+        email body send doesn't use it.
+        """
         text = caption or ""
         text += f"\n\nImage: {image_url}"
         return await self.send(chat_id, text.strip(), reply_to)

gateway/platforms/whatsapp.py

@@ -846,13 +846,20 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
         image_url: str,
         caption: Optional[str] = None,
         reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
     ) -> SendResult:
-        """Download image URL to cache, send natively via bridge."""
+        """Download image URL to cache, send natively via bridge.
+
+        ``metadata`` is accepted to honor the base-class contract — the
+        batch sender ``send_multiple_images`` passes it through to every
+        send path. The bridge media call doesn't use it, matching the
+        sibling overrides (send_video / send_voice / send_document).
+        """
         try:
             local_path = await cache_image_from_url(image_url)
             return await self._send_media_to_bridge(chat_id, local_path, "image", caption)
         except Exception:
-            return await super().send_image(chat_id, image_url, caption, reply_to)
+            return await super().send_image(chat_id, image_url, caption, reply_to, metadata)
 
     async def send_image_file(
         self,
@@ -1136,6 +1143,15 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
             body = data.get("body", "")
             if data.get("isGroup"):
                 body = self._clean_bot_mention_text(body, data)
+
+            # If this is a reply, include the quoted message text so the agent
+            # knows exactly what the user is responding to (fixes "approve" context issue)
+            quoted_text = str(data.get("quotedText") or "").strip()
+            if quoted_text and data.get("hasQuotedMessage"):
+                # Truncate long quoted text to keep prompts reasonable
+                if len(quoted_text) > 300:
+                    quoted_text = quoted_text[:297] + "..."
+                body = f"[Replying to: \"{quoted_text}\"]\n{body}"
             MAX_TEXT_INJECT_BYTES = 100 * 1024
             if msg_type == MessageType.DOCUMENT and cached_urls:
                 for doc_path in cached_urls:

gateway/run.py

@@ -413,6 +413,57 @@ def _resolve_progress_thread_id(platform: Any, source_thread_id: Any, event_mess
     return None
 
 
+def _has_platform_display_override(user_config: dict, platform_key: str, setting: str) -> bool:
+    """Return True when display.platforms.<platform> explicitly sets setting."""
+    display = user_config.get("display") if isinstance(user_config, dict) else None
+    if not isinstance(display, dict):
+        return False
+    platforms = display.get("platforms")
+    if not isinstance(platforms, dict):
+        return False
+    platform_cfg = platforms.get(platform_key)
+    return isinstance(platform_cfg, dict) and setting in platform_cfg
+
+
+def _resolve_gateway_display_bool(
+    user_config: dict,
+    platform_key: str,
+    setting: str,
+    *,
+    default: bool = False,
+    platform: Any = None,
+    require_platform_override_for: set[Any] | None = None,
+) -> bool:
+    """Resolve a boolean display setting with optional platform-only opt-in.
+
+    Some display features expose assistant scratch text rather than deliberate
+    user-facing output.  For high-noise threaded chat surfaces such as
+    Mattermost, a global opt-in is too broad: they must be enabled with an
+    explicit display.platforms.<platform>.<setting> override.
+    """
+    current_platform = _gateway_platform_value(platform or platform_key)
+    platform_only = {
+        _gateway_platform_value(candidate)
+        for candidate in (require_platform_override_for or set())
+    }
+    if (
+        current_platform in platform_only
+        and not _has_platform_display_override(user_config, platform_key, setting)
+    ):
+        return False
+
+    from gateway.display_config import resolve_display_setting
+
+    value = resolve_display_setting(user_config, platform_key, setting, default)
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() in {"true", "yes", "1", "on"}
+    if value is None:
+        return bool(default)
+    return bool(value)
+
+
 def _telegramize_command_mentions(text: str, platform: Any) -> str:
     """Rewrite slash-command mentions to Telegram-valid command names.
 
@@ -8989,17 +9040,24 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     source, session_entry, reason="agent-result-compression",
                 )
 
-            # Prepend reasoning/thinking if display is enabled (per-platform)
+            # Prepend reasoning/thinking if display is enabled (per-platform).
+            # Mattermost requires explicit per-platform opt-in because this is
+            # scratch text, not ordinary final-answer content.
             try:
-                from gateway.display_config import resolve_display_setting as _rds
-                _show_reasoning_effective = _rds(
+                _show_reasoning_effective = _resolve_gateway_display_bool(
                     _load_gateway_config(),
                     _platform_config_key(source.platform),
                     "show_reasoning",
-                    getattr(self, "_show_reasoning", False),
+                    default=bool(getattr(self, "_show_reasoning", False)),
+                    platform=source.platform,
+                    require_platform_override_for={Platform.MATTERMOST},
                 )
             except Exception:
-                _show_reasoning_effective = getattr(self, "_show_reasoning", False)
+                _show_reasoning_effective = (
+                    False
+                    if source.platform == Platform.MATTERMOST
+                    else getattr(self, "_show_reasoning", False)
+                )
             if _show_reasoning_effective and response and not _intentional_silence:
                 last_reasoning = agent_result.get("last_reasoning")
                 if last_reasoning:
@@ -13635,18 +13693,32 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # in chat platforms while opting into concise mid-turn updates.
         interim_assistant_messages_enabled = (
             source.platform != Platform.WEBHOOK
-            and bool(
-                resolve_display_setting(
-                    user_config,
-                    platform_key,
-                    "interim_assistant_messages",
-                    True,
-                )
+            and _resolve_gateway_display_bool(
+                user_config,
+                platform_key,
+                "interim_assistant_messages",
+                default=True,
+                platform=source.platform,
+                require_platform_override_for={Platform.MATTERMOST},
             )
         )
-        
+        # thinking_progress is independent — if enabled, we need the progress
+        # queue even when tool_progress is off (thinking relay uses same infra).
+        # Mattermost requires a per-platform opt-in: global scratch-text display
+        # is too easy to leak into busy public threads.
+        _thinking_enabled = _resolve_gateway_display_bool(
+            user_config,
+            platform_key,
+            "thinking_progress",
+            default=False,
+            platform=source.platform,
+            require_platform_override_for={Platform.MATTERMOST},
+        )
+        needs_progress_queue = tool_progress_enabled or _thinking_enabled
+
+
         # Queue for progress messages (thread-safe)
-        progress_queue = queue.Queue() if tool_progress_enabled else None
+        progress_queue = queue.Queue() if needs_progress_queue else None
         last_tool = [None]  # Mutable container for tracking in closure
         last_progress_msg = [None]  # Track last message for dedup
         repeat_count = [0]  # How many times the same message repeated
@@ -13752,6 +13824,24 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     logger.debug("tool-progress onboarding hint failed: %s", _hint_err)
                 return
 
+            # "_thinking" is assistant scratch text between tool calls.  It
+            # is never ordinary tool progress: only relay it when the platform
+            # explicitly opted into thinking_progress.  Handle both legacy
+            # callback shapes: ("_thinking", text) and
+            # ("reasoning.available", "_thinking", text, ...).
+            if event_type == "_thinking" or tool_name == "_thinking":
+                if not _thinking_enabled:
+                    return
+                thinking_text = preview if tool_name == "_thinking" else tool_name
+                msg = f"💬 {thinking_text}" if thinking_text else None
+                if msg:
+                    progress_queue.put(msg)
+                return
+
+            # If tool_progress is off, only _thinking passes through (above).
+            # Regular tool calls are suppressed.
+            if not tool_progress_enabled:
+                return
 
             # Only act on tool.started events (ignore tool.completed, reasoning.available, etc.)
             if event_type not in {"tool.started",}:
@@ -14783,6 +14873,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
             agent.clarify_callback = _clarify_callback_sync
 
+            # Show assistant thinking between tool calls — independent of
+            # tool_progress mode. Mattermost needs an explicit per-platform
+            # opt-in so global scratch-text display does not leak into threads.
+            agent.thinking_progress = _thinking_enabled
             # Store agent reference for interrupt support
             agent_holder[0] = agent
             # Capture the full tool definitions for transcript logging

gateway/slash_commands.py

@@ -3215,24 +3215,25 @@ class GatewaySlashCommandsMixin:
             lines.append(t("gateway.usage.label_total", count=f"{agent.session_total_tokens:,}"))
             lines.append(t("gateway.usage.label_api_calls", count=agent.session_api_calls))
 
-            # Cost — provider-REPORTED only (OpenRouter usage.cost accumulator
-            # and/or Nous credits-header delta). No estimation: when nothing
-            # was reported the line is omitted entirely, never shown as $0.00.
-            # Subscription-included routes (a billing fact, not a price guess)
-            # still show "included".
+            # Cost estimation
             try:
-                from agent.usage_pricing import real_session_cost_usd, resolve_billing_route
-                real_cost = real_session_cost_usd(agent)
-                if real_cost is not None:
-                    lines.append(t("gateway.usage.label_cost", prefix="", amount=f"{real_cost:.4f}"))
-                else:
-                    route = resolve_billing_route(
-                        agent.model,
-                        provider=getattr(agent, "provider", None),
-                        base_url=getattr(agent, "base_url", None),
-                    )
-                    if route.billing_mode == "subscription_included":
-                        lines.append(t("gateway.usage.label_cost_included"))
+                from agent.usage_pricing import CanonicalUsage, estimate_usage_cost
+                cost_result = estimate_usage_cost(
+                    agent.model,
+                    CanonicalUsage(
+                        input_tokens=input_tokens,
+                        output_tokens=output_tokens,
+                        cache_read_tokens=cache_read,
+                        cache_write_tokens=cache_write,
+                    ),
+                    provider=getattr(agent, "provider", None),
+                    base_url=getattr(agent, "base_url", None),
+                )
+                if cost_result.amount_usd is not None:
+                    prefix = "~" if cost_result.status == "estimated" else ""
+                    lines.append(t("gateway.usage.label_cost", prefix=prefix, amount=f"{float(cost_result.amount_usd):.4f}"))
+                elif cost_result.status == "included":
+                    lines.append(t("gateway.usage.label_cost_included"))
             except Exception:
                 pass
 

gateway/stream_consumer.py

@@ -197,6 +197,30 @@ class GatewayStreamConsumer:
         # this response and route through edit-based for graceful degradation.
         self._draft_failures = 0
 
+    def _metadata_for_send(
+        self,
+        *,
+        final: bool = False,
+        expect_edits: bool = False,
+    ) -> dict | None:
+        """Return per-send metadata for stream-created messages.
+
+        Mattermost treats notify-worthy sends as user-visible final content
+        when deciding whether a broken thread root may fall back flat.  Preview
+        and progress sends keep their original metadata and remain thread-strict.
+
+        ``expect_edits`` preserves the upstream Telegram streaming contract:
+        preview messages that may be edited later must stay on the editable
+        legacy send path, while fresh/fallback final sends can still use richer
+        final-message delivery.
+        """
+        meta = dict(self.metadata) if self.metadata else {}
+        if expect_edits:
+            meta["expect_edits"] = True
+        if final:
+            meta["notify"] = True
+        return meta or None
+
     @property
     def already_sent(self) -> bool:
         """True if at least one message was sent or edited during the run."""
@@ -513,7 +537,11 @@ class GatewayStreamConsumer:
                         chunks_delivered = False
                         reply_to = self._message_id or self._initial_reply_to_id
                         for chunk in chunks:
-                            new_id = await self._send_new_chunk(chunk, reply_to)
+                            new_id = await self._send_new_chunk(
+                                chunk,
+                                reply_to,
+                                final=got_done,
+                            )
                             if new_id is not None and new_id != reply_to:
                                 chunks_delivered = True
                         self._accumulated = ""
@@ -749,7 +777,13 @@ class GatewayStreamConsumer:
         # Strip trailing whitespace/newlines but preserve leading content
         return cleaned.rstrip()
 
-    async def _send_new_chunk(self, text: str, reply_to_id: Optional[str]) -> Optional[str]:
+    async def _send_new_chunk(
+        self,
+        text: str,
+        reply_to_id: Optional[str],
+        *,
+        final: bool = False,
+    ) -> Optional[str]:
         """Send a new message chunk, optionally threaded to a previous message.
 
         Returns the message_id so callers can thread subsequent chunks.
@@ -758,15 +792,11 @@ class GatewayStreamConsumer:
         if not text.strip():
             return reply_to_id
         try:
-            meta = dict(self.metadata) if self.metadata else {}
-            # This chunk becomes the next edit target — adapters that support
-            # rich final sends (Telegram) must keep it on the editable path.
-            meta["expect_edits"] = True
             result = await self.adapter.send(
                 chat_id=self.chat_id,
                 content=text,
                 reply_to=reply_to_id,
-                metadata=meta,
+                metadata=self._metadata_for_send(final=final, expect_edits=True),
             )
             if result.success and result.message_id:
                 self._message_id = str(result.message_id)
@@ -885,7 +915,7 @@ class GatewayStreamConsumer:
                 result = await self.adapter.send(
                     chat_id=self.chat_id,
                     content=chunk,
-                    metadata=self.metadata,
+                    metadata=self._metadata_for_send(final=True),
                 )
                 if result.success:
                     break
@@ -1242,7 +1272,7 @@ class GatewayStreamConsumer:
             result = await self.adapter.send(
                 chat_id=self.chat_id,
                 content=text,
-                metadata=self.metadata,
+                metadata=self._metadata_for_send(final=True),
             )
         except Exception as e:
             logger.debug("Fresh-final send failed, falling back to edit: %s", e)
@@ -1532,7 +1562,10 @@ class GatewayStreamConsumer:
                     chat_id=self.chat_id,
                     content=text,
                     reply_to=self._initial_reply_to_id,
-                    metadata={**(self.metadata or {}), "expect_edits": True},
+                    metadata=self._metadata_for_send(
+                        final=finalize,
+                        expect_edits=True,
+                    ),
                 )
                 if result.success:
                     if result.message_id:

hermes_state.py

@@ -1695,36 +1695,6 @@ class SessionDB:
 
         return self._execute_write(_do) or 0
 
-    def usage_totals(self, days: int = 30) -> Dict[str, Any]:
-        """Aggregate usage for sessions started in the last ``days``.
-
-        ``reported_cost_usd`` sums only provider-REPORTED ``actual_cost_usd``
-        (never estimates) and is None when no session in the window has a
-        reported cost — callers must hide cost rather than print $0.00.
-        """
-        cutoff = time.time() - days * 86400
-        with self._lock:
-            row = self._conn.execute(
-                """SELECT COUNT(*) AS sessions,
-                          COALESCE(SUM(input_tokens), 0)
-                            + COALESCE(SUM(cache_read_tokens), 0)
-                            + COALESCE(SUM(cache_write_tokens), 0) AS input_tokens,
-                          COALESCE(SUM(output_tokens), 0) AS output_tokens,
-                          COALESCE(SUM(api_call_count), 0) AS api_calls,
-                          SUM(actual_cost_usd) AS reported_cost_usd
-                   FROM sessions WHERE started_at >= ?""",
-                (cutoff,),
-            ).fetchone()
-        result = dict(row) if row else {}
-        return {
-            "days": days,
-            "sessions": int(result.get("sessions") or 0),
-            "input_tokens": int(result.get("input_tokens") or 0),
-            "output_tokens": int(result.get("output_tokens") or 0),
-            "api_calls": int(result.get("api_calls") or 0),
-            "reported_cost_usd": result.get("reported_cost_usd"),
-        }
-
     def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
         """Get a session by ID."""
         with self._lock:

plugins/model-providers/openrouter/__init__.py

@@ -82,13 +82,6 @@ class OpenRouterProfile(ProviderProfile):
         if prefs:
             body["provider"] = prefs
 
-        # Usage accounting — makes OpenRouter return the REAL cost it charged
-        # in the response `usage.cost` field (credits are 1:1 USD), instead of
-        # Hermes having to estimate from a pricing table. Captured by
-        # agent.usage_pricing.extract_provider_cost_usd in the conversation
-        # loop. https://openrouter.ai/docs/use-cases/usage-accounting
-        body["usage"] = {"include": True}
-
         # Pareto Code router — model-gated. The plugins block is only
         # meaningful for openrouter/pareto-code; sending it on any other
         # model has no documented effect and would be confusing in logs.

run_agent.py

@@ -636,9 +636,6 @@ class AIAgent:
         self.session_reasoning_tokens = 0
         self.session_api_calls = 0
         self.session_estimated_cost_usd = 0.0
-        # Provider-REPORTED cost only — None means "nothing reported".
-        self.session_actual_cost_usd = None
-        self.session_model_usage = {}
         self.session_cost_status = "unknown"
         self.session_cost_source = "none"
         

scripts/release.py

@@ -90,6 +90,7 @@ AUTHOR_MAP = {
     "290859878+synapsesx@users.noreply.github.com": "synapsesx",
     "157689911+itsflownium@users.noreply.github.com": "itsflownium",
     "dirtyren@users.noreply.github.com": "dirtyren",
+    "evansrory@gmail.com": "zimigit2020",
     "237263164+ft-ioxcs@users.noreply.github.com": "ft-ioxcs",
     "tharushkadinujaya05@gmail.com": "0xneobyte",
     "138671361+Veritas-7@users.noreply.github.com": "Veritas-7",

tests/agent/test_auxiliary_client.py

@@ -1653,6 +1653,37 @@ class TestAuxiliaryFallbackLayering:
         exc.status_code = 402
         return exc
 
+    def test_auto_provider_uses_task_then_main_chain_before_builtin_chain(self, monkeypatch):
+        """Auto aux call failures try per-task then top-level fallback before built-ins."""
+        primary_client = MagicMock()
+        primary_client.chat.completions.create.side_effect = self._make_payment_err()
+
+        main_chain_client = MagicMock()
+        main_chain_client.chat.completions.create.return_value = MagicMock(choices=[
+            MagicMock(message=MagicMock(content="from main fallback chain"))
+        ])
+
+        with patch("agent.auxiliary_client._get_cached_client",
+                   return_value=(primary_client, "qwen/qwen3.5-122b-a10b")), \
+             patch("agent.auxiliary_client._resolve_task_provider_model",
+                   return_value=("auto", None, None, None, None)), \
+             patch("agent.auxiliary_client._try_configured_fallback_chain",
+                   return_value=(None, None, "")) as mock_task_chain, \
+             patch("agent.auxiliary_client._try_main_fallback_chain",
+                   return_value=(main_chain_client, "inclusionai/ring-2.6-1t:free", "openrouter")) as mock_main_chain, \
+             patch("agent.auxiliary_client._try_payment_fallback") as mock_builtin_chain:
+            result = call_llm(
+                task="title_generation",
+                messages=[{"role": "user", "content": "hello"}],
+            )
+
+        assert main_chain_client.chat.completions.create.called
+        mock_task_chain.assert_called_once_with(
+            "title_generation", "auto", reason="payment error")
+        mock_main_chain.assert_called_once_with(
+            "title_generation", "auto", reason="payment error")
+        mock_builtin_chain.assert_not_called()
+
     def test_explicit_provider_uses_configured_chain_first(self, monkeypatch, caplog):
         """When a user has fallback_chain configured, it's tried BEFORE the main agent model."""
         monkeypatch.setenv("OPENROUTER_API_KEY", "or-key")

tests/agent/test_auxiliary_main_first.py

@@ -118,6 +118,64 @@ class TestResolveAutoMainFirst:
         assert client is chain_client
         assert model == "google/gemini-3-flash-preview"
 
+    def test_main_unavailable_uses_task_fallback_chain_before_builtin_chain(self):
+        """Auto aux resolution honors auxiliary.<task>.fallback_chain before built-ins."""
+        task_client = MagicMock()
+        with patch(
+            "agent.auxiliary_client._read_main_provider", return_value="nvidia",
+        ), patch(
+            "agent.auxiliary_client._read_main_model", return_value="qwen/qwen3.5-122b-a10b",
+        ), patch(
+            "agent.auxiliary_client.resolve_provider_client",
+            return_value=(None, None),  # main provider has no client
+        ), patch(
+            "agent.auxiliary_client._try_configured_fallback_chain",
+            return_value=(task_client, "task-free-model", "fallback_chain[0](openrouter)"),
+        ) as mock_task_chain, patch(
+            "agent.auxiliary_client._try_main_fallback_chain",
+        ) as mock_main_chain, patch(
+            "agent.auxiliary_client._try_openrouter",
+        ) as mock_openrouter:
+            from agent.auxiliary_client import _resolve_auto
+
+            client, model = _resolve_auto(task="title_generation")
+
+        assert client is task_client
+        assert model == "task-free-model"
+        mock_task_chain.assert_called_once_with(
+            "title_generation", "nvidia", reason="main provider unavailable")
+        mock_main_chain.assert_not_called()
+        mock_openrouter.assert_not_called()
+
+    def test_main_unavailable_uses_main_fallback_chain_before_builtin_chain(self):
+        """Auto aux resolution honors top-level fallback_providers before built-ins."""
+        main_fallback_client = MagicMock()
+        with patch(
+            "agent.auxiliary_client._read_main_provider", return_value="nvidia",
+        ), patch(
+            "agent.auxiliary_client._read_main_model", return_value="qwen/qwen3.5-122b-a10b",
+        ), patch(
+            "agent.auxiliary_client.resolve_provider_client",
+            return_value=(None, None),  # main provider has no client
+        ), patch(
+            "agent.auxiliary_client._try_configured_fallback_chain",
+            return_value=(None, None, ""),
+        ), patch(
+            "agent.auxiliary_client._try_main_fallback_chain",
+            return_value=(main_fallback_client, "inclusionai/ring-2.6-1t:free", "openrouter"),
+        ) as mock_main_chain, patch(
+            "agent.auxiliary_client._try_openrouter",
+        ) as mock_openrouter:
+            from agent.auxiliary_client import _resolve_auto
+
+            client, model = _resolve_auto(task="title_generation")
+
+        assert client is main_fallback_client
+        assert model == "inclusionai/ring-2.6-1t:free"
+        mock_main_chain.assert_called_once_with(
+            "title_generation", "nvidia", reason="main provider unavailable")
+        mock_openrouter.assert_not_called()
+
     def test_no_main_config_uses_chain_directly(self):
         """No main provider configured → skip step 1, use chain (no regression)."""
         chain_client = MagicMock()

tests/agent/test_provider_cost_capture.py

@@ -1,246 +0,0 @@
-"""Real provider-reported cost capture — never estimated, absent ≠ zero.
-
-Covers the three fixture shapes from the cost-tracking fix:
-  - OpenRouter usage accounting: response ``usage.cost`` present → accumulates.
-  - Nous: ``x-nous-credits-*`` headers present → header delta accumulates.
-  - Provider reports nothing → cost stays None/absent (NOT zero-as-real).
-"""
-
-from types import SimpleNamespace
-
-import pytest
-
-from agent.usage_pricing import extract_provider_cost_usd, nous_header_cost_usd, real_session_cost_usd
-
-
-# ── extract_provider_cost_usd — the per-response REAL cost reader ────────────
-
-
-class TestExtractProviderCost:
-    def test_openrouter_usage_cost_attr(self):
-        usage = SimpleNamespace(prompt_tokens=10, completion_tokens=5, cost=0.001234)
-        assert extract_provider_cost_usd(usage) == pytest.approx(0.001234)
-
-    def test_dict_shaped_usage(self):
-        assert extract_provider_cost_usd({"cost": 0.5}) == pytest.approx(0.5)
-
-    def test_reported_zero_is_real_zero(self):
-        # Free-tier models really cost $0 — distinct from "not reported".
-        usage = SimpleNamespace(cost=0)
-        assert extract_provider_cost_usd(usage) == 0.0
-
-    def test_absent_cost_is_none_not_zero(self):
-        usage = SimpleNamespace(prompt_tokens=10, completion_tokens=5)
-        assert extract_provider_cost_usd(usage) is None
-        assert extract_provider_cost_usd({"prompt_tokens": 10}) is None
-
-    def test_none_usage_is_none(self):
-        assert extract_provider_cost_usd(None) is None
-
-    def test_garbage_cost_values_are_none(self):
-        for bad in ("0.01", True, float("nan"), float("inf"), -0.5, [], {}):
-            assert extract_provider_cost_usd(SimpleNamespace(cost=bad)) is None, bad
-
-
-# ── real_session_cost_usd — the session accumulator surface ─────────────────
-
-
-class _FakeAgent:
-    def __init__(self, actual=None, credits_micros=None):
-        self.session_actual_cost_usd = actual
-        self._credits_micros = credits_micros
-
-    def get_credits_spent_micros(self):
-        return self._credits_micros
-
-
-class TestRealSessionCost:
-    def test_nothing_reported_is_none(self):
-        assert real_session_cost_usd(_FakeAgent()) is None
-
-    def test_openrouter_accumulator_only(self):
-        assert real_session_cost_usd(_FakeAgent(actual=0.42)) == pytest.approx(0.42)
-
-    def test_nous_credits_delta_only(self):
-        # 123_400 micros = $0.1234
-        assert real_session_cost_usd(
-            _FakeAgent(credits_micros=123_400)
-        ) == pytest.approx(0.1234)
-
-    def test_both_sources_sum(self):
-        assert real_session_cost_usd(
-            _FakeAgent(actual=0.10, credits_micros=200_000)
-        ) == pytest.approx(0.30)
-
-    def test_negative_credits_delta_clamped(self):
-        # A mid-session top-up makes the delta negative — never show negative cost.
-        assert real_session_cost_usd(_FakeAgent(credits_micros=-50_000)) == 0.0
-
-    def test_agent_without_credits_method(self):
-        agent = SimpleNamespace(session_actual_cost_usd=None)
-        assert real_session_cost_usd(agent) is None
-
-    def test_non_numeric_actual_ignored(self):
-        agent = _FakeAgent()
-        agent.session_actual_cost_usd = "0.42"  # corrupted attr → ignore
-        assert real_session_cost_usd(agent) is None
-
-
-# ── nous_header_cost_usd — the CHROME status-bar cost (F3: header-only) ──────
-
-
-class TestNousHeaderCost:
-    def test_header_delta_only(self):
-        # 123_400 micros = $0.1234 — the Nous header source feeds the chrome.
-        assert nous_header_cost_usd(_FakeAgent(credits_micros=123_400)) == pytest.approx(0.1234)
-
-    def test_openrouter_accumulator_ignored(self):
-        # The OpenRouter usage.cost accumulator must NOT feed the chrome bar:
-        # a non-Nous session (no header → None) reports no cost even when the
-        # OpenRouter accumulator has a value.
-        assert nous_header_cost_usd(_FakeAgent(actual=0.42)) is None
-
-    def test_no_header_is_none(self):
-        assert nous_header_cost_usd(_FakeAgent()) is None
-
-    def test_negative_delta_clamped(self):
-        # A mid-session top-up makes the delta negative — never show negative.
-        assert nous_header_cost_usd(_FakeAgent(credits_micros=-50_000)) == 0.0
-
-    def test_agent_without_credits_method_is_none(self):
-        agent = SimpleNamespace(session_actual_cost_usd=0.42)
-        assert nous_header_cost_usd(agent) is None
-
-
-# ── Nous header fixture → real accumulator (full _capture_credits path) ─────
-
-
-def _nous_headers(remaining_micros: int) -> dict:
-    return {
-        "x-nous-credits-version": "1",
-        "x-nous-credits-remaining-micros": str(remaining_micros),
-        "x-nous-credits-remaining-usd": f"{remaining_micros / 1_000_000:.2f}",
-        "x-nous-credits-subscription-micros": str(remaining_micros),
-        "x-nous-credits-subscription-usd": f"{remaining_micros / 1_000_000:.2f}",
-        "x-nous-credits-rollover-micros": "0",
-        "x-nous-credits-purchased-micros": "0",
-        "x-nous-credits-purchased-usd": "0.00",
-        "x-nous-credits-denominator-kind": "none",
-        "x-nous-credits-paid-access": "true",
-        "x-nous-credits-as-of-ms": "1717000000000",
-    }
-
-
-def _bare_nous_agent():
-    """Minimal AIAgent shell exercising the real _capture_credits path."""
-    from run_agent import AIAgent
-
-    agent = object.__new__(AIAgent)
-    agent.provider = "nous"
-    agent._credits_state = None
-    agent._credits_session_start_micros = None
-    agent.notice_callback = None
-    agent.notice_clear_callback = None
-    agent.session_actual_cost_usd = None
-    return agent
-
-
-class TestNousHeaderAccumulation:
-    def test_headers_accumulate_into_real_session_cost(self, monkeypatch):
-        monkeypatch.delenv("HERMES_DEV_CREDITS_FIXTURE", raising=False)
-        agent = _bare_nous_agent()
-
-        # First response latches the session-start balance ($10.00).
-        agent._capture_credits(SimpleNamespace(headers=_nous_headers(10_000_000)))
-        assert real_session_cost_usd(agent) == 0.0  # real zero: headers seen, $0 spent
-
-        # Second response: balance dropped by $0.25 → real reported spend.
-        agent._capture_credits(SimpleNamespace(headers=_nous_headers(9_750_000)))
-        assert real_session_cost_usd(agent) == pytest.approx(0.25)
-
-    def test_no_headers_means_no_cost(self, monkeypatch):
-        monkeypatch.delenv("HERMES_DEV_CREDITS_FIXTURE", raising=False)
-        agent = _bare_nous_agent()
-        agent._capture_credits(SimpleNamespace(headers={"content-type": "application/json"}))
-        assert real_session_cost_usd(agent) is None
-
-
-# ── OpenRouter request param — usage accounting must be requested ────────────
-
-
-class TestOpenRouterUsageParam:
-    def test_profile_extra_body_requests_usage_accounting(self):
-        import importlib.util
-        from pathlib import Path
-
-        from providers import get_provider_profile
-
-        profile = get_provider_profile("openrouter")
-        if profile is None:
-            # Force plugin discovery in minimal test envs.
-            plugin = Path(__file__).resolve().parents[2] / "plugins" / "model-providers" / "openrouter" / "__init__.py"
-            spec = importlib.util.spec_from_file_location("_or_plugin", plugin)
-            mod = importlib.util.module_from_spec(spec)
-            spec.loader.exec_module(mod)
-            profile = mod.openrouter
-
-        body = profile.build_extra_body(session_id="s-1")
-        assert body["usage"] == {"include": True}
-
-    def test_legacy_transport_path_requests_usage_accounting(self):
-        from agent.transports.chat_completions import ChatCompletionsTransport
-
-        transport = ChatCompletionsTransport()
-        kwargs = transport.build_kwargs(
-            model="anthropic/claude-sonnet-4.6",
-            messages=[{"role": "user", "content": "hi"}],
-            tools=None,
-            is_openrouter=True,
-        )
-        assert kwargs["extra_body"]["usage"] == {"include": True}
-
-    def test_non_openrouter_does_not_send_usage_param(self):
-        from agent.transports.chat_completions import ChatCompletionsTransport
-
-        transport = ChatCompletionsTransport()
-        kwargs = transport.build_kwargs(
-            model="deepseek-chat",
-            messages=[{"role": "user", "content": "hi"}],
-            tools=None,
-            is_openrouter=False,
-        )
-        assert "usage" not in (kwargs.get("extra_body") or {})
-
-
-# ── nous_credits_compact_line — one-liner for the compact /usage page ───────
-
-
-class TestNousCreditsCompactLine:
-    def test_condenses_snapshot_details(self, monkeypatch):
-        import agent.account_usage as au
-
-        snap = au.AccountUsageSnapshot(
-            provider="nous",
-            source="portal-account",
-            fetched_at=au._utc_now(),
-            title="Nous credits",
-            plan="Ultra",
-            details=(
-                "Subscription credits: $-0.79",
-                "Top-up credits: $988.99",
-                "Total usable: $988.99",
-                "Renews: 2026-06-11T08:14:55.000Z",
-                "Manage / top up: https://portal.nousresearch.com/billing",
-            ),
-        )
-        monkeypatch.setattr(au, "_fetch_nous_credits_snapshot", lambda timeout=10.0: snap)
-        line = au.nous_credits_compact_line()
-        assert line == (
-            "Nous credits (Ultra): Total usable: $988.99 · Renews: 2026-06-11T08:14:55.000Z"
-        )
-
-    def test_none_when_no_snapshot(self, monkeypatch):
-        import agent.account_usage as au
-
-        monkeypatch.setattr(au, "_fetch_nous_credits_snapshot", lambda timeout=10.0: None)
-        assert au.nous_credits_compact_line() is None

tests/cli/test_cli_status_bar.py

@@ -457,7 +457,7 @@ class TestCLIStatusBar:
 
 
 class TestCLIUsageReport:
-    def test_show_usage_reports_real_provider_cost(self, capsys):
+    def test_show_usage_includes_estimated_cost(self, capsys):
         cli_obj = _attach_agent(
             _make_cli(),
             prompt_tokens=10_230,
@@ -469,22 +469,20 @@ class TestCLIUsageReport:
             compressions=1,
         )
         cli_obj.verbose = False
-        # Provider-reported cost (e.g. OpenRouter usage accounting accumulator).
-        cli_obj.agent.session_actual_cost_usd = 0.0640
 
         cli_obj._show_usage()
         output = capsys.readouterr().out
 
         assert "Model:" in output
-        assert "Cost (provider-reported):" in output
+        assert "Cost status:" in output
+        assert "Cost source:" in output
+        assert "Total cost:" in output
         assert "$" in output
         assert "0.064" in output
         assert "Session duration:" in output
         assert "Compressions:" in output
 
-    def test_show_usage_unreported_cost_is_not_a_dollar_figure(self, capsys):
-        """No estimation: when the provider reports nothing, /usage must NOT
-        fabricate a dollar amount — not even $0.00."""
+    def test_show_usage_marks_unknown_pricing(self, capsys):
         cli_obj = _attach_agent(
             _make_cli(model="local/my-custom-model"),
             prompt_tokens=1_000,
@@ -499,15 +497,13 @@ class TestCLIUsageReport:
         cli_obj._show_usage()
         output = capsys.readouterr().out
 
-        assert "not reported by provider" in output
-        assert "Cost (provider-reported):" not in output
-        assert "$0.00" not in output
+        assert "Total cost:" in output
+        assert "n/a" in output
+        assert "Pricing unknown for local/my-custom-model" in output
 
-    def test_show_usage_never_estimates_even_with_known_pricing(self, capsys):
-        """A model with a pricing-table entry must still show NO cost when the
-        provider reported nothing (hard requirement: real cost only)."""
+    def test_zero_priced_provider_models_stay_unknown(self, capsys):
         cli_obj = _attach_agent(
-            _make_cli(model="anthropic/claude-sonnet-4-6"),
+            _make_cli(model="glm-5"),
             prompt_tokens=1_000,
             completion_tokens=500,
             total_tokens=1_500,
@@ -520,8 +516,9 @@ class TestCLIUsageReport:
         cli_obj._show_usage()
         output = capsys.readouterr().out
 
-        assert "not reported by provider" in output
-        assert "Cost (provider-reported):" not in output
+        assert "Total cost:" in output
+        assert "n/a" in output
+        assert "Pricing unknown for glm-5" in output
 
 
 class TestStatusBarWidthSource:

tests/cli/test_worktree.py

@@ -162,11 +162,26 @@ def _has_unpushed_commits(worktree_path, timeout=10):
         return True
 
 
+def _is_dirty(wt_path, timeout=10):
+    """Test version of the worktree dirty-check helper (fail-safe True)."""
+    try:
+        result = subprocess.run(
+            ["git", "status", "--porcelain"],
+            capture_output=True, text=True, timeout=timeout, cwd=wt_path,
+        )
+        if result.returncode != 0:
+            return True
+        return bool(result.stdout.strip())
+    except Exception:
+        return True
+
+
 def _cleanup_worktree(info):
     """Test version of _cleanup_worktree.
 
-    Preserves the worktree only if it has unpushed commits.
-    Dirty working tree alone is not enough to keep it.
+    Mirrors the cli.py contract: preserves the worktree if it has
+    unpushed commits OR uncommitted changes; only deletes the branch
+    after ``git worktree remove`` succeeded.
     """
     wt_path = info["path"]
     branch = info["branch"]
@@ -178,10 +193,16 @@ def _cleanup_worktree(info):
     if _has_unpushed_commits(wt_path, timeout=10):
         return False  # Did not clean up — has unpushed commits
 
-    subprocess.run(
+    if _is_dirty(wt_path):
+        return False  # Did not clean up — uncommitted changes
+
+    result = subprocess.run(
         ["git", "worktree", "remove", wt_path, "--force"],
         capture_output=True, text=True, timeout=15, cwd=repo_root,
     )
+    if result.returncode != 0:
+        return False  # Removal failed — keep the branch
+
     subprocess.run(
         ["git", "branch", "-D", branch],
         capture_output=True, text=True, timeout=10, cwd=repo_root,
@@ -283,17 +304,18 @@ class TestWorktreeCleanup:
         assert result is True
         assert not Path(info["path"]).exists()
 
-    def test_dirty_worktree_cleaned_when_no_unpushed(self, git_repo):
-        """Dirty working tree without unpushed commits is cleaned up.
+    def test_dirty_worktree_preserved_on_cleanup(self, git_repo):
+        """Dirty working tree is preserved even without unpushed commits.
 
-        Agent sessions typically leave untracked files / artifacts behind.
-        Since all real work is in pushed commits, these don't warrant
-        keeping the worktree.
+        Uncommitted changes may be work the user has not retrieved yet —
+        cleanup must never destroy them.
         """
-        info = _setup_worktree(str(git_repo))
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
         assert info is not None
 
-        # Make uncommitted changes (untracked file)
+        # Make uncommitted changes (staged but uncommitted file)
         (Path(info["path"]) / "new-file.txt").write_text("uncommitted")
         subprocess.run(
             ["git", "add", "new-file.txt"],
@@ -301,10 +323,17 @@ class TestWorktreeCleanup:
         )
 
         # The git_repo fixture already has a fake remote ref so the initial
-        # commit is seen as "pushed".  No unpushed commits → cleanup proceeds.
-        result = _cleanup_worktree(info)
-        assert result is True  # Cleaned up despite dirty working tree
-        assert not Path(info["path"]).exists()
+        # commit is seen as "pushed" — only the dirty tree protects it.
+        cli_mod._cleanup_worktree(info)
+        assert Path(info["path"]).exists()  # Preserved despite no unpushed commits
+
+        # Branch and lock are kept too
+        result = subprocess.run(
+            ["git", "branch", "--list", info["branch"]],
+            capture_output=True, text=True, cwd=str(git_repo),
+        )
+        assert info["branch"] in result.stdout
+        assert cli_mod._worktree_is_locked(str(git_repo), info["path"]) is True
 
     def test_worktree_with_unpushed_commits_kept(self, git_repo):
         """Worktree with unpushed commits is preserved."""
@@ -728,47 +757,224 @@ class TestStaleWorktreePruning:
         assert not Path(info["path"]).exists()
 
     def test_force_prunes_very_old_worktree(self, git_repo):
-        """Worktrees older than 72h should be force-pruned regardless."""
+        """Very old (>72h) CLEAN, unlocked, fully-pushed worktrees are pruned."""
         import time
+        import cli as cli_mod
 
-        info = _setup_worktree(str(git_repo))
+        info = cli_mod._setup_worktree(str(git_repo))
         assert info is not None
 
-        # Make an unpushed commit (would normally protect it)
-        (Path(info["path"]) / "work.txt").write_text("stale work")
-        subprocess.run(["git", "add", "work.txt"], cwd=info["path"], capture_output=True)
-        subprocess.run(
-            ["git", "commit", "-m", "old agent work"],
-            cwd=info["path"], capture_output=True,
-        )
+        # _setup_worktree locks the worktree; unlock to simulate a worktree
+        # whose owning session released it (clean + unlocked + pushed).
+        assert cli_mod._unlock_worktree(str(git_repo), info["path"]) is True
 
-        # Make it very old (73h — beyond the 72h hard threshold)
+        # Make it very old (73h)
         old_time = time.time() - (73 * 3600)
         os.utime(info["path"], (old_time, old_time))
 
-        # Simulate the force-prune tier check
-        hard_cutoff = time.time() - (72 * 3600)
-        mtime = Path(info["path"]).stat().st_mtime
-        assert mtime <= hard_cutoff  # Should qualify for force removal
-
-        # Actually remove it (simulates _prune_stale_worktrees force path)
-        branch_result = subprocess.run(
-            ["git", "branch", "--show-current"],
-            capture_output=True, text=True, timeout=5, cwd=info["path"],
-        )
-        branch = branch_result.stdout.strip()
-
-        subprocess.run(
-            ["git", "worktree", "remove", info["path"], "--force"],
-            capture_output=True, text=True, timeout=15, cwd=str(git_repo),
-        )
-        if branch:
-            subprocess.run(
-                ["git", "branch", "-D", branch],
-                capture_output=True, text=True, timeout=10, cwd=str(git_repo),
-            )
+        cli_mod._prune_stale_worktrees(str(git_repo))
 
         assert not Path(info["path"]).exists()
+        # Branch should be gone too
+        result = subprocess.run(
+            ["git", "branch", "--list", info["branch"]],
+            capture_output=True, text=True, cwd=str(git_repo),
+        )
+        assert info["branch"] not in result.stdout
+
+
+class TestWorktreeLocking:
+    """Test git-native worktree locks and the preserve-work contracts.
+
+    These tests exercise the REAL cli.py implementations (not the local
+    reimplementations above), matching the pattern in
+    test_worktree_security.py.
+    """
+
+    def test_setup_worktree_locks(self, git_repo):
+        """_setup_worktree leaves the new worktree locked."""
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
+        assert info is not None
+
+        # Verify via git worktree list --porcelain: the stanza for this
+        # worktree must contain a "locked" line.
+        result = subprocess.run(
+            ["git", "worktree", "list", "--porcelain"],
+            capture_output=True, text=True, cwd=str(git_repo),
+        )
+        target = Path(info["path"]).resolve()
+        current = None
+        locked = False
+        for line in result.stdout.splitlines():
+            if line.startswith("worktree "):
+                current = Path(line[len("worktree "):].strip()).resolve()
+            elif line == "locked" or line.startswith("locked "):
+                if current == target:
+                    locked = True
+        assert locked
+        assert cli_mod._worktree_is_locked(str(git_repo), info["path"]) is True
+
+    def test_unlock_worktree(self, git_repo):
+        """_unlock_worktree releases the lock taken by _setup_worktree."""
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
+        assert info is not None
+        assert cli_mod._worktree_is_locked(str(git_repo), info["path"]) is True
+
+        assert cli_mod._unlock_worktree(str(git_repo), info["path"]) is True
+        assert cli_mod._worktree_is_locked(str(git_repo), info["path"]) is False
+
+    def test_prune_skips_locked_very_old_clean_worktree(self, git_repo):
+        """A locked worktree is never pruned, even >72h old and clean."""
+        import time
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
+        assert info is not None
+        # Still locked from _setup_worktree; clean; fully pushed.
+
+        old_time = time.time() - (80 * 3600)
+        os.utime(info["path"], (old_time, old_time))
+
+        cli_mod._prune_stale_worktrees(str(git_repo))
+
+        assert Path(info["path"]).exists()
+
+    def test_prune_skips_old_dirty_unlocked_worktree(self, git_repo):
+        """An old dirty worktree is not pruned even when unlocked."""
+        import time
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
+        assert info is not None
+        assert cli_mod._unlock_worktree(str(git_repo), info["path"]) is True
+
+        # Uncommitted change (untracked file)
+        (Path(info["path"]) / "wip.txt").write_text("uncommitted work")
+
+        old_time = time.time() - (25 * 3600)
+        os.utime(info["path"], (old_time, old_time))
+
+        cli_mod._prune_stale_worktrees(str(git_repo))
+
+        assert Path(info["path"]).exists()
+        assert (Path(info["path"]) / "wip.txt").exists()
+
+    def test_prune_preserves_very_old_worktree_with_unpushed_commits(self, git_repo):
+        """Unpushed commits protect a worktree at ANY age — the old >72h
+        force-remove tier is gone."""
+        import time
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
+        assert info is not None
+        assert cli_mod._unlock_worktree(str(git_repo), info["path"]) is True
+
+        # Unpushed commit (clean tree afterwards)
+        (Path(info["path"]) / "work.txt").write_text("real work")
+        subprocess.run(["git", "add", "work.txt"], cwd=info["path"], capture_output=True)
+        subprocess.run(
+            ["git", "commit", "-m", "agent work"],
+            cwd=info["path"], capture_output=True,
+        )
+
+        old_time = time.time() - (80 * 3600)
+        os.utime(info["path"], (old_time, old_time))
+
+        cli_mod._prune_stale_worktrees(str(git_repo))
+
+        assert Path(info["path"]).exists()
+        result = subprocess.run(
+            ["git", "branch", "--list", info["branch"]],
+            capture_output=True, text=True, cwd=str(git_repo),
+        )
+        assert info["branch"] in result.stdout
+
+    def test_cleanup_preserves_dirty_worktree(self, git_repo):
+        """_cleanup_worktree keeps a dirty worktree (untracked file)."""
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
+        assert info is not None
+
+        (Path(info["path"]) / "scratch.txt").write_text("not yet committed")
+
+        cli_mod._cleanup_worktree(info)
+
+        assert Path(info["path"]).exists()
+        assert (Path(info["path"]) / "scratch.txt").exists()
+
+    def test_cleanup_removes_clean_locked_worktree(self, git_repo):
+        """_cleanup_worktree unlocks then removes a clean, pushed worktree."""
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
+        assert info is not None
+        assert cli_mod._worktree_is_locked(str(git_repo), info["path"]) is True
+
+        cli_mod._cleanup_worktree(info)
+
+        assert not Path(info["path"]).exists()
+        result = subprocess.run(
+            ["git", "branch", "--list", info["branch"]],
+            capture_output=True, text=True, cwd=str(git_repo),
+        )
+        assert info["branch"] not in result.stdout
+
+    def test_branch_kept_when_worktree_remove_fails(self, git_repo, monkeypatch):
+        """If `git worktree remove` fails, the branch must NOT be deleted."""
+        import subprocess as sp
+        import cli as cli_mod
+
+        info = cli_mod._setup_worktree(str(git_repo))
+        assert info is not None
+
+        real_run = sp.run
+
+        def fake_run(cmd, *args, **kwargs):
+            if (
+                isinstance(cmd, (list, tuple))
+                and list(cmd[:3]) == ["git", "worktree", "remove"]
+            ):
+                return sp.CompletedProcess(
+                    cmd, returncode=1, stdout="", stderr="simulated removal failure"
+                )
+            return real_run(cmd, *args, **kwargs)
+
+        monkeypatch.setattr(sp, "run", fake_run)
+
+        cli_mod._cleanup_worktree(info)
+
+        monkeypatch.undo()
+
+        # Worktree dir still present, branch NOT deleted
+        assert Path(info["path"]).exists()
+        result = subprocess.run(
+            ["git", "branch", "--list", info["branch"]],
+            capture_output=True, text=True, cwd=str(git_repo),
+        )
+        assert info["branch"] in result.stdout
+
+    def test_worktree_is_locked_fail_safe(self, tmp_path):
+        """_worktree_is_locked returns True (fail safe) on a bogus repo_root."""
+        import cli as cli_mod
+
+        bogus = tmp_path / "does-not-exist"
+        assert cli_mod._worktree_is_locked(str(bogus), str(bogus / "wt")) is True
+
+        # An existing directory that is not a git repo is also an error case
+        not_repo = tmp_path / "not-a-repo"
+        not_repo.mkdir()
+        assert cli_mod._worktree_is_locked(str(not_repo), str(not_repo / "wt")) is True
+
+    def test_worktree_is_dirty_fail_safe(self, tmp_path):
+        """_worktree_is_dirty returns True (fail safe) on a bogus path."""
+        import cli as cli_mod
+
+        assert cli_mod._worktree_is_dirty(str(tmp_path / "missing")) is True
 
 
 class TestEdgeCases:

tests/gateway/test_mattermost.py

@@ -6,7 +6,10 @@ import pytest
 from unittest.mock import MagicMock, patch, AsyncMock
 
 from gateway.config import Platform, PlatformConfig
-from gateway.run import _resolve_progress_thread_id
+from gateway.run import (
+    _resolve_gateway_display_bool,
+    _resolve_progress_thread_id,
+)
 
 
 class TestMattermostProgressThreadRouting:
@@ -32,6 +35,97 @@ class TestMattermostProgressThreadRouting:
         ) is None
 
 
+class TestMattermostDisplayHygiene:
+    def test_mattermost_requires_platform_opt_in_for_interim_assistant_messages(self):
+        """Global interim commentary must not make Mattermost leak scratch notes."""
+        user_config = {"display": {"interim_assistant_messages": True}}
+
+        assert _resolve_gateway_display_bool(
+            user_config,
+            "mattermost",
+            "interim_assistant_messages",
+            default=True,
+            platform=Platform.MATTERMOST,
+            require_platform_override_for={Platform.MATTERMOST},
+        ) is False
+
+    def test_mattermost_platform_opt_in_can_enable_interim_assistant_messages(self):
+        """Mattermost can still opt into commentary explicitly per platform."""
+        user_config = {
+            "display": {
+                "interim_assistant_messages": False,
+                "platforms": {
+                    "mattermost": {"interim_assistant_messages": True},
+                },
+            }
+        }
+
+        assert _resolve_gateway_display_bool(
+            user_config,
+            "mattermost",
+            "interim_assistant_messages",
+            default=True,
+            platform=Platform.MATTERMOST,
+            require_platform_override_for={Platform.MATTERMOST},
+        ) is True
+
+    def test_mattermost_requires_platform_opt_in_for_thinking_progress(self):
+        """Global thinking_progress must not surface internal analysis in Mattermost."""
+        user_config = {"display": {"thinking_progress": True}}
+
+        assert _resolve_gateway_display_bool(
+            user_config,
+            "mattermost",
+            "thinking_progress",
+            default=False,
+            platform=Platform.MATTERMOST,
+            require_platform_override_for={Platform.MATTERMOST},
+        ) is False
+
+    def test_mattermost_requires_platform_opt_in_for_show_reasoning(self):
+        """Global show_reasoning must not prepend scratch reasoning in Mattermost."""
+        user_config = {"display": {"show_reasoning": True}}
+
+        assert _resolve_gateway_display_bool(
+            user_config,
+            "mattermost",
+            "show_reasoning",
+            default=False,
+            platform=Platform.MATTERMOST,
+            require_platform_override_for={Platform.MATTERMOST},
+        ) is False
+
+    def test_mattermost_platform_opt_in_can_enable_show_reasoning(self):
+        user_config = {
+            "display": {
+                "show_reasoning": False,
+                "platforms": {"mattermost": {"show_reasoning": True}},
+            }
+        }
+
+        assert _resolve_gateway_display_bool(
+            user_config,
+            "mattermost",
+            "show_reasoning",
+            default=False,
+            platform=Platform.MATTERMOST,
+            require_platform_override_for={Platform.MATTERMOST},
+        ) is True
+
+    def test_global_thinking_progress_still_applies_to_other_platforms(self):
+        """The Mattermost guard must not silently neuter Telegram/other chats."""
+        user_config = {"display": {"thinking_progress": True}}
+
+        assert _resolve_gateway_display_bool(
+            user_config,
+            "telegram",
+            "thinking_progress",
+            default=False,
+            platform=Platform.TELEGRAM,
+            require_platform_override_for={Platform.MATTERMOST},
+        ) is True
+
+
 # ---------------------------------------------------------------------------
 # Platform & Config
 # ---------------------------------------------------------------------------
@@ -347,6 +441,24 @@ class TestMattermostSend:
         payload = self.adapter._api_post.call_args_list[0][0][1]
         assert payload["root_id"] == "root_post"
 
+    @pytest.mark.asyncio
+    async def test_progress_send_with_invalid_thread_root_never_falls_back_flat(self):
+        """Tool/status/progress bubbles must stay quiet when the thread is broken."""
+        self.adapter._reply_mode = "thread"
+        self.adapter._api_get = AsyncMock(return_value={"id": "bad_root", "root_id": ""})
+        self.adapter._api_post = AsyncMock(return_value={})
+
+        result = await self.adapter.send(
+            "channel_1",
+            "⚙️ terminal...",
+            metadata={"thread_id": "bad_root"},
+        )
+
+        assert result.success is False
+        assert self.adapter._api_post.call_count == 1
+        payload = self.adapter._api_post.call_args_list[0][0][1]
+        assert payload["root_id"] == "bad_root"
+
     @pytest.mark.asyncio
     async def test_send_api_failure(self):
         """When API returns error, send should return failure."""

tests/gateway/test_media_metadata_contract.py

@@ -0,0 +1,80 @@
+"""Contract: media-send overrides must accept the ``metadata`` kwarg.
+
+``BasePlatformAdapter.send_multiple_images`` passes ``metadata=metadata``
+to ``send_image`` / ``send_image_file`` / ``send_animation`` on every send.
+An override whose signature stops at ``reply_to`` raises ``TypeError:
+send_image() got an unexpected keyword argument 'metadata'`` at runtime —
+which is exactly how image delivery broke on WhatsApp and email.
+
+This mirrors ``test_discord_media_metadata.py`` but covers the two
+adapters that previously slipped, plus a best-effort sweep over every
+adapter that imports cleanly so the next slip is caught at test time.
+"""
+
+from __future__ import annotations
+
+import importlib
+import inspect
+
+import pytest
+
+
+def _accepts_metadata(method) -> bool:
+    params = inspect.signature(method).parameters
+    if "metadata" in params:
+        return True
+    # A ``**kwargs`` catch-all also absorbs metadata (the convention used by
+    # WhatsApp's send_video / send_voice / send_document overrides).
+    return any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values())
+
+
+# (module, class) for the two adapters this fix targeted. These must import
+# in CI, so assert directly rather than skipping.
+@pytest.mark.parametrize(
+    "module_name, class_name",
+    [
+        ("gateway.platforms.whatsapp", "WhatsAppAdapter"),
+        ("gateway.platforms.email", "EmailAdapter"),
+    ],
+)
+def test_send_image_accepts_metadata(module_name, class_name):
+    cls = getattr(importlib.import_module(module_name), class_name)
+    assert _accepts_metadata(cls.send_image), (
+        f"{class_name}.send_image must accept 'metadata' (or **kwargs) — "
+        f"send_multiple_images passes it on every send"
+    )
+
+
+# Best-effort sweep across all shipped adapters. Modules whose optional
+# platform SDK isn't installed are skipped; an adapter that imports but
+# whose override drops metadata is a hard failure.
+_ALL_ADAPTERS = [
+    ("gateway.platforms.bluebubbles", "BlueBubblesAdapter"),
+    ("gateway.platforms.dingtalk", "DingTalkAdapter"),
+    ("gateway.platforms.discord", "DiscordAdapter"),
+    ("gateway.platforms.email", "EmailAdapter"),
+    ("gateway.platforms.feishu", "FeishuAdapter"),
+    ("gateway.platforms.matrix", "MatrixAdapter"),
+    ("gateway.platforms.mattermost", "MattermostAdapter"),
+    ("gateway.platforms.signal", "SignalAdapter"),
+    ("gateway.platforms.slack", "SlackAdapter"),
+    ("gateway.platforms.telegram", "TelegramAdapter"),
+    ("gateway.platforms.wecom", "WeComAdapter"),
+    ("gateway.platforms.weixin", "WeixinAdapter"),
+    ("gateway.platforms.whatsapp", "WhatsAppAdapter"),
+    ("gateway.platforms.yuanbao", "YuanbaoAdapter"),
+]
+
+
+@pytest.mark.parametrize("module_name, class_name", _ALL_ADAPTERS)
+def test_all_adapters_send_image_metadata_sweep(module_name, class_name):
+    try:
+        module = importlib.import_module(module_name)
+    except Exception as exc:  # optional platform dep not installed
+        pytest.skip(f"{module_name} not importable: {exc}")
+    cls = getattr(module, class_name, None)
+    if cls is None or "send_image" not in cls.__dict__:
+        pytest.skip(f"{class_name} has no send_image override")
+    assert _accepts_metadata(cls.send_image), (
+        f"{class_name}.send_image drops the 'metadata' kwarg"
+    )

tests/gateway/test_stream_consumer_thread_routing.py

@@ -106,6 +106,42 @@ class TestInitialReplyToId:
         assert call_kwargs["metadata"] == {**metadata, "expect_edits": True}
         assert metadata == {"thread_id": "omt_topic789"}
 
+    @pytest.mark.asyncio
+    async def test_final_first_send_marks_metadata_notify_true(self):
+        """Final streaming sends should use the existing notify=True marker."""
+        adapter = _make_adapter()
+        consumer = GatewayStreamConsumer(
+            adapter,
+            "chat_123",
+            metadata={"thread_id": "root_post_123"},
+            initial_reply_to_id="reply_post_456",
+        )
+
+        await consumer._send_or_edit("Final answer", finalize=True)
+
+        call_kwargs = adapter.send.call_args[1]
+        metadata = call_kwargs["metadata"]
+        assert metadata["thread_id"] == "root_post_123"
+        assert metadata["notify"] is True
+        assert "delivery_kind" not in metadata
+        assert "allow_flat_fallback" not in metadata
+
+    @pytest.mark.asyncio
+    async def test_nonfinal_first_send_does_not_mark_notify(self):
+        """Preview/interim streaming sends must not be notify-worthy."""
+        adapter = _make_adapter()
+        consumer = GatewayStreamConsumer(
+            adapter,
+            "chat_123",
+            metadata={"thread_id": "root_post_123"},
+            initial_reply_to_id="reply_post_456",
+        )
+
+        await consumer._send_or_edit("Preview", finalize=False)
+
+        metadata = adapter.send.call_args[1]["metadata"]
+        assert metadata == {"thread_id": "root_post_123", "expect_edits": True}
+
 
 class TestOverflowFirstMessage:
     """Verify thread routing is preserved when the first message overflows."""

tests/gateway/test_telegram_overflow_partial.py

@@ -134,7 +134,7 @@ async def test_stream_consumer_fallback_sends_tail_after_partial_overflow():
 
     adapter.send.assert_awaited_once()
     assert adapter.send.await_args.kwargs["content"] == "world"
-    assert adapter.send.await_args.kwargs["metadata"] == {"thread_id": "77"}
+    assert adapter.send.await_args.kwargs["metadata"] == {"thread_id": "77", "notify": True}
     adapter.delete_message.assert_not_awaited()
     assert consumer.final_response_sent is True
     assert consumer.final_content_delivered is True

tests/gateway/test_tts_media_routing.py

@@ -76,7 +76,7 @@ async def test_base_adapter_routes_telegram_flac_media_tag_to_document_sender(tm
     adapter.send_document.assert_awaited_once_with(
         chat_id="chat-1",
         file_path=str(media_file),
-        metadata=None,
+        metadata={"notify": True},
     )
     adapter.send_voice.assert_not_awaited()
 
@@ -95,7 +95,7 @@ async def test_base_adapter_routes_non_voice_telegram_ogg_media_tag_to_document_
     adapter.send_document.assert_awaited_once_with(
         chat_id="chat-1",
         file_path=str(media_file),
-        metadata=None,
+        metadata={"notify": True},
     )
     adapter.send_voice.assert_not_awaited()
 
@@ -116,7 +116,7 @@ async def test_base_adapter_routes_voice_tagged_telegram_ogg_media_tag_to_voice_
     adapter.send_voice.assert_awaited_once_with(
         chat_id="chat-1",
         audio_path=str(media_file),
-        metadata=None,
+        metadata={"notify": True},
     )
     adapter.send_document.assert_not_awaited()
 

tests/gateway/test_usage_command.py

@@ -21,16 +21,11 @@ def _make_mock_agent(**overrides):
         "session_output_tokens": 10_000,
         "session_cache_read_tokens": 5_000,
         "session_cache_write_tokens": 2_000,
-        # Real provider-reported cost: None = nothing reported (the default).
-        "session_actual_cost_usd": None,
     }
     defaults.update(overrides)
     for k, v in defaults.items():
         setattr(agent, k, v)
 
-    # No Nous credits headers seen unless a test overrides this.
-    agent.get_credits_spent_micros = MagicMock(return_value=None)
-
     # Rate limit state
     rl = MagicMock()
     rl.has_data = True
@@ -77,11 +72,13 @@ class TestUsageCachedAgent:
 
     @pytest.mark.asyncio
     async def test_cached_agent_shows_detailed_usage(self):
-        agent = _make_mock_agent(session_actual_cost_usd=0.1234)
+        agent = _make_mock_agent()
         runner = _make_runner(SK, cached_agent=agent)
         event = MagicMock()
 
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
+        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"), \
+             patch("agent.usage_pricing.estimate_usage_cost") as mock_cost:
+            mock_cost.return_value = MagicMock(amount_usd=0.1234, status="estimated")
             result = await runner._handle_usage_command(event)
 
         assert "claude-sonnet-4.6" in result
@@ -102,7 +99,9 @@ class TestUsageCachedAgent:
         runner = _make_runner(SK, agent=running, cached_agent=cached)
         event = MagicMock()
 
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
+        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"), \
+             patch("agent.usage_pricing.estimate_usage_cost") as mock_cost:
+            mock_cost.return_value = MagicMock(amount_usd=None, status="unknown")
             result = await runner._handle_usage_command(event)
 
         assert "80,000" in result   # running agent's total
@@ -118,7 +117,9 @@ class TestUsageCachedAgent:
         runner._running_agents[SK] = _AGENT_PENDING_SENTINEL
         event = MagicMock()
 
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
+        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"), \
+             patch("agent.usage_pricing.estimate_usage_cost") as mock_cost:
+            mock_cost.return_value = MagicMock(amount_usd=None, status="unknown")
             result = await runner._handle_usage_command(event)
 
         assert "claude-sonnet-4.6" in result
@@ -152,7 +153,9 @@ class TestUsageCachedAgent:
         runner = _make_runner(SK, cached_agent=agent)
         event = MagicMock()
 
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
+        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"), \
+             patch("agent.usage_pricing.estimate_usage_cost") as mock_cost:
+            mock_cost.return_value = MagicMock(amount_usd=None, status="unknown")
             result = await runner._handle_usage_command(event)
 
         assert "Cache read" not in result
@@ -165,7 +168,9 @@ class TestUsageCachedAgent:
         runner = _make_runner(SK, cached_agent=agent)
         event = MagicMock()
 
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
+        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"), \
+             patch("agent.usage_pricing.estimate_usage_cost") as mock_cost:
+            mock_cost.return_value = MagicMock(amount_usd=None, status="included")
             result = await runner._handle_usage_command(event)
 
         assert "Cost: included" in result
@@ -194,7 +199,9 @@ class TestUsageAccountSection:
                 "Session: 85% remaining (15% used)",
             ],
         )
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
+        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"), \
+             patch("agent.usage_pricing.estimate_usage_cost") as mock_cost:
+            mock_cost.return_value = MagicMock(amount_usd=None, status="included")
             result = await runner._handle_usage_command(event)
 
         assert "📊 **Session Token Usage**" in result
@@ -249,42 +256,3 @@ class TestUsageAccountSection:
         assert account_call["kwargs"]["base_url"] == "https://chatgpt.com/backend-api/codex"
         assert "📊 **Session Info**" in result
         assert "📈 **Account limits**" in result
-
-
-class TestUsageRealCostOnly:
-    """Cost lines are provider-REPORTED only — never estimated, never $0.00."""
-
-    @pytest.mark.asyncio
-    async def test_unreported_cost_renders_no_cost_line(self):
-        agent = _make_mock_agent()  # openrouter, nothing reported
-        runner = _make_runner(SK, cached_agent=agent)
-        event = MagicMock()
-
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
-            result = await runner._handle_usage_command(event)
-
-        assert "Cost:" not in result
-        assert "$0.00" not in result
-
-    @pytest.mark.asyncio
-    async def test_nous_credits_delta_renders_as_cost(self):
-        agent = _make_mock_agent(provider="nous", model="Hermes-4.1-405B")
-        agent.get_credits_spent_micros = MagicMock(return_value=123_400)
-        runner = _make_runner(SK, cached_agent=agent)
-        event = MagicMock()
-
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
-            result = await runner._handle_usage_command(event)
-
-        assert "$0.1234" in result
-
-    @pytest.mark.asyncio
-    async def test_openrouter_reported_cost_renders(self):
-        agent = _make_mock_agent(session_actual_cost_usd=0.9876)
-        runner = _make_runner(SK, cached_agent=agent)
-        event = MagicMock()
-
-        with patch("agent.rate_limit_tracker.format_rate_limit_compact", return_value="RPM: 50/60"):
-            result = await runner._handle_usage_command(event)
-
-        assert "$0.9876" in result

tests/providers/test_provider_profiles.py

@@ -110,9 +110,7 @@ class TestOpenRouterProfile:
     def test_extra_body_no_prefs(self):
         p = get_provider_profile("openrouter")
         body = p.build_extra_body()
-        # Usage accounting is always requested (real provider-reported cost);
-        # nothing else should appear without prefs/session.
-        assert body == {"usage": {"include": True}}
+        assert body == {}
 
     def test_pareto_min_coding_score_emitted_for_pareto_model(self):
         """min_coding_score → plugins block when model is openrouter/pareto-code."""

tests/run_agent/test_async_httpx_del_neuter.py

@@ -176,11 +176,16 @@ class TestClientCacheBoundedGrowth:
         """When the loop changes, the old entry should be replaced, not duplicated."""
         from agent.auxiliary_client import (
             _client_cache,
+            _client_cache_key,
             _client_cache_lock,
             _get_cached_client,
         )
 
-        key = ("test_replace", True, "", "", "", (), False, "")
+        key = _client_cache_key(
+            "test_replace",
+            async_mode=True,
+            task="",
+        )
 
         # Simulate a stale entry from a closed loop
         old_loop = asyncio.new_event_loop()

tests/test_hermes_state.py

@@ -158,37 +158,6 @@ class TestSessionLifecycle:
         assert session["api_call_count"] == 5
         assert session["input_tokens"] == 300
 
-    def test_update_token_counts_actual_cost_null_keeps_value(self, db):
-        """A NULL actual_cost_usd delta must not touch the stored REAL cost."""
-        db.create_session(session_id="s1", source="cli")
-        db.update_token_counts("s1", input_tokens=100, actual_cost_usd=0.25)
-        db.update_token_counts("s1", input_tokens=100, actual_cost_usd=None)
-        db.update_token_counts("s1", input_tokens=100, actual_cost_usd=0.10)
-
-        session = db.get_session("s1")
-        assert session["actual_cost_usd"] == pytest.approx(0.35)
-
-    def test_usage_totals_reported_cost_none_when_nothing_reported(self, db):
-        """usage_totals must distinguish 'no reported cost' (None) from $0."""
-        db.create_session(session_id="s1", source="cli")
-        db.update_token_counts("s1", input_tokens=100, output_tokens=50, api_call_count=1)
-
-        totals = db.usage_totals(days=30)
-        assert totals["sessions"] == 1
-        assert totals["input_tokens"] == 100
-        assert totals["output_tokens"] == 50
-        assert totals["reported_cost_usd"] is None
-
-    def test_usage_totals_sums_reported_costs(self, db):
-        db.create_session(session_id="s1", source="cli")
-        db.create_session(session_id="s2", source="tui")
-        db.update_token_counts("s1", input_tokens=100, actual_cost_usd=0.20)
-        db.update_token_counts("s2", input_tokens=300, actual_cost_usd=0.05)
-
-        totals = db.usage_totals(days=30)
-        assert totals["sessions"] == 2
-        assert totals["reported_cost_usd"] == pytest.approx(0.25)
-
     def test_update_token_counts_backfills_model_when_null(self, db):
         db.create_session(session_id="s1", source="telegram")
         db.update_token_counts("s1", input_tokens=10, output_tokens=5, model="openai/gpt-5.4")

tests/test_tui_gateway_server.py

@@ -7,7 +7,6 @@ import time
 import types
 from datetime import datetime
 from pathlib import Path
-import pytest
 from unittest.mock import patch
 
 from hermes_constants import reset_hermes_home_override, set_hermes_home_override
@@ -7498,183 +7497,3 @@ def test_reap_idle_sessions_closes_only_evictable(monkeypatch):
         assert closed == [("stale", "idle_timeout")]
     finally:
         server._sessions.clear()
-
-
-# ── /usage: compact in-process page with real-only costs ─────────────────────
-
-
-def _usage_agent(**overrides):
-    """SimpleNamespace agent with realistic session counters for /usage."""
-    base = dict(
-        model="anthropic/claude-sonnet-4.6",
-        provider="openrouter",
-        base_url="https://openrouter.ai/api/v1",
-        session_input_tokens=35_000,
-        session_output_tokens=10_000,
-        session_cache_read_tokens=5_000,
-        session_cache_write_tokens=2_000,
-        session_reasoning_tokens=0,
-        session_prompt_tokens=40_000,
-        session_completion_tokens=10_000,
-        session_total_tokens=50_000,
-        session_api_calls=5,
-        session_actual_cost_usd=None,
-        session_model_usage={},
-        context_compressor=None,
-    )
-    base.update(overrides)
-    return types.SimpleNamespace(**base)
-
-
-def _mute_usage_externals(monkeypatch):
-    import agent.account_usage as account_usage
-
-    monkeypatch.setattr(server, "_get_db", lambda: None)
-    monkeypatch.setattr(account_usage, "nous_credits_compact_line", lambda **kw: None)
-
-
-def test_get_usage_cost_absent_when_provider_reports_nothing(monkeypatch):
-    """No estimation: even a model with known pricing gets NO cost_usd field."""
-    agent = _usage_agent()
-    usage = server._get_usage(agent)
-    assert "cost_usd" not in usage
-    assert usage["input"] == 35_000
-
-
-def test_get_usage_cost_absent_for_openrouter_accumulator_chrome(monkeypatch):
-    """F3: the chrome status bar shows cost ONLY from the Nous header delta.
-    An OpenRouter usage.cost accumulator alone (no x-nous-credits header) must
-    NOT surface cost_usd in _get_usage — the chrome hides cost off-Nous. (The
-    /usage accounting page still counts the accumulator; that's a separate path.)
-    """
-    agent = _usage_agent(session_actual_cost_usd=0.4321)
-    usage = server._get_usage(agent)
-    assert "cost_usd" not in usage
-
-
-def test_get_usage_cost_from_nous_credits_delta(monkeypatch):
-    agent = _usage_agent(provider="nous")
-    agent.get_credits_spent_micros = lambda: 250_000  # $0.25 real header delta
-    usage = server._get_usage(agent)
-    assert usage["cost_usd"] == pytest.approx(0.25)
-    assert usage["cost_status"] == "actual"
-
-
-def test_compact_usage_per_model_rows_and_real_cost(monkeypatch):
-    _mute_usage_externals(monkeypatch)
-    agent = _usage_agent(
-        session_actual_cost_usd=0.42,
-        session_model_usage={
-            "anthropic/claude-sonnet-4.6": {
-                "calls": 4, "input": 30_000, "output": 9_000,
-                "cache_read": 5_000, "cache_write": 2_000, "cost_usd": 0.42,
-            },
-            "deepseek/deepseek-chat": {
-                "calls": 1, "input": 5_000, "output": 1_000,
-                "cache_read": 0, "cache_write": 0, "cost_usd": None,
-            },
-        },
-    )
-    text = server._compact_usage_text(_session(agent=agent))
-
-    assert "Session — anthropic/claude-sonnet-4.6 (openrouter)" in text
-    sonnet_row = next(l for l in text.splitlines() if "claude-sonnet-4.6" in l and "reqs" in l)
-    assert "reqs 4" in sonnet_row and "$0.4200" in sonnet_row
-    deepseek_row = next(l for l in text.splitlines() if "deepseek-chat" in l)
-    # Cost not reported for this model → no dollar figure on its row.
-    assert "reqs 1" in deepseek_row and "$" not in deepseek_row
-    assert "session cost: $0.4200 (provider-reported)" in text
-    assert "/usage full" in text
-
-
-def test_compact_usage_absent_cost_never_renders_zero(monkeypatch):
-    _mute_usage_externals(monkeypatch)
-    agent = _usage_agent()  # nothing reported
-    text = server._compact_usage_text(_session(agent=agent))
-    assert "session cost: not reported by provider" in text
-    assert "$0.00" not in text
-
-
-def test_compact_usage_no_agent(monkeypatch):
-    _mute_usage_externals(monkeypatch)
-    text = server._compact_usage_text(_session(agent=None) | {"agent": None})
-    assert "no API calls yet" in text
-
-
-def test_compact_usage_recent_summary_and_credits_line(monkeypatch):
-    import agent.account_usage as account_usage
-
-    class _DB:
-        def usage_totals(self, days=30):
-            return {
-                "days": 30, "sessions": 12, "input_tokens": 1_200_000,
-                "output_tokens": 90_000, "api_calls": 64,
-                "reported_cost_usd": 4.5678,
-            }
-
-    monkeypatch.setattr(server, "_get_db", lambda: _DB())
-    monkeypatch.setattr(
-        account_usage, "nous_credits_compact_line",
-        lambda **kw: "Nous credits (Ultra): Total usable: $988.99 · Renews: 2026-06-11",
-    )
-    text = server._compact_usage_text(_session(agent=_usage_agent()))
-    assert "Last 30d: 12 sessions" in text
-    assert "reported cost $4.57" in text
-    assert "Nous credits (Ultra)" in text
-
-
-def test_compact_usage_recent_summary_hides_unreported_cost(monkeypatch):
-    _mute_usage_externals(monkeypatch)
-
-    class _DB:
-        def usage_totals(self, days=30):
-            return {
-                "days": 30, "sessions": 3, "input_tokens": 10_000,
-                "output_tokens": 2_000, "api_calls": 7,
-                "reported_cost_usd": None,
-            }
-
-    monkeypatch.setattr(server, "_get_db", lambda: _DB())
-    text = server._compact_usage_text(_session(agent=_usage_agent()))
-    assert "Last 30d: 3 sessions" in text
-    assert "reported cost" not in text
-
-
-def test_slash_exec_usage_is_answered_in_process(monkeypatch):
-    """/usage must not hit the slash worker (it has no live agent)."""
-    _mute_usage_externals(monkeypatch)
-    server._sessions["sid-usage"] = _session(agent=_usage_agent())
-    try:
-        resp = server.handle_request(
-            {"id": "1", "method": "slash.exec",
-             "params": {"session_id": "sid-usage", "command": "usage"}}
-        )
-        out = resp["result"]["output"]
-        assert "Session — anthropic/claude-sonnet-4.6" in out
-        # Worker untouched.
-        assert server._sessions["sid-usage"]["slash_worker"] is None
-    finally:
-        server._sessions.pop("sid-usage", None)
-
-
-def test_slash_exec_usage_full_falls_through_to_worker(monkeypatch):
-    ran = []
-
-    class _Worker:
-        def run(self, cmd):
-            ran.append(cmd)
-            return "detailed legacy page"
-
-    sess = _session(agent=_usage_agent())
-    sess["slash_worker"] = _Worker()
-    server._sessions["sid-usage-full"] = sess
-    try:
-        monkeypatch.setattr(server, "_mirror_slash_side_effects", lambda *a: "")
-        resp = server.handle_request(
-            {"id": "1", "method": "slash.exec",
-             "params": {"session_id": "sid-usage-full", "command": "usage full"}}
-        )
-        assert resp["result"]["output"] == "detailed legacy page"
-        assert ran == ["usage full"]
-    finally:
-        server._sessions.pop("sid-usage-full", None)

tui_gateway/server.py

@@ -2281,20 +2281,23 @@ def _get_usage(agent) -> dict:
             usage["context_max"] = ctx_max
             usage["context_percent"] = max(0, min(100, round(ctx_used / ctx_max * 100)))
         usage["compressions"] = getattr(comp, "compression_count", 0) or 0
-    # Cost (chrome status bar): Nous portal header delta ONLY (F3, glitch
-    # 2026-06-13). The OpenRouter usage.cost accumulator is deliberately NOT
-    # used here — per-model cache/input/output pricing is unreliable across the
-    # model long tail, so the bar shows cost ONLY on a Nous-portal session and
-    # hides it everywhere else. `cost_usd` is ABSENT (not $0.00) when no header
-    # was seen, and the TUI hides its cost segment. (The /usage accounting page
-    # still uses real_session_cost_usd — both provider-reported sources.)
     try:
-        from agent.usage_pricing import nous_header_cost_usd
+        from agent.usage_pricing import CanonicalUsage, estimate_usage_cost
 
-        real_cost = nous_header_cost_usd(agent)
-        if real_cost is not None:
-            usage["cost_usd"] = real_cost
-            usage["cost_status"] = "actual"
+        cost = estimate_usage_cost(
+            usage["model"],
+            CanonicalUsage(
+                input_tokens=usage["input"],
+                output_tokens=usage["output"],
+                cache_read_tokens=usage["cache_read"],
+                cache_write_tokens=usage["cache_write"],
+            ),
+            provider=getattr(agent, "provider", None),
+            base_url=getattr(agent, "base_url", None),
+        )
+        usage["cost_status"] = cost.status
+        if cost.amount_usd is not None:
+            usage["cost_usd"] = float(cost.amount_usd)
     except Exception:
         pass
     # Dev-only live credits-spent readout (L0 usage-aware-credits). Gated on
@@ -2309,112 +2312,6 @@ def _get_usage(agent) -> dict:
     return usage
 
 
-def _compact_usage_text(session: dict) -> str:
-    """Compact /usage page: current session per-model rows + a tight recent
-    summary + a one-line Nous credits gauge.
-
-    Costs are provider-REPORTED only. When a provider reports nothing the
-    cost is simply omitted (never rendered as $0.00). The detailed legacy
-    page stays reachable via `/usage full` (slash-worker → CLI path).
-    """
-    from agent.usage_pricing import format_token_count_compact as _fmt
-
-    agent = session.get("agent")
-    lines: list[str] = []
-
-    calls = (getattr(agent, "session_api_calls", 0) or 0) if agent is not None else 0
-    if agent is not None and calls > 0:
-        u = _get_usage(agent)
-        header = f"Session — {u['model']}"
-        provider = getattr(agent, "provider", None)
-        if provider:
-            header += f" ({provider})"
-        lines.append(header)
-
-        per_model = getattr(agent, "session_model_usage", None) or {}
-        rows = list(per_model.items()) or [(
-            u["model"],
-            {
-                "calls": u["calls"], "input": u["input"], "output": u["output"],
-                "cache_read": u["cache_read"], "cache_write": u["cache_write"],
-                "cost_usd": None,
-            },
-        )]
-        name_w = max(len(name or "?") for name, _ in rows)
-        for name, row in rows:
-            cells = [
-                f"{(name or '?'):<{name_w}}",
-                f"reqs {row.get('calls', 0)}",
-                f"in {_fmt(int(row.get('input', 0) or 0))}",
-                f"out {_fmt(int(row.get('output', 0) or 0))}",
-            ]
-            cache_read = int(row.get("cache_read", 0) or 0)
-            if cache_read:
-                cells.append(f"cache {_fmt(cache_read)}")
-            cost = row.get("cost_usd")
-            if cost is not None:
-                cells.append(f"${cost:.4f}")
-            lines.append("  " + " · ".join(cells))
-
-        ctx_pct = u.get("context_percent")
-        tail = [f"total {_fmt(int(u['total'] or 0))} tokens", f"{u['calls']} calls"]
-        if ctx_pct is not None:
-            tail.append(f"context {ctx_pct}%")
-        if u.get("compressions"):
-            tail.append(f"compressions {u['compressions']}")
-        lines.append("  " + " · ".join(tail))
-
-        # The /usage page reports the FULL provider-reported cost (OpenRouter
-        # usage.cost accumulator AND/OR the Nous header delta) — NOT the chrome's
-        # Nous-header-only figure (F3 narrowed `_get_usage["cost_usd"]` to the
-        # status bar). Read it straight from real_session_cost_usd here so the
-        # accounting page keeps both sources.
-        try:
-            from agent.usage_pricing import real_session_cost_usd
-
-            cost_usd = real_session_cost_usd(agent)
-        except Exception:
-            cost_usd = None
-        if cost_usd is not None:
-            lines.append(f"  session cost: ${cost_usd:.4f} (provider-reported)")
-        else:
-            lines.append("  session cost: not reported by provider")
-    else:
-        lines.append("Session — no API calls yet")
-
-    # Tight recent summary from the session DB (real costs only).
-    try:
-        db = _get_db()
-        totals = db.usage_totals(days=30) if db is not None else None
-    except Exception:
-        totals = None
-    if totals and totals.get("sessions"):
-        from agent.usage_pricing import format_token_count_compact as _fmt30
-
-        parts = [
-            f"{totals['sessions']} sessions",
-            f"in {_fmt30(totals['input_tokens'])}",
-            f"out {_fmt30(totals['output_tokens'])}",
-        ]
-        reported = totals.get("reported_cost_usd")
-        if reported is not None:
-            parts.append(f"reported cost ${float(reported):.2f}")
-        lines.append("Last 30d: " + " · ".join(parts))
-
-    # Nous credits one-liner (account-level; independent of the live agent).
-    try:
-        from agent.account_usage import nous_credits_compact_line
-
-        credits_line = nous_credits_compact_line()
-    except Exception:
-        credits_line = None
-    if credits_line:
-        lines.append(credits_line)
-
-    lines.append("(/usage full — detailed page)")
-    return "\n".join(lines)
-
-
 def _probe_credentials(agent) -> str:
     """Light credential check at session creation — returns warning or ''."""
     try:
@@ -9292,17 +9189,6 @@ def _(rid, params: dict) -> dict:
         except Exception as e:
             return _ok(rid, {"output": f"Plugin command error: {e}"})
 
-    # /usage — answered in-process from the LIVE agent's session counters.
-    # The slash worker is a separate subprocess that resumes the session
-    # WITHOUT an agent, so it can never see current-session tokens/costs
-    # (it only printed the Nous credits block). `/usage full` still falls
-    # through to the worker for the detailed CLI page.
-    if _cmd_base == "usage" and _cmd_arg.strip().lower() not in {"full", "--full"}:
-        try:
-            return _ok(rid, {"output": _compact_usage_text(session)})
-        except Exception:
-            pass  # fall through to the slash worker
-
     worker = session.get("slash_worker")
     if not worker:
         try:

website/docs/reference/environment-variables.md

@@ -687,7 +687,7 @@ For task-specific direct endpoints, Hermes uses the task's configured API key or
 
 ## Fallback Providers (config.yaml only)
 
-The primary model fallback chain is configured exclusively through `config.yaml` — there are no environment variables for it. Add a top-level `fallback_providers` list with `provider` and `model` keys to enable automatic failover when your main model encounters errors.
+The primary model fallback chain is configured exclusively through `config.yaml` — there are no environment variables for it. Add a top-level `fallback_providers` list with `provider` and `model` keys to enable automatic failover when your main model encounters errors. Auxiliary tasks whose provider is `auto` also consult this chain before Hermes' built-in auxiliary discovery chain.
 
 ```yaml
 fallback_providers:
@@ -695,7 +695,7 @@ fallback_providers:
     model: anthropic/claude-sonnet-4
 ```
 
-The older top-level `fallback_model` single-provider shape is still read for backward compatibility, but new configuration should use `fallback_providers`.
+The older top-level `fallback_model` single-provider shape is still read for backward compatibility, but new configuration should use `fallback_providers`. For task-specific auxiliary policy, use `auxiliary.<task>.fallback_chain` in `config.yaml`; there is no environment variable equivalent.
 
 See [Fallback Providers](/user-guide/features/fallback-providers) for full details.
 

website/docs/user-guide/configuring-models.md

@@ -53,7 +53,7 @@ Click **Show auxiliary** to reveal the 11 task slots:
 
 ![Auxiliary panel expanded](/img/docs/dashboard-models/auxiliary-expanded.png)
 
-Every auxiliary task defaults to `auto` — meaning Hermes uses your main model for that job too. Override a specific task when you want a cheaper or faster model for a side-job.
+Every auxiliary task defaults to `auto` — meaning Hermes tries your main model for that job too. If that route is unavailable or hits a capacity-style failure, `auto` follows any task-specific `auxiliary.<task>.fallback_chain`, then the main `fallback_providers` / `fallback_model` chain, then Hermes' built-in auxiliary discovery chain. Override a specific task when you want a cheaper or faster model for a side-job.
 
 ### Common override patterns
 
@@ -129,7 +129,21 @@ auxiliary:
     # ... other fields unchanged
 ```
 
-`provider: auto` with `model: ''` tells Hermes to use the main model for that task.
+`provider: auto` with `model: ''` tells Hermes to use the main model for that task, while still honoring fallback policy if the main route cannot serve the auxiliary call.
+
+Optional task-specific fallback chains live under the same auxiliary task:
+
+```yaml
+auxiliary:
+  title_generation:
+    provider: auto
+    model: ''
+    fallback_chain:
+      - provider: openrouter
+        model: inclusionai/ring-2.6-1t:free
+```
+
+When `fallback_chain` is absent, `auto` uses the top-level `fallback_providers` chain before the built-in auxiliary discovery chain.
 
 ## When does it take effect?
 

website/docs/user-guide/features/fallback-providers.md

@@ -168,7 +168,7 @@ fallback_providers:
 | Messaging gateway (Telegram, Discord, etc.) | ✔ |
 | Subagent delegation | ✔ (subagents inherit the parent fallback chain) |
 | Cron jobs | ✔ (cron agents inherit configured fallback providers) |
-| Auxiliary tasks (vision, compression) | ✘ (use their own provider chain — see below) |
+| Auxiliary tasks on `provider: auto` | ✔ (try per-task fallback, then the main fallback chain before built-in aux discovery) |
 
 :::tip
 There are no environment variables for the primary fallback chain — configure it exclusively through `config.yaml` or `hermes fallback`. This is intentional: fallback configuration is a deliberate choice, not something a stale shell export should override.
@@ -195,23 +195,30 @@ Hermes uses separate lightweight models for side tasks. Each task has its own pr
 
 ### Auto-Detection Chain
 
-When a task's provider is set to `"auto"` (the default), Hermes tries providers in order until one works:
+When a task's provider is set to `"auto"` (the default), Hermes first tries the main provider + main model for that auxiliary task. If that route is unavailable or later fails with a capacity-style error, Hermes now honors user-configured fallback policy before using the built-in discovery chain:
 
-**For text tasks (compression, web extract, etc.):**
+```text
+Main provider + main model → auxiliary.<task>.fallback_chain →
+fallback_providers / fallback_model → built-in auxiliary discovery chain
+```
+
+The task-specific chain is most precise and wins when present. The top-level `fallback_providers` chain is the same policy the main agent uses, so free-only or same-provider fallback rules apply to auxiliary tasks on `auto` as well.
+
+**Built-in text discovery chain (compression, web extract, title generation, etc.):**
 
 ```text
 OpenRouter → Nous Portal → Custom endpoint → Codex OAuth →
 API-key providers (z.ai, Kimi, MiniMax, Xiaomi MiMo, Hugging Face, Anthropic) → give up
 ```
 
-**For vision tasks:**
+**Built-in vision discovery chain:**
 
 ```text
 Main provider (if vision-capable) → OpenRouter → Nous Portal →
 Codex OAuth → Anthropic → Custom endpoint → give up
 ```
 
-If the resolved provider fails at call time, Hermes also has an internal retry: if the provider is not OpenRouter and no explicit `base_url` is set, it tries OpenRouter as a last-resort fallback.
+Those built-in chains are a convenience fallback for users who have not declared a task-specific or main fallback policy.
 
 ### Configuring Auxiliary Providers
 
@@ -232,6 +239,9 @@ auxiliary:
   compression:
     provider: "auto"
     model: ""
+    fallback_chain:              # optional, task-specific fallback policy
+      - provider: openrouter
+        model: inclusionai/ring-2.6-1t:free
 
   skills_hub:
     provider: "auto"
@@ -242,7 +252,9 @@ auxiliary:
     model: ""
 ```
 
-Every task above follows the same **provider / model / base_url** pattern. Context compression is configured under `auxiliary.compression`:
+Every task above follows the same **provider / model / base_url** pattern. Each task can also declare its own `fallback_chain`; if omitted, `provider: auto` uses the top-level `fallback_providers` chain before Hermes' built-in auxiliary discovery chain.
+
+Context compression is configured under `auxiliary.compression`:
 
 ```yaml
 auxiliary: