commands.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/commands.py
 
"""
Slash-command dispatch and handlers for the OC Chat REPL.
 
Purpose
-------
Register handlers with the :func:`command` decorator; :func:`dispatch`
routes raw user input to the appropriate handler. Includes commands
for model switching, filesystem inspection, session management,
agentic goals, and help.
 
Responsibilities
----------------
- Maintain the ``COMMANDS`` registry and the ``COMMAND_HELP`` table.
- :func:`dispatch` — route ``/<verb> <args>`` to the right handler,
  returning ``True`` when the input was a slash command (so the REPL
  knows to skip the LLM round-trip).
- :class:`Context` — the mutable per-REPL state passed to every handler
  (session, presenter, model manager, tool registry, in-flight goal).
- Handlers themselves: ``/help``, ``/models``, ``/model``, ``/effort``,
  ``/system``, ``/ls``, ``/read``, ``/context``, ``/session``, ``/ui``,
  ``/budget``, ``/react``, ``/goal``, ``/stop``, etc.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — command-handler entry / exit
    L3 — dispatch decisions and arg parsing
    L4 — registry-table dumps
 
Contracts
---------
- Handlers receive a :class:`Context` and the raw arg string; they
  return ``None`` and mutate ``ctx`` in place.
- :func:`dispatch` returns ``True`` for any input starting with ``/``
  (even unknown verbs — handled with a friendly error), ``False``
  otherwise.
- Unknown verbs do not crash the REPL.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.config (shared config constants and ANSI styles)
- oc_chat.core.session (lazy; ChatSession, CANONICAL_POLICY, build_system_prompt, _auto_name - the fresh session for /session new and clear)
- oc_chat.core.defaults (default - the ui_reasoning_tail default)
- oc_chat.core.session_transition (replace_session, load_session_into, rename_session and the endpoint activation pair - session and endpoint changes through the transition seam)
- oc_chat.core.ask_menu (lazy; present_post_answer_menu - the menu after an /ask answer)
- oc_chat.core.cache_usage (lazy; cache_hit_ratio_pct - the /cache hit ratio)
- oc_chat.core.deployment_shape (lazy; check_deployment_shape - refuses a /sandbox set root that overlaps an importable root)
- oc_chat.core.event_bootstrap (lazy; active_event_dir, _slug - where /packet writes its draft)
- oc_chat.core.external_content_gate (lazy; is_canonical_record - the canonical-record check before a session is activated)
- oc_chat.core.generation_options (lazy; GenerationOptions - the per-session /options and their default)
- oc_chat.core.lane_picker (lazy; run_lane_picker, lane_options_from_registry - the /lane drill-down and list)
- oc_chat.core.lane_switch (lazy; switch_lane, LaneKeyMissingError - switching onto the chosen lane)
- oc_chat.core.persistence (lazy; session save, load, export and listing for /session and the palette, and the latest tool artifact for /open)
- oc_chat.core.plan_inventory (lazy; TERMINAL_STATUSES, _expected_filename - plan state and file name for /plan)
- oc_chat.core.plans (lazy; create, find, pause, continue, complete, abort and list plans for /plan)
- oc_chat.core.provider_oauth (lazy; write_credentials - stores or removes the /login and /logout credential)
- oc_chat.core.provider_plugin (lazy; registry - the lane named by /login and /logout)
- oc_chat.core.resource_caps (lazy; the session cost hard cap and thresholds for /budget cost)
- oc_chat.core.risk_profile (lazy; resolve_risk_profile - the profile a /sandbox set root is checked under)
- oc_chat.core.session_checkpoints (lazy; tool_calls_auto_continue, cost_prompts_suppressed - the checkpoint latches for the status bar and /budget cost)
- oc_chat.core.session_mode (lazy; the /mode confirmation mode and remembered answers)
- oc_chat.core.startup (lazy; refresh_variable_context - refreshes the variable context after /cwd)
- oc_chat.core.step_routing (lazy; resolve_output_tokens - the completion caps /budget prints)
- oc_chat.core.tier2_session_grant (lazy; active_session_grant, revoke_session_grants - the Tier-2 grant for the status bar and /grants)
- oc_chat.core.trace_view (lazy; read_session_events, format_trace, build_packet_draft - /trace and /packet)
- oc_chat.tools._sandbox (lazy; set_session_sandbox_root_override, install_resolved - applies the session sandbox root)
- oc_chat.tools.filesystem (lazy; list_dir, read_file, _sandbox_root_info - /ls, /read, /context and the sandbox display)
- oc_chat.tools.planner (lazy; get_status, stop_plan - /status and /stop)
- prompt_toolkit (lazy; DummyHistory - keeps the pasted login code out of the prompt input history)
- rich (lazy; Console, Markdown - the /last pager and its Markdown rendering)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §1.5 Context — owner of the command-handler
aggregate dataclass.
"""
from __future__ import annotations
 
import uuid
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any, TypeVar
 
from oc_diagnostics import _dbg
 
_F = TypeVar("_F", bound=Callable[..., Any])
 
from oc_chat.core.config import (
    DEFAULT_MAX_REACT_ITERATIONS,
    REASONING_EFFORT_LEVELS,
    TOKEN_PROFILES,
)
from oc_chat.core.defaults import default
from oc_chat.core.session_transition import (
    ChatLogLoadError,
    TransitionRejectedError,
    commit_endpoint_activation,
    load_session_into,
    prepare_endpoint_activation,
    rename_session,
    replace_session,
)
 
# ── Registry ──────────────────────────────────────────────────────────────────
 
COMMANDS: dict = {}
 
# Human-readable usage strings shown by /help  (verb → "usage - description")
COMMAND_HELP: dict[str, str] = {
    "/help":    "/help                    - show this help message",
    "/models":  "/models                  - list available models and their capabilities",
    "/model":   "/model <name>            - switch to any model (keeps conversation)",
    "/use":     "/use <name>              - alias for /model",
    "/login":   "/login <lane>           - sign in using a pasted authorization code",
    "/logout":  "/logout <lane>          - remove the stored OAuth credential",
    "/lane":    "/lane [list|<name>]      - drill down to a lane and model, or switch to a named lane",
    "/pro":     "/pro                     - switch to the active provider's reasoning-tier model",
    "/flash":   "/flash                   - switch to the active provider's cheap-tier model",
    "/effort":  "/effort [low|medium|high]- show or set reasoning effort level",
    "/sandbox": "/sandbox [set <path>|unset] - show or set the session sandbox root",
    "/grants":  "/grants [list|revoke]    - show or revoke the session Tier-2 python grant",
    "/mode":    "/mode [ask|auto|forget]  - confirmation mode for this session; 'forget' drops the remembered answers",
    "/system":  "/system [prompt]         - show or set the system prompt",
    "/ls":      "/ls [-r] [path]          - list files/dirs (add -r for recursive)",
    "/read":    "/read <file>             - print a file's contents",
    "/context": "/context <file>          - inject file content into conversation (works with any model)",
    "/cwd":     "/cwd [<path>]            - re-anchor this session to a new working directory",
    "/open":    "/open last               - print the most recent tool artifact for this session",
    "/last":    "/last [answer|raw|reasoning|tool] - page the latest answer or related output",
    "/session": "/session <subcommand>    - manage sessions (new, list, current, save, load, export, switch, clear)",
    "/ui":      "/ui <subcommand>         - configure the terminal UI (reasoning visibility; theme/stats in later phases)",
    "/palette": "/palette [query]         - fuzzy-search commands and recent sessions (Ctrl+P at the prompt)",
    "/goal":    "/goal <task>             - give the agent an autonomous goal to execute",
    "/status":  "/status                  - show the current plan's step-by-step status",
    "/stop":    "/stop                    - halt the active plan",
    "/cache":   "/cache                   - prompt cache hit/miss (session + recent turns)",
    "/trace":   "/trace [N|all]           - show this session's recorded event stream (default last 200)",
    "/packet":  "/packet                  - write a draft merge-readiness packet (packet-draft/v0) for this session",
    "/budget":  "/budget [cheap|balanced|deep|cost] - token profile for completion caps; 'cost' shows session spend, checkpoints and the hard cap",
    "/react":   "/react [N|reset]         - show or set max ReAct iterations (1–99)",
    "/ask":     "/ask <question>          - one-shot question (no session log, no tool calls)",
    "/exit":    "/exit  (or /quit)        - save conversation log and exit",
}
 
 
def _default_gen_opts():
    from oc_chat.core.generation_options import GenerationOptions
    return GenerationOptions()
 
 
def _latest_message_of_role(messages: list[dict], role: str) -> dict | None:
    """Return the latest message with *role*, or ``None``."""
    func = "_latest_message_of_role"
    for message in reversed(messages):
        if message.get("role") == role:
            _dbg("OC_CHAT.MAIN", func, f"found role={role}", 4)
            return message
    _dbg("OC_CHAT.MAIN", func, f"missing role={role}", 4)
    return None
 
 
def _latest_message_with_field(
    messages: list[dict],
    role: str,
    field_name: str,
) -> dict | None:
    """Return the latest role-matched message with a non-empty *field_name*."""
    func = "_latest_message_with_field"
    for message in reversed(messages):
        if message.get("role") == role and message.get(field_name):
            _dbg("OC_CHAT.MAIN", func, f"found role={role} field={field_name}", 4)
            return message
    _dbg("OC_CHAT.MAIN", func, f"missing role={role} field={field_name}", 4)
    return None
 
 
def _message_content_to_text(content: object) -> str:
    """Convert stored message content into display text."""
    func = "_message_content_to_text"
    if content is None:
        return ""
    if isinstance(content, str):
        return content
    import json
    try:
        return json.dumps(content, ensure_ascii=False, indent=2)
    except TypeError as exc:
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {type(exc).__name__}: {exc}", 3)
        return str(content)
 
 
def _page_last_content(content: str, *, markdown: bool) -> None:
    """Render *content* in a pager when interactive, else print inline."""
    func = "_page_last_content"
    import sys
    from rich.console import Console
    from rich.markdown import Markdown
 
    is_tty = bool(getattr(sys.stdout, "isatty", lambda: False)())
    console = Console(
        force_terminal=is_tty,
        color_system="truecolor" if is_tty else None,
    )
 
    def _print_content() -> None:
        if markdown:
            console.print(Markdown(content))
        else:
            console.print(content, markup=False)
 
    if not is_tty:
        _dbg("OC_CHAT.MAIN", func, "non-tty fallback", 3)
        _print_content()
        return
 
    _dbg("OC_CHAT.MAIN", func, f"pager markdown={markdown}", 3)
    with console.pager(styles=True):
        _print_content()
 
 
def command(*names: str) -> Callable[[_F], _F]:
    """Decorator that registers a handler for one or more command verbs."""
    def deco(fn: _F) -> _F:
        for name in names:
            COMMANDS[name] = fn
        return fn
    return deco
 
 
@dataclass
class Context:
    """Aggregate object passed to every command handler.
 
    Bundles the live session, model manager, tool registry, presenter,
    and API client so commands can interact with the full system.
    """
    session: object       # ChatSession
    model_manager: object # ModelManager
    tool_registry: dict
    presenter: object     # ConsolePresenter
    client: object        # DeepSeekClient
    tool_schemas: list = field(default_factory=list)
 
    # ── /goal tracking ────────────────────────────────────────────────────
    in_goal: bool = False
    """``True`` when the current chat turn was triggered by ``/goal``."""
    goal_description: str = ""
    """The goal text passed to ``/goal``, used as the changelog scope."""
    goal_files_changed: list = field(default_factory=list)
    """File paths (relative strings) written during the current goal turn."""
 
    # ── /plan tracking (Phase J) ───────────────────────────────────────────
    in_plan: bool = False
    """``True`` when the current chat turn was triggered by ``/plan``."""
    plan_description: str = ""
    """The description passed to ``/plan`` (or empty after a ``--continue``)."""
    plan_files_changed: list = field(default_factory=list)
    """File paths (relative strings) written during the current plan turn."""
 
    # ── /options (Phase O4) ───────────────────────────────────────────────
    generation_options: object = field(default_factory=lambda: _default_gen_opts())
    """Per-session generation options (temperature, top_k, etc.)."""
 
    # ── /endpoint (Phase O2) ──────────────────────────────────────────────
    endpoint_registry: object = None
    """Optional ``EndpointRegistry`` bound at startup in main.py. ``None``
    when initialisation failed or no provider endpoints were registered;
    ``/endpoint`` then surfaces a friendly hint."""
 
    # ── /sandbox (M0.4) ──────────────────────────────────────────────
    sandbox_root_override: str | None = None
    """Optional per-session sandbox root set via ``/sandbox set``."""
 
    # ── Iteration stats sink (Plan 152 D4) ─────────────────────────────────
    iteration_stats_sink: object = None
    """Optional process-owned ``IterationStatsSink`` opened at startup in
    main.py when ``--stats-file`` is given. Sessions BORROW its writer at every
    transition commit, so the knobs survive session replacement and the
    finaliser closes exactly one handle instead of leaking one per session."""
 
    # ── Realtime status bar (Plan 147 D1) ───────────────────────────────────
    turn_stats: object = None
    """Optional ``TurnStats`` (``oc_ui.terminal.input``) bound at startup in
    main.py. Read by :func:`refresh_presenter_bindings` to build the
    ``bind_status`` provider closure; ``None`` when no turn/session timing
    is available (e.g. a headless caller that never built one)."""
 
    # ── /lane (plan 217 T4) ────────────────────────────────────────────────
    agent: object = None
    """Optional back-reference to the owning ``Agent``, bound in main.py.
 
    ``Agent.client`` is the object ``run_turn`` hands the provider, so a
    lane switch that updated only ``ctx.client`` would leave the NEXT TURN
    on the old lane while every command reported the new one. ``None`` for
    headless callers that never build an Agent."""
 
    prompt_driver: object = None
    # OCT-LINT: disable=no_hardcoded_secrets reason="Context field docstring; high-entropy English prose, not a credential (operator-ruled 2026-08-31)"
    """Optional live ``PromptDriver``, bound in main.py once the real one
    exists. ``/lane`` reads its picker answers through it so the drill-down
    shares the REPL's input backend; ``None`` falls back to ``input()``."""
 
 
def _metadata_for_activation(ctx: "Context", name: str) -> tuple[dict, str]:
    """Load *name*'s metadata at the LIVE load/switch boundary (plan 152 D9).
 
    This is the ONLY boundary where in-process F67 trust inheritance is
    possible — a fresh process starts with an empty trust set — so the
    ``session_id -> canonical record`` rule is enforced here and startup, resume
    and batch stay untouched.
 
    A record is NON-canonical when any of these holds:
 
    (a) its persisted name disagrees with the store location it was resolved
        from — which is what a metadata file copied into another session's
        directory looks like;
    (b) it presents a ``session_id`` this process already binds to a DIFFERENT
        record;
    (c) it carries the rename ``superseded_by`` mark, making it non-canonical
        by construction rather than by load order.
 
    For a non-canonical record we mint a FRESH canonical id before hydration, so
    ``apply_session_metadata_fields`` keeps that id and resets trust before the
    identity is exposed. Minting explicitly matters here: load/switch hydrate
    IN PLACE, so simply dropping the persisted key would leave the session
    carrying the PREVIOUS session's id — and its trust.
 
    Residual, stated honestly: a cross-run whole-directory clone with an edited
    name is indistinguishable without a global uniqueness index. That closes
    when the sessions store makes ``session_id`` a primary key.
    """
    func = "_metadata_for_activation"
    from oc_chat.core.external_content_gate import is_canonical_record
    from oc_chat.core.persistence import load_session_metadata_located
 
    loaded = load_session_metadata_located(name)
    meta = dict(loaded.metadata)
    locator = loaded.canonical_record_locator
    sid = meta.get("session_id")
 
    persisted_name = meta.get("session_name")
    reason = ""
    if isinstance(persisted_name, str) and persisted_name and persisted_name != name:
        reason = "record name disagrees with the store location it was resolved from"
    elif meta.get("superseded_by"):
        reason = "record was superseded by a rename"
    elif not is_canonical_record(sid, locator):
        reason = "session_id is already owned by a different record in this process"
 
    if reason:
        meta.pop("session_id", None)
        ctx.session.session_id = str(uuid.uuid4())
        _dbg("OC_CHAT.MAIN", func,
             f"SECURITY_ERROR: non-canonical record for {name!r} ({reason}); "
             f"minted a fresh identity and trust will reset before exposure", 2, override=True)
    return meta, locator
 
 
def refresh_presenter_bindings(ctx: "Context") -> None:
    """Re-bind the presenter to the current session + status snapshot (Plan 147 D1).
 
    Calls ``presenter.bind_session(ctx.session)`` and
    ``presenter.bind_status(provider)`` together. Route EVERY session-
    mutation path through this single helper — startup, the pre-turn
    rebind, and every ``/session`` subcommand that replaces or reloads
    ``ctx.session`` — so presenter chrome (the enhanced header, the D3
    status bar) never shows a stale session after ``/session new`` /
    ``load`` / ``switch`` / ``clear``.
 
    The ``bind_status`` provider is a fresh closure over *ctx* built on
    every call — it reads ``ctx.session`` at RENDER time (late-bound),
    never captures the ``ChatSession`` object itself, so a later
    ``/session new`` (which reassigns ``ctx.session``) is picked up
    without re-binding. ``TurnStats.rebind_session`` resets the session-
    elapsed clock only when the session identity actually changed —
    calling this helper on every turn is safe and idempotent.
    """
    presenter = ctx.presenter
    if hasattr(presenter, "bind_session"):
        presenter.bind_session(ctx.session)
    if ctx.turn_stats is not None:
        ctx.turn_stats.rebind_session(ctx.session.session_id)
 
        def _status_provider():
            session = ctx.session
            turn_stats = ctx.turn_stats
            if session is None or turn_stats is None:
                return None
            session_tools = session.stats.tool_call_count if hasattr(session, "stats") else 0
            # Plan 235 (SHELL-02): read at RENDER time like every other field, so
            # taking or revoking a grant shows up on the very next repaint without
            # a second wiring path to keep in sync.
            from oc_chat.core.tier2_session_grant import active_session_grant
            granted = active_session_grant()
            # Plan 237 (CAPS-01): money and the checkpoint latch, read at RENDER
            # time for the same reason the grant is — the very next repaint after
            # a priced sample or an "a" keystroke shows it, with no second wiring
            # path to keep in sync. ``None`` cost renders nothing at all.
            from oc_chat.core.session_checkpoints import tool_calls_auto_continue
            _cost = getattr(session, "stats", None)
            _cost = getattr(_cost, "cost", None)
            _priced = getattr(_cost, "priced_usd", 0.0) if _cost is not None else 0.0
            _unpriced = getattr(_cost, "unpriced_samples", 0) if _cost is not None else 0
            return turn_stats.build_status(
                model=session.model,
                session_name=session.name,
                effort=getattr(session, "reasoning_effort", "high"),
                session_tools=session_tools,
                session_grant=granted.name if granted is not None else None,
                session_cost_usd=(_priced if (_priced or _unpriced) else None),
                unpriced_samples=_unpriced,
                checkpoint_auto=tool_calls_auto_continue(),
            )
 
        presenter.bind_status(_status_provider)
 
 
#: The four outcomes ``dispatch`` can actually observe (plan 152 ADR-7).
NOT_HANDLED = "NOT_HANDLED"
HANDLED_NO_TURN = "HANDLED_NO_TURN"
HANDLED_TURN_QUEUED = "HANDLED_TURN_QUEUED"
INTERNAL_FAILURE = "INTERNAL_FAILURE"
 
#: What a failing handler left behind.
EFFECT_NONE = "NONE"
EFFECT_COMMITTED = "COMMITTED"
EFFECT_UNKNOWN = "UNKNOWN"
 
#: Handlers that mutate state AND queue a turn, so they carry their own
#: catch/classify/compensate boundary and are covered by dispatch's backstop.
#: Every OTHER handler keeps today's propagation semantics exactly — several of
#: them mutate too, and a blanket catch would re-prompt into an unclassified
#: live hybrid, which is worse than today's exit (plan 152 R11-01).
INSTRUMENTED_VERBS = frozenset({"/goal", "/plan"})
 
 
@dataclass(frozen=True)
class DispatchResult:
    """What ``dispatch`` observed, as a typed value rather than a bool.
 
    ``kind`` is one of the four constants above. ``HANDLED_TURN_QUEUED`` is
    produced ONLY by a handler explicitly returning a ``queued_turn_token`` — it
    is NEVER derived from a message-count delta. That heuristic looked
    equivalent but was not: an in-place ``/session load`` whose restored log ends
    in a user message would have fired an unrequested LLM turn, and ``/context``
    (a fourth appender) would have fired one on its own instead of letting its
    injected content ride into the next user-initiated turn.
 
    ``effect_disposition`` says what a failure left behind: ``NONE`` (the handler
    guarantees no mutation ran), ``COMMITTED`` (honour it — never compensate it
    away), or ``UNKNOWN`` (fail closed and report a reconciliation path).
 
    Truthiness matches the old ``bool`` return, so existing ``if handled:``
    call sites and tests keep working.
    """
 
    kind: str
    effect_disposition: str = EFFECT_NONE
    queued_turn_token: object = None
    warning: str = ""
 
    def __bool__(self) -> bool:
        """``True`` when the input was consumed as a command."""
        return self.kind != NOT_HANDLED
 
 
def dispatch(ctx: "Context", raw_input: str) -> DispatchResult:
    """Dispatch a ``/command``. Returns a typed :class:`DispatchResult`.
 
    The backstop below is scoped to :data:`INSTRUMENTED_VERBS` ONLY. It is a
    LAST-RESORT net for a handler that raised past its own handler-local
    containment: it classifies nothing and fails closed with
    ``effect_disposition=UNKNOWN``. An exception escaping any OTHER handler
    propagates exactly as it does today — no new containment, no behaviour
    change. ``SystemExit`` is never caught.
    """
    func = "dispatch"
    parts = raw_input.split()
    if not parts:
        return DispatchResult(kind=NOT_HANDLED)
    verb = parts[0].lower()
    _dbg("OC_CHAT.MAIN", func, f"verb={verb}", 2)
    handler = COMMANDS.get(verb)
    if handler:
        if verb in INSTRUMENTED_VERBS:
            try:
                outcome = handler(ctx, raw_input)
            except Exception as exc:
                _dbg("OC_CHAT.MAIN", func,
                     f"SYSTEM_ERROR: {verb} raised past its own containment: "
                     f"{type(exc).__name__}: {exc}", 1, override=True)
                ctx.presenter.show_error(
                    f"{verb} failed unexpectedly: {type(exc).__name__}: {exc}. "
                    f"Its effects are INDETERMINATE — check /plan --list and the "
                    f"session state before retrying."
                )
                return DispatchResult(
                    kind=INTERNAL_FAILURE, effect_disposition=EFFECT_UNKNOWN,
                )
            if isinstance(outcome, DispatchResult):
                return outcome
            return DispatchResult(kind=HANDLED_NO_TURN)
        handler(ctx, raw_input)
        return DispatchResult(kind=HANDLED_NO_TURN)
    if verb.startswith("/"):
        ctx.presenter.show_error(
            f"Unknown command: {verb}. Type /help for available commands."
        )
        return DispatchResult(kind=HANDLED_NO_TURN)
    return DispatchResult(kind=NOT_HANDLED)
 
 
# ── Model / session commands ──────────────────────────────────────────────────
 
@command("/models")
def cmd_models(ctx: "Context", _: str) -> None:
    """List all known models and their capabilities."""
    added = ctx.model_manager.fetch_live_if_needed(ctx.client)
    if added > 0:
        ctx.presenter.show_info(f"Loaded {added} new model(s) from API.")
    models = ctx.model_manager.known_models()
    ctx.presenter.show_available_models(models)
 
 
def _warn_if_no_tools(ctx, model_name: str) -> None:
    """Print a dimmed warning if *model_name* does not support tool use."""
    caps = ctx.model_manager.capabilities(model_name)
    if not caps.get("supports_tools", True):
        from oc_chat.core.config import STYLE_DIM, STYLE_RESET
        print(f"\n{STYLE_DIM}[Note: {model_name} does not support tool use — agent tools are disabled for this model]{STYLE_RESET}")
 
 
def _refresh_presenter_for_model(ctx) -> None:
    """Plan 134 N5: re-bind the session so a context-aware presenter (enhanced)
    refreshes its ctx-gauge window for the NEW model. bind_session is idempotent
    and safe; presenters without it (plain console) are skipped.
    """
    bind = getattr(ctx.presenter, "bind_session", None)
    if callable(bind):
        bind(ctx.session)
 
 
def _switch_to_tier(ctx, tier_name: str) -> None:
    """Switch the session model to whatever the active provider maps the
    given semantic tier to. If the provider has no model for that tier,
    surface a friendly message and leave the session model unchanged.
    """
    target = ctx.model_manager.provider.tier(tier_name)
    if not target:
        ctx.presenter.show_info(
            f"Active provider has no '{tier_name}' tier — command unavailable."
        )
        return
    ctx.session.set_model(target)  # plan 135 §3.5: refreshes the env block in place
    ctx.model_manager.ensure(target)
    _refresh_presenter_for_model(ctx)
    ctx.presenter.show_info(f"Switched to: {target} (conversation preserved)")
    _warn_if_no_tools(ctx, target)
 
 
@command("/pro")
def cmd_pro(ctx: "Context", _: str) -> None:
    """Switch to the active provider's reasoning-tier model."""
    _switch_to_tier(ctx, "reasoning")
 
 
@command("/flash")
def cmd_flash(ctx: "Context", _: str) -> None:
    """Switch to the active provider's cheap-tier model."""
    _switch_to_tier(ctx, "cheap")
 
 
@command("/model", "/use")
def cmd_switch_model(ctx: "Context", raw: str) -> None:
    """Switch to a model specified by name."""
    parts = raw.split(maxsplit=1)
    if len(parts) < 2:
        ctx.presenter.show_info("Usage: /model <model_name>")
        return
    name = parts[1].strip()
    if name not in ctx.model_manager.known_models():
        # Try fetching live models in case this is a new API model
        ctx.model_manager.fetch_live_if_needed(ctx.client)
        if name not in ctx.model_manager.known_models():
            ctx.presenter.show_info(f"Unknown model: {name}. Use /models to list available.")
            return
    ctx.session.set_model(name)  # plan 135 §3.5: sets dirty + refreshes the env block
    _refresh_presenter_for_model(ctx)
    ctx.presenter.show_info(f"Switched to: {name} (conversation preserved)")
    _warn_if_no_tools(ctx, name)
 
 
def _lane_options_for(ctx: "Context"):
    """Level-one rows for the live registry (plan 217 T4)."""
    from oc_chat.core.lane_picker import lane_options_from_registry
 
    return lane_options_from_registry()
 
 
def _apply_lane_selection(ctx: "Context", lane_name: str, model_id: str = "") -> bool:
    """Switch *ctx* onto *lane_name*; report and return ``False`` on refusal.
 
    Every failure mode surfaces as a message rather than a traceback: an
    unset key names the variable and points at the wizard (P3), and an
    unknown lane lists what is available.
    """
    func = "_apply_lane_selection"
    from oc_chat.core.lane_switch import LaneKeyMissingError, switch_lane
 
    try:
        prepared = switch_lane(ctx, lane_name, model_id)
    except LaneKeyMissingError as exc:
        # SECURITY_ERROR-adjacent (S6 L1 "refuse, never route around"), but the
        # operator-facing fix is mundane, so the message carries it.
        _dbg("OC_CHAT.MAIN", func, f"refused lane {lane_name!r}: no key", 2)
        ctx.presenter.show_error(f"  {exc}")
        return False
    except KeyError as exc:
        # VALIDATION_ERROR (S6 L3) — a name that is not a registered lane.
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {exc}", 3)
        ctx.presenter.show_error(f"  {exc}")
        return False
    except (PermissionError, OSError, ValueError) as exc:
        # EXTERNAL_ERROR (S6 L2) — the lane's client refused to build.
        _dbg("OC_CHAT.MAIN", func,
             f"EXTERNAL_ERROR: {type(exc).__name__}: {exc}", 2, override=True)
        ctx.presenter.show_error(f"  Could not switch to {lane_name!r}: {exc}")
        return False
 
    _refresh_presenter_for_model(ctx)
    ctx.presenter.show_info(
        f"Switched to lane {prepared.lane_name} "
        f"(model: {prepared.model_id}, conversation preserved)"
    )
    _warn_if_no_tools(ctx, prepared.model_id)
    return True
 
 
@command("/lane")
def cmd_lane(ctx: "Context", raw: str) -> None:
    """Choose a provider lane and model, or switch to a named lane.
 
    /lane            — open the drill-down picker (lane, free/paid, family, model)
    /lane list       — list every registered lane with its key state
    /lane <name>     — switch straight to a lane, on its default model
 
    The picker mutates nothing until its final confirmation, and stepping
    back off the first level leaves the session exactly as it was
    (plan 217 P4).
    """
    from oc_chat.core.lane_picker import run_lane_picker
 
    parts = raw.split(maxsplit=1)
    argument = parts[1].strip() if len(parts) > 1 else ""
 
    if argument.lower() in ("list", "ls"):
        lines = ["Provider lanes:"]
        for option in _lane_options_for(ctx):
            state = "ready" if option.has_key else "no key"
            key_note = option.api_key_env_var or "(no key needed)"
            lines.append(
                f"  {option.name:<18}{option.display_label:<30}"
                f"{option.badge:<5} {len(option.models):>3} curated  "
                f"{state} [{key_note}]"
            )
        ctx.presenter.show_info("\n".join(lines))
        return
 
    if argument:
        _apply_lane_selection(ctx, argument)
        return
 
    selection = run_lane_picker(
        ctx.presenter, prompt_driver=getattr(ctx, "prompt_driver", None)
    )
    if selection is None:
        ctx.presenter.show_info("Lane unchanged.")
        return
    _apply_lane_selection(ctx, selection.lane_name, selection.model_id)
 
 
class _OAuthCallbacks:
    """Render provider callbacks through the existing terminal seams."""
 
    def __init__(self, ctx: "Context") -> None:
        self.ctx = ctx
 
    def on_auth(self, url: str) -> None:
        self.ctx.presenter.show_info(url)
 
    def on_prompt(self, prompt: str) -> str:
        driver = getattr(self.ctx, 'prompt_driver', None)
        if driver is None or getattr(driver, 'using_prompt_toolkit', None) is False:
            from getpass import getpass
            value = getpass(prompt)
        elif getattr(driver, 'using_prompt_toolkit', None) is True:
            from prompt_toolkit.history import DummyHistory
            terminal = driver._pt
            previous = (terminal.history, terminal.default_buffer.history,
                        terminal.is_password, terminal.completer, terminal.auto_suggest)
            terminal.history = terminal.default_buffer.history = DummyHistory()
            terminal.is_password = True
            terminal.completer = terminal.auto_suggest = None
            try:
                value = driver.read_user_input(None, prompt_text=prompt)
            finally:
                # Clear the code before restoring the ordinary chat history.
                terminal.default_buffer.reset()
                (terminal.history, terminal.default_buffer.history,
                 terminal.is_password, terminal.completer, terminal.auto_suggest) = previous
        else:
            value = driver.read_user_input(None, prompt_text=prompt)
        if not value or not value.strip():
            raise EOFError
        return value.strip()
 
 
def _oauth_provider(ctx: "Context", raw: str) -> Any:
    """Select exactly one known lane without echoing untrusted arguments."""
    from oc_chat.core.provider_plugin import registry
    parts = raw.split()
    if len(parts) != 2:
        ctx.presenter.show_error('Use /login <lane> or /logout <lane>.')
        return None
    try:
        return registry.get(parts[1])
    except KeyError:
        ctx.presenter.show_error('Unknown provider lane.')
        return None
 
 
@command("/login")
def cmd_login(ctx: "Context", raw: str) -> None:
    """Run paste-only provider login and atomically store its credential."""
    from oc_chat.core.provider_oauth import write_credentials
    provider = _oauth_provider(ctx, raw)
    if provider is None:
        return
    capability = getattr(provider, 'oauth', None)
    if capability is None:
        ctx.presenter.show_error('This lane does not support OAuth login.')
        return
    try:
        credentials = capability.oauth_login(_OAuthCallbacks(ctx))
        write_credentials(provider.name, credentials)
    except (EOFError, KeyboardInterrupt):
        ctx.presenter.show_info('Login cancelled.')
        return
    except Exception:
        # Provider and storage errors may embed a credential. Do not render them.
        ctx.presenter.show_error('Login failed; no new credential was saved.')
        return
    ctx.presenter.show_info(f'Signed in to {provider.name}. Use /lane {provider.name} to select it.')
 
 
@command("/logout")
def cmd_logout(ctx: "Context", raw: str) -> None:
    """Remove only the selected lane from the OAuth sidecar."""
    from oc_chat.core.provider_oauth import write_credentials
    provider = _oauth_provider(ctx, raw)
    if provider is None:
        return
    try:
        write_credentials(provider.name, None)
    except Exception:
        ctx.presenter.show_error('Logout failed; check the OAuth sidecar.')
        return
    ctx.presenter.show_info('Stored OAuth credential removed. Existing clients keep their current key until rebuilt.')
 
 
@command("/model-info")
def cmd_model_info(ctx: "Context", raw: str) -> None:
    """Show detailed metadata for a model from the provider.
 
    /model-info             — info for the current model
    /model-info <name>      — info for any named model
    """
    parts = raw.split(None, 1)
    model = parts[1].strip() if len(parts) > 1 else ctx.session.model
    info = ctx.client.provider.model_info(model, ctx.client._client)
    if info is None:
        ctx.presenter.show_info(
            f"Provider {ctx.client.provider.name!r} does not support model info."
        )
        return
    lines = [
        f"Name:           {info.name}",
        f"Family:         {info.family or '—'}",
        f"Parameters:     {info.parameter_count or '—'}",
        f"Quantization:   {info.quantization or '—'}",
        f"Context window: {info.context_window or '—'}",
    ]
    ctx.presenter.show_info("\n".join(lines))
 
 
@command("/effort")
def cmd_effort(ctx: "Context", raw: str) -> None:
    """Show or set the reasoning effort level."""
    parts = raw.split()
    if len(parts) == 1:
        ctx.presenter.show_info(
            f"Reasoning effort: {ctx.session.reasoning_effort}  "
            f"(options: {', '.join(REASONING_EFFORT_LEVELS)})"
        )
    else:
        level = parts[1].lower()
        if level in REASONING_EFFORT_LEVELS:
            ctx.session.reasoning_effort = level
            ctx.session.dirty = True
            ctx.presenter.show_info(f"Reasoning effort set to: {level}")
        else:
            ctx.presenter.show_info(f"Invalid level. Choose from: {', '.join(REASONING_EFFORT_LEVELS)}")
 
 
# NOTE: /clear has been removed as a standalone command.
# Use /session clear instead — it provides verification, backup, and save options.
 
 
@command("/endpoint")
def cmd_endpoint(ctx: "Context", raw: str) -> None:
    """Show, list, or switch the active provider endpoint (Phase O2).
 
    /endpoint            — show the current endpoint
    /endpoint list       — list every known endpoint with LAN policy
    /endpoint use <name> — switch the active endpoint and invalidate caches
    """
    registry = getattr(ctx, "endpoint_registry", None)
    if registry is None or not registry.entries:
        ctx.presenter.show_info(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Command help strings and templates, not secrets"
            "No endpoint registry configured for the current provider. "
            "Set OLLAMA_BASE_URLS or populate ~/.oc_chat/endpoints.toml."
        )
        return
 
    parts = raw.split(maxsplit=2)
    sub = parts[1].strip().lower() if len(parts) > 1 else ""
 
    if sub == "" or sub == "show":
        active = registry.active()
        ctx.presenter.show_info(
            f"Active endpoint: {active.name} -> {active.base_url} [{active.lan_policy_status}]"
        )
        return
 
    if sub == "list":
        lines = ["Endpoints:"]
        active_name = registry.active_name
        for name in sorted(registry.entries):
            ep = registry.entries[name]
            marker = "*" if name == active_name else " "
            lines.append(f" {marker} {name:<16} {ep.base_url}  [{ep.lan_policy_status}]")
        ctx.presenter.show_info("\n".join(lines))
        return
 
    if sub == "use":
        if len(parts) < 3 or not parts[2].strip():
            ctx.presenter.show_info("Usage: /endpoint use <name>")
            return
        target = parts[2].strip()
        # Plan 152 R10-03 — route through the SHARED coordinator instead of the
        # old five-step split commit (switch -> rebind_to -> invalidate, plus the
        # session field and dirty flag). Those five installs ran under the same
        # unguarded dispatch as the session seam, so a Ctrl+C between any two of
        # them split the endpoint tuple exactly like the session desync. Now every
        # fallible step happens in PREPARE and the installs commit together under
        # the transition guard, so the tuple is all-OLD or all-NEW.
        #
        # It also closes a policy hole: the old path reached `entries[name]` +
        # `rebind_to`, and `rebind_to` enforces only the forbidden-host check,
        # NOT the "blocked" LAN policy. `resolve` enforces both, in the owner.
        try:
            prepared = prepare_endpoint_activation(ctx, target)
        except KeyError as exc:
            ctx.presenter.show_error(str(exc))
            return
        except PermissionError as exc:
            ctx.presenter.show_error(str(exc))
            return
        commit_endpoint_activation(ctx, prepared)
        new_active = prepared.endpoint
        ctx.presenter.show_info(
            f"Switched to endpoint '{new_active.name}' ({new_active.base_url})"
        )
        return
 
    ctx.presenter.show_info(
        "Usage: /endpoint | /endpoint list | /endpoint use <name>"
    )
 
 
# The /queue and /add-agent slash commands retired with the queue machinery
# (plan 223 P3, design 4.8). They were the REPL surface of the Markdown queue
# and the agent-pool sidecar, both gone: dispatch is now the only path that
# assigns work, and the pool is a registry view rendered by
# `python -m oc_chat.scripts.plan_manage agents active`.
 
 
@command("/system")
def cmd_system(ctx: "Context", raw: str) -> None:
    """Show or set the system prompt."""
    parts = raw.split(maxsplit=1)
    if len(parts) < 2:
        ctx.presenter.show_info(f"System prompt: {ctx.session.system_prompt}")
        return
    ctx.session.set_system_prompt(parts[1].strip())
    ctx.presenter.show_info("System prompt updated.")
 
 
# ── Filesystem commands ────────────────────────────────────────────────────────
 
@command("/ls")
def cmd_ls(ctx: "Context", raw: str) -> None:
    """List files and directories (supports -r for recursive)."""
    from oc_chat.tools.filesystem import list_dir
    tokens = raw.split()[1:]  # strip "/ls"
    recursive  = "--recursive" in tokens or "-r" in tokens
    path_parts = [t for t in tokens if not t.startswith("-")]
    path       = path_parts[0] if path_parts else "."
    max_depth  = 5 if recursive else 0
 
    try:
        result = list_dir(path=path, max_depth=max_depth)
        for dir_key, contents in sorted(result.items()):
            print(f"\n{dir_key}/")
            for d in contents["dirs"]:
                print(f"  {d}/")
            for f in contents["files"]:
                print(f"  {f}")
    except Exception as e:
        ctx.presenter.show_error(str(e))
 
 
@command("/read")
def cmd_read(ctx: "Context", raw: str) -> None:
    """Print a file's contents to the terminal."""
    from oc_chat.tools.filesystem import read_file
    parts = raw.split(maxsplit=1)
    if len(parts) < 2:
        ctx.presenter.show_info("Usage: /read <file_path>")
        return
    result = read_file(parts[1].strip())
    if "error" in result:
        ctx.presenter.show_error(result["error"])
    else:
        print(result["content"])
        if result.get("truncated"):
            ctx.presenter.show_info("[File truncated at limit]")
 
 
@command("/open")
def cmd_open(ctx: "Context", raw: str) -> None:
    """Print the most-recently-spilled tool artifact for the current session.
 
    Phase 1.C ships only ``/open last`` — useful after a long
    ``read_file`` / ``run_command`` was spilled to a file in
    ``oc_chat/sessions/<name>/artifacts/``.
    """
    parts = raw.split(maxsplit=1)
    target = parts[1].strip() if len(parts) > 1 else "last"
 
    if target.lower() != "last":
        ctx.presenter.show_info(
            f"Only '/open last' is supported in this phase (got: {target!r})."
        )
        return
 
    from oc_chat.core.persistence import find_latest_artifact
    path = find_latest_artifact(ctx.session.name)
    if path is None:
        ctx.presenter.show_info(
            f"No tool artifacts found for session '{ctx.session.name}'."
        )
        return
 
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as exc:
        ctx.presenter.show_error(f"Could not read {path}: {exc}")
        return
 
    ctx.presenter.show_info(f"Artifact: {path}")
    print(text)
 
 
@command("/last")
def cmd_last(ctx: "Context", raw: str) -> None:
    """Page the latest answer, raw answer, reasoning text, or tool result."""
    func = "cmd_last"
    parts = raw.split(maxsplit=1)
    variant = parts[1].strip().lower() if len(parts) > 1 else "answer"
    if not variant:
        variant = "answer"
 
    messages = getattr(ctx.session, "messages", []) or []
    markdown = False
    kind = "answer"
    content = ""
 
    if variant == "answer":
        message = _latest_message_of_role(messages, "assistant")
        content = _message_content_to_text(message.get("content")) if message else ""
        markdown = True
    elif variant == "raw":
        message = _latest_message_of_role(messages, "assistant")
        content = _message_content_to_text(message.get("content")) if message else ""
    elif variant == "reasoning":
        kind = "reasoning"
        message = _latest_message_with_field(
            messages,
            "assistant",
            "reasoning_content",
        )
        content = (
            _message_content_to_text(message.get("reasoning_content"))
            if message else ""
        )
    elif variant == "tool":
        kind = "tool result"
        message = _latest_message_of_role(messages, "tool")
        content = _message_content_to_text(message.get("content")) if message else ""
    else:
        _dbg("OC_CHAT.MAIN", func, f"unknown variant={variant}", 3)
        ctx.presenter.show_error("Usage: /last [answer|raw|reasoning|tool]")
        return
 
    if not content.strip():
        ctx.presenter.show_info(f"No {kind} available yet this session.")
        return
 
    _page_last_content(content, markdown=markdown)
 
 
@command("/context")
def cmd_context(ctx: "Context", raw: str) -> None:
    """Inject a file's content into the conversation as a user message."""
    from oc_chat.tools.filesystem import read_file
    parts = raw.split(maxsplit=1)
    if len(parts) < 2:
        ctx.presenter.show_info("Usage: /context <file_path>")
        return
    path = parts[1].strip()
    result = read_file(path)
    if "error" in result:
        ctx.presenter.show_error(result["error"])
        return
    content = result["content"]
    ctx.session.messages.append({
        "role": "user",
        "content": f"File context — {path}:\n\n{content}",
    })
    ctx.session.dirty = True
    ctx.presenter.show_info(f"Injected {path} into conversation ({len(content)} chars).")
    if result.get("truncated"):
        ctx.presenter.show_info("[File was truncated at read limit]")
 
 
# ── Helper: format token count nicely ─────────────────────────────────────────
 
def _fmt_tokens(n: int) -> str:
    """Format a token count with commas, switching to ``M`` for millions."""
    if n >= 1_000_000:
        return f"{n/1_000_000:.1f}M"
    return f"{n:,}"
 
 
def _fmt_tokens_display(stats) -> str:
    """Return token string, or ``'--'`` when tokens were never tracked for a session that has messages."""
    if stats.total_tokens == 0 and stats.message_count > 0:
        return "--"
    return _fmt_tokens(stats.total_tokens)
 
 
# ── /session command (unified session management) ─────────────────────────────
 
def _sync_sandbox_override(ctx: "Context") -> None:
    """Sync the active session sandbox override into the sandbox helper."""
    override = getattr(ctx, "sandbox_root_override", None)
    if override is None:
        override = getattr(ctx.session, "sandbox_root_override", None)
        ctx.sandbox_root_override = override
    from oc_chat.tools._sandbox import set_session_sandbox_root_override
    set_session_sandbox_root_override(override)
 
 
def _cwd_inside_sandbox(root: object) -> bool:
    """Return True when cwd resolves under *root*."""
    from pathlib import Path
    cwd = Path.cwd().resolve(strict=False)
    try:
        cwd.relative_to(Path(root).resolve(strict=False))
        return True
    except ValueError:
        return False
 
 
@command("/sandbox")
def cmd_sandbox(ctx: "Context", raw: str) -> None:
    """Show or set the current session's filesystem sandbox root."""
    from pathlib import Path
    from oc_chat.tools import filesystem
    from oc_chat.tools._sandbox import set_session_sandbox_root_override
 
    parts = raw.split(maxsplit=2)
    sub = parts[1].lower() if len(parts) > 1 else ""
 
    if sub in {"", "show", "current"}:
        _sync_sandbox_override(ctx)
        root, source = filesystem._sandbox_root_info()
        inside = "yes" if _cwd_inside_sandbox(root) else "no"
        ctx.presenter.show_info(
            f"Sandbox root: {root}\nSource: {source}\nCWD inside sandbox: {inside}"
        )
        return
 
    if sub == "set":
        if len(parts) < 3 or not parts[2].strip():
            ctx.presenter.show_info("Usage: /sandbox set <existing-directory>")
            return
        candidate = Path(parts[2].strip().strip('"').strip("'")).expanduser()
        root = candidate.resolve(strict=False)
        if not root.is_dir():
            ctx.presenter.show_error(f"Sandbox root does not exist or is not a directory: {candidate}")
            return
        # WU-7 / G-1: under guarded/isolated, refuse a new sandbox that overlaps a
        # harness importable root (write->import in-process ACE). permissive: no-op.
        from oc_chat.core.risk_profile import resolve_risk_profile
        from oc_chat.core.deployment_shape import check_deployment_shape, DeploymentShapeError
        try:
            check_deployment_shape(root, resolve_risk_profile())
        except DeploymentShapeError as exc:
            ctx.presenter.show_error(f"Refused: {exc}")
            return
        override = str(root)
        ctx.sandbox_root_override = override
        ctx.session.sandbox_root_override = override
        ctx.session.dirty = True
        set_session_sandbox_root_override(override)
        ctx.presenter.show_info(f"Sandbox root set to {root}")
        return
 
    if sub in {"unset", "reset"}:
        ctx.sandbox_root_override = None
        ctx.session.sandbox_root_override = None
        ctx.session.dirty = True
        set_session_sandbox_root_override(None)
        root, source = filesystem._sandbox_root_info()
        ctx.presenter.show_info(f"Sandbox override cleared. Effective root: {root} ({source})")
        return
 
    ctx.presenter.show_info("Usage: /sandbox [set <existing-directory>|unset]")
 
 
@command("/grants")
def cmd_grants(ctx: "Context", raw: str) -> None:
    """Show or revoke the session-long Tier-2 python grant (plan 235 / SHELL-02).
 
    /grants          — print the granted directory, or "none"
    /grants list     — alias for the above
    /grants revoke   — drop the grant; the next python run asks the human again
 
    The grant is what the third key at the Tier-2 confirm prompt records: for the
    rest of the session, python scripts resolving inside that one directory run
    without a keystroke. It is process-lifetime state — never persisted — and it
    also drops on ``/session clear`` and on a session load/switch.
    """
    from oc_chat.core.tier2_session_grant import active_session_grant, revoke_session_grants
 
    parts = raw.split(maxsplit=1)
    sub = parts[1].strip().lower() if len(parts) > 1 else ""
 
    if sub in {"", "list", "show"}:
        active = active_session_grant()
        ctx.presenter.show_info(
            f"Session Tier-2 python grant: {active}\n"
            "  Any .py file the agent writes there runs without asking. "
            "Drop it with /grants revoke."
            if active else "Session Tier-2 python grant: none"
        )
        return
 
    if sub == "revoke":
        active = active_session_grant()
        revoke_session_grants()
        ctx.presenter.show_info(
            f"Revoked the session Tier-2 python grant on {active}."
            if active else "No session Tier-2 python grant to revoke."
        )
        return
 
    ctx.presenter.show_info("Usage: /grants [list|revoke]")
 
 
@command("/mode")
def cmd_mode(ctx: "Context", raw: str) -> None:
    """Show or set this session's confirmation mode (plan 283 / SHELL-04).
 
    /mode          — print the mode and every remembered answer
    /mode ask      — every confirmable call reaches the human (the default)
    /mode auto     — the routine asks are answered without a keystroke
    /mode forget   — drop the remembered answers, so the human is asked again
 
    This command, the ``--auto`` launch flag and the ``u`` key at the Tier-2 ask
    are the ONLY ways in. No tool, no model output, no settings file and no
    environment variable can set the mode — which is the point of it: it is the
    human's switch, not the agent's. Auto mode never answers a protected write,
    never turns an argv-policy denial into a run, and is inert outside the
    ``permissive`` risk profile.
    """
    func = "cmd_mode"
    from oc_chat.core import session_mode
 
    parts = raw.split(maxsplit=1)
    sub = parts[1].strip().lower() if len(parts) > 1 else ""
 
    if sub in {"", "list", "show"}:
        remembered = session_mode.remembered_programs()
        lines = [f"Confirmation mode: {session_mode.current_mode()}"]
        if session_mode.current_mode() == session_mode.AUTO and not session_mode.auto_mode_available():
            lines.append("  (inert here: Auto mode applies only under the "
                         "permissive risk profile)")
        if remembered:
            lines.append(f"  Remembered answers ({len(remembered)}), "
                         f"drop them with /mode forget:")
            for program, answer in sorted(remembered.items()):
                verdict = (f"always run (budget {answer.budget_s} s)"
                           if answer.run else "never run")
                lines.append(f"    {program}: {verdict}")
        else:
            lines.append("  Remembered answers: none")
        ctx.presenter.show_info("\n".join(lines))
        return
 
    if sub == "forget":
        dropped = session_mode.forget_remembered()
        ctx.presenter.show_info(
            f"Dropped {dropped} remembered answer(s); the next Tier-2 command "
            f"asks again." if dropped else "No remembered answers to drop.")
        return
 
    if sub in session_mode.VALID_MODES:
        if sub == session_mode.AUTO and not session_mode.auto_mode_available():
            # Refuse where it could not apply, rather than setting a flag that
            # silently does nothing — the operator would read the confirmation
            # line as a change that happened.
            ctx.presenter.show_error(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Operator-facing refusal prose naming an env var; high-entropy English, not a credential (operator-ruled 2026-09-16)"
                "Auto mode applies only under the permissive risk profile. "
                "The active profile is set out of band by OC_CHAT_PROFILE and "
                "cannot be changed from here; the mode is unchanged.")
            _dbg("OC_CHAT.MAIN", func, "refused /mode auto off permissive", 2)
            return
        session_mode.set_mode(sub)
        ctx.presenter.show_info(
            f"Confirmation mode: {sub}."
            + ("  The routine asks are answered without a keystroke; a "
               "protected write still always reaches you. /mode ask switches back."
               if sub == session_mode.AUTO else
               "  Every confirmable call reaches you again.")
        )
        return
 
    ctx.presenter.show_info("Usage: /mode [ask|auto|forget]")
 
 
@command("/cwd")
def cmd_cwd(ctx: "Context", raw: str) -> None:
    """Re-anchor this oc_chat to a new working directory.
 
    /cwd          — print current anchors (cwd, sandbox root, source label)
    /cwd <path>   — re-anchor to <path>, which must be an existing directory
 
    Uses the 152 entry lease (transition_entry_lease) so a mid-turn or
    reentrant call is rejected with zero effects. Commits chdir + sandbox
    override (both mirrors) + dirty flag, then refreshes variable context.
    """
    import os as _os
    from pathlib import Path
    from oc_chat.tools import filesystem
    from oc_chat.tools import _sandbox
    from oc_chat.core.session_transition import (
        TransitionRejectedError, transition_entry_lease,
    )
    from oc_chat.core.startup import refresh_variable_context
 
    parts = raw.split(maxsplit=1)
 
    if len(parts) == 1 or not parts[1].strip():
        # No args: print current anchors
        _sync_sandbox_override(ctx)
        root, source = filesystem._sandbox_root_info()
        ctx.presenter.show_info(
            f"CWD: {Path.cwd()}\nSandbox root: {root}\nSource: {source}"
        )
        return
 
    # Validate: resolve path, must be existing directory
    candidate = Path(parts[1].strip().strip('"').strip("'")).expanduser()
    resolved = candidate.resolve(strict=False)
 
    if not resolved.is_dir():
        ctx.presenter.show_error(
            f"/cwd: {candidate} does not exist or is not a directory"
        )
        return
 
    # Commit under the 152 entry lease
    original_cwd = _os.getcwd()
    try:
        with transition_entry_lease():
            _os.chdir(resolved)
            override = str(resolved)
            ctx.sandbox_root_override = override
            ctx.session.sandbox_root_override = override
            ctx.session.dirty = True
            _sandbox.install_resolved(resolved)
            # Sticky re-anchor (operator decision 2026-07-22): also set the
            # process-level sandbox root so a subsequent `/session new` — which
            # resets the per-session override to the fresh session's empty
            # projection (152 D-model) — still resolves to this directory via
            # the env rung (filesystem._sandbox_root_info precedence #2). An
            # explicit per-session override still wins over this. Delivers the
            # plan's "/cwd then /session new is born anchored" flow.
            _os.environ["OC_CHAT_SANDBOX_ROOT"] = override
 
        # Post-transition refresh (best-effort)
        refresh_variable_context(ctx.session)
 
        root, source = filesystem._sandbox_root_info()
        ctx.presenter.show_info(
            f"re-anchored \u2192 {resolved} (sandbox: {source})"
        )
    except TransitionRejectedError as exc:
        ctx.presenter.show_error(f"/cwd rejected: {exc}")
    except OSError as exc:
        # chdir failed — restore
        try:
            _os.chdir(original_cwd)
        except OSError:
            pass
        ctx.presenter.show_error(f"/cwd: cannot change to {resolved}: {exc}")
 
 
def _session_help(ctx):
    """Print help for ``/session`` subcommands."""
    print("\n/session subcommands:")
    print("  /session new [name]       - start a new named session")
    print("  /session list             - list all saved sessions")
    print("  /session current          - show current session info")
    print("  /session save [name]      - save current session")
    print("  /session load <name>      - load a saved session (auto-saves current)")
    print("  /session export [--html] [file]  - export current session as Markdown (default) or HTML")
    print("  /session switch <name>    - switch to another session (auto-saves current)")
    print("  /session clear            - clear conversation history with verification & backup options")
    print()
 
 
@command("/session")
def cmd_session(ctx: "Context", raw: str) -> None:
    """Dispatch ``/session <subcommand> [...]``."""
    from oc_chat.core.persistence import (
        save_chat_log, save_session_metadata, load_session_metadata,
        export_markdown, list_sessions, session_exists, DEFAULT_SESSIONS_DIR,
    )
    from oc_chat.core.session import (
        _auto_name,
        CANONICAL_POLICY,
        ChatSession,
        build_system_prompt,
    )
 
    parts = raw.split(maxsplit=2)
    sub = parts[1].lower() if len(parts) > 1 else ""
 
    # ── /session new [name] ───────────────────────────────────────────────
    if sub == "new":
        name = parts[2].strip() if len(parts) > 2 else _auto_name()
        if session_exists(name):
            ctx.presenter.show_error(
                f"Session '{name}' already exists. "
                f"Use /session switch {name} to resume it, or pick a different name."
            )
            return
        # Auto-save current session if dirty
        if ctx.session.dirty:
            save_chat_log(ctx.session)
            save_session_metadata(ctx.session)
            ctx.presenter.show_info(f"Auto-saved current session.")
        # Create fresh session (immutable system prefix includes tool schemas)
        system_text = build_system_prompt(CANONICAL_POLICY, ctx.tool_schemas)
        new_session = ChatSession(
            name=name,
            messages=[{"role": "system", "content": system_text}],
        )
        new_session.created = name.split("_", 1)[1] if "_" in name else "unknown"
        # Plan 152 D3: replace through the transition seam, never by bare
        # assignment. `Context.session` is a plain dataclass field, so
        # `ctx.session = ...` invokes nothing — the agent, the event writer, the
        # F67 trust key, the sandbox global and the generation-options
        # projection all kept pointing at the OLD session, which is the desync
        # this plan exists to fix. The seam also performs the presenter refresh
        # (post-transaction, best-effort), so the explicit call is gone.
        replace_session(ctx, new_session)
        # M8.0 F67: external-content trust is session-scoped; the react-loop resets
        # it on the next turn (external_content_gate.bind_active_session sees the new
        # session), so a source trusted in the prior session does not carry over.
        ctx.presenter.show_info(f"New session: {name} (model: {new_session.model})")
        return
 
    # ── /session list ─────────────────────────────────────────────────────
    if sub == "list":
        sessions = list_sessions()
        ctx.presenter.show_session_list(sessions, current_name=ctx.session.name)
        return
 
    # ── /session current ──────────────────────────────────────────────────
    if sub == "current":
        s = ctx.session
        status = "(unsaved changes)" if s.dirty else "(clean)"
        if s.filepath is None and not s.dirty:
            status = "(never saved)"
        print(f"\nSession: {s.name}")
        print(f"Model: {s.model}")
        print(f"Reasoning effort: {s.reasoning_effort}")
        print(f"Messages: {s.stats.message_count} {status}")
        print(f"Tool calls: {s.stats.tool_call_count}")
        if s.stats.total_tokens == 0 and s.stats.message_count > 0:
            print("Token usage: -- (not tracked for this session)")
        else:
            print(f"Token usage: prompt={_fmt_tokens(s.stats.total_prompt_tokens)}, "
                  f"completion={_fmt_tokens(s.stats.total_completion_tokens)}, "
                  f"total={_fmt_tokens(s.stats.total_tokens)}")
        if s.chat_logs:
            print(f"Chat logs ({len(s.chat_logs)}):")
            for cl in s.chat_logs:
                print(f"  - {cl}")
        if s.filepath:
            print(f"Metadata file: {s.filepath}")
        print()
        return
 
    # ── /session save [name] ──────────────────────────────────────────────
    if sub == "save":
        name = parts[2].strip() if len(parts) > 2 else None
        if name and name != ctx.session.name and session_exists(name):
            ctx.presenter.show_error(
                f"Session '{name}' already exists. Use a different name."
            )
            return
        # Plan 152 D8 — rename/save-as is its OWN transition class. It must
        # capture the outgoing name's event slug BEFORE the name mutation, and
        # supersede-mark the old on-disk record so the same-session_id sibling a
        # rename inevitably creates is non-canonical by construction rather than
        # adjudicated by load order. Trust is NOT reset: the canonical
        # session_id is preserved, so this is the same identity renamed.
        log_path, meta_path = rename_session(ctx, name or ctx.session.name)
        ctx.presenter.show_info(f"Saved chat log to {log_path}")
        ctx.presenter.show_info(f"Saved metadata to {meta_path}")
        return
 
    # ── /session load <name> ──────────────────────────────────────────────
    if sub == "load":
        if len(parts) < 3:
            ctx.presenter.show_info("Usage: /session load <name>")
            return
        name = parts[2].strip()
        if not session_exists(name):
            ctx.presenter.show_error(f"No session named '{name}' found in sessions directory.")
            return
        # Plan 152-b D2/D3 — the WHOLE load is one atomic prepare-then-commit
        # transition through the seam, replacing the pre-152-b in-place hydration.
        # The candidate is fully built and its event writer re-keyed to the LOADED
        # id (closing the audit-plane orphaning), the outgoing session is saved
        # from an immutable snapshot, and identity publishes last. A same-name
        # `/session load` is a full same-record reload through this path (D3).
        # Presenter notices are DEFERRED to post-commit: nothing is emitted until
        # the transition has actually succeeded.
        try:
            outcome = load_session_into(ctx, name)
        except (ChatLogLoadError, TransitionRejectedError) as e:
            ctx.presenter.show_error(str(e))
            return
        except Exception as e:
            ctx.presenter.show_error(str(e))
            return
        if outcome.outgoing_saved:
            ctx.presenter.show_info("Auto-saved current session.")
        if outcome.migration_notice:
            ctx.presenter.show_info(outcome.migration_notice)
        ctx.presenter.show_info(
            f"Loaded session '{ctx.session.name}' — "
            f"{ctx.session.stats.message_count} messages, "
            f"{ctx.session.stats.tool_call_count} tool calls, "
            f"{_fmt_tokens(ctx.session.stats.total_tokens)} tokens "
            f"(model: {ctx.session.model})"
        )
        return
 
    # ── /session export [filename] ────────────────────────────────────────
    if sub == "export":
        # Parse: /session export [--html] [filename]
        rest = parts[2].strip() if len(parts) > 2 else ""
        tokens = rest.split()
        as_html = "--html" in tokens
        filename_tokens = [t for t in tokens if not t.startswith("--")]
        filename = filename_tokens[0] if filename_tokens else None
        try:
            if as_html:
                from oc_chat.core.persistence import export_html
                path = export_html(ctx.session, filepath=filename)
            else:
                path = export_markdown(ctx.session, filepath=filename)
            ctx.presenter.show_info(f"Exported to {path}")
        except ImportError as e:
            ctx.presenter.show_error(
                f"--html requires the optional `rich` dependency: {e}.  "
                f"Install with: pip install -e .[rich]"
            )
        except Exception as e:
            ctx.presenter.show_error(str(e))
        return
 
    # ── /session switch <name> ────────────────────────────────────────────
    if sub == "switch":
        if len(parts) < 3:
            ctx.presenter.show_info("Usage: /session switch <name>")
            return
        name = parts[2].strip()
        if name == ctx.session.name:
            ctx.presenter.show_info(f"Already on session '{name}'.")
            return
        if not session_exists(name):
            ctx.presenter.show_error(f"No session named '{name}' found in sessions directory.")
            return
        # Plan 152-b D2/D3 — `/session switch` reads the SAME persisted record
        # source as `/session load` (there is no in-memory switch source), so it
        # runs the identical atomic seam. The same-name early no-op above is
        # RETAINED for compatibility (D3); everything else is one prepare-then-
        # commit transition with deferred post-commit notices.
        try:
            outcome = load_session_into(ctx, name)
        except (ChatLogLoadError, TransitionRejectedError) as e:
            ctx.presenter.show_error(str(e))
            return
        except Exception as e:
            ctx.presenter.show_error(str(e))
            return
        if outcome.outgoing_saved:
            ctx.presenter.show_info("Auto-saved current session.")
        if outcome.migration_notice:
            ctx.presenter.show_info(outcome.migration_notice)
        ctx.presenter.show_info(
            f"Switched to session: {ctx.session.name} — "
            f"{ctx.session.stats.message_count} messages, "
            f"{ctx.session.stats.tool_call_count} tool calls, "
            f"{_fmt_tokens(ctx.session.stats.total_tokens)} tokens "
            f"(model: {ctx.session.model})"
        )
        return
 
    # ── /session clear ────────────────────────────────────────────────────
    if sub == "clear":
        msg_count = ctx.session.stats.message_count
        print(f"\nSession '{ctx.session.name}' has {msg_count} message(s).")
        print("What would you like to do before clearing?")
        print()
        print("  1) Save an archived backup, then clear")
        print("  2) Save as a new session & switch to a fresh one")
        print("  3) Clear without saving")
        print("  4) Cancel")
 
        choice = input("\nChoice (1-4): ").strip()
 
        if choice == "1":
            # Save first to ensure the live files reflect the latest in-memory
            # state, then clear and save again.  Each save archives the
            # previous version to oc_chat/sessions/archive/{name}/, so the
            # pre-clear conversation is preserved in archive/.
            save_chat_log(ctx.session)
            save_session_metadata(ctx.session)
            ctx.session.clear()
            save_chat_log(ctx.session)
            save_session_metadata(ctx.session)
            ctx.presenter.show_info(
                "Conversation history cleared.  Pre-clear snapshot is in "
                f"oc_chat/sessions/archive/{ctx.session.name}/."
            )
 
        elif choice == "2":
            # Save current session, then start a fresh one
            save_chat_log(ctx.session)
            save_session_metadata(ctx.session)
            ctx.presenter.show_info(f"Current session saved.")
            new_name = _auto_name()
            system_text = build_system_prompt(CANONICAL_POLICY, ctx.tool_schemas)
            new_session = ChatSession(
                name=new_name,
                messages=[{"role": "system", "content": system_text}],
            )
            new_session.created = new_name.split("_", 1)[1] if "_" in new_name else "unknown"
            # Plan 152 D3 — replacement path 2 of 3; see the note at `/session new`.
            replace_session(ctx, new_session)
            ctx.presenter.show_info(f"Started new session: {new_name} (model: {new_session.model})")
 
        elif choice == "3":
            # Clear without saving — ask for final confirmation
            confirm = input("Clearing will permanently delete all messages. Are you sure? (y/N): ").strip().lower()
            if confirm in ("y", "yes"):
                ctx.session.clear()
                ctx.presenter.show_info("Conversation history cleared.")
            else:
                ctx.presenter.show_info("Clear cancelled.")
 
        else:
            ctx.presenter.show_info("Clear cancelled.")
 
        return
 
    # ── unknown / missing subcommand ──────────────────────────────────────
    _session_help(ctx)
 
 
# ── UI configuration command ──────────────────────────────────────────────────
 
# Valid values for /ui reasoning — kept in module scope so future phases can
# import the same source of truth from RichPresenter / TextualPresenter.
UI_REASONING_VALUES = ("show", "dim", "hide")
UI_REASONING_DEFAULT = "dim"
# Plan 147 D5: the 'dim' tail line count, persisted as the separate typed
# key ui.reasoning_tail — independent of ui.show_reasoning so switching
# modes never silently resets a previously-chosen tail length.
UI_REASONING_TAIL_DEFAULT = default("ui_reasoning_tail")
 
 
def _ui_help(ctx) -> None:
    """Print help for /ui subcommands."""
    print("\n/ui subcommands:")
    print("  /ui reasoning [show|dim [N]|hide]  - control how reasoning / chain-of-thought is rendered")
    print(f"                                    ('dim' shows the last N lines, default {UI_REASONING_TAIL_DEFAULT};")
    print("                                    takes effect with --ui rich; persisted regardless)")
    print()
    print("  Future phases will add:  /ui theme <name>, /ui stats <compact|verbose|off>,")
    print("                            /ui set <key> <value>")
    print()
 
 
def _read_ui_pref(session, key: str, default):
    """Read ``meta.json[ui][<key>]`` from a live session (None-safe)."""
    ui = getattr(session, "ui_prefs", None)
    if ui is None:
        return default
    return ui.get(key, default)
 
 
def _write_ui_pref(session, key: str, value) -> None:
    """Write ``meta.json[ui][<key>]`` and mark session dirty so it persists on save."""
    if not hasattr(session, "ui_prefs") or session.ui_prefs is None:
        session.ui_prefs = {}
    session.ui_prefs[key] = value
    session.dirty = True
 
 
@command("/ui")
def cmd_ui(ctx: "Context", raw: str) -> None:
    """Dispatch ``/ui <subcommand> [args...]``.
 
    Phase 0 implements only the ``reasoning`` subcommand.  The dispatcher
    is structured to mirror :func:`cmd_session` so Phases 1+ can add
    ``theme``, ``stats``, ``set``, etc. as one-clause additions.
    """
    parts = raw.split(maxsplit=2)
    sub = parts[1].lower() if len(parts) > 1 else ""
 
    if sub == "":
        _ui_help(ctx)
        return
 
    if sub == "reasoning":
        if len(parts) < 3:
            current = _read_ui_pref(ctx.session, "show_reasoning", UI_REASONING_DEFAULT)
            tail = _read_ui_pref(ctx.session, "reasoning_tail", UI_REASONING_TAIL_DEFAULT)
            ctx.presenter.show_info(
                f"reasoning: {current} (dim tail: {tail} lines)  "
                f"(options: {', '.join(UI_REASONING_VALUES)})"
            )
            return
 
        # /ui reasoning <mode> [N] — N is only meaningful (and only accepted) for 'dim'.
        arg_parts = parts[2].strip().split(maxsplit=1)
        value = arg_parts[0].lower()
        if value not in UI_REASONING_VALUES:
            ctx.presenter.show_error(
                f"Invalid value '{value}' for /ui reasoning.  "
                f"Choose one of: {', '.join(UI_REASONING_VALUES)}"
            )
            return
 
        tail_arg = arg_parts[1].strip() if len(arg_parts) > 1 else ""
        tail_note = ""
        if tail_arg:
            if value != "dim":
                ctx.presenter.show_error(
                    f"A tail line count is only valid with 'dim' (got '{value} {tail_arg}')."
                )
                return
            try:
                tail_n = int(tail_arg)
            except ValueError:
                ctx.presenter.show_error(
                    f"Invalid tail line count '{tail_arg}' — must be a positive integer."
                )
                return
            if tail_n < 1:
                ctx.presenter.show_error(f"Tail line count must be >= 1 (got {tail_n}).")
                return
            _write_ui_pref(ctx.session, "reasoning_tail", tail_n)
            tail_note = f" (tail: {tail_n} lines)"
 
        _write_ui_pref(ctx.session, "show_reasoning", value)
        ctx.presenter.show_info(
            f"reasoning: {value}{tail_note}  "
            f"(persisted in session metadata; takes visible effect with --ui rich)"
        )
        return
 
    ctx.presenter.show_error(f"Unknown /ui subcommand: '{sub}'")
    _ui_help(ctx)
 
 
# ── Command palette ────────────────────────────────────────────────────────────
 
@command("/palette")
def cmd_palette(ctx: "Context", raw: str) -> None:
    """Fuzzy-search commands and recent sessions.
 
    This is the slash-command equivalent of ``Ctrl+P`` at the prompt.
    If a query is provided, matching commands and session-load actions
    are printed.  Without a query, the full list is shown.
    """
    func = "cmd_palette"
    parts = raw.split(maxsplit=1)
    query = parts[1].strip().lower() if len(parts) > 1 else ""
 
    # Collect palette entries
    entries: list[tuple[str, str]] = []  # (display, action)
 
    for verb, help_text in COMMAND_HELP.items():
        display = help_text.split("  - ", 1)[1] if "  - " in help_text else verb
        entries.append((f"{verb:<20} {display}", verb))
 
    # Add recent session load actions
    try:
        from oc_chat.core.persistence import list_sessions
        for s in list_sessions()[:10]:  # limit to 10 most recent
            name = s["name"]
            entries.append((
                f"/session load {name:<25} load saved session",
                f"/session load {name}",
            ))
    except Exception as e:
        # VALIDATION_ERROR — sessions dir absent / unreadable; degrade to verb-only palette
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
 
    # Filter by query
    if query:
        entries = [
            (display, action)
            for display, action in entries
            if query in display.lower() or query in action.lower()
        ]
 
    if not entries:
        ctx.presenter.show_info("No matching commands or sessions found.")
        return
 
    lines = [f"\n  Command palette{' — ' + query if query else ''}:", ""]
    for display, _ in entries:
        lines.append(f"    {display}")
    lines.append("")
    for line in lines:
        ctx.presenter.show_info(line)
    ctx.presenter.show_info(
        "Tip: Press Ctrl+P at the prompt for interactive fuzzy search."
    )
 
 
# ── Agentic commands ───────────────────────────────────────────────────────────
 
@command("/goal")
def cmd_goal(ctx: "Context", raw: str) -> DispatchResult:
    """Inject an autonomous goal as a user message; main loop will run the turn.
 
    Plan 152 D7/R11-04 — containment is HANDLER-LOCAL. The mutation unit (the
    goal flags plus the appended message) runs CONTIGUOUSLY and BEFORE any
    raising call, and this handler catches its own failures while its
    compensation tokens are still in scope. ``dispatch`` cannot do that job: a
    raised handler returns nothing, so the tokens would never reach it.
 
    A post-commit REPORTING failure is NOT a command failure — the turn stays
    queued and runs exactly once, carrying a ``warning``. Mapping it to a
    no-turn failure would strand a committed queue.
    """
    func = "cmd_goal"
    parts = raw.split(maxsplit=1)
    if len(parts) < 2:
        ctx.presenter.show_info("Usage: /goal <task description>")
        return DispatchResult(kind=HANDLED_NO_TURN)
    goal = parts[1].strip()
 
    # Compensation tokens, captured BEFORE the mutation unit.
    prior_in_goal = ctx.in_goal
    prior_description = ctx.goal_description
    prior_files = list(ctx.goal_files_changed or [])
    message_index = len(ctx.session.messages)
 
    try:
        # ── The mutation unit: goal tracking + the queued message, contiguous ──
        ctx.in_goal = True
        ctx.goal_description = goal
        ctx.goal_files_changed = []
        ctx.session.add_user(
            f"Please accomplish this goal using the available tools. "
            f"Break it into steps, execute each step, and report results.\n\nGoal: {goal}"
        )
    except Exception as exc:
        # Mid-unit failure: unwind our own tokens so no orphan message and no
        # stranded in_goal flag poisons the next REPL iteration.
        _dbg("OC_CHAT.MAIN", func,
             f"SYSTEM_ERROR: goal unit failed: {type(exc).__name__}: {exc}", 1, override=True)
        ctx.in_goal = prior_in_goal
        ctx.goal_description = prior_description
        ctx.goal_files_changed = prior_files
        del ctx.session.messages[message_index:]
        ctx.presenter.show_error(f"Could not queue the goal: {exc}")
        return DispatchResult(kind=INTERNAL_FAILURE, effect_disposition=EFFECT_NONE)
 
    # ── Post-commit reporting. A failure here does NOT cancel the queued turn ──
    warning = ""
    try:
        ctx.presenter.show_info(f"Goal queued: {goal}")
    except Exception as exc:
        warning = f"goal queued, but the confirmation could not be displayed: {exc}"
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: {warning}", 2, override=True)
    return DispatchResult(
        kind=HANDLED_TURN_QUEUED,
        effect_disposition=EFFECT_COMMITTED,
        queued_turn_token=message_index,
        warning=warning,
    )
 
 
@command("/plan")
def cmd_plan(ctx: "Context", raw: str) -> "DispatchResult | None":
    """Plan-document management — create, pause, resume, list, etc.
 
    Usage:
        /plan <description>       — create a new plan (alias: --new <desc>)
        /plan --status            — show the active plan's state
        /plan --pause             — pause the active plan
        /plan --continue          — resume a paused plan
        /plan --complete          — mark the active plan complete
        /plan --abort             — abandon the active plan
        /plan --list              — list all plans, grouped by status
        /plan --switch <id>       — change the active plan
    """
    parts = raw.split(maxsplit=1)
    rest = parts[1].strip() if len(parts) > 1 else ""
 
    # No-arg invocation prints usage
    if not rest:
        ctx.presenter.show_info(cmd_plan.__doc__ or "")
        return
 
    # Flag-shaped invocation: dispatch to the matching sub-handler.
    # Bare invocation (`/plan <free text>`) creates a new plan.
    if rest.startswith("--new"):
        # Strip "--new" and optional whitespace, treat remainder as description
        desc = rest[len("--new"):].strip()
        return _cmd_plan_new(ctx, desc)
    if rest == "--status":
        return _cmd_plan_status(ctx)
    if rest == "--pause":
        return _cmd_plan_pause(ctx)
    if rest == "--continue":
        return _cmd_plan_continue(ctx)
    if rest == "--complete":
        return _cmd_plan_complete(ctx)
    if rest == "--abort":
        return _cmd_plan_abort(ctx)
    if rest == "--list":
        return _cmd_plan_list(ctx)
    if rest.startswith("--switch"):
        target = rest[len("--switch"):].strip()
        return _cmd_plan_switch(ctx, target)
    if rest.startswith("--"):
        ctx.presenter.show_info(
            "Unknown /plan flag. Run /plan without args for usage."
        )
        return
    # Free-text description → new plan
    return _cmd_plan_new(ctx, rest)
 
 
def _cmd_plan_new(ctx: "Context", desc: str) -> DispatchResult:
    """Create a new plan and enter plan mode."""
    func = "cmd_plan_new"
    if not desc:
        ctx.presenter.show_info("Usage: /plan <description>")
        return DispatchResult(kind=HANDLED_NO_TURN)
 
    # Refuse if another plan is already active or paused in this session
    from oc_chat.core.plans import create_plan, find_plan_by_id
    from oc_chat.core.plan_inventory import TERMINAL_STATUSES
 
    if ctx.session.active_plan_id:
        existing = find_plan_by_id(ctx.session.active_plan_id)
        if existing is not None and existing.status not in TERMINAL_STATUSES:
            ctx.presenter.show_info(
                f"Another plan is active in this session: "
                f"{ctx.session.active_plan_id} (status: {existing.status}). "
                "Run /plan --complete or /plan --abort first, "
                "or /plan --switch to a different plan."
            )
            return DispatchResult(kind=HANDLED_NO_TURN)
 
    # Compensation tokens, captured BEFORE the durable call (plan 152 R11-04/R11-06).
    prior_active_plan_id = ctx.session.active_plan_id
    prior_dirty = ctx.session.dirty
    prior_in_plan = ctx.in_plan
    prior_plan_description = ctx.plan_description
    prior_plan_files = list(ctx.plan_files_changed or [])
    message_index = len(ctx.session.messages)
    plan_id = ""
 
    try:
        # ── The mutation unit, contiguous and before any raising call ──────
        plan_id, path = create_plan(desc, session_name=ctx.session.name)
        ctx.session.active_plan_id = plan_id
        ctx.session.dirty = True
        ctx.in_plan = True
        ctx.plan_description = desc
        ctx.plan_files_changed = []
        ctx.session.add_user(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Command help strings and templates, not secrets"
            f"Design an implementation plan for: {desc}\n\n"
            f"Follow the template at `oc_chat/plans/PLAN_TEMPLATE.md`. "
            f"Read only files necessary to design the plan; "
            f"**do NOT modify any source files**. "
            f"Output the complete plan as a single Markdown file to "
            f"`{path.as_posix()}` using `write_file`. "
            f"When finished, reply with the filename only."
        )
    except Exception as exc:
        _dbg("OC_CHAT.MAIN", func,
             f"SYSTEM_ERROR: plan-new unit failed: {type(exc).__name__}: {exc}",
             1, override=True)
        ctx.session.active_plan_id = prior_active_plan_id
        ctx.session.dirty = prior_dirty
        ctx.in_plan = prior_in_plan
        ctx.plan_description = prior_plan_description
        ctx.plan_files_changed = prior_plan_files
        del ctx.session.messages[message_index:]
        # Retract OUR OWN orphan via the owner, so the "another plan is active"
        # guard above is not poisoned for the next /plan --new. The record is
        # retained in a terminal state, never deleted — this is a --new orphan,
        # and ONLY a --new orphan may be retracted this way.
        disposition = EFFECT_NONE
        if plan_id:
            try:
                from oc_chat.core.plans import abort_plan
                abort_plan(plan_id)
            except Exception as abort_exc:
                disposition = EFFECT_UNKNOWN
                _dbg("OC_CHAT.MAIN", func,
                     f"SYSTEM_ERROR: could not retract orphan {plan_id}: {abort_exc}",
                     1, override=True)
        ctx.presenter.show_error(
            f"Could not create the plan: {exc}"
            + ("" if disposition == EFFECT_NONE else
               f" The orphaned record {plan_id} could NOT be retracted — its state is "
               f"indeterminate. Check /plan --list and clear it with /plan --abort "
               f"before creating another plan.")
        )
        return DispatchResult(kind=INTERNAL_FAILURE, effect_disposition=disposition)
 
    warning = ""
    try:
        ctx.presenter.show_info(f"Plan queued: {plan_id} ({path.as_posix()})")
    except Exception as exc:
        warning = f"plan {plan_id} created, but the confirmation could not be displayed: {exc}"
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: {warning}", 2, override=True)
    _dbg("OC_CHAT.MAIN", func, f"created {plan_id!r}", 2)
    return DispatchResult(
        kind=HANDLED_TURN_QUEUED,
        effect_disposition=EFFECT_COMMITTED,
        queued_turn_token=message_index,
        warning=warning,
    )
 
 
def _plan_checkbox_tally(path) -> tuple[int, int]:
    """Count `- [ ]` (open) and `- [x]` (done) lines in a plan markdown file.
 
    Returns ``(done, open)``; returns ``(0, 0)`` if the file is missing
    or unreadable.
    """
    try:
        content = path.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        return (0, 0)
    done = 0
    open_ = 0
    for line in content.splitlines():
        stripped = line.lstrip()
        if stripped.startswith("- [ ]"):
            open_ += 1
        elif stripped.startswith("- [x]") or stripped.startswith("- [X]"):
            done += 1
    return (done, open_)
 
 
def _active_record_or_warn(ctx: "Context"):
    """Resolve session.active_plan_id to a PlanRecord; warn if missing.
 
    Returns ``None`` and prints a presenter message in the no-active-plan
    case, so callers can `if rec is None: return` cleanly.
    """
    from oc_chat.core.plans import find_plan_by_id
 
    pid = ctx.session.active_plan_id
    if not pid:
        ctx.presenter.show_info("No active plan in this session.")
        return None
    rec = find_plan_by_id(pid)
    if rec is None:
        ctx.presenter.show_info(
            f"Inventory has no record for active plan {pid!r}. "
            "Run /plan --list to see existing plans."
        )
        return None
    return rec
 
 
def _cmd_plan_status(ctx: "Context") -> None:
    """Show the active plan's filename, state, description, timestamps,
    and checkbox tally."""
    from pathlib import Path
    from oc_chat.core.plans import DEFAULT_PLANS_DIR
    from oc_chat.core.plan_inventory import _expected_filename
 
    rec = _active_record_or_warn(ctx)
    if rec is None:
        return
 
    expected = _expected_filename(rec.plan_id, rec.status)
    path = DEFAULT_PLANS_DIR / expected
    if path.exists():
        done, open_ = _plan_checkbox_tally(path)
        tally = f"{done}/{done + open_} steps complete"
    else:
        tally = "file missing"
 
    lines = [
        f"Plan:        {rec.plan_id}",
        f"Status:      {rec.status}",
        f"Description: {rec.description}",
        f"Filename:    {expected}",
        f"Created:     {rec.created_at}",
    ]
    if rec.completed_at:
        lines.append(f"Completed:   {rec.completed_at}")
    if rec.aborted_at:
        lines.append(f"Aborted:     {rec.aborted_at}")
    lines.append(f"Progress:    {tally}")
    for line in lines:
        ctx.presenter.show_info(line)
 
 
def _cmd_plan_pause(ctx: "Context") -> None:
    """Pause the active plan."""
    from oc_chat.core.plans import pause_plan
 
    rec = _active_record_or_warn(ctx)
    if rec is None:
        return
    try:
        new_rec = pause_plan(rec.plan_id)
    except ValueError as exc:
        ctx.presenter.show_info(f"Cannot pause: {exc}")
        return
    ctx.presenter.show_info(f"Paused: {new_rec.plan_id}")
 
 
def _cmd_plan_continue(ctx: "Context") -> DispatchResult:
    """Resume a paused plan and inject a continuation message.
 
    Compensation here is NON-DESTRUCTIVE, and deliberately different from
    ``--new`` (plan 152 R11-06). A ``--new`` orphan may be abort-retracted
    because nothing valid existed before it. A ``--continue`` failure must NEVER
    abort the plan: the user has a real, valid, paused plan, and terminalising
    it because queue assembly or reporting failed would destroy work and make it
    unretryable. Either the plan stays paused and retryable, or — if the durable
    paused->active transition already committed and only later steps failed —
    the result reports COMMITTED and the queue reconciles. It never acquires
    ``aborted_at`` because a message could not be displayed.
    """
    func = "cmd_plan_continue"
    from oc_chat.core.plans import continue_plan
 
    rec = _active_record_or_warn(ctx)
    if rec is None:
        return DispatchResult(kind=HANDLED_NO_TURN)
    try:
        new_rec = continue_plan(rec.plan_id)
    except ValueError as exc:
        # The durable transition did not happen; the plan is untouched.
        ctx.presenter.show_info(f"Cannot continue: {exc}")
        return DispatchResult(kind=HANDLED_NO_TURN)
 
    # The durable paused->active transition HAS committed past this point.
    prior_in_plan = ctx.in_plan
    prior_plan_description = ctx.plan_description
    prior_plan_files = list(ctx.plan_files_changed or [])
    message_index = len(ctx.session.messages)
 
    try:
        ctx.in_plan = True
        ctx.plan_description = new_rec.description
        ctx.plan_files_changed = []
        ctx.session.add_user(
            f"Continue work on plan `{new_rec.plan_id}.md`. "
            "Re-read it and proceed from where you left off."
        )
    except Exception as exc:
        _dbg("OC_CHAT.MAIN", func,
             f"SYSTEM_ERROR: continue unit failed after the durable transition: "
             f"{type(exc).__name__}: {exc}", 1, override=True)
        ctx.in_plan = prior_in_plan
        ctx.plan_description = prior_plan_description
        ctx.plan_files_changed = prior_plan_files
        del ctx.session.messages[message_index:]
        # NOT aborted. The plan is active on disk; say so and let the user retry.
        ctx.presenter.show_error(
            f"Plan {new_rec.plan_id} was resumed on disk, but the continuation "
            f"message could not be queued: {exc}. The plan is ACTIVE and "
            f"retryable — run /plan --continue again."
        )
        return DispatchResult(
            kind=INTERNAL_FAILURE, effect_disposition=EFFECT_COMMITTED,
        )
 
    warning = ""
    try:
        ctx.presenter.show_info(f"Resumed: {new_rec.plan_id}")
    except Exception as exc:
        warning = f"plan {new_rec.plan_id} resumed, but the confirmation could not be displayed: {exc}"
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: {warning}", 2, override=True)
    return DispatchResult(
        kind=HANDLED_TURN_QUEUED,
        effect_disposition=EFFECT_COMMITTED,
        queued_turn_token=message_index,
        warning=warning,
    )
 
 
def _cmd_plan_complete(ctx: "Context") -> None:
    """Mark the active plan complete and clear the session pointer."""
    from oc_chat.core.plans import complete_plan
 
    rec = _active_record_or_warn(ctx)
    if rec is None:
        return
    try:
        new_rec = complete_plan(rec.plan_id)
    except ValueError as exc:
        ctx.presenter.show_info(f"Cannot complete: {exc}")
        return
    ctx.session.active_plan_id = None
    ctx.session.dirty = True
    ctx.presenter.show_info(f"Completed: {new_rec.plan_id}")
 
 
def _cmd_plan_abort(ctx: "Context") -> None:
    """Abort the active plan and clear the session pointer."""
    from oc_chat.core.plans import abort_plan
 
    rec = _active_record_or_warn(ctx)
    if rec is None:
        return
    try:
        new_rec = abort_plan(rec.plan_id)
    except ValueError as exc:
        ctx.presenter.show_info(f"Cannot abort: {exc}")
        return
    ctx.session.active_plan_id = None
    ctx.session.dirty = True
    ctx.presenter.show_info(f"Aborted: {new_rec.plan_id}")
 
 
def _cmd_plan_list(ctx: "Context") -> None:
    """List all plans grouped by status, marking the active one."""
    from oc_chat.core.plans import list_plans
 
    records = list_plans()
    if not records:
        ctx.presenter.show_info("No plans found.")
        return
 
    active_id = ctx.session.active_plan_id
    groups: dict[str, list] = {"active": [], "paused": [], "complete": [], "aborted": []}
    for r in records:
        groups.setdefault(r.status, []).append(r)
 
    for status in ("active", "paused", "complete", "aborted"):
        bucket = groups.get(status, [])
        if not bucket:
            continue
        ctx.presenter.show_info(f"[{status}]")
        for r in bucket:
            marker = "*" if r.plan_id == active_id else " "
            desc = r.description if len(r.description) <= 60 else r.description[:57] + "..."
            ctx.presenter.show_info(f"  {marker} {r.plan_id}  {desc}")
 
 
def _cmd_plan_switch(ctx: "Context", target: str) -> None:
    """Change session.active_plan_id to a different (non-terminal) plan."""
    from oc_chat.core.plans import find_plan_by_id
    from oc_chat.core.plan_inventory import TERMINAL_STATUSES
 
    if not target:
        ctx.presenter.show_info("Usage: /plan --switch <plan-id-or-prefix>")
        return
    try:
        rec = find_plan_by_id(target)
    except ValueError as exc:
        ctx.presenter.show_info(f"Cannot switch: {exc}")
        return
    if rec is None:
        ctx.presenter.show_info(f"No plan found for {target!r}.")
        return
    if rec.status in TERMINAL_STATUSES:
        ctx.presenter.show_info(
            f"Cannot switch to {rec.plan_id} — status is {rec.status!r} (terminal). "
            "Choose an active or paused plan instead."
        )
        return
    ctx.session.active_plan_id = rec.plan_id
    ctx.session.dirty = True
    ctx.presenter.show_info(f"Switched to: {rec.plan_id} (status: {rec.status})")
 
 
@command("/options")
def cmd_options(ctx: "Context", raw: str) -> None:
    """Manage per-session generation options (temperature, top_k, etc.).
 
    /options            — show current options
    /options set <field> <value>   — set a field
    /options reset      — clear all options
    """
    from oc_chat.core.generation_options import GenerationOptions
 
    parts = raw.strip().split(None, 3)
    sub = parts[1].lower() if len(parts) > 1 else ""
 
    if sub == "reset":
        ctx.generation_options = GenerationOptions()
        ctx.session.generation_options = {}
        ctx.presenter.show_info("Generation options cleared (using provider defaults).")
        return
 
    if sub == "set":
        if len(parts) < 4:
            ctx.presenter.show_info("Usage: /options set <field> <value>")
            return
        fname, raw_val = parts[2].lower(), parts[3]
        if fname not in GenerationOptions.model_fields:
            valid = ", ".join(GenerationOptions.model_fields)
            ctx.presenter.show_info(f"Unknown option {fname!r}. Valid: {valid}")
            return
        try:
            ann = GenerationOptions.model_fields[fname].annotation
            # annotation may be e.g. float | None — pick the non-None type
            base = next((t for t in getattr(ann, "__args__", [ann]) if t is not type(None)), float)
            value = base(raw_val)
        except (ValueError, TypeError):
            ctx.presenter.show_info(f"Cannot convert {raw_val!r} for {fname}")
            return
        ctx.generation_options = ctx.generation_options.model_copy(update={fname: value})
        ctx.session.generation_options = ctx.generation_options.model_dump(exclude_none=True)
        ctx.presenter.show_info(f"Set {fname} = {value}")
        return
 
    # Default: show current options
    opts = ctx.generation_options
    if not opts.any_set():
        ctx.presenter.show_info("No generation options set (using provider defaults).")
    else:
        lines = [f"  {k} = {v}" for k, v in opts.model_dump(exclude_none=True).items()]
        ctx.presenter.show_info("Generation options:\n" + "\n".join(lines))
 
 
@command("/status")
def cmd_status(ctx: "Context", _: str) -> None:
    """Display the current plan's step-by-step status as JSON."""
    from oc_chat.tools.planner import get_status
    import json
    print(json.dumps(get_status(), indent=2, ensure_ascii=False))
 
 
@command("/stop")
def cmd_stop(ctx: "Context", _: str) -> None:
    """Halt the active plan."""
    from oc_chat.tools.planner import stop_plan
    stop_plan()
    ctx.presenter.show_info("Active plan stopped.")
 
 
# ── Phase 4: cache / budget / react caps ─────────────────────────────────────
 
@command("/cache")
def cmd_cache(ctx: "Context", _raw: str) -> None:
    """Show prompt disk-cache stats for this session."""
    from oc_chat.core.cache_usage import cache_hit_ratio_pct
 
    s = ctx.session
    st = s.stats
    hit, miss = st.prompt_cache_hit_tokens, st.prompt_cache_miss_tokens
    pct = cache_hit_ratio_pct(hit, miss)
    print("\n── Prompt cache (session cumulative) ──")
    print(f"  cache_hit_tokens:   {hit:,}")
    print(f"  cache_miss_tokens:  {miss:,}")
    if pct is not None:
        print(f"  hit ratio:          {pct:.1f}%")
    else:
        print("  hit ratio:          (no cache data yet)")
 
    snaps = list(getattr(s, "cache_turn_snapshots", []) or [])
    if snaps:
        print("\n── Recent turns (newest last) ──")
        tail = snaps[-12:]
        for i, row in enumerate(tail, start=len(snaps) - len(tail) + 1):
            h, m = int(row.get("hit", 0)), int(row.get("miss", 0))
            p = int(row.get("prompt", 0))
            r = cache_hit_ratio_pct(h, m)
            rs = f"{r:.0f}%" if r is not None else "—"
            print(f"  turn {i:>2}:  hit={h:>6}  miss={m:>6}  ratio={rs:>5}  prompt≈{p:,}")
    else:
        print("\n(no per-turn cache snapshots yet — complete a model turn first)")
    print()
 
 
@command("/trace")
def cmd_trace(ctx: "Context", raw: str) -> None:
    """Render this session's persisted M2 event stream (M2.6 /trace)."""
    from oc_chat.core.trace_view import format_trace, read_session_events
 
    parts = raw.split(maxsplit=1)
    limit = 200  # default tail of most-recent events
    if len(parts) > 1:
        arg = parts[1].strip().lower()
        if arg in ("all", "full"):
            limit = None
        elif arg.isdigit():
            limit = max(1, int(arg))
    events = read_session_events(ctx.session)
    print("\n" + format_trace(events, limit=limit) + "\n")
 
 
@command("/packet")
def cmd_packet(ctx: "Context", _raw: str) -> None:
    """Write a draft merge-readiness packet (packet-draft/v0) for this session (M2.6 /packet)."""
    import json
    from datetime import datetime
 
    from oc_chat.core.event_bootstrap import _slug, active_event_dir
    from oc_chat.core.trace_view import build_packet_draft, read_session_events
 
    s = ctx.session
    events = read_session_events(s)
    packet = build_packet_draft(s, events)
    out_dir = active_event_dir(s)
    out_dir.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    path = out_dir / f"{_slug(s.name)}-{stamp}.packet.json"
    path.write_text(json.dumps(packet, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
    print(f"\n[packet-draft written to {path}]")
    print(f"  events: {sum(packet['event_counts'].values())}  turns: {packet['turn_count']}\n")
 
 
def _show_resolved_output_caps(ctx: "Context") -> None:
    """Print the completion caps the profile actually resolves to.
 
    The caps derive from the model's declared ``max_output_tokens`` and the
    operator ceiling, so the profile name alone does not tell you what you
    get. Printing the numbers also explains why ``deep`` can equal
    ``balanced``: a bucket already at the ceiling has nowhere to scale.
    """
    func = "_show_resolved_output_caps"
    try:
        from oc_chat.core.config import get_output_tokens_ceiling
        from oc_chat.core.step_routing import resolve_output_tokens
 
        model = ctx.session.model
        profile = getattr(ctx.session, "token_profile", "balanced")
        rows = [
            ("planning / final answer", "planning"),
            ("tool iterations", "thinking"),
            ("after a plain read", "simple"),
        ]
        caps = ", ".join(
            f"{label} {resolve_output_tokens(ctx.model_manager, model, bucket, profile):,}"
            for label, bucket in rows
        )
        ctx.presenter.show_info(
            f"Completion caps for {model} at '{profile}': {caps} "
            f"(operator ceiling {get_output_tokens_ceiling():,})"
        )
    except Exception as e:  # noqa: BLE001 — an informational line must never abort the command
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
 
 
def _show_session_cost(ctx: "Context") -> None:
    """Print the plan-237 money block: spend, checkpoints, cap, and the knob.
 
    Everything an operator needs to answer "how much has this cost, and what
    stops it?" without reading the source — including the residual, because a
    total that silently excludes unpriced samples is a number that lies by
    omission.
    """
    func = "_show_session_cost"
    try:
        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 (
            cost_prompts_suppressed, tool_calls_auto_continue,
        )
 
        cost = ctx.session.stats.cost
        cap = resolve_session_cost_hard_cap_usd()
        thresholds = resolve_cost_thresholds_usd()
        crossed = [t for t in thresholds if cost.has_fired(t)]
        pending = [t for t in thresholds if not cost.has_fired(t)]
        _plural = cost.unpriced_samples != 1
        unpriced = (
            f" ({cost.unpriced_samples} sample{'s' if _plural else ''} could not "
            f"be priced and {'are' if _plural else 'is'} NOT included, so real "
            f"spend may be higher)"
            if cost.unpriced_samples else ""
        )
        latches = []
        if tool_calls_auto_continue():
            latches.append("tool-call checkpoint: not asking again this session")
        if cost_prompts_suppressed():
            latches.append("cost prompts: suppressed this session (alerts and the "
                           "hard cap still apply)")
        lines = [
            f"Session spend: ${cost.priced_usd:.2f}{unpriced}",
            f"  checkpoints crossed : "
            + (", ".join(f"${t:.2f}" for t in crossed) if crossed else "none"),
            f"  checkpoints ahead   : "
            + (", ".join(f"${t:.2f}" for t in pending) if pending else "none"),
            f"  hard cap            : ${cap} - at or above it no further provider "
            f"call is made this session",
            f"  raise the cap       : "
            + (f"{_COST_CEILING_ENV_VAR}=<whole dollars>, out of band only "
               f"(a settings file can only LOWER it)"
               if _COST_CEILING_ENV_VAR else "no out-of-band raise path"),
        ]
        lines.extend(f"  {note}" for note in latches)
        ctx.presenter.show_info("\n".join(lines))
    except Exception as e:  # noqa: BLE001 — an informational line must never abort the command
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
 
 
@command("/budget")
def cmd_budget(ctx: "Context", raw: str) -> None:
    """Show or set the session token profile, and show the session money bounds.
 
    Plan 237 (CAPS-01 P7) asked for a ``/budget`` that prints session spend, the
    checkpoints already crossed, the hard cap and the env variable that raises
    it. ``/budget`` was NOT a new command — it has selected the token profile
    since phase 4 — so rather than take the name and break that, the bare form
    now prints BOTH blocks (profile + money), ``/budget cost`` prints the money
    block alone, and ``/budget cheap|balanced|deep`` is unchanged.
    """
    parts = raw.split(maxsplit=1)
    s = ctx.session
    if len(parts) < 2:
        ctx.presenter.show_info(
            f"Token profile: {getattr(s, 'token_profile', 'balanced')} "
            f"(options: {', '.join(TOKEN_PROFILES)}). Example: /budget cheap"
        )
        _show_resolved_output_caps(ctx)
        _show_session_cost(ctx)
        return
    val = parts[1].strip().lower()
    if val == "cost":
        _show_session_cost(ctx)
        return
    if val not in TOKEN_PROFILES:
        ctx.presenter.show_error(
            f"Unknown profile. Use one of: {', '.join(TOKEN_PROFILES)} — "
            f"or 'cost' for this session's spend against the hard cap."
        )
        return
    s.token_profile = val
    s.dirty = True
    ctx.presenter.show_info(f"Token profile set to: {val}")
    _show_resolved_output_caps(ctx)
 
 
@command("/react")
def cmd_react(ctx: "Context", raw: str) -> None:
    """Inspect or override the ReAct iteration cap for this session."""
    parts = raw.split(maxsplit=1)
    s = ctx.session
    cap = getattr(s, "max_react_iterations", None)
    eff = cap if isinstance(cap, int) and cap >= 1 else DEFAULT_MAX_REACT_ITERATIONS
    if len(parts) < 2:
        ctx.presenter.show_info(
            f"ReAct cap: {eff} (default {DEFAULT_MAX_REACT_ITERATIONS}). "
            f"Set with /react 15  or  /react reset"
        )
        return
    arg = parts[1].strip().lower()
    if arg == "reset":
        s.max_react_iterations = None
        s.dirty = True
        ctx.presenter.show_info(f"ReAct cap reset to default ({DEFAULT_MAX_REACT_ITERATIONS}).")
        return
    try:
        n = int(parts[1].strip())
    except ValueError:
        ctx.presenter.show_error("Usage: /react [1-99|reset]")
        return
    if not 1 <= n <= 99:
        ctx.presenter.show_error("ReAct cap must be between 1 and 99.")
        return
    s.max_react_iterations = n
    s.dirty = True
    ctx.presenter.show_info(f"ReAct cap set to {n} iterations for this session.")
 
 
# ── One-shot ask (K.5a) ───────────────────────────────────────────────────────
 
@command("/ask")
def cmd_ask(ctx: "Context", raw: str) -> None:
    """Issue a single non-streaming question without touching the session log.
 
    No ReAct iteration, no tool schemas in the system prompt (the model
    cannot dispatch tool calls in one-shot mode), and ``session.messages``
    /``session.stats`` are left untouched. Phase K.5b wires the
    post-answer menu onto this handler.
    """
    func = "cmd_ask"
    parts = raw.split(maxsplit=1)
    if len(parts) < 2 or not parts[1].strip():
        ctx.presenter.show_error("Usage: /ask <question>")
        return
    question = parts[1].strip()
    _dbg("OC_CHAT.MAIN", func, f"one-shot question_len={len(question)}", 2)
 
    system_prompt = (
        getattr(ctx.session, "system_prompt", "")
        or "You are a helpful assistant."
    )
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": question},
    ]
 
    try:
        response = ctx.client.one_shot(
            messages,
            ctx.session.model,
            effort="low",
            max_tokens=4096,
        )
    except Exception as e:  # noqa: BLE001 — surface every error uniformly
        _dbg("OC_CHAT.MAIN", func, f"one_shot raised {type(e).__name__}: {e}", 1, override=True)
        ctx.presenter.show_error(f"/ask failed: {type(e).__name__}: {e}")
        return
 
    ctx.presenter.show_content(response)
    ctx.presenter.show_info("(one-shot — no session log written)")
    _dbg("OC_CHAT.MAIN", func, f"one-shot response_len={len(response)}", 2)
 
    # K.5b — post-answer menu (5 options: follow-up / save / copy / new / quit).
    from oc_chat.core.ask_menu import present_post_answer_menu
    chosen = present_post_answer_menu(ctx, question, response)
    _dbg("OC_CHAT.MAIN", func, f"post-answer option chosen: {chosen}", 2)
 
 
# ── Help ───────────────────────────────────────────────────────────────────────
 
@command("/help")
def cmd_help(ctx: "Context", _: str) -> None:
    """Print the full command listing, session subcommands, and registered tools."""
    tools = sorted(ctx.tool_registry) if ctx.tool_registry else None
    ctx.presenter.show_help(COMMAND_HELP, tools=tools)
    # The /session subcommand details are still printed plainly — they're a
    # fixed text block that doesn't benefit from a table.  PR 1.B+ may move
    # these into the presenter too.
    _session_help(ctx)