main.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/main.py
 
"""
Main entry point for the OC Chat REPL — wiring shell only.
 
Purpose
-------
Parse argv, initialise diagnostics, select a presenter, run the
storage migration, drive the startup session selection menu, build
an :class:`~oc_chat.core.agent.Agent` and a slash-command
:class:`~oc_chat.core.commands.Context`, and run the read-eval loop
that ferries user input into ``agent.run_turn(...)`` and slash
commands into :func:`dispatch`. All the per-turn ReAct logic, tool
dispatch, registry construction, startup wizard, CHANGELOG updates,
and presentation formatting now live in their own modules; this
file is the orchestration shell that wires them together.
 
Responsibilities
----------------
- :func:`main` — the wiring sequence above.
- Slash-command loop body: dispatch slash commands, otherwise
  drive ``agent.run_turn(user_input)``; on ``/goal`` completion,
  trigger :func:`append_session_log` against the agent's session.
 
The decomposition lives at:
- :mod:`oc_chat.core.react_loop` (G.6.A) — ``run_turn`` + tool dispatch.
- :mod:`oc_chat.core.tool_registry` (G.6.A) — ``build_tool_registry``.
- :mod:`oc_chat.core.startup` (G.6.B) — ``startup_session_prompt``.
- :mod:`oc_chat.core.session_log` (G.6.B) — ``append_session_log``.
- :mod:`oc_ui.terminal.formatting` (G.6.B) — presentation formatters.
- :mod:`oc_chat.cli` (G.6.C) — argparse, presenter selection,
  terminal clear.
- :mod:`oc_chat.core._version` (G.6.C) — version-string resolver.
- :mod:`oc_chat.core.agent` (G.6.C) — :class:`Agent` and
  :class:`AgentConfig`, the runtime building block this shell wires.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — REPL lifecycle events.
    L3 — semantic details.
    L4 — deep tracing (currently unused).
 
Contracts
---------
- ``oc_diagnostics.init(intercept_streams=False,
  intercept_exceptions=False)`` runs once at the top of ``main()``.
  Stream intercept is disabled so the REPL keeps clean stdout/stderr.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.cli (argparse, presenter selection, terminal clear)
- oc_chat.core.agent (Agent + AgentConfig — runtime building block)
- oc_chat.core.session_log (post-goal SESSION_LOG.md writer)
- oc_chat.core.commands (slash-command dispatch)
- oc_chat.core.persistence (migrate_legacy_layout)
- oc_chat.core.provider_plugin (ProviderPlugin ABC and registry singleton)
- oc_chat.core.startup (startup_session_prompt — interactive bootstrap)
- oc_chat.core._version (resolve_version — banner/CHANGELOG)
- providers.deepseek (DeepSeek provider plugin — default backend)
- oc_ui.terminal.input (PromptDriver, TurnStats, prompt_prefix builder)
- oc_chat.core.config_validation (validate_runtime_config and format_config_issue - the startup configuration check and its messages)
- oc_chat.core.session (ChatSession, TokenStats and the metadata and system-prompt helpers - rebuilding a session for the crash auto-reload)
- oc_chat.core.session_transition (IterationStatsSink and the turn-phase marks - the process-owned stats sink and the deferred Ctrl+C)
- oc_chat.tools.ask_user (bind_interaction - binds the ask_user tool to the live prompt driver)
- providers.ollama (Ollama provider plugin - registered on import)
- providers.openai_compat (the seven OCON lane providers - registered on import)
- oc_chat.core.batch_runner (lazy; run_batch_turn - the headless --input / --output turn)
- oc_chat.core.cancellation (lazy; signal_tool_cancel - stops in-flight tools on the second Ctrl+C)
- oc_chat.core.deployment_shape (lazy; check_deployment_shape and DeploymentShapeError - refuses a sandbox that overlaps an importable root)
- oc_chat.core.endpoint_registry (lazy; EndpointRegistry - restores the session's last-used endpoint)
- oc_chat.core.event_bootstrap (lazy; attach_default_writer - the NDJSON event writer on the default bus)
- oc_chat.core.external_content_gate (lazy; blast_radius_banner - the permissive profile's startup banner)
- oc_chat.core.generation_options (lazy; GenerationOptions - restores the session's persisted generation options)
- oc_chat.core.lane_switch (lazy; switch_lane - applies a lane chosen in the opening menu to the Agent)
- oc_chat.core.lane_tree (lazy; serving_host_of - names the host that served a turn)
- oc_chat.core.presenter_events (lazy; attach_presenter_bridge - renders the approval and sub-agent events in the terminal)
- oc_chat.core.prompt_layers (lazy; _git_branch - the branch in the startup anchor banner)
- oc_chat.core.risk_profile (lazy; resolve_risk_profile - the active risk profile for the deployment-shape check)
- oc_chat.core.session_mode (lazy; the --auto launch flag's switch into auto mode)
- oc_chat.core.settings (lazy; init_runtime_settings - resolves the composition mode once from the settings file and the CLI)
- oc_chat.tools.filesystem (lazy; _sandbox_root and _sandbox_root_info - the sandbox root for the anchor banner and the deployment-shape check)
- oc_ui.presenter_base (lazy; INTERRUPT_NOTICES - the stderr fallback for a Ctrl+C notice)
 
"""
 
from oc_diagnostics import init as _diagnostics_init, _dbg
import os
import signal
import sys
 
from oc_chat.cli import build_arg_parser, clear_terminal, select_presenter
from oc_chat.core._version import __version__, resolve_version
 
# Plan 233 (P2): the module-level ``EnhancedPresenter`` import that used
# to sit here was REDUNDANT — nothing in this module referenced the name.
# It also pulled ``rich`` transitively (enhanced_presenter -> rich_presenter)
# at import time, defeating cli.py's ImportError-to-plain fallback. The one
# place that builds an EnhancedPresenter, ``cli.select_presenter``, already
# imports it lazily inside the branch that owns that fallback.
from oc_chat.core.agent import Agent, AgentConfig
from oc_chat.tools import ask_user
from oc_chat.core.session_log import append_session_log
from oc_chat.core.commands import (
    Context, EFFECT_UNKNOWN, HANDLED_NO_TURN, HANDLED_TURN_QUEUED,
    INTERNAL_FAILURE, dispatch, refresh_presenter_bindings,
)
from oc_chat.core.session_transition import (
    IterationStatsSink, consume_pending_sigint, mark_between_turns, mark_in_turn,
)
from oc_chat.core.config_validation import (
    format_config_issue,
    validate_runtime_config,
)
from oc_chat.core.persistence import (
    DEFAULT_SESSIONS_DIR,
    apply_tail_recovery,
    detect_incomplete_tail,
    load_chat_log,
    load_session_metadata,
    migrate_legacy_layout,
    read_crash_signal,
    write_clean_exit_marker,
)
from oc_chat.core import provider_plugin
from oc_chat.core.session import (
    ChatSession,
    TokenStats,
    VARIABLE_CONTEXT_SENTINEL,
    apply_session_metadata_fields,
    ensure_system_includes_tool_reference,
    _migrate_legacy_system_prompt,
)
from oc_chat.core.startup import (
    _inject_variable_context,
    migrate_session_model,
    refresh_variable_context,
    startup_session_prompt,
)
import providers.deepseek  # noqa: F401 — import triggers DeepSeekProvider.register()
import providers.ollama  # noqa: F401 — import triggers OllamaProvider.register()
import providers.openai_compat  # noqa: F401 — import registers the seven OCON lane providers (plan 215)
from oc_ui.terminal.input import PromptDriver, TurnStats
 
 
_VERSION = resolve_version()
 
 
# Phase O3e: module-level holder for the currently-active Agent so the
# SIGINT handler (installed at the top of main()) can find and flip its
# CancellationToken even though the handler is installed before the
# Agent is constructed. ``register_active_agent`` is called once just
# after ``Agent(...)`` construction below.
_active_agent_ref: dict[str, "object | None"] = {"agent": None}
 
 
def register_active_agent(agent: "object | None") -> None:
    """Register *agent* as the SIGINT-flippable target (Phase O3e).
 
    Public for tests; main() also calls this after building its Agent.
    """
    _active_agent_ref["agent"] = agent
 
 
def _announce_interrupt(agent: "object | None", stage: str) -> None:
    """Announce a Ctrl+C *stage* through the presenter, falling back to stderr.
 
    Plan 188 T5. The handler used to write this notice inline, which meant the
    component that OWNS the terminal — the presenter, possibly mid-``Live`` —
    had no say in how or where it landed. It now goes through
    :meth:`~oc_ui.presenter_base.Presenter.show_interrupt_notice`, whose
    default is the same stderr write, so a presenter that does not override it
    behaves exactly as before.
 
    Two properties matter because this runs inside a signal handler:
 
    * **it never raises** — a broken presenter must not turn Ctrl+C into a
      crash, so any failure degrades to the stderr notice and the
      cancellation below still proceeds;
    * **it stays one short write** — no prompting, no blocking, no
      renderable pipeline (see the hook's signal-safety contract).
    """
    func = "_announce_interrupt"
    presenter = getattr(agent, "presenter", None) if agent is not None else None
    if presenter is not None:
        try:
            presenter.show_interrupt_notice(stage)
            return
        except Exception as exc:  # noqa: BLE001 — a signal handler must not raise
            _dbg("OC_CHAT.MAIN", func,
                 f"presenter notice failed: {type(exc).__name__}: {exc}", 2)
    from oc_ui.presenter_base import INTERRUPT_NOTICES
    text = INTERRUPT_NOTICES.get(stage)
    if text is None:
        return
    sys.stderr.write(text)
    sys.stderr.flush()
 
 
def _o3_sigint_handler(signum, frame):
    """Two-stage cooperative Ctrl+C (Phase O3e + TUI-polish).
 
    During an active turn the Agent owns a ``CancellationToken``:
 
    * **1st Ctrl+C** → request a cooperative interrupt and NOTHING ELSE
      (operator decision 2026-08-12): the in-flight tool keeps running, and the
      turn is asked "Stop the current action? (y/N)" at its next safe
      checkpoint — which arrives after that tool returns. "n" continues with
      nothing lost; nothing was cancelled before the answer. A one-line notice
      says the current tool keeps running and a second press force-stops.
    * **2nd Ctrl+C** (an interrupt is already pending, or a hard cancel is set)
      → force: flip the hard-cancel flag, signal the process-wide tool-cancel
      flag (the ``[cancelled by user]`` / returncode −2 machinery in
      ``tools/_proc.py`` — so the child dies cleanly, not orphaned), and raise
      ``KeyboardInterrupt`` so blocking I/O unwinds immediately.
 
    **Plan 188 T5.** Both notices go through :func:`_announce_interrupt` →
    the active presenter's ``show_interrupt_notice``, instead of the raw
    ``stderr`` write this handler used to do inline. The mechanism above is
    unchanged; what changed is that the component owning the terminal (Rich,
    possibly mid-``Live``) now chooses how and where the message lands, with
    the previous stderr write as the base-class default and the fallback.
 
    Between turns (no active token) Ctrl+C raises ``KeyboardInterrupt`` as before
    — the prompt's own handler cancels the input line without exiting OC Chat.
 
    **Transition deferral (plan 152 R09-01).** While a session transition is
    committing, the interrupt is RECORDED and the handler returns without
    raising. That is what makes the commit block genuinely uninterruptible:
    ``dispatch`` is covered by no ``KeyboardInterrupt`` guard, and
    ``KeyboardInterrupt`` is a ``BaseException`` an ``except Exception`` wrapper
    misses, so a press landing between two "no-fail" installs would otherwise
    strand a mixed tuple with no recovery. Repeated presses coalesce to one
    pending flag; it is consumed exactly once at the next prompt boundary or
    turn launch.
    """
    # FIRST statement, and read LIVE through the getter — importing the flag by
    # value would snapshot False at import time and defeat the deferral entirely.
    from oc_chat.core import session_transition as _st
    if _st.transition_in_progress():
        _st.note_pending_sigint()
        return
 
    agent = _active_agent_ref["agent"]
    token = getattr(agent, "cancellation", None) if agent is not None else None
    if token is not None:
        # The Ctrl+C handler owns the process-wide tool-cancel flag so in-flight
        # tools (running in worker threads, which cannot be force-killed) stop at
        # their next poll point.
        from oc_chat.core.cancellation import signal_tool_cancel
        if token.interrupt_requested or token.is_cancelled:
            token.cancel()  # 2nd press — hard stop
            signal_tool_cancel()  # ensure tools stop too
            # Announce BEFORE the raise: the KeyboardInterrupt unwinds blocking
            # I/O immediately, so a notice after it would never be reached.
            _announce_interrupt(agent, "force")
            raise KeyboardInterrupt
        # 1st press — cooperative request ONLY (operator decision 2026-08-12,
        # 188-T5 v2): nothing is cancelled until the checkpoint's y/N confirms.
        # The in-flight tool keeps running — a stray Ctrl+C costs nothing, and
        # "n" resumes with the tool result intact. The tool-cancel flag now
        # belongs exclusively to the 2nd press (and to nothing else).
        token.request_interrupt()
        # Plan 188 T5: the presenter owns the terminal, so it owns the notice
        # (default behaviour is the same stderr write this line used to do).
        _announce_interrupt(agent, "cancel")
        return
    raise KeyboardInterrupt
 
 
def _active_sessions_dir() -> str:
    return os.environ.get("OC_CHAT_SESSIONS_DIR", DEFAULT_SESSIONS_DIR)
 
 
def _read_startup_input(
    prompt_driver, prompt_text: str, default: str, *, presenter
) -> str:
    """Read one answer at a pre-session prompt; a Ctrl+C asks before deciding.
 
    Plan 220 P3 / TUI-02 finding 3, at ``main``'s own two pre-session
    prompts: the crash auto-reload and the incomplete-tail recovery. Both
    used to fold a cancelled read into *default*, so a Ctrl+C here
    auto-reloaded a session, or dropped a saved tail, that nobody asked
    for — and one more press carried the operator into the REPL.
 
    EOF still yields *default*: a closed stdin is not an operator choice
    and there is nobody left to confirm with. A cancel is the opposite —
    somebody is at the keyboard — so it goes to
    :func:`~oc_chat.core.startup.confirm_quit_or_continue`, which either
    exits or returns to re-ask this same question.
    """
    from oc_chat.core.startup import confirm_quit_or_continue
 
    while True:
        try:
            value = prompt_driver.read_user_input(None, prompt_text=prompt_text)
        except EOFError:
            return default
        except KeyboardInterrupt:
            confirm_quit_or_continue(presenter, prompt_driver)
            continue
        if value is None:
            # The driver's report of a Ctrl+C at a menu prompt — an
            # interrupt, not an empty line (see the docstring above).
            confirm_quit_or_continue(presenter, prompt_driver)
            continue
        return value.strip().lower()
 
 
def _show_serving_host(presenter, client) -> None:
    """Name the host that actually served the turn — where the lane says so.
 
    Plan 220 P4 / TUI-02 finding 4. A routed lane picks its upstream per
    request, so the serving host is knowable only from the response and
    only after the fact (the plan-215 catalogue's provenance caveat). It is
    therefore reported here, after the turn, and never claimed in advance
    at selection — where the model VENDOR is shown instead.
 
    A lane reports one by leaving a mapping on its provider's
    ``last_response_report``; a lane that reports nothing renders nothing.
    Duck-typed exactly as the plan-215 lane panel is, so ``main`` stays
    free of any import of an optional provider bundle: a bundle that
    learns to report a serving host starts showing one here with no change
    to this function.
    """
    func = "_show_serving_host"
    from oc_chat.core.lane_tree import serving_host_of
 
    provider = getattr(client, "provider", None)
    host = serving_host_of(getattr(provider, "last_response_report", None))
    if not host:
        return
    lane = getattr(provider, "name", "") or "lane"
    presenter.show_info(f"[{lane} — served by {host}]")
    _dbg("OC_CHAT.MAIN", func, f"lane={lane} reported serving host {host!r}", 3)
 
 
def _load_session_by_name_for_startup(
    name: str,
    presenter,
    tool_schemas: list | None,
    sessions_dir: str,
) -> ChatSession:
    """Load a named session without entering the interactive picker."""
    func = "_load_session_by_name_for_startup"
    meta = load_session_metadata(name, sessions_dir=sessions_dir)
    session_name = meta.get("session_name") or name
 
    # Plan 135 §3.5/§4.7: set the persisted model via the constructor — model
    # writes go through ChatSession.set_model, never a raw attribute assignment.
    model = meta.get("model")
    session = (
        ChatSession(name=session_name, model=model)
        if model else ChatSession(name=session_name)
    )
    session.reasoning_effort = meta.get("reasoning_effort", session.reasoning_effort)
    session.chat_logs = meta.get("chat_logs", [])
    session.stats = TokenStats.from_dict(meta.get("stats", {}))
    session.created = meta.get("created") or (
        session_name.split("_", 1)[1] if "_" in session_name else "unknown"
    )
    session.ui_prefs = dict(meta.get("ui", {}) or {})
    apply_session_metadata_fields(session, meta)
    session.dirty = False
 
    try:
        log_data = load_chat_log(session_name, sessions_dir=sessions_dir)
        session.messages = log_data.get("messages", session.messages)
    except FileNotFoundError as exc:
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {session_name}: {exc}", 3)
        presenter.show_info(
            f"[Note: Session '{session_name}' has no chat log on disk. "
            "Starting with fresh context.]"
        )
    except Exception as exc:
        _dbg(
            "OC_CHAT.MAIN",
            func,
            f"SYSTEM_ERROR: parse {session_name}: {type(exc).__name__}: {exc}",
            1,
            override=True,
        )
        session.load_failed = True
        presenter.show_error(
            f"[Warning: Could not parse chat log for session '{session_name}' "
            f"({exc}). On-disk chat log will NOT be overwritten on save until "
            "reload succeeds.]"
        )
 
    # Plan 135 §4.3: migrate any legacy embedded-schema messages[0], then ensure
    # the compact tool reference is present (both idempotent).
    _migrate_legacy_system_prompt(session, tool_schemas)
    ensure_system_includes_tool_reference(session, tool_schemas)
 
    # Plan 135 §3.5: refresh the loaded variable context IN PLACE (preserves all
    # conversation history); inject only when the saved session predates it.
    has_varctx = any(
        m.get("role") == "user"
        and (m.get("content") or "").startswith(VARIABLE_CONTEXT_SENTINEL)
        for m in session.messages
    )
    if has_varctx:
        refresh_variable_context(session)
    else:
        _inject_variable_context(session)
 
    presenter.show_info(
        f"Loaded session '{session_name}' — "
        f"{session.stats.message_count} messages, "
        f"{session.stats.tool_call_count} tool calls, "
        f"{session.stats.total_tokens:,} tokens "
        f"(model: {session.model})"
    )
    return session
 
 
def _maybe_autoreload_crashed_session(
    agent: Agent,
    presenter,
    prompt_driver,
    sessions_dir: str,
) -> ChatSession | None:
    signal_data = read_crash_signal(sessions_dir=sessions_dir)
    if signal_data is None:
        return None
 
    name = signal_data["last_session_name"]
    last_modified = signal_data["last_modified"]
    presenter.show_info(
        "OC Chat may have exited unexpectedly during your last session "
        f"('{name}', last modified {last_modified})."
    )
 
    while True:
        choice = _read_startup_input(
            prompt_driver,
            "Auto-reload that session? [Y/n]: ",
            "y",
            presenter=presenter,
        )
        if choice in ("", "y", "yes"):
            return _load_session_by_name_for_startup(
                name,
                presenter,
                agent.tool_schemas,
                sessions_dir,
            )
        if choice in ("n", "no"):
            return None
        presenter.show_info("Please enter Y or N.")
 
 
def _recover_incomplete_session_tail(session: ChatSession, presenter, prompt_driver) -> bool:
    report = detect_incomplete_tail(session.messages)
    if report is None:
        return False
 
    presenter.show_info("Session appears to have an incomplete saved turn.")
    presenter.show_info(report["summary"])
 
    while True:
        choice = _read_startup_input(
            prompt_driver,
            "Recover by [D]ropping tail, [K]eeping tail, or opening [F]resh? [D/k/f]: ",
            "d",
            presenter=presenter,
        )
        if choice in ("", "d", "drop"):
            recovery_choice = "drop"
            break
        if choice in ("k", "keep"):
            recovery_choice = "keep"
            break
        if choice in ("f", "fresh"):
            recovery_choice = "fresh"
            break
        presenter.show_info("Please enter D, K, or F.")
 
    before_count = len(session.messages)
    recovered_messages = apply_tail_recovery(session.messages, recovery_choice)
    session.messages = recovered_messages
    if recovery_choice in ("drop", "fresh") and len(recovered_messages) != before_count:
        session.dirty = True
 
    if recovery_choice == "drop":
        removed = before_count - len(recovered_messages)
        presenter.show_info(f"Session recovered: dropped {removed} incomplete tail message(s).")
    elif recovery_choice == "fresh":
        presenter.show_info("Session recovered: opened fresh context for this session.")
    else:
        presenter.show_info("Session recovery skipped: incomplete tail kept unchanged.")
 
    return True
 
 
def _show_config_validation(presenter, result) -> bool:
    for issue in result.warnings:
        presenter.show_info(format_config_issue(issue))
    for issue in result.errors:
        presenter.show_error(format_config_issue(issue))
    return result.ok
 
 
def _select_presenter_once(args) -> "object":
    """Pick the presenter exactly once for this :func:`main` invocation.
 
    Delegates to :func:`oc_chat.cli.select_presenter`, which already
    honours both ``--ui`` and the legacy ``--enhanced-ui`` flag (the
    latter via ``getattr(args, "enhanced_ui", False)``). The named
    indirection makes the single-call invariant explicit at the call
    site — :func:`main` previously selected the presenter twice (once
    for config validation, once after) which silently dropped the
    ``--enhanced-ui`` choice. Keep this helper as the only path that
    constructs a presenter inside :func:`main`.
    """
    return select_presenter(args)
 
 
def _safe_presenter_close(presenter) -> None:
    """Call ``presenter.close()`` if present; swallow + log on failure.
 
    Phase TUI runtime-prep. The future TUI presenter needs reliable
    teardown so it can restore the ANSI scroll region, leave the
    alt-screen, and drop its key bindings — even when :func:`main` is
    exiting via an unhandled exception. Extracted as a helper so:
 
    1. :func:`main`'s ``finally`` block stays tidy.
    2. The contract is unit-testable in isolation
       (``test_main_cleanup.py``).
 
    ``hasattr`` keeps the older :class:`ConsolePresenter` (which has
    no ``close()``) compatible. ``try/except Exception`` keeps the
    finalizer robust when cleanup itself fails mid-exception-handling.
    """
    func = "_safe_presenter_close"
    if not hasattr(presenter, "close"):
        return
    try:
        presenter.close()
    except Exception as exc:
        _dbg(
            "OC_CHAT.MAIN",
            func,
            f"SYSTEM_ERROR: presenter.close: {type(exc).__name__}: {exc}",
            1,
            override=True,
        )
 
 
def _assert_stdin_is_tty_or_input_mode(args) -> None:
    """Refuse to enter the interactive REPL with redirected stdin.
 
    The REPL reads from ``sys.stdin``; if stdin is a regular file (not a
    TTY) and the user didn't ask for batch mode via ``--input``/``--output``,
    each newline in that file becomes a separate user prompt and the loop
    only exits on EOF. Almost always a mistake (e.g. accidentally pointing
    stdin at ``.oc_chat/launch/agent-*.md`` or another launch artifact).
 
    Bypasses on any of: stdin is a TTY, ``--input PATH`` was passed, or
    ``--output PATH`` was passed. See plan 115 ADR-3 — no env-var override.
    """
    if not sys.stdin.isatty() and args.input is None and args.output is None:
        print(
            "OC_Chat: stdin is not a TTY but no --input was passed. "
            "To run a prompt from a file, use --input PATH (or --input - for stdin). "
            "The interactive REPL refuses redirected stdin to prevent accidental "
            "file-as-prompt loops.",
            file=sys.stderr,
        )
        raise SystemExit(2)
 
 
def _anchor_banner_line(root: object, source: str, branch: "str | None") -> str:
    """Compose the one-line pre-menu anchor banner (plan 173 item 2 / F19).
 
    ``sandbox root: <abs> (<source label>) | branch: <branch or ->``. ASCII-only
    (F4): the separator is `` | ``, never a non-cp1252 middot, so it never crashes
    a stock Windows console.
    """
    return f"sandbox root: {root} ({source}) | branch: {branch or '-'}"
 
 
def _emit_startup_anchor_banner(presenter, args) -> None:
    """Print the pre-menu anchor banner; on a ``--workdir`` vs resolved-root
    disagreement, warn and require an explicit y/N before the menu (plan 173 / F19).
 
    F19: three consecutive mis-anchored launches (a resumed-session persisted-root
    override + an omitted ``--workdir``) preceded the successful one because the
    anchor block existed only in the model's context, discovered after pasting.
    This surfaces ``root · source · branch`` to the operator BEFORE the menu.
    """
    func = "_emit_startup_anchor_banner"
    from pathlib import Path
    from oc_chat.tools.filesystem import _sandbox_root_info
    from oc_chat.core.prompt_layers import _git_branch
 
    root, source = _sandbox_root_info()
    branch = _git_branch(root)
    presenter.show_info(_anchor_banner_line(root, source, branch))
 
    workdir = getattr(args, "workdir", None)
    if not workdir:
        return
    try:
        wd = Path(workdir).expanduser().resolve(strict=False)
    except (OSError, ValueError):
        return
    if wd == root:
        return
    presenter.show_info(
        f"WARNING: --workdir {wd} but the resolved sandbox root is {root} "
        f"({source}); files under --workdir may resolve elsewhere."
    )
    if not sys.stdin.isatty():
        return
    try:
        from oc_ui.terminal.input import read_single_key
        answer = read_single_key("Continue with this sandbox root? (y/N) ", valid="yn")
    except (EOFError, KeyboardInterrupt, OSError):
        answer = ""
    if answer != "y":
        _dbg("OC_CHAT.MAIN", func, "operator declined the --workdir/root mismatch", 2)
        presenter.show_info("Aborted at the anchor check; re-launch with the intended --workdir.")
        raise SystemExit(2)
 
 
def main(argv: list[str] | None = None) -> None:
    """Start the OC Chat REPL.
 
    *argv* is exposed for testing; defaults to ``sys.argv[1:]`` when None.
    """
    func = "main"
    clear_terminal()
 
    # Activate oc_diagnostics side effects (loads debug_config.json, enables
    # _dbg routing). intercept_streams/intercept_exceptions stay off so the
    # interactive REPL keeps clean stdout/stderr — _dbg() is the only path
    # that writes diagnostic output.
    _diagnostics_init(intercept_streams=False, intercept_exceptions=False)
    _dbg("OC_CHAT.MAIN", func, f"oc_chat v{_VERSION} starting", 2)
 
    # Phase O3e: install the SIGINT handler before any code that might
    # build an Agent. The handler reads the module-level _active_agent_ref
    # holder, so it's safe to install now and register the agent later
    # (see register_active_agent call after Agent construction below).
    signal.signal(signal.SIGINT, _o3_sigint_handler)
 
    args = build_arg_parser().parse_args(argv)
 
    # Resolve the process composition mode once (settings file default + CLI
    # override). --option-c is the convenience alias for --mode option-c.
    from oc_chat.core.settings import init_runtime_settings
    _cli_mode = "option-c" if getattr(args, "option_c", False) else getattr(args, "mode", None)
    init_runtime_settings(_cli_mode)
 
    # M8.0 WU-7 / G-1: the guarded/isolated deployment-shape invariant. Refuse to
    # start (interactive OR batch) if the write sandbox overlaps a harness importable
    # root — a write-then-import would be in-process code execution. permissive: no-op.
    from oc_chat.core.risk_profile import resolve_risk_profile
    from oc_chat.core.deployment_shape import check_deployment_shape, DeploymentShapeError
    from oc_chat.tools.filesystem import _sandbox_root
    _active_profile = resolve_risk_profile()
    _raw_profile = os.environ.get("OC_CHAT_PROFILE", "").strip()
    if _raw_profile and _raw_profile.lower() != _active_profile:
        # An explicitly-set but unrecognized profile silently falls to permissive
        # (WU-8 spec). Make that downgrade LOUD so a typo (e.g. 'guareed') cannot
        # silently disable the guarded T2 protections.
        print(f"\nWARNING: OC_CHAT_PROFILE={_raw_profile!r} is not a recognized risk "
              f"profile; running '{_active_profile}'. Valid: permissive, guarded, isolated.\n",
              file=sys.stderr)
    try:
        check_deployment_shape(_sandbox_root(), _active_profile)
    except DeploymentShapeError as _shape_exc:
        print(f"\nERROR: {_shape_exc}\n", file=sys.stderr)
        raise SystemExit(2)
 
    # Plan 283 (SHELL-04): --auto is the launch spelling of /mode auto, and one
    # of only three ways into the mode (the command, this flag, and the `u` key
    # at the ask). Set AFTER the profile is resolved, so the refusal below can
    # tell the operator their flag will not apply here instead of leaving them
    # to discover it at the first prompt that still appears.
    if getattr(args, "auto", False):
        from oc_chat.core import session_mode
        if session_mode.auto_mode_available():
            session_mode.set_mode(session_mode.AUTO)
        else:
            # OCT-LINT: disable=no_hardcoded_secrets reason="Operator-facing launch warning; high-entropy English prose, not a credential (operator-ruled 2026-09-16)"
            print(f"\nWARNING: --auto applies only under the permissive risk "
                  f"profile; running '{_active_profile}', so every confirmable "
                  f"call will still reach you.\n", file=sys.stderr)
 
    _assert_stdin_is_tty_or_input_mode(args)
 
    # ``--no-color`` was already pre-parsed in ``OC_Chat.py`` to set
    # NO_COLOR before any oc_chat import.  We honour the flag here too in
    # case ``oc_chat.main:main`` is invoked directly (e.g. from tests).
    if args.no_color and os.getenv("NO_COLOR") is None:
        os.environ["NO_COLOR"] = "1"
 
    sessions_dir = _active_sessions_dir()
 
    # API-key pre-flight — fail fast with a clear message before constructing
    # Agent (whose ClientInterface would otherwise build with an empty key).
    provider_name = os.environ.get("OC_CHAT_PROVIDER", "deepseek")
 
    if args.input is not None or args.output is not None:
        from oc_chat.core.batch_runner import run_batch_turn
 
        raise SystemExit(run_batch_turn(
            session_name=args.session,
            input_path=args.input,
            output_path=args.output,
            provider_name=provider_name,
            presenter=select_presenter(args),
            argv=argv,
        ))
    
    # Presenter selection — exactly one call per main() invocation.
    # _select_presenter_once handles --ui plain/rich/enhanced AND the
    # legacy --enhanced-ui flag in one place; previously main() picked
    # twice (once before config validation, once after) which dropped
    # --enhanced-ui silently. Stay disciplined: do not reassign
    # ``presenter`` below.
    presenter = _select_presenter_once(args)
 
    try:
        provider = provider_plugin.registry.get(provider_name)
    except KeyError as e:
        validation = validate_runtime_config(
            provider_name=provider_name,
            provider_names=provider_plugin.registry.names(),
            sessions_dir=sessions_dir,
        )
        if not _show_config_validation(presenter, validation):
            return
        presenter.show_error(f"ERROR: {e}")
        return
 
    validation = validate_runtime_config(
        provider_name=provider_name,
        provider_names=provider_plugin.registry.names(),
        provider=provider,
        sessions_dir=sessions_dir,
    )
    if not _show_config_validation(presenter, validation):
        return
 
    # ── One-shot migration: flat layout → per-session-directory layout ────
    try:
        mig = migrate_legacy_layout(sessions_dir)
        if mig["migrated"] or mig["recovered"] or mig["orphans_moved"]:
            parts = []
            if mig["migrated"]:
                parts.append(f"{mig['migrated']} session(s) migrated")
            if mig["recovered"]:
                parts.append(f"{mig['recovered']} chat log(s) recovered from backup")
            if mig["orphans_moved"]:
                parts.append(f"{mig['orphans_moved']} orphan log(s) moved to legacy/")
            presenter.show_info("Storage upgrade: " + ", ".join(parts) + ".")
        if mig["errors"]:
            presenter.show_error(
                "Migration warnings: " + "; ".join(mig["errors"][:3])
                + (f"  (+{len(mig['errors']) - 3} more)" if len(mig["errors"]) > 3 else "")
            )
    except Exception as exc:
        # SYSTEM_ERROR — storage migration is best-effort; surface and continue
        _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: migration: {type(exc).__name__}: {exc}", 1, override=True)
        presenter.show_error(f"Storage migration failed: {exc}")
 
    # ── Build the agent (engine objects + a placeholder session) ──────────
    # The agent's session is replaced once the startup menu picks/loads one.
    agent = Agent(
        config=AgentConfig(provider_name=provider_name),
        presenter=presenter,
    )
    # Phase O3e: tell the SIGINT handler (installed at the top of main())
    # which Agent owns the in-flight CancellationToken.
    register_active_agent(agent)
 
    # Build the input driver early so the startup flow can use it too.
    turn_stats = TurnStats()
    startup_driver = PromptDriver(None, turn_stats=turn_stats)
 
    # Plan 173 item 2 (F19): surface the resolved sandbox anchor BEFORE the menu,
    # and confirm on a --workdir/root disagreement. Banner-only — no turn-control.
    _emit_startup_anchor_banner(presenter, args)
 
    # ── Startup session selection ─────────────────────────────────────────
    session = _maybe_autoreload_crashed_session(
        agent,
        presenter,
        startup_driver,
        sessions_dir,
    )
    if session is None:
        def _apply_lane_selection(selection):
            """Plan 217 T4: apply a confirmed lane selection to the Agent.
 
            ``main`` owns the Agent whose client and model manager the switch
            rebuilds, so the opening menu delegates the apply step here rather
            than importing the Agent into ``startup``. Returning the rebuilt
            pair lets the menu rebind its own locals — otherwise the very next
            new-session prompt would list the OLD lane's models.
            """
            from oc_chat.core.lane_switch import switch_lane
 
            switch_lane(agent, selection.lane_name, selection.model_id)
            return agent.model_manager, agent.client
 
        session = startup_session_prompt(
            presenter,
            version=_VERSION,
            model_manager=agent.model_manager,
            tool_schemas=agent.tool_schemas,
            prompt_driver=startup_driver,
            client=agent.client,
            lane_switch=_apply_lane_selection,
        )
    _recover_incomplete_session_tail(session, presenter, startup_driver)
    agent.session = session
 
    # Plan 217 T4: the menu may have switched lanes, which rebuilt the Agent's
    # client. ``provider`` was resolved BEFORE the menu, so re-read it from the
    # live client — the endpoint registry below is provider-scoped and must not
    # be built for the lane the operator just left.
    provider = agent.client.provider
 
    # M2.3: attach the NDJSON event writer to the process-wide default bus once
    # the active session is bound (plan 101 ADR-2). Idempotent and best-effort —
    # event persistence must never block REPL startup.
    try:
        from oc_chat.core.event_bootstrap import attach_default_writer
        attach_default_writer(session)
    except Exception as exc:
        _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: event writer attach: {type(exc).__name__}: {exc}", 1, override=True)
 
    # Plan 188 T2: attach the terminal read side of the bus — the approval and
    # sub-agent lanes the loop already publishes but never rendered. Same
    # idempotent, best-effort shape as the writer above: a UI subscriber must
    # never block REPL startup. Interactive only; batch mode stays quiet
    # (its presenter default is plain for exactly that reason — D3).
    try:
        from oc_chat.core.presenter_events import attach_presenter_bridge
        attach_presenter_bridge(presenter)
    except Exception as exc:
        _dbg("OC_CHAT.MAIN", func,
             f"SYSTEM_ERROR: presenter bridge attach: {type(exc).__name__}: {exc}",
             1, override=True)
 
    # Open the per-iteration analytics sink. Plan 152 D4/ADR-4: the sink is
    # PROCESS-owned, not session-owned. Under session ownership a replacement
    # silently dropped --stats-file / --no-iteration-stats, the finaliser closed
    # only whichever session happened to be current (leaking every other), and a
    # blind append-reopen rewrote the CSV header mid-file. The sink writes the
    # header only for a NEW or EMPTY path, and every session BORROWS its writer.
    iteration_stats_sink = None
    if args.stats_file:
        try:
            iteration_stats_sink = IterationStatsSink(
                args.stats_file, show=not args.no_iteration_stats
            )
        except (OSError, TypeError) as e:
            presenter.show_error(f"Cannot open stats file {args.stats_file}: {e}")
 
    session.show_iteration_stats = not args.no_iteration_stats
    if iteration_stats_sink is not None:
        # Bridge the startup session onto the process sink; every later
        # transition re-binds the candidate at commit.
        iteration_stats_sink.bind_to(session)
    # Plan 135 §4.6: migrate a retired session model id (deepseek-chat/-reasoner)
    # to its V4 replacement before the model is ensured/used.
    migrate_session_model(session, agent.model_manager.provider, presenter)
    agent.model_manager.ensure(session.model)
 
    ctx = Context(
        session=session,
        model_manager=agent.model_manager,
        tool_registry=agent.tool_registry,
        presenter=presenter,
        client=agent.client,
        tool_schemas=agent.tool_schemas,
        turn_stats=turn_stats,
        iteration_stats_sink=iteration_stats_sink,
        # Plan 217 T4: /lane rebuilds the client and model manager, and
        # ``Agent.client`` is what run_turn passes — so the switch has to reach
        # both. The back-reference is how it does, without agent.py knowing
        # about the picker.
        agent=agent,
    )
 
    # Tell the presenter which session is active so it can pick up
    # ui_prefs and write artifacts into the right directory, and bind the
    # realtime status-snapshot provider (plan 147 D1). Optional —
    # ConsolePresenter doesn't define bind_session.
    refresh_presenter_bindings(ctx)
 
    # Restore persisted generation options from session into ctx
    _gen_opts_dict = getattr(session, "generation_options", None) or {}
    if _gen_opts_dict:
        from oc_chat.core.generation_options import GenerationOptions
        ctx.generation_options = GenerationOptions(**_gen_opts_dict)
 
    # Phase O2 — build the endpoint registry, overlay env + TOML, and
    # restore the session's last-used endpoint (best effort; falls back
    # to the registry's default on failure).
    try:
        from pathlib import Path as _Path
        from oc_chat.core.endpoint_registry import EndpointRegistry as _EndpointRegistry
        _ep_registry = _EndpointRegistry(provider)
        _ep_registry.load_env(os.environ)
        _ep_registry.load_toml(_Path.home() / ".oc_chat" / "endpoints.toml")
        _saved_ep = getattr(session, "endpoint_name", "") or ""
        if _saved_ep and _saved_ep in _ep_registry.entries:
            try:
                _ep_registry.switch(_saved_ep)
                ctx.client.rebind_to(_ep_registry.active())
            except PermissionError as _ep_exc:
                _dbg("OC_CHAT.MAIN", func, f"saved endpoint blocked: {_ep_exc}", 1, override=True)
        ctx.endpoint_registry = _ep_registry
    except Exception as _ep_exc:
        # SYSTEM_ERROR — endpoint registry init is best-effort; carry on
        # with no registry (the /endpoint command surfaces a hint).
        _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: endpoint registry init: {type(_ep_exc).__name__}: {_ep_exc}", 1, override=True)
        ctx.endpoint_registry = None
 
    agent.ctx = ctx
    agent.turn_stats = turn_stats
 
    presenter.show_welcome(session)
 
    # M8.0 F68: the permissive profile shows its blast-radius banner at startup
    # (present only while the F67 gate is active, AC-V09-6). The external-content
    # trust set itself is session-scoped and reset by the react-loop per turn
    # (external_content_gate.bind_active_session).
    from oc_chat.core.external_content_gate import blast_radius_banner
    _blast_banner = blast_radius_banner()
    if _blast_banner:
        presenter.show_info(f"[security] {_blast_banner}")
 
    # Re-create the input driver with the real context so commands and
    # models are available for tab completion.
    prompt_driver = PromptDriver(ctx, turn_stats=turn_stats)
    # Plan 217 T4: /lane's drill-down reads its answers through the REPL's own
    # input backend rather than a bare input(), so history and key handling
    # behave as they do at every other prompt.
    ctx.prompt_driver = prompt_driver
 
    # Plan 163 §Wiring (the orchestrator's merge step): bind the ask_user
    # bridge to the live driver. Cancels (Ctrl+C / EOF) become None so the
    # tool's tested "unanswered" contract handles them and the turn continues.
    def _ask_user_read(text: str) -> str | None:
        try:
            return prompt_driver.read_user_input(None, prompt_text=text)
        except (EOFError, KeyboardInterrupt):
            return None
 
    ask_user.bind_interaction(_ask_user_read)
 
    if prompt_driver.using_prompt_toolkit:
        presenter.show_info(
            "[prompt_toolkit enabled — Tab completes commands, \u2191/\u2193 cycles history, "
            "Alt+Enter inserts newline, Ctrl+P opens command palette]"
        )
 
    # Plan 218 T2 — the session-opening rule. Everything below it belongs to
    # a turn, so the very first prompt echo already sits inside a boundary
    # instead of floating above the first rule, which the REPL would
    # otherwise only draw once the user had already pressed Enter.
    presenter.show_session_start()
 
    try:
        while True:
            # Plan 152 R10-02 — consume a Ctrl+C that the transition critical
            # section deferred. NEVER re-raise here: this region catches only
            # EOFError, and the enclosing try has no `except` at all, so a raise
            # would EXIT OC Chat. Re-prompting is the whole point.
            if consume_pending_sigint():
                presenter.show_info("[Ctrl+C — the session change completed; nothing was interrupted]")
                continue
 
            try:
                # D7/D2 — read the CURRENT session, not the startup loop-local.
                # A slash command may have replaced it, and the prompt driver
                # stores what it is handed as the session it reports.
                user_input = prompt_driver.read_user_input(ctx.session)
            except EOFError:
                break
 
            if user_input is None:
                # Ctrl+C cancelled the current input line — re-prompt.
                continue
 
            user_input = user_input.strip()
            if user_input.lower() in ("/exit", "/quit", "exit", "quit"):
                break
            if not user_input:
                continue
 
            result = dispatch(ctx, user_input)
 
            # Plan 152 ADR-7 — map the four kinds dispatch can actually produce.
            # The old message-count delta heuristic is GONE: a queued turn now
            # requires an explicit token from the handler, so an in-place
            # /session load whose restored log ends in a user message can no
            # longer fire an unrequested LLM turn.
            if result.kind == INTERNAL_FAILURE:
                if result.effect_disposition == EFFECT_UNKNOWN:
                    # Fail CLOSED: the error and its reconciliation path were
                    # already surfaced by the handler. Continue on the stated
                    # indeterminate state rather than pretending it succeeded.
                    _dbg("OC_CHAT.MAIN", func,
                         "command failed with INDETERMINATE effects", 1, override=True)
                continue
            if result.kind == HANDLED_NO_TURN:
                continue
            if result.warning:
                presenter.show_info(f"[{result.warning}]")
            # A committed queue runs EXACTLY once; a reporting failure is a
            # warning, not a cancellation of valid business work.
            turn_input = None if result.kind == HANDLED_TURN_QUEUED else user_input
 
            agent.compress_if_needed()
 
            # Re-bind so any /ui or /session change between turns takes effect
            # (plan 147 D1). Reads through ctx.session (not the loop-local
            # ``session``) — a slash command may have replaced ctx.session.
            refresh_presenter_bindings(ctx)
 
            # D7 — snapshot the session the turn will run against. The cancel
            # and error handlers below pop from THIS object, not from whatever
            # is current when they fire: an intervening replacement would
            # otherwise pop a real message off a different session.
            turn_session = ctx.session
            # Plan 152 Fable R11-04 — a Ctrl+C deferred during the command's own
            # commit converts to a cooperative interrupt AT TURN ENTRY, so the
            # notice precedes the turn and the committed queue still runs,
            # stopping at its first checkpoint.
            if consume_pending_sigint():
                presenter.show_info("[Ctrl+C — the queued turn will stop at its first checkpoint]")
                token = getattr(agent, "cancellation", None)
                if token is not None:
                    token.request_interrupt()
 
            mark_in_turn()
            try:
                if turn_input is not None:
                    agent.run_turn(turn_input)
            except KeyboardInterrupt:
                # Ctrl+C during streaming. Phase A.5a's react_loop handler
                # already showed the cancellation summary and recorded
                # partial cost; this backstop ensures the REPL never
                # crashes — pop the orphaned user message and loop back
                # to the prompt.
                _dbg("OC_CHAT.MAIN", func, "turn cancelled by user", 2, override=True)
                if turn_session.messages and turn_session.messages[-1]["role"] == "user":
                    turn_session.messages.pop()
            except Exception as e:
                # SYSTEM_ERROR — surface stream/transport failure and trim the
                # orphaned user message so the next turn isn't poisoned.
                _dbg("OC_CHAT.MAIN", func,
                     f"SYSTEM_ERROR: turn: {type(e).__name__}: {e}", 1, override=True)
                presenter.show_error(str(e))
                if turn_session.messages and turn_session.messages[-1]["role"] == "user":
                    turn_session.messages.pop()
            finally:
                # Restore the phase unconditionally: leaving IN_TURN set would
                # make every later transition fail the entry lease.
                mark_between_turns()
 
            # Plan 220 T4: the host that actually served this turn, if the
            # lane reported one. Silent otherwise — never a claim in advance,
            # never an invented one after. Reads the LIVE agent.client: a
            # /lane switch mid-session replaces it.
            _show_serving_host(presenter, agent.client)
 
            # ── After a /goal turn, record it in SESSION_LOG.md if files changed ──
            if ctx.in_goal:
                if ctx.goal_files_changed:
                    try:
                        append_session_log(ctx)
                    except Exception as e:
                        # SYSTEM_ERROR — the session-log write must not abort the REPL
                        _dbg("OC_CHAT.MAIN", func,
                             f"SYSTEM_ERROR: session_log: {type(e).__name__}: {e}", 1, override=True)
                        presenter.show_error(f"SESSION_LOG update failed: {e}")
                ctx.in_goal = False
                ctx.goal_description = ""
                ctx.goal_files_changed = []
 
            # ── After a /plan turn, reset plan-mode flags (Phase J) ──
            # Planning does not modify source files, so no CHANGELOG
            # update — just reset the flags.
            if ctx.in_plan:
                ctx.in_plan = False
                ctx.plan_description = ""
                ctx.plan_files_changed = []
 
    finally:
        # Plan 152 D4 — close the ONE process-lifetime sink exactly once, rather
        # than whichever session's descriptor happened to be current. Sessions
        # only ever hold a BORROWED reference and never close it.
        if iteration_stats_sink is not None:
            iteration_stats_sink.close()
        agent.save()
        if sys.exc_info()[0] is None:
            try:
                write_clean_exit_marker(sessions_dir)
            except Exception as exc:
                _dbg(
                    "OC_CHAT.MAIN",
                    func,
                    f"SYSTEM_ERROR: clean-exit marker: {type(exc).__name__}: {exc}",
                    1,
                    override=True,
                )
        # Phase TUI runtime-prep: call presenter.close() so the future
        # TUI presenter can restore the ANSI scroll region / leave the
        # alt-screen / drop key bindings even on abnormal exit.
        _safe_presenter_close(presenter)