react_loop.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/react_loop.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/unit/test_react_loop_events.py plus plan 112 API-context cap regression tests"
"""
The ReAct inner loop — one user turn through up to N iterations.
 
Purpose
-------
:func:`run_turn` executes a single user turn: streams from the API,
collects tool calls, dispatches them in parallel via
:func:`_dispatch_tool`, accumulates per-iteration usage onto the
session, and either commits the final assistant answer or hits the
ReAct iteration cap. This is the file the rest of the runtime hands
control to once a user message is added to the session.
 
Both :func:`run_turn` and :func:`_dispatch_tool` form one tightly
coupled unit (the loop calls dispatch on every iteration that
yields tool calls), which is why they share a module rather than
splitting into ``react_loop.py`` + ``tool_dispatch.py``. The
*registry* — a separate concern — lives in
:mod:`oc_chat.core.tool_registry`.
 
Responsibilities
----------------
- :func:`run_turn` — the per-turn ReAct loop; calls the API,
  dispatches tool calls, accumulates per-iteration usage, applies
  step routing, and surfaces results. Owns the "one user turn"
  contract.
- :func:`_dispatch_tool` — single tool dispatch with
  :func:`dbg_timed` instrumentation and structured error handling.
- :func:`_run_bracketed` — the ONE definition of what a running tool
  looks like: the presenter's dispatch spinner plus the
  :data:`~oc_chat.core.tool_activity.TOOL_ACTIVITY` record, opened
  around every dispatch and closed in ``finally``. Used by the
  allowed path and by all six re-dispatch sites in the three
  confirmation resolvers (plan 279).
- :class:`_TurnAccumulator` — private dataclass that replaces the
  8 turn-level locals previously declared at the top of the loop;
  carries the per-iteration sums into the closing turn-summary block
  and the per-turn cache snapshot list.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — turn START / END, ReAct iteration markers, final-answer
         received, ReAct cap hit (override=True at L1 on cap).
    L3 — per-iteration routing decisions, tool dispatch summary,
         per-tool OK/ERROR results.
    L4 — deep tracing (currently unused).
 
Tool execution timing flows under ``OC_CHAT.TOOLS`` via
``dbg_timed("OC_CHAT.TOOLS", "_dispatch_tool", label=tool_name)``.
 
Contracts
---------
- The ReAct loop respects ``session.max_react_iterations`` (clamped
  to 1–99) — falling back to ``DEFAULT_MAX_REACT_ITERATIONS``.
- Multiple tool calls in one assistant message run in parallel via
  :class:`ThreadPoolExecutor`; results are reordered to match the
  original tool_call order before being appended to history.
- The session's ``last_turn_max_prompt_tokens`` is updated each turn
  so :func:`maybe_compress` has an accurate trigger.
- All file-modifying tool calls (``write_file``) are recorded in
  ``ctx.goal_files_changed`` when a goal is in flight, for later
  changelog updates.
- Per-turn auto-save happens at the end of every turn via
  :func:`save_chat_log` + :func:`save_session_metadata` so a crash
  mid-session at most loses the in-flight turn.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.cache_usage (DeepSeek prompt-cache hit ratio reporting)
- oc_chat.core.api_context (build the message list sent to the API)
- oc_chat.core.session (ensure_system_includes_tool_reference helper)
- oc_chat.core.step_routing (per-iteration model/effort/max_tokens routing)
- oc_chat.core.persistence (per-turn auto-save)
- oc_chat.core.event_bus (default_bus for synchronous M2 event publication)
- oc_chat.core.events (make_* TurnEvent factories and new_event_id)
- oc_chat.core.tool_activity (the terminal-only in-flight tool record)
- oc_chat.core.cancellation (two-stage Ctrl+C: cancellation token, stop confirmation, tool-cancel reset)
- oc_chat.core.config (get_max_context_tokens for the context-size gauge)
- oc_chat.core.context (_estimate_tokens fallback for the context usage display)
- oc_chat.core.iteration_stat (IterationStat per-iteration statistics record)
- oc_chat.core.permission_validator (permission_decision fields for auto-allow, cap and stall denials, and tool capability)
- oc_chat.core.provider_plugin (PerformanceStats on the turn accumulator)
- oc_chat.core.sub_loop (run_sub_loop to offload a sub-task)
- oc_chat.core.turn_cost (TurnCostAccumulator and price_usage for the per-sample USD estimate)
- oc_ui.terminal.enhanced_presenter (EnhancedPresenter type of the optional run_turn argument)
- oc_chat.core.external_content_gate (lazy; external-fetch confirmation and binding source trust to the active session)
- oc_chat.core.resource_caps (lazy; cost thresholds, per-turn tool-call cap and worker clamp)
- oc_chat.core.session_checkpoints (lazy; mini report and the cost, tool-call and stall checkpoints)
- oc_chat.core.session_mode (lazy; remembered answers, Auto mode and budgets for Tier-2 confirmation)
- oc_chat.core.settings (lazy; runtime settings and mode for the verification config)
- oc_chat.core.stall_detector (lazy; StallDetector and stall_sentence for repeated-call denial)
- oc_chat.core.tier2_session_grant (lazy; the session work-directory grant for Tier-2 python runs)
- oc_chat.core.tokenizer (lazy; token-count estimate of a cancelled stream with no usage chunk)
- oc_chat.core.verification (lazy; verification baseline, runs and changed-path tracking)
- oc_chat.tools._argv_policy (lazy; classify_command to re-derive the Tier-2 argv)
- oc_chat.tools._sandbox (lazy; redact_secrets on error text and sub-loop results)
- oc_chat.tools.filesystem (lazy; sandbox root and the protected-write confirmation)
- oc_chat.tools.shell (lazy; human-confirmed command and budget for Tier-2 re-dispatch)
- oc_pricing (lazy; rate_at quote for the provider budget event)
- oc_ui.presenter_base (lazy; Presenter type hint)
- oc_ui.terminal.input (lazy; read_single_key prompts and the TurnStats type hint)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §5 Streaming chunk tuples — the
``("reasoning"|"content"|"tool_call"|"usage"|"finish_reason", data)``
shapes the loop consumes; §6 Tool-call dict shape — the dict shape
passed to :func:`_dispatch_tool`.
"""
from __future__ import annotations
 
import hashlib
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Callable, TypeVar
 
from oc_diagnostics import _dbg, dbg_timed
from oc_ui.terminal.enhanced_presenter import EnhancedPresenter
from oc_chat.core.cancellation import (
    CancellationError,
    CancellationToken,
    confirm_or_continue,
    reset_tool_cancel,
)
from oc_chat.core.provider_plugin import PerformanceStats
from oc_chat.core.turn_cost import TurnCostAccumulator, price_usage
 
from oc_chat.core.event_bus import default_bus
from oc_chat.core.events import (
    make_assistant_message, make_error, make_permission_decision,
    make_prompt_rendered, make_provider_budget_consumed,
    make_tool_call_completed, make_tool_call_proposed, make_tool_call_started,
    make_user_input_received, new_event_id,
)
from oc_chat.core.api_context import messages_for_api, resolve_tool_result_cap
from oc_chat.core.config import get_max_context_tokens
from oc_chat.core.context import _estimate_tokens
from oc_chat.core.cache_usage import cache_hit_ratio_pct
from oc_chat.core.permission_validator import (
    auto_allow_decision_fields, cap_deny_decision_fields, link_started_to_decision,
    stall_deny_decision_fields, tool_capability,
)
from oc_chat.core.persistence import save_chat_log, save_session_metadata
from oc_chat.core.session import ensure_system_includes_tool_reference
from oc_chat.core.step_routing import resolve_iteration_routing, should_use_sub_loop
from oc_chat.core.sub_loop import run_sub_loop
from oc_chat.core.tool_activity import TOOL_ACTIVITY
import time
from oc_chat.core.iteration_stat import IterationStat
from datetime import datetime, timezone
 
 
#: Hard backstop on sub-loop offloads per turn (plan 132). This is a runaway
#: guard, NOT a behaviour limiter: the real fix is removing the would_switch_model
#: coupling that caused the iter-0→119 infinite offload. Set well above legitimate
#: use (multiple sub-loops per turn for distinct sub-tasks is designed, tested
#: M2.4 behaviour — see test_sub_loop_event_chain) and well below runaway. Even
#: if a guard regresses, a turn can never spend itself offloading. Tunable from
#: dogfooding data; revisit with the targeted-offload redesign (tracked in the
#: living ROADMAP-EXTENDED.md backlog: read-size/intent-driven, opt-in).
_MAX_SUB_LOOPS_PER_TURN: int = 5
 
#: Placeholder ``content`` for a reasoning-only completion — an iteration that
#: streamed chain-of-thought but was cut off before emitting a single content
#: token (typically the output-token cap being spent entirely on reasoning).
#: The API needs a non-empty string here; the model reads it as the record of
#: what happened on the previous hop.
REASONING_ONLY_CONTENT: str = (
    "[This reply was cut off before any answer text was produced — the "
    "reasoning above is all that was generated. Continue from it.]"
)
 
#: Operator-facing notice for the same condition. Distinct from the
#: nothing-at-all case, which stays "[No response content received]".
#: ``{cause}`` comes from the provider's finish_reason — a confirmed cap hit
#: reads differently from a stop the provider declined to explain.
REASONING_ONLY_NOTICE: str = (
    "[No answer text — the model produced only reasoning ({n} chars) before the "
    "response ended{cause}. The reasoning is kept in this session; ask it to "
    "continue, or raise the output cap with /budget deep.]"
)
 
#: ``{cause}`` fillers. Absence of a finish_reason is "unknown", never "stop" —
#: saying "hit the cap" without the provider's word for it would be a guess.
CAUSE_LENGTH: str = ", having hit its {cap}-token output cap"
CAUSE_UNKNOWN: str = " (the provider reported no finish reason)"
 
#: The plain truncation the operator sees most: a real answer cut mid-sentence.
TRUNCATED_ANSWER_NOTICE: str = (
    "[Answer truncated — the model hit its {cap}-token output cap mid-response. "
    "Ask it to continue, or raise the cap with /budget deep.]"
)
 
#: Truncation that lands mid tool-call leaves ``arguments`` an unparseable JSON
#: fragment. _dispatch_tool already answers the model with a re-issue error;
#: this tells the human why a tool call failed for no apparent reason.
TRUNCATED_TOOL_CALL_NOTICE: str = (
    "[Tool call(s) truncated — the model hit its {cap}-token output cap while "
    "emitting {n} tool call(s), so their arguments may be incomplete.]"
)
 
 
def _extract_last_user_task(messages: list[dict]) -> str:
    """Extract the most recent user message content for sub-loop task analysis."""
    for m in reversed(messages):
        if m.get("role") == "user" and m.get("content"):
            return m["content"]
    return ""
 
 
def _detect_response_repetition(messages: list[dict]) -> bool:
    """Detect if recent assistant responses are too similar (repetition bug)."""
    recent_assistant_messages = [
        m.get("content", "") for m in reversed(messages) 
        if m.get("role") == "assistant" and m.get("content")
    ]
    
    if len(recent_assistant_messages) < 3:
        return False
    
    # Check similarity between last 3 responses
    last_three = recent_assistant_messages[:3]
    similarity_threshold = 0.8  # 80% similarity threshold
    
    for i in range(len(last_three) - 1):
        for j in range(i + 1, len(last_three)):
            # Simple similarity check based on common words
            words1 = set(last_three[i].lower().split())
            words2 = set(last_three[j].lower().split())
            
            if words1 and words2:
                intersection = words1.intersection(words2)
                union = words1.union(words2)
                similarity = len(intersection) / len(union)
                
                if similarity > similarity_threshold:
                    return True
    
    return False
    from oc_ui.presenter_base import Presenter
    from oc_ui.terminal.input import TurnStats
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
@dataclass
class _TurnAccumulator:
    """Per-turn running totals carried across ReAct iterations.
 
    Replaces the 8 turn-level ``int`` locals previously declared at
    the top of :func:`run_turn`. No behaviour change — purely a
    typed shape that makes the contract obvious to anyone reading
    and lays groundwork for a public ``TurnResult`` in G.6.D.
    """
 
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    tool_calls: int = 0
    iterations: int = 0
    prompt_cached: int = 0
    cache_hit_tokens: int = 0
    cache_miss_tokens: int = 0
    max_prompt_in_turn: int = 0
    perf_stats: PerformanceStats = field(default_factory=PerformanceStats)
    # Plan 188 T4: the honesty-gated USD estimate. Folded per SAMPLE (each
    # iteration at its own route.model, each sub-loop at its own) rather than
    # from the turn's summed tokens, so a routed or offloaded turn is priced
    # exactly. One unpriceable sample suppresses the whole turn's line.
    cost: TurnCostAccumulator = field(default_factory=TurnCostAccumulator)
    # Plan 137 (verification backpressure) turn-scoped state — reset each turn via a
    # fresh _TurnAccumulator (never session-scoped). ``changed_files`` is CUMULATIVE
    # across the turn's iterations (so later writes + the model's fix are reverified);
    # ``pending_findings`` carries the last non-clean result for the cap-hit notice.
    changed_files: set = field(default_factory=set)
    verification_runs: int = 0
    pending_findings: "VerificationResult | None" = None
    # Plan 173 item 3 (F22): one-shot latch so the per-turn tool-call cap-approach
    # warning is injected to the model at most once per turn (not every iteration).
    cap_warning_emitted: bool = False
    # Plan 237 (CAPS-01) turn-scoped checkpoint state.
    # ``cap_extensions`` counts the cap-widths the operator has bought this turn,
    # so the next checkpoint lands at 200, 300, … rather than re-asking at 100.
    # The other two feed the mini report; they are populated unconditionally
    # (unlike ``changed_files``, which is gated on the plan-137 flag) because the
    # checkpoint must be able to say what the turn DID whether or not
    # verification is on. Counting a name and a path per call is free.
    cap_extensions: int = 0
    tool_name_counts: dict = field(default_factory=dict)
    files_written: list = field(default_factory=list)
 
 
def _now_iso() -> str:
    """Millisecond-precision local ISO-8601 timestamp for event envelopes."""
    return datetime.now().astimezone().isoformat(timespec="milliseconds")
 
 
def _now_utc() -> datetime:
    """The pricing clock: a tz-AWARE UTC instant (plan 202 decision 3).
 
    oc_pricing treats a naive ``at`` as ``None``, and every model in its
    table is effective-dated/TOU-banded — a naive instant would therefore
    turn EVERY sample unpriceable: a silent pricing blackout dressed as an
    honest "unknown". This seam is the single source of pricing time, and
    each sample captures it ONCE at the send seam so the ``$``-line fold
    and the budget event can never price the same sample under two clock
    reads that straddle a tariff/band boundary. Monkeypatchable in tests.
    """
    return datetime.now(timezone.utc)
 
 
def _safe_tool_args(tc: dict) -> dict:
    """Best-effort parse of a tool call's JSON ``arguments`` into a dict."""
    try:
        parsed = json.loads(tc.get("arguments") or "{}")
    except (json.JSONDecodeError, TypeError):
        return {"_raw": tc.get("arguments", "")}
    return parsed if isinstance(parsed, dict) else {"_raw": tc.get("arguments", "")}
 
 
def _publish_pre_dispatch_decisions(
    tool_calls: list[dict],
    allowed: int,
    ev: Callable[[], dict],
    stall_denied: set[int] | None = None,
) -> list[str]:
    """M8.1 WU-0a: decide every proposed call **before** any of them starts.
 
    The classify half of the classify → decide → emit-started-for-allowed lifecycle
    (§7.2 / F7 / F31). Exactly two verdicts are computable pre-dispatch, and both are
    emitted here as one ``permission_decision`` per call, carrying its ``tool_call_id``:
 
    - calls at or beyond the per-turn budget (``idx >= allowed``, the same predicate
      :func:`_dispatch_or_cap` denies on) — a **cap-deny**, carrying the fail-closed
      ``risk_class="critical"`` sentinel and ``denial_reason_code``;
    - every other call — an **auto-allow**, whose rung is derived from the tool's
      capability alone.
 
    Tier-2 / protected-write / external-content verdicts are computed inside the tool
    bodies and stay post-dispatch until WU-0b (§A.9(b) narrows "exactly one
    ``permission_decision`` per call" to "per call whose verdict is computable
    pre-dispatch"), so this function never consults a gate.
 
    Parameters
    ----------
    tool_calls : list[dict]
        This iteration's proposed calls, in call order.
    allowed : int
        The per-turn budget's allowed prefix length (``_allowed_calls``).
    ev : Callable[[], dict]
        The turn's envelope-kwargs factory (``event_id``/``parent``/``session``/``ts``).
 
    Returns
    -------
    list[str]
        The ``permission_decision_id`` of each call's decision, positionally aligned
        with *tool_calls*, so the started event can carry its deciding id.
    """
    func = "_publish_pre_dispatch_decisions"
    decision_ids: list[str] = []
    stalled = stall_denied or set()
    for idx, tc in enumerate(tool_calls):
        if idx in stalled:
            build = stall_deny_decision_fields
        elif idx >= allowed:
            build = cap_deny_decision_fields
        else:
            build = auto_allow_decision_fields
        fields = build(tool_call_id=tc["id"], tool_name=tc["name"], args=_safe_tool_args(tc))
        pd_id = new_event_id()
        decision_ids.append(pd_id)
        default_bus().publish(make_permission_decision(
            **ev(), permission_decision_id=pd_id, **fields))
    _dbg("OC_CHAT.TOOLS", func,
         f"decided {len(decision_ids)} call(s) pre-dispatch "
         f"({allowed} in cap, {len(stalled)} stall-denied, "
         f"{max(0, len(tool_calls) - allowed)} cap-denied)", 2)
    return decision_ids
 
 
#: Return type of whatever :func:`_run_bracketed` is handed; the bracket is
#: transparent, so what goes in comes out.
_T = TypeVar("_T")
 
 
def _run_bracketed(presenter, label: str, fn: "Callable[[], _T]") -> _T:
    """Run *fn* with the terminal's busy chrome open, and close it whatever happens.
 
    ONE place decides what a running tool looks like (plan 279 P3). The chrome
    is two things: the presenter's dispatch spinner, which hosts the status bar
    and its ticking timers, and :data:`~oc_chat.core.tool_activity.TOOL_ACTIVITY`,
    the record a child's output lines land on. Before this existed only the
    allowed-path dispatch was bracketed, so a Tier-2 command — which returns
    ``requires_confirmation`` in milliseconds and is re-dispatched by a
    confirmation resolver AFTER that bracket has closed — ran with no chrome at
    all. A twelve-minute suite showed nothing.
 
    Transparent by construction: the value *fn* returns is returned unchanged,
    an exception propagates unchanged, and nothing about the call's result, its
    events or its logging passes through here. Both halves close in ``finally``,
    so a raising tool strands neither a Live display nor a stale tail.
    """
    handle = presenter.spinner_start(label)
    TOOL_ACTIVITY.begin(label)
    try:
        return fn()
    finally:
        TOOL_ACTIVITY.end()
        presenter.spinner_stop(handle)
 
 
def _dispatch_tool(tc: dict, tool_registry: dict) -> tuple[dict, dict]:
    """Execute a single tool call. Returns ``(tc, result)`` preserving original *tc* for ordering."""
    func = "_dispatch_tool"
    _dbg("OC_CHAT.TOOLS", func, f"executing {tc['name']}", 3)
    with dbg_timed("OC_CHAT.TOOLS", func, label=tc["name"], level=3):
        tool_fn = tool_registry.get(tc["name"])
        if not tool_fn:
            _dbg("OC_CHAT.TOOLS", func, f"unknown tool: {tc['name']}", 1, override=True)
            return tc, {"error": f"Unknown tool: {tc['name']}"}
        # Parse arguments separately from execution: a malformed/truncated
        # arguments JSON (observed with write_file + DeepSeek — the streamed
        # tool call ends mid-string) must NOT surface as a cryptic
        # JSONDecodeError the model can't act on. Instrument the raw payload
        # for root-cause analysis (length cut vs early stream end vs model
        # glitch) and hand the model an actionable retry message.
        raw_args = tc.get("arguments") or "{}"
        try:
            args = json.loads(raw_args)
        except json.JSONDecodeError as e:
            _dbg("OC_CHAT.TOOLS", func,
                 f"MALFORMED tool args for {tc['name']}: {type(e).__name__}: {e}; "
                 f"len={len(raw_args)} raw={raw_args[:300]!r}", 1, override=True)
            return tc, {"error": (
                f"Your `{tc['name']}` tool-call arguments were not valid JSON "
                f"({e}). They appear truncated or malformed. Re-issue the call "
                f"with complete, valid JSON arguments; for large file content, "
                f"write it in smaller pieces."
            )}
        try:
            result = tool_fn(**args)
        except Exception as e:
            _dbg("OC_CHAT.TOOLS", func,
                 f"tool {tc['name']} raised {type(e).__name__}: {e}", 1, override=True)
            result = {"error": str(e)}
    return tc, result
 
 
def _stdin_is_interactive() -> bool:
    """True only when a real interactive TTY is attached (a human is present).
 
    Fail-closed default: any non-TTY (sub-loop, batch, piped/redirected stdin,
    pytest capture) or any probe error returns ``False``. Isolated as a tiny
    seam so the M8.0-C confirmation relay is testable without faking ``sys.stdin``.
    """
    import sys as _sys
    try:
        return bool(_sys.stdin and _sys.stdin.isatty())
    except (ValueError, OSError, AttributeError):
        return False
 
 
def _budget_text(seconds: "int | None") -> str:
    """How a budget reads in an operator-facing line (plan 283).
 
    ``None`` is a real answer here — "run until it finishes" — and must never
    render as the word ``None`` in a line whose whole job is to say how long the
    terminal is about to be busy.
    """
    return "no limit" if seconds is None else f"{seconds} s"
 
 
def _run_confirmed_tier2(
    tc: dict, tool_registry: dict, presenter, command: str,
    program: str, budget: "int | None",
) -> dict:
    """Re-dispatch an approved Tier-2 call under its approval and its budget.
 
    The ONE place a plan-283 answer becomes a running command, so the three
    approval routes — a key at the prompt, this session's memory, and Auto mode
    — cannot drift apart in what they authorise or how long they allow. The
    authority is still and only :func:`shell.human_confirmed_command` with the
    exact string; the budget rides beside it and decides nothing.
 
    An expiry hook is installed only when there is a budget to expire. "Run
    until it finishes" has no expiry, so there is nothing to ask about, and
    passing a hook there would arm a prompt that could never fire.
    """
    from oc_chat.core import session_mode
    from oc_chat.tools import shell
 
    on_expired = None
    if budget is not None:
        def on_expired() -> "int | None":                      # noqa: F811
            return session_mode.ask_budget_extension(presenter, program, budget)
 
    with shell.human_confirmed_command(command):
        with shell.human_budget(budget, on_expired=on_expired):
            # Plan 279: the bracket the twelve-minute suite was missing.
            _, confirmed_result = _run_bracketed(
                presenter, f"running {tc['name']}…",
                lambda: _dispatch_tool(tc, tool_registry))
    return confirmed_result
 
 
def _resolve_tier2_confirmation(
    tc: dict, result: dict, tool_registry: dict, presenter,
) -> dict:
    """M8.0-C: out-of-band human-confirm channel for a Tier-2 ``requires_confirmation``.
 
    A Tier-2 command (``run_command`` over python/pytest/pip-install/oct) returns
    ``{"requires_confirmation": True, ...}`` that no model-reachable flag can satisfy
    (M8.0-A). Here -- in the single-threaded post-dispatch loop, never inside the
    parallel executor -- the decision is relayed to the *human*:
 
    * interactive TTY -> show the exact command + reason, prompt 'y/N'; on 'y'
      re-dispatch the SAME call with the command authorized via
      :func:`shell.human_confirmed_command` (a ContextVar the model cannot set);
    * headless / sub-loop / batch / piped stdin (no TTY) -> fail closed.
 
    Either denial path hands the model a terminal ``error`` it can act on (stop
    asking), not the raw ``requires_confirmation`` it cannot satisfy. Any non-
    confirmation result is returned unchanged (identity).
 
    Plan 235 (SHELL-02) adds a THIRD answer, ``s``, offered only when the pending
    command is ``python <script inside the work directory>``: it grants that
    DIRECTORY for the session, so the agent's next probe script there runs with no
    keystroke (:mod:`oc_chat.core.tier2_session_grant`). The grant only decides
    whether the human is ASKED — a grant-satisfied run still re-dispatches under
    :func:`shell.human_confirmed_command`, so the ContextVar exact-string contract
    remains the single authorisation seam. It is consulted only AFTER the TTY
    check above, so the headless path is unchanged, and never off ``permissive``.
 
    Plan 283 (SHELL-04) widens the ANSWER, never the authority. ``y`` now runs
    under the program's default budget (:mod:`oc_chat.core.session_mode`), ``c``
    runs until it finishes, ``t`` takes a typed budget, ``a`` and ``d`` record a
    standing yes or no for that program, and ``u`` runs it and switches the
    session to Auto mode. Every one of them re-dispatches through
    :func:`shell.human_confirmed_command` exactly as ``y`` always has; what is
    new is only how long the human is willing to wait and how often they are
    willing to be asked. The order below is deliberate: the TTY check first (so
    headless is untouched byte for byte), then the plan-235 grant (unchanged),
    then this session's memory, then Auto mode, and only then the prompt.
    """
    func = "_resolve_tier2_confirmation"
    if not (isinstance(result, dict) and result.get("requires_confirmation")):
        return result
 
    from oc_chat.tools import shell
 
    command = result.get("command", "")
    reason = result.get("reason", "")
 
    # Fail closed unless a real interactive human is present to approve.
    if not _stdin_is_interactive():
        _dbg("OC_CHAT.TOOLS", func, f"no TTY -> Tier-2 denied: {command!r}", 2, override=True)
        return shell.deny_unconfirmed_tier2(
            result, note="No interactive terminal is attached, so a human cannot approve it here.")
 
    # Plan 235: re-derive the argv from the same classifier `run_command` used, so
    # what the grant inspects is what would execute — never the raw string.
    from oc_chat.core.tier2_session_grant import (
        active_session_grant, grant_work_dir_for_session, grantable_work_dir,
        session_grant_satisfies,
    )
    from oc_chat.tools._argv_policy import classify_command
    from oc_chat.tools.filesystem import _sandbox_root
    sandbox_root = _sandbox_root()
    _decision = classify_command(command, sandbox_root)
    argv = _decision.argv if _decision.decision == "confirm" else []
 
    if session_grant_satisfies(argv, sandbox_root):
        granted = active_session_grant()
        # Say so on EVERY such run: the transcript must show the gate was down.
        presenter.show_info(
            f"[grant] running under the session grant on {granted} "
            f"(no confirmation asked — /grants revoke to drop it):\n    {command}"
        )
        _dbg("OC_CHAT.TOOLS", func,
             f"session grant satisfied Tier-2 without a prompt: {command!r} "
             f"(granted dir {granted})", 2, override=True)
        with shell.human_confirmed_command(command):
            # Plan 279: bracketed, so a granted run animates exactly like an
            # allowed one. The bracket is transparent — `granted_result` is the
            # tool's own object, unchanged.
            _, granted_result = _run_bracketed(
                presenter, f"running {tc['name']}…",
                lambda: _dispatch_tool(tc, tool_registry))
        return granted_result
 
    # Plan 283: what this session already decided about this program, and what
    # Auto mode would decide, are both consulted before the human is disturbed.
    # Neither is authority — each only answers the question the prompt asks.
    from oc_chat.core import session_mode
 
    program = session_mode.program_key(argv)
    default_budget = session_mode.default_budget_for(program)
 
    remembered = session_mode.remembered_answer(program)
    if remembered is not None:
        if not remembered.run:
            presenter.show_info(
                f"[remembered] denied {program} under the session answer "
                f"(/mode forget to drop it):\n    {command}")
            _dbg("OC_CHAT.TOOLS", func,
                 f"session answer denied Tier-2 without a prompt: {command!r}", 2,
                 override=True)
            return shell.deny_unconfirmed_tier2(
                result, note="A human reviewed the request and declined to run it.")
        presenter.show_info(
            f"[remembered] running {program} under the session answer "
            f"(budget {_budget_text(remembered.budget_s)}; /mode forget to drop it):"
            f"\n    {command}")
        _dbg("OC_CHAT.TOOLS", func,
             f"session answer satisfied Tier-2 without a prompt: {command!r}", 2,
             override=True)
        return _run_confirmed_tier2(
            tc, tool_registry, presenter, command, program, remembered.budget_s)
 
    if session_mode.auto_mode_active():
        auto = session_mode.auto_answer_for(program)
        presenter.show_info(
            f"[auto] running {program} in Auto mode "
            f"(budget {_budget_text(auto.budget_s)}; /mode ask to stop):"
            f"\n    {command}")
        _dbg("OC_CHAT.TOOLS", func,
             f"Auto mode answered the Tier-2 ask: {command!r}", 2, override=True)
        return _run_confirmed_tier2(
            tc, tool_registry, presenter, command, program, auto.budget_s)
 
    presenter.show_info(
        f"[confirm] The agent requests a Tier-2 command:\n    {command}"
        + (f"\n  Reason: {reason}" if reason else "")
    )
    # The third key is offered ONLY when pressing it would satisfy this command;
    # otherwise the prompt is byte-identical to the pre-235 two-key one.
    offer_dir = grantable_work_dir(argv, sandbox_root)
    # ``u`` is offered on exactly the terms the mode itself runs on: permissive
    # only, like the grant. Offering a key that could not take effect would be
    # the worst of both — a keystroke that looks like it did something.
    offer_auto = session_mode.auto_mode_available()
 
    # ONE legend naming every key the prompt below offers (P2, v1.1).
    legend = [
        f"  Budget for {program}: {default_budget} s.",
        f"  [y] run with the {default_budget} s budget"
        f"   [c] run until it finishes   [t] type a budget",
        f"  [a] always run {program} this session"
        f"   [d] never ask about {program} again",
    ]
    if offer_auto:
        legend.append("  [u] run it and switch this session to Auto mode")
    legend.append("  [N] deny (default)")
    presenter.show_info("\n".join(legend))
 
    if offer_dir is not None:
        presenter.show_info(
            f"  [s] grant python on {offer_dir} for this session (any script the "
            f"agent writes there runs without asking until /grants revoke or the "
            f"session ends)"
        )
    keys = ["y", "N", "c", "t", "a", "d"]
    if offer_auto:
        keys.append("u")
    if offer_dir is not None:
        keys.append("s")
    prompt = f"Run this command? ({'/'.join(keys)}) "
    valid = "".join(key.lower() for key in keys)
    from oc_ui.terminal.input import read_single_key
    try:
        answer = read_single_key(prompt, valid=valid)
    except (EOFError, KeyboardInterrupt):
        # Interrupt at the prompt -> fail closed (deny) and return a well-formed
        # result, rather than propagating and orphaning the tool_call.
        return shell.deny_unconfirmed_tier2(result, note="Confirmation interrupted; denied (fail-closed).")
 
    # Every key that is not an approval falls through to the SAME denial the
    # two-key prompt returned, including a key whose precondition has gone —
    # the offer flags are re-checked here rather than trusted from the prompt.
    budget: "int | None" = default_budget
    if answer == "s" and offer_dir is not None:
        # 's' records the grant AND runs the current command, under today's
        # clamped timeout: the plan-235 path gains no budget (P4).
        grant_work_dir_for_session(offer_dir)
        _dbg("OC_CHAT.TOOLS", func, f"human approved Tier-2: {command!r}", 2, override=True)
        with shell.human_confirmed_command(command):
            _, confirmed_result = _run_bracketed(
                presenter, f"running {tc['name']}…",
                lambda: _dispatch_tool(tc, tool_registry))
        return confirmed_result
    if answer == "c":
        budget = None
    elif answer == "t":
        budget = session_mode.read_typed_budget(presenter)
        if budget is None:
            _dbg("OC_CHAT.TOOLS", func,
                 f"no usable budget typed -> Tier-2 denied: {command!r}", 2, override=True)
            return shell.deny_unconfirmed_tier2(
                result, note="A human reviewed the request and declined to run it.")
    elif answer == "a":
        session_mode.remember_answer(
            program, session_mode.RememberedAnswer(run=True, budget_s=default_budget))
    elif answer == "d":
        session_mode.remember_answer(
            program, session_mode.RememberedAnswer(run=False))
        _dbg("OC_CHAT.TOOLS", func,
             f"human declined Tier-2 and will not be asked again: {command!r}", 2,
             override=True)
        return shell.deny_unconfirmed_tier2(
            result, note="A human reviewed the request and declined to run it.")
    elif answer == "u" and offer_auto:
        # v1.1: the mode is reachable from the answer itself, exactly as
        # ``/mode auto`` would set it — the same one setter, no second path.
        session_mode.set_mode(session_mode.AUTO)
        presenter.show_info(
            "  Auto mode is ON for this session: the routine asks are answered "
            "without a keystroke. /mode ask switches back; a protected write "
            "still always reaches you.")
    elif answer != "y":
        _dbg("OC_CHAT.TOOLS", func, f"human declined Tier-2: {command!r}", 2, override=True)
        return shell.deny_unconfirmed_tier2(
            result, note="A human reviewed the request and declined to run it.")
 
    # Approved: re-dispatch the SAME call with this exact command authorized.
    _dbg("OC_CHAT.TOOLS", func, f"human approved Tier-2: {command!r}", 2, override=True)
    return _run_confirmed_tier2(
        tc, tool_registry, presenter, command, program, budget)
 
 
def _resolve_external_content_confirmation(
    tc: dict, result: dict, tool_registry: dict, presenter,
) -> dict:
    """M8.0 F67: relay an external-content gate block to the human (§7.3).
 
    ``fetch_url`` / ``web_search`` return ``{"gate": "external_content", ...}`` when
    the source is not session-trusted. Here -- single-threaded, with stdin/presenter
    -- a three-choice prompt is shown:
 
    * ``a`` allow once  -> re-dispatch under :func:`ecg.allow_external_source_once`
      (a one-shot ContextVar grant, NOT persisted as trust);
    * ``t`` trust this source for the session -> :func:`ecg.trust_source_for_session`
      then re-dispatch (later fetches from it skip the prompt);
    * ``d`` / Enter (default) deny -> terminal error.
 
    Headless / sub-loop / batch (no TTY) fail closed. Non-gate results pass through.
    """
    func = "_resolve_external_content_confirmation"
    if not (isinstance(result, dict) and result.get("gate") == "external_content"):
        return result
 
    from oc_chat.core import external_content_gate as ecg
 
    source = result.get("source", "")
    tool = result.get("tool", tc.get("name", ""))
    detail = result.get("detail", "")
 
    if not _stdin_is_interactive():
        _dbg("OC_CHAT.TOOLS", func, f"no TTY -> external fetch denied: {source!r}", 2, override=True)
        return ecg.deny_external_fetch(
            result, note="No interactive terminal is attached, so a human cannot approve it here.")
 
    # Plan 279: both approved paths are bracketed — a slow fetch is exactly the
    # case where the terminal must not look dead.
    _label = f"running {tc['name']}…"
 
    # Plan 283: Auto mode answers this ask with ``a`` — allow ONCE — and never
    # with ``t``. Trust is a standing grant on a source and outlives the mode;
    # a one-shot allow expires with the call, which is the most a mode should be
    # able to hand out on the human's behalf.
    from oc_chat.core import session_mode
    if session_mode.auto_mode_active():
        presenter.show_info(
            f"[auto] allowing ONE fetch from '{source}' in Auto mode "
            f"(not trusted for the session; /mode ask to stop)")
        _dbg("OC_CHAT.TOOLS", func,
             f"Auto mode allowed one external fetch: {source!r}", 2, override=True)
        with ecg.allow_external_source_once(source):
            _, approved = _run_bracketed(
                presenter, _label, lambda: _dispatch_tool(tc, tool_registry))
        return approved
 
    presenter.show_info(
        f"[confirm] {tool} wants to pull EXTERNAL, untrusted content from '{source}'"
        + (f":\n    {detail}" if detail else "")
        + "\n  External content can carry injected instructions."
    )
    from oc_ui.terminal.input import read_single_key
    try:
        answer = read_single_key(
            f"  [a]llow once / [t]rust '{source}' this session / [d]eny (default deny): ", valid="atd")
    except (EOFError, KeyboardInterrupt):
        return ecg.deny_external_fetch(result, note="Confirmation interrupted; denied (fail-closed).")
 
    if answer == "t":
        ecg.trust_source_for_session(source)
        _dbg("OC_CHAT.TOOLS", func, f"human trusted source: {source!r}", 2, override=True)
        # now session-trusted -> passes
        _, approved = _run_bracketed(
            presenter, _label, lambda: _dispatch_tool(tc, tool_registry))
        return approved
    if answer == "a":
        _dbg("OC_CHAT.TOOLS", func, f"human allowed once: {source!r}", 2, override=True)
        with ecg.allow_external_source_once(source):
            _, approved = _run_bracketed(
                presenter, _label, lambda: _dispatch_tool(tc, tool_registry))
        return approved
    _dbg("OC_CHAT.TOOLS", func, f"human denied external fetch: {source!r}", 2, override=True)
    return ecg.deny_external_fetch(result, note="A human reviewed the request and declined.")
 
 
def _resolve_protected_write_confirmation(
    tc: dict, result: dict, tool_registry: dict, presenter,
) -> dict:
    """M8.0 F70/F70a: relay a human-confirm protected-write block to the human.
 
    ``write_file`` returns ``{"gate": "protected_write", ...}`` for a config-as-code
    or importable-tree target (a write-then-import in-process code-execution risk).
    Here -- single-threaded, with stdin/presenter:
 
    * interactive TTY -> show the target + reason, prompt 'y/N'; on 'y' re-dispatch
      the SAME call under :func:`filesystem.human_confirmed_write` (a ContextVar the
      model cannot set);
    * headless (no TTY) -> **F70a**: denied unless a signed ``PlanAutonomyEnvelope``
      covers the write (M8.3; none in M8.0), never a silent auto-allow.
 
    Non-protected-write results pass through unchanged.
 
    **This resolver has no Auto-mode path, and must never grow one (plan 283
    P3).** Its two siblings consult :func:`oc_chat.core.session_mode` and answer
    themselves when the operator has switched the session to ``auto``; this one
    does not, because the paths it guards are the ruled set — the security layer,
    the thresholds, any octrc, ``oct/oct/core/``, MCP and the scanner — and those
    are the decisions the project requires a human to make one at a time. A write
    there is a protected write today or it is a gap in
    :func:`oc_chat.tools._sandbox.is_protected_write_path`, which is where the
    covered paths are defined; either way the answer is never "a mode said yes".
    ``test_283_session_mode.py::test_w4d_a_protected_write_still_reads_a_key_in_auto``
    is the negative control that fails the moment this changes.
    """
    func = "_resolve_protected_write_confirmation"
    if not (isinstance(result, dict) and result.get("gate") == "protected_write"):
        return result
 
    from oc_chat.tools import filesystem
 
    path = result.get("path", "")
    reason = result.get("reason", "")
 
    # Plan 279: every approved re-dispatch in this resolver is bracketed. On the
    # headless path the presenter's spinner hooks are no-ops, so it costs nothing
    # there and keeps all six sites one shape instead of five.
    _label = f"running {tc['name']}…"
 
    if not _stdin_is_interactive():
        if filesystem._plan_envelope_covers_write(path):
            _dbg("OC_CHAT.TOOLS", func, f"headless write covered by envelope: {path!r}", 2, override=True)
            with filesystem.human_confirmed_write(path):
                _, approved = _run_bracketed(
                    presenter, _label, lambda: _dispatch_tool(tc, tool_registry))
            return approved
        _dbg("OC_CHAT.TOOLS", func, f"headless -> protected-write denied (no envelope): {path!r}",
             2, override=True)
        return filesystem.deny_protected_write(
            result, note="No interactive terminal and no signed PlanAutonomyEnvelope (M8.3); denied.")
 
    # Show a CONTENT preview so the human can inspect WHAT is written (e.g. the
    # endpoints.toml base_url / API host) — a path-only prompt is a weak guard for the
    # config-as-code tier. oct_new has no `content` arg (it generates the file), so the
    # preview is empty there.
    _content = ""
    try:
        _content = str((json.loads(tc.get("arguments") or "{}") or {}).get("content", ""))
    except (ValueError, TypeError):
        _content = ""
    _preview = _content if len(_content) <= 800 else _content[:800] + "\n   …[truncated]…"
    presenter.show_info(
        f"[confirm] {tc.get('name', 'write')} wants to write '{path}' ({reason})."
        "\n  config-as-code / importable-tree write — a write-then-import can execute "
        "code in the harness process."
        + (f"\n  --- content ---\n{_preview}\n  --- end content ---" if _content else "")
    )
    from oc_ui.terminal.input import read_single_key
    try:
        answer = read_single_key(f"  Allow this write to '{path}'? (y/N) ", valid="yn")
    except (EOFError, KeyboardInterrupt):
        return filesystem.deny_protected_write(result, note="Confirmation interrupted; denied (fail-closed).")
    if answer != "y":
        _dbg("OC_CHAT.TOOLS", func, f"human declined write: {path!r}", 2, override=True)
        return filesystem.deny_protected_write(result, note="A human reviewed the write and declined.")
 
    _dbg("OC_CHAT.TOOLS", func, f"human approved write: {path!r}", 2, override=True)
    with filesystem.human_confirmed_write(path):
        _, approved = _run_bracketed(
            presenter, _label, lambda: _dispatch_tool(tc, tool_registry))
    return approved
 
 
def _provider_name(client) -> str:
    """The active provider's name, or ``""`` when the client exposes none."""
    return getattr(getattr(client, "provider", None), "name", "") or ""
 
 
def _tool_error_summary(err: object, rc: object) -> Optional[str]:
    """Why a tool call failed, or ``None`` when it did not (plan 203 / G10).
 
    Built from exactly the two failure signals the loop already holds at the
    completion seam — the ``error`` key and a non-zero ``returncode`` — so
    the summary can never disagree with the ``outcome`` computed beside it.
    A successful call yields ``None``, never an empty string — "no error" and
    "an error with no message" are different facts.
 
    **Secrets are scrubbed here, and the cap is not what protects them.**
    A tool error is built from an exception the tool did not sanitise
    (``oct_tools`` returns ``{"error": str(exc)}`` verbatim), so a failed
    authenticated operation can quote a credential — a git auth failure
    naming a token in its remote URL is the concrete case, and it is only
    ~100 characters, comfortably inside ``ERROR_SUMMARY_MAX_CHARS``. The
    cap bounds *volume*, not *sensitivity*; without this scrub the token
    would land verbatim in the durable ``events.jsonl``, which is the
    designated forensic substrate and the mining corpus. This is the same
    control ``shell.py`` already applies to command stdout/stderr and to
    ``str(e)`` at the identical exception→string seam.
 
    Redaction runs BEFORE the contract's truncation, so a pattern is matched
    against the whole string rather than a clipped one, and a marker can
    never be cut in half in a way that resurrects the tail of a token.
    """
    from oc_chat.tools._sandbox import redact_secrets  # noqa: PLC0415
 
    if err:
        return redact_secrets(str(err))
    if isinstance(rc, int) and rc != 0:
        # No error string on this branch — the F27 gate-refusal case. The
        # returncode is not sensitive, but it goes through the same scrub so
        # the function has one exit contract rather than two.
        return redact_secrets(f"tool exited with returncode {rc}")
    return None
 
 
def _publish_provider_budget(env: dict, provider: str, model: str, *,
                             cache_hit_tokens: int, cache_miss_tokens: int,
                             prompt_tokens: int, completion_tokens: int,
                             at: datetime, session=None) -> None:
    """Publish ``provider_budget_consumed`` for ONE usage sample (plan 190 T1).
 
    The STEP-0 audit confirmed plan 188's finding: this kind had zero
    producers. The data it needs already exists at both ``acc.cost.add``
    seams, so publication reuses exactly the inputs the accumulator prices
    — the event and the turn's ``$`` line can never disagree about a sample.
 
    **The honesty gate (188 T4, inherited).** ``price_usage`` returns
    ``None`` when the sample cannot be priced. A ``0.0`` must never stand in
    for "unknown" — it would be summed downstream as real spend.
 
    **Plan 203 closes 190's loop.** ``cost_usd`` is now ``Optional[float]``
    in the contract, so an unpriceable sample finally HAS a truthful
    encoding: we publish the event with ``cost_usd=None``. The tokens — which
    we do know — reach the stream either way, and the projection counts the
    sample as unpriced instead of the event vanishing. Silence was the honest
    answer only while the payload could not say "unknown"; now that it can,
    silence would itself be the lie, because a consumer could not distinguish
    "no spend" from "spend we could not price".
 
    **Plan 202: the pricing instant travels in.** *at* is the sample's ONE
    captured instant (the send-seam ``_now_utc()`` read) — the same value
    the accumulator folds, so the event and the turn's ``$`` line can
    never price the same sample under two clock reads that disagree
    across the 2026-08-16T16:00Z cutover or a peak-window edge.
 
    *env* is one ``_ev()`` envelope dict (fresh ``event_id`` per call).
    """
    from oc_pricing import rate_at  # noqa: PLC0415
 
    func = "_publish_provider_budget"
    usd = price_usage(
        provider, model,
        cache_hit_tokens=cache_hit_tokens, cache_miss_tokens=cache_miss_tokens,
        prompt_tokens=prompt_tokens, completion_tokens=completion_tokens,
        at=at,
    )
    if usd is None:
        _dbg("OC_CHAT.MAIN", func,
             f"sample unpriceable ({provider}/{model}); publishing cost_usd=None", 3)
    quote = rate_at(provider, model, at=at) if usd is not None else None
    audit_model = model if usd is not None else None
    audit_cache_read = cache_hit_tokens if usd is not None else None
    audit_cache_miss = cache_miss_tokens if usd is not None else None
    audit_priced_at = at.isoformat() if usd is not None else None
    paid_quote = quote if quote is not None and quote.billing_kind != "free" else None
    rate_effective_from = (
        paid_quote.effective_from.isoformat() if paid_quote is not None else None
    )
    rate_verified_at = paid_quote.verified_at if paid_quote is not None else None
    # Plan 237 (CAPS-01 P2): the SESSION fold happens here, off the very value
    # this event carries. One pricing source, one number — a second formula over
    # tokens would eventually disagree with the event stream, and the figure that
    # can stop a session must be the one an auditor can reproduce from the bus.
    # ``None`` increments the unpriced count and adds nothing: an unpriceable
    # sample is unknown spend, never zero spend.
    if session is not None:
        try:
            session.stats.add_cost_sample(usd)
        except AttributeError as exc:  # pragma: no cover - a session without stats
            _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {exc}", 3)
    default_bus().publish(make_provider_budget_consumed(
        **env, provider_name=provider,
        input_tokens=prompt_tokens, output_tokens=completion_tokens,
        cost_usd=usd,
        model=audit_model,
        cache_read_tokens=audit_cache_read,
        cache_miss_tokens=audit_cache_miss,
        priced_at=audit_priced_at,
        rate_effective_from=rate_effective_from,
        rate_verified_at=rate_verified_at,
    ))
 
 
def _mini_report(session, acc: "_TurnAccumulator", turn_t0: float):
    """Build the plan-237 checkpoint mini report from live turn state."""
    from oc_chat.core.session_checkpoints import build_mini_report
 
    cost = session.stats.cost
    return build_mini_report(
        turn_tool_calls=acc.tool_calls,
        tool_name_counts=acc.tool_name_counts,
        files_written=acc.files_written,
        priced_usd=cost.priced_usd,
        unpriced_samples=cost.unpriced_samples,
        elapsed_ms=int((time.monotonic() - turn_t0) * 1000),
    )
 
 
def _cost_gate(session, presenter, acc: "_TurnAccumulator", turn_t0: float):
    """Evaluate the plan-237 dollar hard cap and checkpoints for this session.
 
    Called after every folded sample (main loop and sub-loop), so the session
    stops on the sample that crossed the line rather than one request later.
    Returns a ``CostVerdict``; the caller owns every side effect.
    """
    from oc_chat.core.resource_caps import (
        _COST_CEILING_ENV_VAR, resolve_cost_thresholds_usd,
        resolve_session_cost_hard_cap_usd,
    )
    from oc_chat.core.session_checkpoints import evaluate_cost_checkpoints
 
    return evaluate_cost_checkpoints(
        session_id=session.session_id,
        cost=session.stats.cost,
        presenter=presenter,
        report=_mini_report(session, acc, turn_t0),
        is_interactive=_stdin_is_interactive(),
        hard_cap_usd=resolve_session_cost_hard_cap_usd(),
        thresholds=resolve_cost_thresholds_usd(),
        env_var_name=_COST_CEILING_ENV_VAR,
    )
 
 
def _end_turn_on_cost(session, presenter, reason: str, ev) -> None:
    """Close a turn cleanly at a cost bound: say why, in every channel.
 
    The refusal is appended as a normal assistant message BEFORE the loop
    breaks, so the transcript is well-formed (no assistant message carrying
    tool_calls has been added for this iteration yet, so nothing is orphaned)
    and the model reads the reason on the next turn instead of being cut off
    without explanation.
    """
    func = "_end_turn_on_cost"
    presenter.show_error(reason)
    session.add_assistant(reason)
    default_bus().publish(make_assistant_message(**ev(), text=reason, is_final=True))
    _dbg("OC_CHAT.MAIN", func, "turn ended by a cost checkpoint", 1, override=True)
 
 
def _prompt_layer_digests(api_messages: list, tools) -> "tuple[list[str], dict[str, str]]":
    """Hash the request layers that are exactly recoverable at the send seam.
 
    Returns ``(layer_ids, layer_hashes)`` for
    :func:`~oc_chat.core.events.make_prompt_rendered`. Only layers whose
    bytes we can read off the assembled request are hashed — nothing is
    inferred:
 
    - ``system_prompt`` — the system message actually sent. In this product
      that one message carries the inventory's ``base_system_prompt`` +
      ``compact_tool_map`` + environment block concatenated
      (:mod:`oc_chat.core.session`), so it is hashed as the single unit it
      is on the wire rather than split into layers we cannot separate.
    - ``tool_schemas`` — the tool definitions sent alongside, which sit in
      the same cached prefix. Omitted entirely when no tools are sent.
 
    Hashes are truncated SHA-256 (16 hex chars): identifiers for change
    detection, not a way to recover the prompt text.
    """
    layer_ids: list[str] = []
    layer_hashes: dict[str, str] = {}
    system = next((m for m in api_messages if m.get("role") == "system"), None)
    if system is not None:
        layer_ids.append("system_prompt")
        layer_hashes["system_prompt"] = hashlib.sha256(
            str(system.get("content", "")).encode("utf-8")
        ).hexdigest()[:16]
    if tools:
        layer_ids.append("tool_schemas")
        layer_hashes["tool_schemas"] = hashlib.sha256(
            json.dumps(tools, sort_keys=True, default=str).encode("utf-8")
        ).hexdigest()[:16]
    return layer_ids, layer_hashes
 
 
def _prompt_version_for(layer_ids: list, layer_hashes: dict) -> str:
    """Derive the ``prompt_version`` from the hashed layers (plan 190 T1).
 
    ``make_prompt_rendered`` requires a ``prompt_version``, and the product
    declares none — the STEP-0 audit found zero non-test references to the
    field anywhere. Rather than stamp a fabricated literal, the version is
    *content-addressed*: a digest over the layer digests, so it changes
    exactly when a hashed layer changes and is identical across processes
    for an identical prefix. That is precisely the property cache
    governance needs (synthesis §5).
    """
    joined = "|".join(f"{lid}={layer_hashes.get(lid, '')}" for lid in layer_ids)
    return "sha256:" + hashlib.sha256(joined.encode("utf-8")).hexdigest()[:16]
 
 
def _fold_sub_loop_usage(session, acc, turn_stats, client, sub_result,
                         sub_model: str = "", *, at: datetime) -> None:
    """Fold a sub-loop's token spend into parent COST accounting (plan 134 B2).
 
    Cost only: ``session.record_usage`` + the turn accumulator's cost fields +
    ``turn_stats`` for the live display.  Deliberately does **not** touch
    ``acc.max_prompt_in_turn`` (M1) — sub-loop prompt tokens are cost, not
    main-loop context, and would inflate C2's ``ctx:%`` gauge.
 
    Plan 188 T4: *sub_model* is the model the sub-loop actually ran on, so its
    USD sample is priced at that model's rates rather than the parent's.
 
    Plan 202: *at* is the sub-loop sample's ONE captured pricing instant —
    the caller reads ``_now_utc()`` once and hands the same value here and
    to ``_publish_provider_budget``, so the fold and the event agree.
    """
    usage = sub_result.usage
    if not usage:
        return
    session.record_usage(usage)
    acc.prompt_tokens     += int(usage.get("prompt_tokens", 0) or 0)
    acc.completion_tokens += int(usage.get("completion_tokens", 0) or 0)
    acc.total_tokens      += int(usage.get("total_tokens", 0) or 0)
    ptd = usage.get("prompt_tokens_details")
    if isinstance(ptd, dict):
        acc.prompt_cached += int(ptd.get("cached_tokens", 0) or 0)
    sub_cache = client.parse_usage(usage)
    acc.cache_hit_tokens  += sub_cache.cache_hit_tokens
    acc.cache_miss_tokens += sub_cache.cache_miss_tokens
    if sub_model:
        acc.cost.add(
            _provider_name(client), sub_model,
            cache_hit_tokens=sub_cache.cache_hit_tokens,
            cache_miss_tokens=sub_cache.cache_miss_tokens,
            prompt_tokens=int(usage.get("prompt_tokens", 0) or 0),
            completion_tokens=int(usage.get("completion_tokens", 0) or 0),
            at=at,
        )
    if turn_stats is not None:
        turn_stats.add_usage(usage)
 
 
def _warn_unknown_usage_fields(session, presenter, provider_name, model,
                               unknown_raw) -> None:
    """Warn once per (provider, model, field) about unrecognised usage fields.
 
    Throttle state lives on the **session** (transient, non-persisted) so the
    sub-loop needs no session handle and we avoid a process-lifetime module
    global (plan 134 N4 — the same anti-pattern A3 deletes).  Always-on: a new
    usage field is an accounting-integrity signal, not verbose noise.
    """
    func = "_warn_unknown_usage_fields"
    if not unknown_raw:
        return
    seen = getattr(session, "_warned_usage_fields", None)
    if seen is None:
        seen = set()
        session._warned_usage_fields = seen
    for entry in unknown_raw:
        field_name = entry.get("field")
        key = (provider_name, model, field_name)
        if key in seen:
            continue
        seen.add(key)
        _dbg("OC_CHAT.MAIN", func,
             f"unknown sub-loop usage field {field_name!r} "
             f"(provider={provider_name} model={model}); carried raw, not merged",
             1, override=True)
        presenter.show_info(
            f"[accounting] new usage field '{field_name}' from "
            f"{provider_name}/{model} — carried raw, not folded into cost."
        )
 
 
def _schema_tool_name(schema: dict) -> str:
    """Tool name from an OpenAI-style ``{"function":{"name":…}}`` or flat schema."""
    if not isinstance(schema, dict):
        return ""
    fn = schema.get("function", schema)
    return str(fn.get("name", "")) if isinstance(fn, dict) else ""
 
 
def _filter_readonly_tools(
    tool_registry: dict, tool_schemas: list, mutating: "tuple[str, ...]",
) -> "tuple[dict, list]":
    """Drop the *mutating* tools from BOTH the registry and the schemas (plan 137
    §3.10) so a sub-loop is never even told it has them. Returns new objects; the
    inputs are not mutated."""
    mut = set(mutating)
    reg = {k: v for k, v in tool_registry.items() if k not in mut}
    schemas = [s for s in (tool_schemas or []) if _schema_tool_name(s) not in mut]
    return reg, schemas
 
 
#: M8.0 WU-9c / F35 — the sub-loop's deny-by-default POSITIVE allowlist (least
#: authority): only pure read / list / search / recall tools. Everything else —
#: ``fetch_url``/``web_search`` (egress), ``run_command`` (exec), ``write_file``/
#: ``remember`` (mutation), every ``oct_*`` (subprocess), plan mutation, and any
#: future tool — is denied by default. A subtractive filter would leak a newly added
#: tool to the sub-loop; this does not (F10/F35).
_SUBLOOP_ALLOWED_TOOLS: frozenset[str] = frozenset({
    "read_file", "read_chunks", "file_info", "list_dir", "search_files", "text_search", "recall",
})
 
 
def _subloop_positive_allowlist(
    tool_registry: dict, tool_schemas: list,
) -> "tuple[dict, list]":
    """Restrict a sub-loop to the :data:`_SUBLOOP_ALLOWED_TOOLS` positive allowlist
    (deny-by-default). Returns new ``(registry, schemas)``; the inputs are unchanged."""
    reg = {k: v for k, v in tool_registry.items() if k in _SUBLOOP_ALLOWED_TOOLS}
    schemas = [s for s in (tool_schemas or []) if _schema_tool_name(s) in _SUBLOOP_ALLOWED_TOOLS]
    return reg, schemas
 
 
def _maybe_run_verification(
    session: "ChatSession",
    acc: "_TurnAccumulator",
    vcfg: "VerificationConfig",
    presenter: "Presenter",
    baseline: object | None,
) -> None:
    """Plan 137 §3.2/§3.4 — the post-mutating-dispatch additive verification hook.
 
    Caller has already confirmed ``vcfg.enabled``. Runs the deterministic sensor on
    the turn's CUMULATIVE ``changed_files`` (capped at ``max_runs_per_turn``), and on
    NEW findings appends one compacted ``[verification]`` user message to the volatile
    tail (never ``messages[0]``/``[1]``). Degraded ⇒ a one-line note, never a block.
    Fully additive: it only appends; it never alters control flow.
    """
    func = "_maybe_run_verification"
    if acc.verification_runs >= vcfg.max_runs_per_turn:
        _dbg("OC_CHAT.MAIN", func, f"per-turn cap reached ({vcfg.max_runs_per_turn})", 3)
        return
    if not acc.changed_files:
        return
    from oc_chat.core.verification import run_verification, render_observation
    acc.verification_runs += 1
    result = run_verification(sorted(acc.changed_files), baseline, vcfg, presenter=presenter)
    if result.findings:
        acc.pending_findings = result
        session.messages.append({
            "role": "user",
            "content": render_observation(result, vcfg.max_findings),
        })
        session.dirty = True
        _dbg("OC_CHAT.MAIN", func,
             f"injected {len(result.findings)} new finding(s) over {len(acc.changed_files)} file(s)", 2)
    elif result.degraded:
        presenter.show_info(f"[verification] skipped/degraded: {result.skip_reason or 'unavailable'}")
        _dbg("OC_CHAT.MAIN", func, f"degraded: {result.skip_reason}", 3)
    else:
        _dbg("OC_CHAT.MAIN", func, "clean — no new findings", 3)
 
 
def run_turn(
    session: "ChatSession",
    client: "ClientInterface",
    presenter: "Presenter",
    model_manager: "ModelManager",
    tool_registry: dict,
    tool_schemas: list,
    ctx: "Context | None" = None,
    turn_stats: "TurnStats | None" = None,
    enhanced_presenter: "EnhancedPresenter | None" = None,
    cancellation: "CancellationToken | None" = None,
) -> None:
    """Execute one user-turn through up to ``MAX_REACT_ITERATIONS`` ReAct cycles.
 
    Multiple tool calls in a single model response are executed in
    parallel.  Captures usage data and records it on the session.
    Shows per-iteration usage and a final aggregated turn summary.
 
    If *ctx* is provided and ``ctx.in_goal`` is ``True``, file-modifying
    tool calls (``write_file``) are recorded in ``ctx.goal_files_changed``
    for later changelog updates.
 
    If *turn_stats* is provided (a :class:`TurnStats`), its counters are
    updated so the prompt's rprompt can show live token counts.
    """
    func = "run_turn"
    # Plan 147 D1: the summary-ordering pin — show_turn_summary's duration is
    # computed directly from this turn-start monotonic timestamp at call
    # time, independent of TurnStats.finish_turn (called after, below).
    _turn_t0 = time.monotonic()
    # ── Turn-level accumulator ─────────────────────────────────────────────
    acc = _TurnAccumulator()
    from oc_chat.core.stall_detector import StallDetector
    stall_detector = StallDetector.from_defaults()
 
    # ── M2.3: event-stream instrumentation ─────────────────────────────────
    # turn_id groups the turn's events and is the synthetic non-None parent the
    # contract requires (a dedicated session-root event remains future work).
    # sid is the session's bound, persistent identity (M2.5 — the _session_id_for
    # shim has been retired). _ev() supplies the shared envelope kwargs; each
    # emission still names its make_* factory explicitly (ADR-3).
    # user_input_received fires at turn entry, before any client call (ADR-5).
    turn_id = new_event_id()
    sid = session.session_id
 
    def _ev() -> dict:
        return {"event_id": new_event_id(), "parent_event_id": turn_id,
                "session_id": sid, "ts": _now_iso(), "turn_id": turn_id}
 
    default_bus().publish(make_user_input_received(
        **_ev(), text=_extract_last_user_task(session.messages)))
 
    # Phase O2: rebind the client to the active endpoint at turn start so
    # /endpoint use <name> takes effect on the next user turn. Guarded
    # so tests / callers without an endpoint registry are unaffected.
    _ep_registry = getattr(ctx, "endpoint_registry", None) if ctx is not None else None
    if _ep_registry is not None and _ep_registry.entries:
        try:
            client.rebind_to(_ep_registry.active())
        except KeyError:
            # Empty registry caught here defensively; the .entries guard
            # above should already prevent it. Continue with current client.
            pass
 
    if turn_stats is not None:
        turn_stats.start_turn()
 
    react_cap = session.react_cap()
    _dbg("OC_CHAT.MAIN", func, "turn START", 2)
 
    # M8.0 F67: the external-content trust set is session-scoped. Re-bind it to the
    # active session each turn so a /session new|load|switch (which changes the
    # session between turns) starts with an empty trust set — trust never leaks
    # across sessions in one process.
    from oc_chat.core.external_content_gate import bind_active_session
    bind_active_session(session)
 
    presenter.start_turn()
 
    ensure_system_includes_tool_reference(session, tool_schemas)
 
    # ── Plan 137: verification backpressure gate (off ⇒ no new code path) ───────
    # Resolve the typed config once per turn. When disabled, _verif_enabled is
    # False and neither the changed-files tracking nor the sensor hook below runs,
    # so run_turn is byte-identical to pre-137. The baseline is captured lazily on
    # the first enabled turn (≈ session start — nothing has mutated yet) and cached
    # on the session instance for reuse.
    from oc_chat.core.settings import runtime_settings, runtime_mode, verification_config
    _vcfg = verification_config(runtime_settings(), runtime_mode())
    _verif_enabled = _vcfg.enabled
    _verif_baseline = None
    if _verif_enabled and _vcfg.baseline:
        _verif_baseline = getattr(session, "_verification_baseline", None)
        if _verif_baseline is None:
            from oc_chat.core.verification import capture_baseline
            _bspin = presenter.spinner_start("capturing verification baseline…")
            try:
                _verif_baseline = capture_baseline(_vcfg)
            finally:
                presenter.spinner_stop(_bspin)
            try:
                session._verification_baseline = _verif_baseline
            except (AttributeError, TypeError):
                pass  # session forbids dynamic attrs — proceed with an unsaved baseline
 
    sub_loops_this_turn = 0  # plan 132: per-turn offload backstop
 
    # Clear any stale cooperative tool-cancel flag from a prior turn so tools in
    # this turn start runnable; a Ctrl+C during this turn re-sets it.
    reset_tool_cancel()
 
    def _confirm_stop() -> bool:
        # Ask-before-cancelling (operator decision 2026-08-12, 188-T5 v2):
        # press 1 cancelled NOTHING — the handler only requested this
        # checkpoint — so "n" genuinely resumes with the in-flight tool's
        # result intact. 'y' stops; a non-interactive interrupt (no TTY)
        # stops, never loops.
        import sys as _sys
        from oc_ui.terminal.input import read_single_key
 
        if not _sys.stdin.isatty():
            return True
        return read_single_key("\nStop the current action? (y/N) ", valid="yn") == "y"
 
    for _iteration in range(react_cap):
        # Two-stage Ctrl+C: a first press set ``interrupt_requested``; confirm it
        # here at the iteration boundary (no Live region active). A hard cancel
        # (second press) raises CancellationError straight through.
        confirm_or_continue(cancellation, _confirm_stop)
 
        # Plan 237 (CAPS-01 P4): the hard cap is a PRE-FLIGHT, not only a
        # post-fold check. A session that reached the cap on its previous turn
        # must make no provider call on this one either — "no further provider
        # call is made this session" is the whole promise, and it has to survive
        # a turn boundary. Priced spend only; the refusal says so.
        _precheck = _cost_gate(session, presenter, acc, _turn_t0)
        if _precheck.stop:
            _end_turn_on_cost(session, presenter, _precheck.reason, _ev)
            break
 
        t0 = time.monotonic()
 
        route = resolve_iteration_routing(
            session_model=session.model,
            session_reasoning_effort=session.reasoning_effort,
            model_manager=model_manager,
            iteration=_iteration,
            is_finalize_nudge=(_iteration == react_cap - 1),
            messages=session.messages,
            token_profile=getattr(session, "token_profile", "balanced") or "balanced",
        )
 
        _dbg("OC_CHAT.MAIN", func,
             f"iter {_iteration}/{react_cap} model={route.model} effort={route.effort} max_tokens={route.max_tokens}", 3)
 
        caps  = model_manager.capabilities(route.model)
        tools = tool_schemas if (tool_registry and caps.get("supports_tools", True)) else None
 
        tool_calls_this_iter: list = []
        full_content               = ""
        full_reasoning             = ""
        last_usage                 = None
        # None == the provider did not report one. Absence is "unknown", never
        # "stop" — only an explicit "length" proves a max_tokens truncation.
        finish_reason: str | None  = None
 
        api_messages = messages_for_api(
            session.messages,
            session_name=session.name,
            inject_finalize_nudge=(_iteration == react_cap - 1),
            model=route.model,
        )
 
        # ── Sub-loop delegation ──────────────────────────────────
        # Enhanced sub-loop triggering logic with task complexity analysis
        should_use_sub = should_use_sub_loop(
            task=_extract_last_user_task(session.messages),
            iteration_count=_iteration,
            recent_messages=session.messages[-10:]  # Last 10 messages for analysis
        )
        
        # Check for response repetition bug (E4) and trigger context refresh if needed
        is_repeating = _detect_response_repetition(session.messages)
        
        # Plan 132: route.would_switch_model is NO LONGER a sub-loop trigger.
        # The cheap-tier signal kept reclassifying the same stale read batch
        # into an offload every iteration (sub-loop results are injected as
        # user messages, so last_assistant_tool_names never advances), causing
        # the iter-0→119 infinite offload in terminal-out-01.md. The sub-loop
        # now fires ONLY on the guarded should_use_sub_loop() or is_repeating,
        # and is hard-capped per turn. The would_switch_model / cheap_model
        # fields remain (step_routing unchanged) for a future targeted-offload
        # redesign — see the living ROADMAP-EXTENDED.md backlog (read-size/
        # intent-driven, opt-in). DO NOT
        # re-add would_switch_model here without that redesign + a re-trigger guard.
        # Plan 134 interim model-lock: sub-loops run on the SESSION model, not
        # route.cheap_model — no silent flash usage until the targeted, opt-in
        # cheap-model offload redesign lands (which stays deferred). This is the
        # one-line interim; route.cheap_model is intentionally not used here now.
        sub_model = session.model
        wants_sub_loop = should_use_sub or is_repeating
        if wants_sub_loop and sub_loops_this_turn >= _MAX_SUB_LOOPS_PER_TURN:
            presenter.show_info(
                f"[sub-loop] iter={_iteration}: offload suppressed "
                f"(per-turn cap {_MAX_SUB_LOOPS_PER_TURN} reached)"
            )
            _dbg("OC_CHAT.MAIN", func,
                 f"sub-loop suppressed at iter {_iteration}: per-turn cap reached", 2)
        if wants_sub_loop and sub_loops_this_turn < _MAX_SUB_LOOPS_PER_TURN:
            sub_loops_this_turn += 1
            presenter.show_info(
                f"[sub-loop] iter={_iteration}: running sub-loop on {sub_model} "
                f"(complexity={should_use_sub} repetition={is_repeating})"
            )
            _dbg("OC_CHAT.MAIN", func,
                 f"triggering sub-loop on {sub_model} complexity_task={should_use_sub} repetition={is_repeating}", 2)
            # Extract the most recent user request as the task
            sub_task = _extract_last_user_task(session.messages)
            if not sub_task:
                sub_task = "Complete the assigned task using available tools."
            
            # If we're repeating, add context refresh instruction
            if is_repeating:
                # OCT-LINT: disable=no_hardcoded_secrets reason="Static event labels, prompts, and model-routing strings are not secrets"
                sub_task += "\n\n[CONTEXT_REFRESH: The previous responses were too similar. Please provide a different perspective or approach.]"
            
            # Integrate enhanced UI into react_loop for real-time updates
            if enhanced_presenter is not None:
                enhanced_presenter.show_sub_loop_trigger(sub_model, sub_task)
                
                # Update enhanced UI statistics
                if hasattr(enhanced_presenter, 'get_stats_tracker'):
                    stats_tracker = enhanced_presenter.get_stats_tracker()
                    if stats_tracker:
                        # Prefer API-reported prompt_tokens; fall back to estimate
                        api_prompt = last_usage.get("prompt_tokens", 0) if last_usage else 0
                        if api_prompt > 0:
                            stats_tracker.update_context_usage(api_prompt)
                        else:
                            estimated_tokens = _estimate_tokens(session.messages)
                            stats_tracker.update_context_usage(estimated_tokens)
                        
                        # Update turn statistics. Plan 147 D1: no duration_ms here —
                        # last_usage['dur_ms'] is never populated by any provider
                        # (the dead always-zero path this removes); StatisticsTracker's
                        # duration_ms param defaults to 0 and is unused by any reader.
                        prompt_tokens = last_usage.get('prompt_tokens', 0) if last_usage else 0
                        completion_tokens = last_usage.get('completion_tokens', 0) if last_usage else 0
                        cache_hits = last_usage.get('cache_hit_tokens', 0) if last_usage else 0
                        tool_calls = len(tool_calls_this_iter) if tool_calls_this_iter else 0
 
                        stats_tracker.update_turn_stats(
                            prompt_tokens=prompt_tokens,
                            completion_tokens=completion_tokens,
                            cache_hits=cache_hits,
                            tool_calls=tool_calls
                        )
            
            # ── M8.0 WU-9c / F35: least-authority sub-loop surface ─────────────
            # The sub-loop ALWAYS gets a deny-by-default POSITIVE allowlist (pure
            # read/list/search/recall) — independent of the verification config and
            # strictly stronger than the old subtractive read-only filter: it never
            # leaks fetch_url/run_command/oct_*/write_file/remember (or a newly added
            # tool) to a sub-agent whose result re-enters as model_generated.
            _sub_registry, _sub_schemas = _subloop_positive_allowlist(
                tool_registry, tool_schemas or [])
            sub_result = run_sub_loop(
                client=client,
                task=sub_task,
                model=sub_model,
                tool_registry=_sub_registry,
                tool_schemas=_sub_schemas,
                session_name=session.name,
                presenter=presenter,  # Phase H.4a-iii: wire spinner through to sub-loop
                parent_event_id=turn_id, parent_session_id=sid,  # M2.4: sub-loop event chain
            )
            # Inject sub-loop result as a user message so the main loop can
            # reason over it next iteration (user role avoids the tool-role
            # API validation error). Bound it at the MAIN model's tool-result
            # cap (plan 134 A4) — it's a user message, so api_context's
            # tool/assistant truncation never touches it; this is the only
            # bound. On truncation, point at the archived full transcript.
            # M8.0 WU-9 review fix (F46/F51): the sub-loop's synthesized text re-enters
            # the deciding context as a user message — redact secret patterns at that
            # boundary (a sub-loop tool like recall can surface unredacted memory).
            from oc_chat.tools._sandbox import redact_secrets
            sub_text = redact_secrets(sub_result.text)
            _sub_cap = resolve_tool_result_cap(session.model)
            if len(sub_text) > _sub_cap:
                if sub_result.artifact_path:
                    _marker = (
                        f"\n[OUTPUT_TRUNCATED at {_sub_cap} chars; full sub-loop "
                        f"transcript archived at {sub_result.artifact_path}]"
                    )
                else:
                    _marker = f"\n[OUTPUT_TRUNCATED at {_sub_cap} chars]"
                sub_text = sub_text[:_sub_cap] + _marker
            session.messages.append({
                "role": "user",
                "content": f"[Sub-loop result]\n{sub_text}",
            })
 
            # Update enhanced UI statistics if available
            if hasattr(presenter, 'get_stats_tracker'):
                stats_tracker = presenter.get_stats_tracker()
                if stats_tracker:
                    # Prefer API-reported prompt_tokens; fall back to estimate
                    api_prompt = last_usage.get("prompt_tokens", 0) if last_usage else 0
                    if api_prompt > 0:
                        stats_tracker.update_context_usage(api_prompt)
                    else:
                        estimated_tokens = _estimate_tokens(session.messages)
                        stats_tracker.update_context_usage(estimated_tokens)
                    
                    # Update turn statistics. Plan 147 D1: no duration_ms here —
                    # last_usage['dur_ms'] is never populated by any provider
                    # (the dead always-zero path this removes); StatisticsTracker's
                    # duration_ms param defaults to 0 and is unused by any reader.
                    prompt_tokens = last_usage.get('prompt_tokens', 0) if last_usage else 0
                    completion_tokens = last_usage.get('completion_tokens', 0) if last_usage else 0
                    cache_hits = last_usage.get('cache_hit_tokens', 0) if last_usage else 0
                    tool_calls = len(tool_calls_this_iter) if tool_calls_this_iter else 0
 
                    stats_tracker.update_turn_stats(
                        prompt_tokens=prompt_tokens,
                        completion_tokens=completion_tokens,
                        cache_hits=cache_hits,
                        tool_calls=tool_calls
                    )
            # Fold the sub-loop's real token spend into parent COST + live
            # display (B2) — not just a tool-call tick. Cost only; the ctx:%
            # gauge stays main-loop-only (M1, handled inside the helper).
            acc.tool_calls += 1
            # Plan 202 decision 3: ONE pricing instant for the sub-loop
            # sample — the fold below and the publish share it, so the two
            # cannot disagree across a tariff/band boundary.
            at_sub = _now_utc()
            _fold_sub_loop_usage(session, acc, turn_stats, client, sub_result,
                                 sub_model=sub_model, at=at_sub)
            # Plan 190 T1: a sub-loop's spend is a sample too. Published at
            # the CALL SITE (not inside the fold helper) so that helper stays
            # pure accounting, and guarded on sub_model exactly as the fold's
            # own pricing is — an unknown sub-model is unpriceable, not free.
            if sub_model and sub_result.usage:
                _sub_cache = client.parse_usage(sub_result.usage)
                _publish_provider_budget(
                    _ev(), _provider_name(client), sub_model,
                    cache_hit_tokens=_sub_cache.cache_hit_tokens,
                    cache_miss_tokens=_sub_cache.cache_miss_tokens,
                    prompt_tokens=int(sub_result.usage.get("prompt_tokens", 0) or 0),
                    completion_tokens=int(sub_result.usage.get("completion_tokens", 0) or 0),
                    at=at_sub, session=session,
                )
            # Plan 237: a sub-loop's spend is session spend, so the same bounds
            # apply to it. Evaluated here rather than only on the main-loop
            # sample, or a turn that offloads its work would spend past the cap
            # without ever being asked.
            _sub_verdict = _cost_gate(session, presenter, acc, _turn_t0)
            if _sub_verdict.stop:
                _end_turn_on_cost(session, presenter, _sub_verdict.reason, _ev)
                break
            _warn_unknown_usage_fields(
                session, presenter, _provider_name(client), sub_model,
                sub_result.unknown_usage_raw)
            _dbg("OC_CHAT.MAIN", func,
                 f"sub-loop returned {len(sub_result.text)} chars "
                 f"(iters={sub_result.iterations} tools={sub_result.tool_calls})", 2)
            continue  # next main-loop iteration uses sub-loop result
 
        # Merge provider-translated generation options into extra_body
        _extra_body = route.extra_body
        if ctx is not None and ctx.generation_options is not None:
            _gen_opts = client.provider.translate_generation_options(ctx.generation_options)
            if _gen_opts:
                _extra_body = {**(_extra_body or {}), **_gen_opts}
 
        # Plan 190 T1: prompt_rendered at the SEND seam, not at the
        # messages_for_api() call above — the sub-loop branch assembles a
        # request it then discards, and an event claiming that prompt was
        # rendered for the provider would be false. Here the request is
        # final (messages + tools + model) and about to go out.
        _layer_ids, _layer_hashes = _prompt_layer_digests(api_messages, tools)
        default_bus().publish(make_prompt_rendered(
            **_ev(),
            prompt_version=_prompt_version_for(_layer_ids, _layer_hashes),
            layer_ids=_layer_ids, layer_hashes=_layer_hashes,
        ))
 
        # Plan 202 decision 3: THIS iteration's pricing instant, captured
        # ONCE at the send seam. The usage that comes back from this
        # request is priced at the tariff in force now — by both the
        # accumulator fold and the budget event below, which must never
        # read the clock again (two reads could straddle the live
        # 2026-08-16T16:00Z cutover or a peak-window edge and disagree).
        at_sample = _now_utc()
 
        try:
            with presenter.turn():
                presenter.show_waiting()
                for chunk_type, data in client.stream_chat(
                    api_messages,
                    route.model,
                    effort=route.effort,
                    extra_body=_extra_body,
                    tools=tools,
                    max_tokens=route.max_tokens,
                    cancellation=cancellation,
                ):
                    if chunk_type == "reasoning":
                        full_reasoning += data
                        presenter.show_reasoning(data)
                    elif chunk_type == "content":
                        full_content += data
                        presenter.show_content(data)
                    elif chunk_type == "tool_call":
                        tool_calls_this_iter.append(data)
                    elif chunk_type == "usage":
                        last_usage = data
                    elif chunk_type == "finish_reason":
                        finish_reason = data
        except (KeyboardInterrupt, CancellationError):
            # User pressed Ctrl+C (KeyboardInterrupt) or cooperative
            # cancellation was signalled (CancellationError). Record
            # partial cost so session totals stay accurate.
            _dbg("OC_CHAT.MAIN", func,
                 "stream cancelled by user (Ctrl+C)", 2, override=True)
 
            if last_usage is not None:
                # API-confirmed usage chunk arrived before the interrupt
                partial_usage = last_usage
                cost_label = "exact"
            else:
                # Estimate via Phase P.1a tokenizer; cl100k_base is close
                # enough to DeepSeek tokenization for cost estimation.
                from oc_chat.core.tokenizer import count_message_tokens, count_tokens
                est_prompt = count_message_tokens(api_messages)
                est_completion = (
                    count_tokens(full_content) + count_tokens(full_reasoning)
                )
                partial_usage = {
                    "prompt_tokens": est_prompt,
                    "completion_tokens": est_completion,
                    "total_tokens": est_prompt + est_completion,
                    "_estimated": True,
                }
                cost_label = "approx"
 
            session.record_usage(partial_usage)
            if turn_stats is not None:
                turn_stats.add_usage(partial_usage)
 
            # Keep prompt-token tracker in sync so compression heuristics
            # still trigger correctly on the next turn.
            pt = int(partial_usage.get("prompt_tokens") or 0)
            session.last_turn_max_prompt_tokens = max(
                int(getattr(session, "last_turn_max_prompt_tokens", 0) or 0), pt
            )
            acc.prompt_tokens     += int(partial_usage.get("prompt_tokens", 0))
            acc.completion_tokens += int(partial_usage.get("completion_tokens", 0))
            acc.total_tokens      += int(partial_usage.get("total_tokens", 0))
            acc.max_prompt_in_turn = max(acc.max_prompt_in_turn, pt)
 
            p = partial_usage.get("prompt_tokens", 0)
            c = partial_usage.get("completion_tokens", 0)
            presenter.show_info(
                f"Cancelled. ({cost_label} cost: {p} prompt + {c} completion tokens)"
            )
 
            # M2.3: terminal error event before re-raise (ADR-4).
            default_bus().publish(make_error(
                **_ev(), error_summary="cancelled: stream cancelled by user"))
            # Re-raise so the REPL backstop pops the orphaned user message
            raise
        except Exception as e:
            # SYSTEM_ERROR — surfacing stream/transport failure with override
            _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: {type(e).__name__}: {e}", 1, override=True)
            # M2.3: terminal error event before re-raise (ADR-4).
            default_bus().publish(make_error(
                **_ev(), error_summary=f"{type(e).__name__}: {e}"))
            raise
 
        acc.iterations += 1
 
        # ── Truncation telemetry ───────────────────────────────────────────
        # "length" is the provider saying it stopped because the completion
        # budget ran out, not because the model was finished. Logged at L1
        # with the routing that produced it so the caps can be sized from
        # real hit-rates instead of guessed. A truncated tool-call batch gets
        # a notice here because the failure surfaces later as an opaque
        # "invalid JSON" tool error the operator can't otherwise explain.
        if finish_reason == "length":
            _dbg("OC_CHAT.MAIN", func,
                 f"TRUNCATED at max_tokens: iter={_iteration} model={route.model} "
                 f"effort={route.effort} max_tokens={route.max_tokens} "
                 f"content={len(full_content)}c reasoning={len(full_reasoning)}c "
                 f"tool_calls={len(tool_calls_this_iter)}", 1, override=True)
            if tool_calls_this_iter:
                presenter.show_info(TRUNCATED_TOOL_CALL_NOTICE.format(
                    cap=route.max_tokens, n=len(tool_calls_this_iter)))
 
        # ── Per-iteration usage stats ──────────────────────────────────────
        usage_stats = None  # default; set below
        if last_usage:
            presenter.show_usage(last_usage)
            session.record_usage(last_usage)
            # Accumulate turn-level totals
            acc.prompt_tokens     += last_usage.get("prompt_tokens", 0)
            acc.completion_tokens += last_usage.get("completion_tokens", 0)
            acc.total_tokens      += last_usage.get("total_tokens", 0)
            if last_usage.get("prompt_tokens_details"):
                pd = last_usage["prompt_tokens_details"]
                if isinstance(pd, dict):
                    acc.prompt_cached += pd.get("cached_tokens", 0)
 
            # Provider-aware cache stats: DeepSeek's parse_usage uses the
            # generic helper; future providers can override.
            usage_stats = client.parse_usage(last_usage)
            perf = client.parse_performance(last_usage)
            if not perf.is_default():
                acc.perf_stats = perf
            _dbg("OC_CHAT.CLIENT", func,
                 f"usage: prompt={last_usage.get('prompt_tokens', 0)} "
                 f"completion={last_usage.get('completion_tokens', 0)} "
                 f"cache_hit={usage_stats.cache_hit_tokens} "
                 f"cache_miss={usage_stats.cache_miss_tokens}", 3)
            acc.cache_hit_tokens  += usage_stats.cache_hit_tokens
            acc.cache_miss_tokens += usage_stats.cache_miss_tokens
 
            # Plan 188 T4: price THIS sample at the model that produced it —
            # and (plan 202) at the ONE instant captured at its send seam.
            acc.cost.add(
                _provider_name(client), route.model,
                cache_hit_tokens=usage_stats.cache_hit_tokens,
                cache_miss_tokens=usage_stats.cache_miss_tokens,
                prompt_tokens=int(last_usage.get("prompt_tokens") or 0),
                completion_tokens=int(last_usage.get("completion_tokens") or 0),
                at=at_sample,
            )
            # Plan 190 T1: the same sample on the bus (cost_usd=None if
            # unpriceable) — same tokens, same model, same at_sample.
            _publish_provider_budget(
                _ev(), _provider_name(client), route.model,
                cache_hit_tokens=usage_stats.cache_hit_tokens,
                cache_miss_tokens=usage_stats.cache_miss_tokens,
                prompt_tokens=int(last_usage.get("prompt_tokens") or 0),
                completion_tokens=int(last_usage.get("completion_tokens") or 0),
                at=at_sample, session=session,
            )
            # Plan 237: evaluate the money bounds on the sample that just
            # landed. This runs BEFORE this iteration's tool calls are appended
            # to history, so a stop here leaves no assistant message carrying
            # unanswered tool_calls — the transcript stays well-formed.
            _verdict = _cost_gate(session, presenter, acc, _turn_t0)
            if _verdict.stop:
                _end_turn_on_cost(session, presenter, _verdict.reason, _ev)
                break
 
            pt = int(last_usage.get("prompt_tokens") or 0)
            acc.max_prompt_in_turn = max(acc.max_prompt_in_turn, pt)
 
            # Update thread-safe turn stats for live rprompt
            if turn_stats is not None:
                turn_stats.add_usage(last_usage)
 
 
        # ── Per-iteration stat emission (G.6.D) ───────────────────────
        dur_ms = int((time.monotonic() - t0) * 1000)
        _u = last_usage or {}
        _cache_hit = usage_stats.cache_hit_tokens if usage_stats else 0
        _cache_miss = usage_stats.cache_miss_tokens if usage_stats else 0
        _now = datetime.now().astimezone()
        _ts = _now.isoformat(timespec='milliseconds')
        stat = IterationStat(
            ts=_ts,
            iter=_iteration,
            model=route.model,
            effort=route.effort or "",
            tools=tuple(tc["name"] for tc in tool_calls_this_iter),
            prompt_tokens=_u.get("prompt_tokens", 0),
            completion_tokens=_u.get("completion_tokens", 0),
            cache_hit_tokens=_cache_hit,
            cache_miss_tokens=_cache_miss,
            dur_ms=dur_ms,
        )
        if getattr(session, 'show_iteration_stats', True):
            presenter.show_iteration_stat(stat)
        if session.iteration_stats_file is not None:
            try:
                _tools_csv = ",".join(stat.tools) if stat.tools else "-"
                session.iteration_stats_file.write(
                    f'{stat.ts},{stat.iter},{stat.model},{stat.effort},{_tools_csv},'
                    f'{stat.prompt_tokens},{stat.completion_tokens},'
                    f'{stat.cache_hit_tokens},{stat.cache_miss_tokens},{stat.dur_ms}\n'
                )
                session.iteration_stats_file.flush()
            except Exception:
                pass  # best-effort
 
        if tool_calls_this_iter:
            _dbg("OC_CHAT.TOOLS", func,
                 f"dispatching {len(tool_calls_this_iter)} tool(s): "
                 f"{[tc['name'] for tc in tool_calls_this_iter]}", 2)
            # M8.0 WU-9a (Appendix B / F38): per-turn tool-call cap. Calls beyond the
            # budget are DENIED (not executed) but still answered, so every requested
            # tool_call_id gets a result and the API pairing stays well-formed.
            from oc_chat.core.resource_caps import (
                partition_call_budget,
                resolve_max_tool_calls_per_turn,
            )
            _tool_cap = resolve_max_tool_calls_per_turn()  # absolute OOB ceiling; a settings write may lower, never raise (F33)
            # Plan 237 (CAPS-01 P1): the count is now a CHECKPOINT, not a wall.
            # The effective budget is the resolved cap plus one cap-width per
            # extension the operator has already bought this turn, so the next
            # pause lands at 200, 300, … With no extension (the headless path,
            # and every turn before the first checkpoint) this IS ``_tool_cap``
            # and every line below behaves exactly as it did at BASE.
            _effective_cap = _tool_cap * (1 + acc.cap_extensions)
            if acc.tool_calls + len(tool_calls_this_iter) > _effective_cap:
                from oc_chat.core.session_checkpoints import tool_call_checkpoint
                if tool_call_checkpoint(
                    presenter=presenter,
                    report=_mini_report(session, acc, _turn_t0),
                    effective_cap=_effective_cap,
                    is_interactive=_stdin_is_interactive(),
                    session_id=session.session_id,
                ):
                    acc.cap_extensions += 1
                    _effective_cap = _tool_cap * (1 + acc.cap_extensions)
                    _dbg("OC_CHAT.TOOLS", func,
                         f"per-turn tool-call budget extended to {_effective_cap}", 2)
            _allowed_calls, _denied_calls = partition_call_budget(
                acc.tool_calls, len(tool_calls_this_iter), cap=_effective_cap)
            # Plan 246: completed observations make the NEXT identical proposal
            # occurrence N. Deny it and the remaining batch tail before dispatch.
            from oc_chat.core.stall_detector import stall_sentence
            _stall_denied_indices: set[int] = set()
            _stall_hit = None
            for _stall_idx, _stall_tc in enumerate(tool_calls_this_iter):
                if _stall_idx >= _allowed_calls:
                    break
                _stall = stall_detector.check_proposal(
                    _stall_tc["name"], _safe_tool_args(_stall_tc),
                )
                if _stall is not None:
                    _stall_denied_indices.update(
                        range(_stall_idx, _allowed_calls),
                    )
                    _stall_hit = (
                        _stall_tc,
                        _stall,
                        stall_sentence(_stall),
                    )
                    break
            if _denied_calls:
                _dbg("OC_CHAT.TOOLS", func,
                     f"per-turn tool-call cap: {_denied_calls} call(s) denied "
                     f"(budget {_effective_cap}/turn)", 1, override=True)
            acc.tool_calls += len(tool_calls_this_iter)
            # Plan 237: per-tool counts for the checkpoint mini report. Counted
            # over the PROPOSED calls, the same population ``acc.tool_calls``
            # counts, so the report's "N tool calls" and its tool breakdown can
            # never tell two different stories.
            for _tc_named in tool_calls_this_iter:
                _name = _tc_named.get("name", "?")
                acc.tool_name_counts[_name] = acc.tool_name_counts.get(_name, 0) + 1
 
            # Plan 173 item 3 (F22): warn the MODEL as it approaches the per-turn
            # tool-call cap so it can checkpoint instead of being truncated silently.
            # One-shot per turn, fired at >=80% of the budget but before the cap
            # (the at-cap case is already answered by the per-call denial result
            # from _dispatch_or_cap, which tells the model the call did not run).
            _cap_warn_at = max(1, (_tool_cap * 4) // 5)  # 80% of the budget
            if (not acc.cap_warning_emitted
                    and acc.tool_calls >= _cap_warn_at
                    and acc.tool_calls < _tool_cap):
                acc.cap_warning_emitted = True
                _calls_left = _tool_cap - acc.tool_calls
                session.messages.append({
                    "role": "user",
                    "content": (
                        f"[system] Approaching the per-turn tool-call limit: "
                        f"{acc.tool_calls}/{_tool_cap} calls used this turn, {_calls_left} "
                        f"remain. Checkpoint your progress and make only the essential "
                        f"remaining calls; work past the limit is deferred to a new turn."
                    ),
                })
                session.dirty = True
                _dbg("OC_CHAT.TOOLS", func,
                     f"cap-approach warning injected at {acc.tool_calls}/{_tool_cap}", 2)
 
            # Append the assistant message with tool_calls so history is well-formed.
            # reasoning_content must be included when thinking mode is on — the API
            # rejects subsequent turns if it is absent from the history.
            assistant_msg: dict = {
                "role":    "assistant",
                "content": full_content or None,
                "tool_calls": [
                    {
                        "id":   tc["id"],
                        "type": "function",
                        "function": {"name": tc["name"], "arguments": tc["arguments"]},
                    }
                    for tc in tool_calls_this_iter
                ],
            }
            if full_reasoning:
                assistant_msg["reasoning_content"] = full_reasoning
            session.add_assistant_with_tool_calls(assistant_msg)
 
            # M2.3: proposed events before any presenter side-effect (ADR-5).
            for _tc in tool_calls_this_iter:
                default_bus().publish(make_tool_call_proposed(
                    **_ev(), tool_name=_tc["name"], requested_args=_safe_tool_args(_tc)))
 
            # M8.1 WU-0a (F7/F31): the batch barrier — classify every proposed call
            # and apply the per-turn cap BEFORE any tool_call_started, so a cap-denied
            # call can no longer announce itself as started.
            _decision_ids = _publish_pre_dispatch_decisions(
                tool_calls_this_iter,
                _allowed_calls,
                _ev,
                stall_denied=_stall_denied_indices,
            )
 
            # Announce every tool call before executing. Plan 147 D4: the
            # per-tool show_tool_progress() call that used to follow here is
            # removed — it was this method's only production call site, and
            # the spinner_start bracket a few lines below carries the same
            # (now dispatch-chrome-rendering) information; show_tool_progress
            # itself is unchanged as standalone public presenter API.
            for tc in tool_calls_this_iter:
                presenter.show_tool_call(tc["name"], tc["arguments"])
 
            # M2.3: all started events before dispatch (and before any
            # executor.submit) so event order matches call order (ADR-5).
            # M8.1 WU-0a: published for ALLOWED calls only, each carrying the
            # permission_decision_id that allowed it — the decision↔call linkage
            # plan 190 documented as underivable.
            for _idx, _tc in enumerate(tool_calls_this_iter):
                if _idx >= _allowed_calls or _idx in _stall_denied_indices:
                    continue  # cap-denied above: decided, never started
                default_bus().publish(link_started_to_decision(
                    make_tool_call_started(
                        **_ev(), tool_call_id=_tc["id"], tool_name=_tc["name"]),
                    _decision_ids[_idx]))
 
            # Execute tools — parallel when there are multiple calls.
            # Phase H.4a-ii: bracket the dispatch with a presenter spinner so
            # slow tools (run_command, fetch_url) don't leave the terminal
            # silent. The spinner is a no-op on plain/non-TTY presenters.
            from oc_chat.core.resource_caps import clamp_workers
 
            def _dispatch_or_cap(tc: dict, idx: int) -> "tuple[dict, dict]":
                # WU-9a: deny calls beyond the per-turn budget without executing them.
                if idx >= _allowed_calls:
                    return tc, {"error": (
                        f"Per-turn tool-call cap reached ({_effective_cap} "
                        f"calls/turn); this call was not executed. Continue in a new turn."
                    )}
                if idx in _stall_denied_indices:
                    return tc, {"error": _stall_hit[2]}
                return _dispatch_tool(tc, tool_registry)
 
            # Plan 279: the same `_run_bracketed` the confirmation resolvers use.
            # The labels and therefore the presenter call sequence are unchanged;
            # what is new is that TOOL_ACTIVITY is open for the duration, so the
            # bar can show the elapsed time and the last line a child printed.
            # On the multi-tool path the record carries whichever child wrote
            # last, which is the honest answer to "what is happening right now".
            if len(tool_calls_this_iter) > 1:
                _spinner_label = f"running {len(tool_calls_this_iter)} tools…"
 
                def _dispatch_all() -> "list[tuple[dict, dict]]":
                    with ThreadPoolExecutor(max_workers=clamp_workers(len(tool_calls_this_iter))) as executor:
                        futures  = [executor.submit(_dispatch_or_cap, tc, i) for i, tc in enumerate(tool_calls_this_iter)]
                        return [f.result() for f in futures]  # preserves submission order
 
                ordered = _run_bracketed(presenter, _spinner_label, _dispatch_all)
            else:
                _spinner_label = f"running {tool_calls_this_iter[0]['name']}…"
                ordered = _run_bracketed(
                    presenter, _spinner_label,
                    lambda: [_dispatch_or_cap(tool_calls_this_iter[0], 0)])
 
            _observed_indices = [
                idx for idx in range(len(ordered))
                if idx < _allowed_calls and idx not in _stall_denied_indices
            ]
            _last_observed_idx = (
                _observed_indices[-1] if _observed_indices else None
            )
            for _result_idx, (tc, result) in enumerate(ordered):
                # M8.0 confirmation relays (single-threaded, with stdin/presenter):
                # F67 external-content gate (3-choice) and the M8.0-C Tier-2 gate
                # (y/N). Both re-dispatch on approval, fail closed headless/declined,
                # and are identity otherwise.
                if isinstance(result, dict) and result.get("requires_confirmation"):
                    _gate = result.get("gate")
                    if _gate == "external_content":
                        result = _resolve_external_content_confirmation(tc, result, tool_registry, presenter)
                    elif _gate == "protected_write":
                        result = _resolve_protected_write_confirmation(tc, result, tool_registry, presenter)
                    else:
                        result = _resolve_tier2_confirmation(tc, result, tool_registry, presenter)
                    # Plan 173 item 1 (F11 / retired plan-160): emit the audit
                    # permission_decision event at every Tier-2 / protected-write /
                    # external-content confirm point (the defined-but-never-emitted M2
                    # kind). decided_by="human" when an interactive TTY answered;
                    # "policy" when a headless run fails closed. decision="deny" iff the
                    # relay blocked (a residual error / still-requires-confirmation).
                    _pd_blocked = isinstance(result, dict) and (
                        result.get("requires_confirmation") or result.get("error") is not None)
                    _pd_by = "human" if _stdin_is_interactive() else "policy"
                    # Plan 235 P7: name HOW an allowed relay was authorised, inside
                    # the closed AuthorizationMode vocabulary. A run satisfied by a
                    # 235 session grant is a "human_confirm" too — the human DID
                    # decide, once, for the directory. It is deliberately NOT
                    # "grant": that value denotes the M8.1 signed one-use grant and
                    # the validator requires the grant fingerprint fields with it, so
                    # using it here would misattest. The Literal cannot currently
                    # distinguish a session-grant run from a keystroke; a
                    # "session_grant" member is a reported follow-up, not this plan.
                    # Denials keep the pre-235 three-field shape exactly.
                    default_bus().publish(make_permission_decision(
                        **_ev(),
                        permission_decision_id=new_event_id(),
                        decision="deny" if _pd_blocked else "allow",
                        decided_by=_pd_by,
                        authorization_mode=(
                            "human_confirm" if (not _pd_blocked and _pd_by == "human") else None),
                    ))
                _err = result.get("error") if isinstance(result, dict) else None
                # Plan 173 item 1 (F27): a non-zero returncode is a failure too, so a
                # gate-refused oct_git_commit (returncode != 0, no "error" key) is no
                # longer indistinguishable in the event stream from one that landed.
                _rc = result.get("returncode") if isinstance(result, dict) else None
                _failed = bool(_err) or (isinstance(_rc, int) and _rc != 0)
                _status = ("ERROR: " + _err) if _err else (f"rc={_rc}" if _failed else "OK")
                _dbg("OC_CHAT.TOOLS", func, f"tool={tc['name']} {_status}", 1 if _failed else 3)
                # M2.3: completion event before presenter side-effect (ADR-5); outcome
                # was the payload's only failure signal — returncode-aware (F27) and,
                # since plan 203 (audit G10), accompanied by WHY it failed.
                _error_summary = _tool_error_summary(_err, _rc)
                default_bus().publish(make_tool_call_completed(
                    **_ev(), tool_call_id=tc["id"], outcome="error" if _failed else "ok",
                    error_summary=_error_summary))
                if _result_idx in _observed_indices:
                    _capability = tool_capability(tc["name"])
                    stall_detector.observe(
                        tool=tc["name"],
                        arguments=_safe_tool_args(tc),
                        outcome="error" if _failed else "ok",
                        error_summary=_error_summary,
                        is_write=_capability not in {"read", "list"},
                        iteration=_iteration,
                        iteration_complete=_result_idx == _last_observed_idx,
                    )
                presenter.show_tool_result(tc["name"], result)
 
                # ── Track file changes for /goal changelog updates ─────────
                if ctx is not None and ctx.in_goal and tc["name"] == "write_file":
                    try:
                        args = json.loads(tc["arguments"] or "{}")
                        path = args.get("path", "")
                        if path and path not in ctx.goal_files_changed:
                            ctx.goal_files_changed.append(path)
                    except (json.JSONDecodeError, KeyError):
                        pass
 
                # ── Track file changes for /plan turns (Phase J) ──────────
                if ctx is not None and ctx.in_plan and tc["name"] == "write_file":
                    try:
                        args = json.loads(tc["arguments"] or "{}")
                        path = args.get("path", "")
                        if path and path not in ctx.plan_files_changed:
                            ctx.plan_files_changed.append(path)
                    except (json.JSONDecodeError, KeyError):
                        pass
 
                # ── Plan 237: files written this turn, for the checkpoint mini
                # report. Ungated (unlike the plan-137 block below, which needs
                # the verification flag): the operator answering "continue?"
                # must see what the turn has already changed on disk whether or
                # not verification is configured. Successful writes only.
                if not _failed and tc["name"] == "write_file":
                    _written = _safe_tool_args(tc).get("path", "")
                    if _written and _written not in acc.files_written:
                        acc.files_written.append(_written)
 
                # ── Plan 137: track changed files from SUCCESSFUL mutating results
                # (cumulative, turn-scoped; works regardless of ctx — incl. headless).
                # Gated on the verification flag so the off path is byte-identical.
                if _verif_enabled and tc["name"] in _vcfg.mutating_tools:
                    from oc_chat.core.verification import changed_paths_from_result
                    for _p in changed_paths_from_result(tc["name"], _safe_tool_args(tc), result):
                        acc.changed_files.add(_p)
 
                tool_result_msg = {
                    "role":         "tool",
                    "tool_call_id": tc["id"],
                    "content":      json.dumps(result, ensure_ascii=False),
                }
                session.add_tool_result(tool_result_msg)
 
            if _stall_hit is not None:
                from oc_chat.core.session_checkpoints import stall_checkpoint
                _stall_tc, _stall, _stall_reason = _stall_hit
                if stall_checkpoint(
                    presenter=presenter,
                    report=_mini_report(session, acc, _turn_t0),
                    sentence=_stall_reason,
                    is_interactive=_stdin_is_interactive(),
                    session_id=session.session_id,
                ):
                    stall_detector.reset_shape(
                        _stall_tc["name"], _safe_tool_args(_stall_tc),
                    )
                else:
                    default_bus().publish(make_error(
                        **_ev(), error_summary=_stall_reason,
                    ))
                    presenter.show_error(_stall_reason)
                    break
 
            # ── Plan 137: post-mutating-dispatch verification hook (additive) ──
            # After all tool results are appended, before the loop continues. Off ⇒
            # never entered (byte-identical). On a successful mutating iteration it
            # runs the sensor on the CUMULATIVE changed files and appends one
            # [verification] observation on new findings. The existing continue-to-
            # next-iteration is unchanged; the final-answer break is never touched.
            if _verif_enabled and acc.changed_files:
                _maybe_run_verification(session, acc, _vcfg, presenter, _verif_baseline)
 
        else:
            # No tool calls — this is the final answer
            _dbg("OC_CHAT.MAIN", func, "final answer received", 2)
            if full_content:
                msg_entry = {
                    "role": "assistant",
                    "content": full_content,
                }
                if full_reasoning:
                    msg_entry["reasoning_content"] = full_reasoning
                session.messages.append(msg_entry)
                session.dirty = True
                session.stats.add_message()
                # M2.3: final answer after it is durable in the session (ADR-5).
                default_bus().publish(make_assistant_message(
                    **_ev(), text=full_content, is_final=True))
                # The answer is complete prose as far as the loop is concerned,
                # but the provider says it ran out of budget mid-sentence. Say
                # so — this is the plain truncation an operator hits first.
                if finish_reason == "length":
                    presenter.show_info(
                        TRUNCATED_ANSWER_NOTICE.format(cap=route.max_tokens))
            elif full_reasoning:
                # Reasoning-only completion. The model spent its whole output
                # budget on chain-of-thought and stopped before the first
                # content token — the user watched it stream, so the work is
                # real and visible. Persist it: the empty branch below pops the
                # user message, which for this shape erased the prompt AND the
                # reasoning, leaving session.messages byte-identical to before
                # the turn (the "it forgot everything" report). Mirrors the
                # tool-call branch above, which already keeps reasoning_content
                # alongside absent content.
                msg_entry = {
                    "role": "assistant",
                    "content": REASONING_ONLY_CONTENT,
                    "reasoning_content": full_reasoning,
                }
                session.messages.append(msg_entry)
                session.dirty = True
                session.stats.add_message()
                # M2.3: durable before the presenter side-effect (ADR-5).
                default_bus().publish(make_assistant_message(
                    **_ev(), text=REASONING_ONLY_CONTENT, is_final=True))
                _dbg("OC_CHAT.MAIN", func,
                     f"reasoning-only completion: {len(full_reasoning)} chars kept, "
                     f"no content token received (finish_reason={finish_reason})",
                     1, override=True)
                _cause = (CAUSE_LENGTH.format(cap=route.max_tokens)
                          if finish_reason == "length" else CAUSE_UNKNOWN)
                presenter.show_info(REASONING_ONLY_NOTICE.format(
                    n=len(full_reasoning), cause=_cause))
            else:
                if session.messages and session.messages[-1]["role"] == "user":
                    session.messages.pop()
                # M2.3: empty completion is a terminal error for replay (ADR-4).
                default_bus().publish(make_error(
                    **_ev(), error_summary="empty_completion: no response content received"))
                presenter.show_info("[No response content received]")
            break
    else:
        _dbg("OC_CHAT.MAIN", func,
             f"hit ReAct cap ({react_cap} iterations) without final answer", 1, override=True)
        # Plan 137 §3.2: a mutation on the final react_cap iteration is verified +
        # recorded but the model gets no same-turn chance to reconcile — flag it
        # here (gated; off ⇒ the original message, byte-identical).
        _cap_msg = (
            f"[Reached the maximum of {react_cap} ReAct iterations without a final reply — "
            "try narrowing the task, /session clear, /session new, or starting a new session.]"
        )
        if _verif_enabled and acc.pending_findings is not None:
            _cap_msg += (
                " Note: verification found unreconciled issues in the files changed "
                "this turn — review them before continuing."
            )
        presenter.show_info(_cap_msg)
 
    session.last_turn_max_prompt_tokens = acc.max_prompt_in_turn
 
    n_user_msgs = sum(1 for m in session.messages if m.get("role") == "user")
    if n_user_msgs >= 12 and not session.rollover_hint_shown:
        session.rollover_hint_shown = True
        presenter.show_info(
            "[Tip] This conversation has many user turns. For a shorter prefix and better cache "
            "efficiency, consider `/session save` then `/session new` (or export) and continue in a fresh session."
        )
 
    presenter.end_turn()
    # ── Turn summary (aggregated across all iterations) ────────────────────
    _dbg("OC_CHAT.MAIN", func,
         f"turn END iters={acc.iterations} tools={acc.tool_calls} "
         f"prompt={acc.prompt_tokens} completion={acc.completion_tokens} "
         f"cached={acc.prompt_cached}", 2)
    if acc.iterations > 0:
        pc = acc.prompt_cached if acc.prompt_cached > 0 else None
        has_cache = acc.cache_hit_tokens + acc.cache_miss_tokens > 0
        pct = cache_hit_ratio_pct(acc.cache_hit_tokens, acc.cache_miss_tokens) if has_cache else None
        saved = acc.cache_hit_tokens if has_cache else None
        # Plan 134 C2: a REAL context-size gauge — the turn's MAIN-LOOP prompt
        # high-water mark over the model's context window. Uses acc.max_prompt_in_turn
        # only (NOT summed parent+sub-loop usage — B2 separation), so a sub-loop turn
        # does not inflate it. The caller computes the %; presenters just render it.
        ctx_window = get_max_context_tokens(session.model)
        context_used_pct = None
        if ctx_window > 0 and acc.max_prompt_in_turn > 0:
            context_used_pct = max(0.0, min(100.0,
                acc.max_prompt_in_turn / ctx_window * 100.0))
        presenter.show_turn_summary(
            prompt=acc.prompt_tokens,
            completion=acc.completion_tokens,
            total=acc.total_tokens,
            iterations=acc.iterations,
            tool_calls=acc.tool_calls,
            prompt_cached=pc,
            cache_hit_pct=pct,
            cache_saved_tokens=saved,
            performance_stats=acc.perf_stats,
            model=session.model,
            context_used_pct=context_used_pct,
            duration_ms=int((time.monotonic() - _turn_t0) * 1000),
            # Plan 188 T4: None when any sample was unpriceable — presenters
            # then render NO cost rather than a partial (i.e. wrong) total.
            turn_cost=acc.cost.result(),
        )
        if acc.cache_hit_tokens + acc.cache_miss_tokens > 0:
            snap = {
                "hit": acc.cache_hit_tokens,
                "miss": acc.cache_miss_tokens,
                "prompt": acc.prompt_tokens,
            }
            session.cache_turn_snapshots.append(snap)
            session.cache_turn_snapshots = session.cache_turn_snapshots[-30:]
            session.dirty = True
 
    # Mark turn finished in thread-safe stats
    if turn_stats is not None:
        session_cache_pct = cache_hit_ratio_pct(
            session.stats.prompt_cache_hit_tokens,
            session.stats.prompt_cache_miss_tokens,
        )
        turn_stats.finish_turn(
            session.stats.total_tokens,
            iterations=acc.iterations,
            tool_calls=acc.tool_calls,
            session_cache_pct=session_cache_pct,
            session_cache_saved=session.stats.prompt_cache_hit_tokens,
        )
 
    # ── Per-turn auto-save ────────────────────────────────────────────────
    # Each completed turn snapshots both chat log and metadata.  The persistence
    # archives the previous version first, so a crash mid-session at most
    # loses the in-flight turn.
    try:
        save_chat_log(session)
        save_session_metadata(session)
    except RuntimeError as exc:
        # save_chat_log raises this when load_failed is set — preserve the
        # original chat log on disk and just save metadata.
        presenter.show_info(f"[auto-save: {exc}]")
        try:
            save_session_metadata(session)
        except Exception as exc2:
            presenter.show_error(f"Auto-save metadata failed: {exc2}")
    except Exception as exc:
        presenter.show_error(f"Auto-save failed: {exc}")