session_transition.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/session_transition.py
 
"""
The ONE owner of "the current session": prepare-then-commit session transitions.
 
Purpose
-------
Replacing the active session used to mean assigning ``ctx.session`` and hoping
every other holder of "the current session" noticed. They did not: the agent,
the REPL loop-local, the event-writer singleton, the F67 trust key, the sandbox
process-global and the generation-options projection each kept their own copy,
so ``/session new`` continued the conversation in the OLD session while the UI
reported the new one — and a URL trusted in the old session stayed trusted in
the new one.
 
This module makes ``ctx.session`` the sole authority and gives every other
holder a two-phase seam to move through:
 
- **PREPARE** — non-mutating. Touches NO live process state. Every fallible
  operation (endpoint policy checks, building an SDK client, calling the
  provider for model caps, ``Path.resolve``, mkdir/open of the event file,
  ``session_id`` validation) happens here, against a CANDIDATE. A failure
  leaves the whole OLD tuple intact.
- **COMMIT** — a TRANSITION CRITICAL SECTION. Uninterruptible, and every step a
  non-throwing install of a value that PREPARE already built. Because nothing
  between the first swap and the identity publish can raise OR be interrupted,
  there is no rollback machinery — it would be unreachable.
 
Both halves are required for atomicity. Making every step synchronously no-fail
is not enough on its own: this REPL wires Ctrl+C to ``raise KeyboardInterrupt``
between turns, and ``dispatch`` is covered by no ``KeyboardInterrupt`` guard, so
an interrupt landing between two "no-fail" assignments would strand a mixed
tuple with no recovery. Hence the deferral guard below.
 
Responsibilities
----------------
- :class:`TransitionRejectedError` + :func:`transition_entry_lease` — the ONE
  entry precondition (main thread, between turns, non-reentrant), enforced
  BEFORE any prepare work so a rejected call has zero effects.
- The runtime PHASE (:func:`mark_in_turn` / :func:`mark_between_turns`), owned
  here and marked by ``main.py`` around ``run_turn``. Initialises to
  ``BETWEEN_TURNS`` so a transition issued as the first-ever command succeeds.
- The ``transition_in_progress`` guard + deferred-SIGINT bookkeeping
  (:func:`transition_in_progress`, :func:`note_pending_sigint`,
  :func:`consume_pending_sigint`).
- :class:`PreparedSessionTransition` + :func:`prepare_transition` (a
  resource-owning context that aborts in REVERSE order on ANY ``BaseException``)
  + :func:`commit_transition` + the :func:`replace_session` entry point.
- :class:`IterationStatsSink` — the PROCESS-owned CSV stats sink that sessions
  borrow, replacing the per-session file descriptor that leaked across
  transitions.
- The SHARED endpoint coordinator (:class:`PreparedEndpointActivation` +
  :func:`commit_endpoint_activation`) so ``/endpoint use`` commits inside the
  same critical section instead of its own split five-step sequence.
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L1 — a transition aborted (SYSTEM_ERROR / SECURITY_ERROR class).
    L2 — transition lifecycle: lease taken, prepare done, commit done.
    L3 — per-owner prepare/install detail; deferred-SIGINT bookkeeping.
    L4 — module-load coverage marker.
 
Contracts
---------
- PREPARE calls NO mutating live API — not ``rebind_to``,
  ``set_session_sandbox_root_override``, ``ModelManager.invalidate``,
  ``EndpointRegistry.switch``, ``attach_default_writer`` / ``detach_default_writer``,
  nor any COMMIT install primitive.
- COMMIT installs process-globals and F67 trust BEFORE publishing identity, so
  the externally visible set never shows a new identity carrying old trust.
- ``ctx.session = candidate`` is the LAST state mutation of the transaction.
- On ANY ``BaseException`` before commit, the full old tuple stands and this
  module leaks no descriptor, no filesystem artifact and no lease.
- This module never writes another module's private fields; every install goes
  through an owner-defined method.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.external_content_gate (lazy; F67 prepared binding + no-throw install)
- oc_chat.core.event_bootstrap (lazy; prepared event-writer install)
- oc_chat.tools._sandbox (lazy; no-fail resolved-override install)
- oc_chat.core.commands (lazy; refresh_presenter_bindings after the commit)
- oc_chat.core.event_writer (lazy; EventWriter for the prepared writer and its type hint)
- oc_chat.core.generation_options (lazy; GenerationOptions from the persisted options dict)
- oc_chat.core.models (lazy; PreparedModelCache type hint)
- oc_chat.core.persistence (lazy; load, save and supersede the session records)
- oc_chat.core.provider_plugin (lazy; ProviderEndpoint type hint)
- oc_chat.core.session (lazy; ChatSession construction and metadata hydration of the candidate)
- oc_chat.core.startup (lazy; model migration and variable-context refresh of the candidate)
"""
from __future__ import annotations
 
import threading
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Iterator, Optional
 
from oc_diagnostics import _dbg
 
if TYPE_CHECKING:                                    # pragma: no cover - typing only
    from oc_chat.core.event_writer import PreparedEventWriter
    from oc_chat.core.external_content_gate import (
        PreparedF67Binding, PreparedRecordBinding,
    )
    from oc_chat.core.models import PreparedModelCache
    from oc_chat.core.provider_plugin import ProviderEndpoint
    from oc_chat.core.session import ChatSession
 
_dbg("OC_CHAT.CORE", "session_transition", "module", 4)
 
 
# ── Runtime phase ───────────────────────────────────────────────────────────
 
#: The REPL is between turns (a transition may run) vs inside ``run_turn``.
BETWEEN_TURNS = "between_turns"
IN_TURN = "in_turn"
 
#: Initialised to BETWEEN_TURNS at module init, NOT inferred from cancellation
#: state, so a transition issued as the FIRST-ever command succeeds.
_phase: str = BETWEEN_TURNS
 
_lease_held = False
_transition_in_progress = False
_pending_sigint = False
 
 
class TransitionRejectedError(RuntimeError):
    """The ONE typed rejection for a transition attempted out of contract.
 
    Raised BEFORE any prepare work, so a rejected call has NO filesystem, bus,
    registry or process-global effect — in particular no event directory or file
    was created. There is no serialization alternative: an out-of-contract call
    is rejected, never queued.
    """
 
 
def current_phase() -> str:
    """Return the runtime phase. Read live; never cache this."""
    return _phase
 
 
def mark_in_turn() -> None:
    """Mark entry into ``run_turn`` (``main.py``, immediately before the call)."""
    global _phase
    _phase = IN_TURN
 
 
def mark_between_turns() -> None:
    """Mark the ``run_turn`` boundary exit (``main.py``, in a ``finally``)."""
    global _phase
    _phase = BETWEEN_TURNS
 
 
def transition_in_progress() -> bool:
    """Live getter for the COMMIT guard.
 
    ``main.py``'s SIGINT handler MUST read this through the function on every
    signal — never import the flag by value, which would snapshot ``False`` at
    import time and defeat the deferral entirely.
    """
    return _transition_in_progress
 
 
def note_pending_sigint() -> None:
    """Record a Ctrl+C deferred by the critical section. Repeats COALESCE."""
    global _pending_sigint
    _pending_sigint = True
 
 
def consume_pending_sigint() -> bool:
    """Consume-once, NO-THROW: was a Ctrl+C deferred while we were committing?
 
    Returns ``True`` at most once per deferred interrupt. The caller maps that
    to a re-prompt (at the prompt boundary) or to
    ``token.request_interrupt()`` (when a turn is already queued) — NEVER to a
    re-raise. Re-raising at the prompt boundary would land in a region where
    only ``EOFError`` is caught and would EXIT OC Chat; and an injected bare
    ``KeyboardInterrupt`` behaves differently in the two prompt backends (one
    re-raises, the other converts it to ``None``), so the consumed flag is the
    only backend-safe surface.
    """
    global _pending_sigint
    was_pending = _pending_sigint
    _pending_sigint = False
    return was_pending
 
 
@contextmanager
def uninterruptible_commit() -> Iterator[None]:
    """Run a no-fail install block with Ctrl+C DEFERRED (plan 217 T3).
 
    The same critical section :func:`commit_endpoint_activation` opens
    inline, exposed so a coordinator living in another module can share it
    rather than reach for the private flag. ``main.py``'s SIGINT handler
    reads :func:`transition_in_progress` live, so a press landing inside
    the block is recorded and replayed at the next prompt boundary instead
    of splitting the tuple being installed.
 
    Nesting is safe: the flag is restored to whatever it was on entry, so
    an inner block cannot clear an outer one's deferral.
    """
    func = "uninterruptible_commit"
    global _transition_in_progress
    previous = _transition_in_progress
    _transition_in_progress = True
    try:
        yield
    finally:
        _transition_in_progress = previous
        _dbg("OC_CHAT.CORE", func, "commit block closed", 3)
 
 
def _reset_for_tests() -> None:
    """Test-only: restore module state to a freshly-imported baseline."""
    global _phase, _lease_held, _transition_in_progress, _pending_sigint
    _phase = BETWEEN_TURNS
    _lease_held = False
    _transition_in_progress = False
    _pending_sigint = False
 
 
# ── Entry lease ─────────────────────────────────────────────────────────────
 
@contextmanager
def transition_entry_lease() -> Iterator[None]:
    """Acquire the ONE entry precondition, or reject. Released on ANY exit.
 
    Three conditions, all checked BEFORE any prepare work:
 
    - the caller is on the MAIN thread;
    - the runtime phase is ``BETWEEN_TURNS`` (explicitly marked by ``main.py``
      around ``run_turn`` — not inferred from cancellation state);
    - the lease is not already held (non-reentrancy).
 
    Any violation raises :class:`TransitionRejectedError` with zero effects. The
    lease is released in a ``finally`` that also covers ``BaseException``, so a
    ``KeyboardInterrupt`` or ``SystemExit`` mid-transition cannot wedge every
    later transition.
 
    This precondition is what forecloses the event-bus swap race: with no
    publisher able to be mid-dispatch between turns on another thread, the
    linearizable ``replace_subscription`` needs no in-flight drain.
    """
    func = "transition_entry_lease"
    global _lease_held
    if threading.current_thread() is not threading.main_thread():
        raise TransitionRejectedError(
            "session transitions are main-thread only; "
            f"called from {threading.current_thread().name!r}"
        )
    if _phase != BETWEEN_TURNS:
        raise TransitionRejectedError(
            f"session transitions run between turns only; runtime phase is {_phase!r}"
        )
    if _lease_held:
        raise TransitionRejectedError("a session transition is already in progress (reentrant call)")
    _lease_held = True
    _dbg("OC_CHAT.CORE", func, "entry lease acquired", 2)
    try:
        yield
    finally:
        _lease_held = False
        _dbg("OC_CHAT.CORE", func, "entry lease released", 3)
 
 
# ── Process-owned iteration stats sink (D4 / ADR-4) ─────────────────────────
 
#: The CSV header the REPL writes for a new iteration-stats file.
STATS_CSV_HEADER = (
    "ts,iter,model,effort,tools,prompt_tokens,completion_tokens,"
    "cache_hit_tokens,cache_miss_tokens,dur_ms\n"
)
 
 
class IterationStatsSink:
    """A PROCESS-owned iteration-stats CSV file that sessions BORROW.
 
    Replaces the session-owned file descriptor. Under session ownership a
    replacement silently dropped ``--stats-file`` / ``--no-iteration-stats``,
    the finaliser closed only whichever session happened to be current (leaking
    the others), and a blind append-reopen rewrote the CSV header mid-file.
 
    Ownership is explicit: this object opens, flushes and closes exactly once;
    a session holds a BORROWED writer reference it must never close. The actual
    write path (``react_loop``) is unchanged — it still reads
    ``session.iteration_stats_file`` / ``session.show_iteration_stats``, which
    the transition commit points at this sink for every session. That bridge is
    what makes process ownership real without editing the loop.
    """
 
    def __init__(self, path: str | Path, *, show: bool = True) -> None:
        func = "IterationStatsSink.__init__"
        if not isinstance(path, (str, Path)):
            raise TypeError(
                "IterationStatsSink requires a filesystem path, not a file-like "
                "object; the sink owns open/close and cannot adopt a pathless handle"
            )
        self.path = Path(path)
        self.show = bool(show)
        self._fh: Any = None
        self._closed = False
        # The header belongs to a NEW or EMPTY file only — appending it to an
        # existing log would corrupt it with a header line mid-file.
        needs_header = not self.path.exists() or self.path.stat().st_size == 0
        self._fh = open(self.path, "a", encoding="utf-8", newline="")
        if needs_header:
            self._fh.write(STATS_CSV_HEADER)
            self._fh.flush()
        _dbg("OC_CHAT.CORE", func, f"opened {self.path} (header={needs_header})", 2)
 
    @property
    def writer(self) -> Any:
        """The borrowed writer handle a session's ``iteration_stats_file`` points at."""
        return self._fh
 
    def flush(self) -> None:
        """Best-effort flush; never raises into a caller's happy path."""
        func = "IterationStatsSink.flush"
        try:
            if self._fh is not None and not self._closed:
                self._fh.flush()
        except Exception as exc:
            _dbg("OC_CHAT.CORE", func, f"EXTERNAL_ERROR: flush failed: {exc}", 2)
 
    def close(self) -> None:
        """Flush + close EXACTLY ONCE. Idempotent and non-raising."""
        func = "IterationStatsSink.close"
        if self._closed:
            return
        self._closed = True
        try:
            if self._fh is not None:
                self._fh.flush()
                self._fh.close()
        except Exception as exc:
            _dbg("OC_CHAT.CORE", func, f"EXTERNAL_ERROR: close failed: {exc}", 2)
        finally:
            self._fh = None
        _dbg("OC_CHAT.CORE", func, "closed", 2)
 
    def bind_to(self, session: "ChatSession") -> None:
        """Point *session* at this sink (the pinned react_loop bridge).
 
        Pure assignment of a borrowed reference — cannot fail, so it is safe
        inside the commit critical section. The session never closes it.
        """
        session.iteration_stats_file = self._fh
        session.show_iteration_stats = self.show
 
 
# ── Prepared values ─────────────────────────────────────────────────────────
 
@dataclass
class PreparedSessionTransition:
    """Everything a session replacement needs, built while nothing was mutated.
 
    Each member is the finished product of one owner's fallible work. COMMIT
    installs them in a pinned order and does nothing else, which is what makes
    the commit block genuinely no-fail rather than merely called no-fail.
    """
 
    candidate: "ChatSession"
    #: Unattached SDK client from ``ClientInterface.build_candidate``.
    client: Any = None
    #: The resolved endpoint the candidate activates.
    endpoint: Optional["ProviderEndpoint"] = None
    #: Fully-built next model cache from ``ModelManager.prepare_cache``.
    model_cache: Optional["PreparedModelCache"] = None
    #: Opened-but-unsubscribed event writer + its ownership receipt.
    event_writer: Optional["PreparedEventWriter"] = None
    #: ALREADY-RESOLVED sandbox override (the one fallible resolve ran in PREPARE).
    sandbox_override: Optional[Path] = None
    #: ``GenerationOptions`` converted from the candidate's persisted dict.
    generation_options: Any = None
    #: Validated F67 trust rotation, ready for the no-throw install.
    f67_binding: Optional["PreparedF67Binding"] = None
    #: Prebuilt ``session_id -> canonical record`` binding for load/switch
    #: (plan 152-b D4). ``None`` for every non-load replacement path, which is
    #: what keeps the commit step below byte-compatible for ``replace_session``.
    record_binding: Optional["PreparedRecordBinding"] = None
 
 
@dataclass
class PreparedEndpointActivation:
    """An ``/endpoint use`` activation prepared for the shared coordinator.
 
    ``/endpoint use`` used to commit five things in sequence — client, model
    cache, registry ``active_name``, ``session.endpoint_name``, ``dirty`` —
    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.
    Preparing them together and committing under the same lease closes that.
    """
 
    endpoint: "ProviderEndpoint"
    client: Any
    model_cache: "PreparedModelCache"
    session: "ChatSession"
 
 
# ── PREPARE ─────────────────────────────────────────────────────────────────
 
@contextmanager
def prepare_transition(
    ctx: Any,
    candidate: "ChatSession",
) -> Iterator[PreparedSessionTransition]:
    """Build a :class:`PreparedSessionTransition` without touching live state.
 
    A RESOURCE-OWNING context. On ANY ``BaseException`` — not just
    ``Exception`` — the owners abort in REVERSE order and the transition leaves
    the old tuple untouched. ``BaseException`` matters because the real SIGINT
    handler raises ``KeyboardInterrupt`` between turns and the COMMIT deferral
    guard does not exist yet during PREPARE, so an interrupt here would
    otherwise strand an open descriptor and a half-created event directory.
 
    Cleanup is idempotent, so a repeated Ctrl+C during the unwind is safe.
    Ownership transfers to COMMIT only when the block exits normally.
    """
    func = "prepare_transition"
    prepared = PreparedSessionTransition(candidate=candidate)
    committed = False
    try:
        # 1. Endpoint — resolve + policy-check in the OWNER. Non-mutating: no
        #    switch(), no active_name write. Unknown / missing / blocked /
        #    forbidden-host are rejected here.
        registry = getattr(ctx, "endpoint_registry", None)
        if registry is not None and getattr(registry, "entries", None):
            prepared.endpoint = registry.resolve(getattr(candidate, "endpoint_name", "") or "")
            # 2. Client — built UNATTACHED; the live _client is untouched.
            prepared.client = ctx.client.build_candidate(prepared.endpoint)
            # 3. Model cache — the fallible provider.models() / default_caps().
            prepared.model_cache = ctx.model_manager.prepare_cache(
                getattr(candidate, "model", "") or ""
            )
 
        # 4. Sandbox — the ONE fallible resolve happens HERE, not at commit.
        prepared.sandbox_override = _resolve_override(
            getattr(candidate, "sandbox_root_override", None)
        )
 
        # 5. Generation options — the dict->GenerationOptions conversion is
        #    pinned to PREPARE because ctx holds the typed value while the
        #    session persists a dict, and react_loop reads the ctx projection.
        prepared.generation_options = _convert_generation_options(
            getattr(candidate, "generation_options", None)
        )
 
        # 6. F67 — validated trust key, fully built so the install allocates,
        #    validates and logs NOTHING.
        from oc_chat.core.external_content_gate import prepare_session_binding
        prepared.f67_binding = prepare_session_binding(candidate)
 
        # 7. Event writer — the real file I/O (mkdir, resume scan, open) with an
        #    ownership receipt, but NO bus subscribe.
        from oc_chat.core.event_bootstrap import _resolve_base_dir
        from oc_chat.core.event_writer import EventWriter
        prepared.event_writer = EventWriter(_resolve_base_dir(candidate)).open_unsubscribed()
 
        _dbg("OC_CHAT.CORE", func, "prepare complete (no live state mutated)", 2)
        yield prepared
        committed = True
    except BaseException as exc:
        _dbg("OC_CHAT.CORE", func,
             f"SYSTEM_ERROR: prepare aborted: {type(exc).__name__}: {exc}", 1, override=True)
        raise
    finally:
        if not committed:
            _abort_prepared(prepared)
 
 
def _abort_prepared(prepared: PreparedSessionTransition) -> None:
    """Release a prepared transition's resources in REVERSE build order.
 
    Only the event writer holds anything: the built client and prepared cache
    are inert values that simply fall away, and no bus/registry/global was ever
    touched. Idempotent and non-raising.
    """
    if prepared.event_writer is not None:
        prepared.event_writer.abort()
 
 
def _resolve_override(raw: Any) -> Optional[Path]:
    """Resolve a sandbox override string ONCE, in PREPARE. May raise here."""
    if raw is None or str(raw).strip() == "":
        return None
    return Path(str(raw)).expanduser().resolve(strict=False)
 
 
def _convert_generation_options(raw: Any) -> Any:
    """Convert the persisted generation-options dict into ``GenerationOptions``."""
    from oc_chat.core.generation_options import GenerationOptions
    if raw is None:
        return GenerationOptions()
    if isinstance(raw, dict):
        return GenerationOptions(**raw)
    return raw                                       # already the typed value
 
 
# ── COMMIT ──────────────────────────────────────────────────────────────────
 
def commit_transition(ctx: Any, prepared: PreparedSessionTransition) -> None:
    """Install a prepared transition. UNINTERRUPTIBLE, and every step no-fail.
 
    The whole block runs under the process-owned ``transition_in_progress``
    guard, so ``main.py``'s SIGINT handler DEFERS ``KeyboardInterrupt`` instead
    of raising it between two installs. Order is pinned and load-bearing:
 
    1. client, 2. sandbox (ctx mirror AND process-global), 3. generation
    options, 4. model cache, 5. event writer (linearizable bus swap),
    6. stats-sink bridge, 7. endpoint activation + the CANDIDATE's
    ``endpoint_name``/``dirty``, 8. F67 trust rotation, 9. ``ctx.session``.
 
    Identity publishes LAST and trust rotates BEFORE it, so there is never a
    moment where the new identity carries the old session's trust set. The
    externally visible set includes the process-globals, not just
    ``ctx.session``, which is why they precede the publish too.
 
    Step 10 — the presenter/TurnStats refresh — is POST-transaction and
    best-effort: it calls arbitrary collaborator methods that a custom presenter
    could make raise, so it is excluded from the atomic tuple. Stale UI at
    worst, never a state hybrid.
    """
    func = "commit_transition"
    global _transition_in_progress
    candidate = prepared.candidate
    _transition_in_progress = True
    try:
        # (1) client
        if prepared.client is not None:
            ctx.client.install_prepared(prepared.client)
        # (2) sandbox — BOTH projections. _sync_sandbox_override reads the ctx
        #     mirror FIRST, so resetting only the global re-poisons from the
        #     stale mirror on the next sync.
        from oc_chat.tools import _sandbox
        resolved = prepared.sandbox_override
        ctx.sandbox_root_override = str(resolved) if resolved is not None else None
        _sandbox.install_resolved(resolved)
        # (3) generation options — react_loop reads the CTX projection.
        ctx.generation_options = prepared.generation_options
        # (4) model cache
        if prepared.model_cache is not None:
            ctx.model_manager.apply_prepared(prepared.model_cache)
        # (5) event writer — one lock, exactly one wildcard writer throughout.
        if prepared.event_writer is not None:
            from oc_chat.core.event_bootstrap import install_prepared_writer
            install_prepared_writer(prepared.event_writer)
        # (6) stats sink — the candidate BORROWS the process sink's writer.
        sink = getattr(ctx, "iteration_stats_sink", None)
        if sink is not None:
            sink.bind_to(candidate)
        # (7) endpoint activation — the committed OBJECT, and the CANDIDATE's
        #     fields (ctx.session is still the OLD session until step 9).
        if prepared.endpoint is not None:
            registry = getattr(ctx, "endpoint_registry", None)
            if registry is not None:
                registry.commit_endpoint_activation(prepared.endpoint)
            candidate.endpoint_name = prepared.endpoint.name
        # (8) F67 trust rotation — BEFORE identity.
        if prepared.f67_binding is not None:
            from oc_chat.core.external_content_gate import install_prepared_session_binding
            install_prepared_session_binding(prepared.f67_binding)
        # (8b) canonical-record binding (plan 152-b D4) — the ONE sanctioned
        #      amendment to the 152 critical section. A no-throw dict assignment
        #      that installs the load/switch candidate's session_id->locator map
        #      immediately before identity, so _id_to_locator joins the atomic
        #      old/new tuple. ``None`` for every non-load path (replace_session),
        #      so this step is a no-op there and the 152 order is unchanged.
        if prepared.record_binding is not None:
            from oc_chat.core.external_content_gate import install_prepared_record_binding
            install_prepared_record_binding(prepared.record_binding)
        # (9) identity publish — truly last.
        ctx.session = candidate
    finally:
        _transition_in_progress = False
    _dbg("OC_CHAT.CORE", func, f"committed session {getattr(candidate, 'name', '?')!r}", 2)
    _post_transaction(ctx, prepared)
 
 
def _post_transaction(ctx: Any, prepared: PreparedSessionTransition) -> None:
    """Best-effort projections AFTER the transaction succeeded. Never fatal.
 
    ``turn_stats.rebind_session`` runs FIRST: it is an infallible id assignment
    carrying token accounting, and if a raising presenter went first it would be
    skipped and usage would keep being attributed to the OLD ``session_id``.
    Then the deferred F67 lifecycle diagnostic (moved out of the critical
    section because ``_dbg`` can raise under warnings-as-errors before
    diagnostics are initialised), then the presenter rebind.
    """
    func = "_post_transaction"
    try:
        turn_stats = getattr(ctx, "turn_stats", None)
        if turn_stats is not None:
            turn_stats.rebind_session(getattr(ctx.session, "session_id", ""))
        if prepared.f67_binding is not None:
            from oc_chat.core.external_content_gate import log_session_binding_lifecycle
            log_session_binding_lifecycle(prepared.f67_binding)
        from oc_chat.core.commands import refresh_presenter_bindings
        refresh_presenter_bindings(ctx)
    except Exception as exc:
        # The transaction already SUCCEEDED — new state is coherent. A UI
        # projection failure is reported, never reclassified as a failed
        # transition.
        _dbg("OC_CHAT.CORE", func,
             # OCT-LINT: disable=no_hardcoded_secrets reason="Diagnostic _dbg prose trips the Shannon entropy heuristic (4.50 bits/char at the 4.5 threshold); no secret material"
             f"EXTERNAL_ERROR: post-transaction refresh failed: {type(exc).__name__}: {exc}",
             2, override=True)
 
 
def rename_session(ctx: Any, new_name: str) -> tuple[str, str]:
    """Rename / save-as — its OWN transition class (plan 152 D8).
 
    A rename has DIFFERENT invariants from a replacement, which is why it is a
    separate named transition rather than a reuse of :func:`replace_session`:
 
    - the canonical ``session_id`` is PRESERVED, so F67 trust must NOT reset —
      it is the same identity under a new display name. That falls out for free
      from keying trust on the id;
    - sandbox, generation options, endpoint and the stats sink are UNCHANGED;
    - the OLD name is retained until the commit point.
 
    PREPARE captures the current name's SLUG **before** the name mutation —
    ``/session save <name>`` overwrites the name and then persists, so any
    post-mutation capture is already too late and the old event history would
    become unreachable.
 
    COMMIT order is deliberate: write the NEW record first, THEN supersede-mark
    the old one. If the mark fails, two same-id records exist with no mark, and
    the deterministic newest-meta-timestamp rule resolves to the new record —
    the same answer. Marking first and then failing to save would instead
    demote a record that is still canonical.
 
    Returns ``(log_path, meta_path)``. On a failure before the name mutation the
    old name and old files stand.
    """
    func = "rename_session"
    from oc_chat.core.event_bootstrap import _slug
    from oc_chat.core.persistence import (
        mark_record_superseded, save_chat_log, save_session_metadata,
    )
 
    session = ctx.session
    old_name = getattr(session, "name", "")
 
    # Not a rename — a plain re-save. No alias, no supersede mark.
    if not new_name or new_name == old_name:
        return save_chat_log(session), save_session_metadata(session)
 
    # PREPARE — capture the outgoing slug BEFORE the name changes.
    old_slug = _slug(old_name)
    aliases = list(getattr(session, "legacy_event_slugs", None) or [])
    new_slug = _slug(new_name)
    if old_slug and old_slug not in aliases and old_slug != new_slug:
        aliases.append(old_slug)
 
    # COMMIT — mutate, persist under the new name, then demote the old record.
    session.name = new_name
    session.legacy_event_slugs = aliases
    log_path = save_chat_log(session)
    meta_path = save_session_metadata(session)
    mark_record_superseded(old_name, new_name)
    _dbg("OC_CHAT.CORE", func,
         f"renamed {old_name!r} -> {new_name!r}; aliases={len(aliases)}", 2)
    return log_path, meta_path
 
 
class ChatLogLoadError(RuntimeError):
    """A load/switch candidate's chat log could not be read/parsed (plan 152-b D3).
 
    v2 pins the load/switch chat-log contract to fail CLOSED for a parse/read
    error: rather than the pre-152-b warn-and-load-empty behaviour (which quietly
    replaced a corrupt-but-present history with an empty buffer), the whole
    transition aborts with the old session intact. A MISSING file is still the
    nonfatal empty-candidate case and does NOT raise.
    """
 
 
@dataclass(frozen=True)
class _ResolvedActivation:
    """The pure RESOLVE result for a load/switch (plan 152-b D2, Sol P1-2).
 
    Computed values only — normalized metadata, the canonical record locator, and
    whether the record was non-canonical (id popped). Building it mutates NO
    ``ctx``, process-global, or live-session state, unlike the legacy
    ``_metadata_for_activation`` which writes a fresh id onto the LIVE session.
    """
 
    metadata: dict
    locator: str
    noncanonical_reason: str
 
 
def _resolve_activation_pure(name: str) -> _ResolvedActivation:
    """Resolve *name*'s activation metadata WITHOUT touching live state (RESOLVE).
 
    Replaces the use of ``commands._metadata_for_activation`` for the seam paths:
    that helper mints a fresh id directly onto ``ctx.session`` for a
    malformed / superseded / duplicate-id record, which is unusable in a pure
    RESOLVE. Here the same non-canonical decision is made, but the only effect is
    popping ``session_id`` from the RETURNED dict — the fresh candidate then keeps
    its own construction-minted canonical id under ``pure`` hydration.
    """
    func = "_resolve_activation_pure"
    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)
        _dbg("OC_CHAT.CORE", func,
             f"SECURITY_ERROR: non-canonical record for {name!r} ({reason}); the "
             f"candidate keeps a freshly-minted identity and trust rotates at commit",
             2, override=True)
    return _ResolvedActivation(metadata=meta, locator=locator, noncanonical_reason=reason)
 
 
def _hydrate_candidate_messages(candidate: "ChatSession", name: str) -> None:
    """Load *name*'s chat log onto the CANDIDATE (PREPARE; fallible).
 
    Contract (plan 152-b D3, Sol P2-3): a MISSING file is the nonfatal
    empty-candidate case (today's behaviour, retained); a parse/read FAILURE
    aborts the transition fail-closed via :class:`ChatLogLoadError` rather than
    silently replacing an unreadable-but-present history with an empty buffer.
    """
    from oc_chat.core.persistence import load_chat_log
    candidate.load_failed = False
    try:
        log_data = load_chat_log(name)
    except FileNotFoundError:
        candidate.messages = []
        return
    except Exception as exc:
        raise ChatLogLoadError(
            f"chat log for {name!r} could not be read/parsed ({type(exc).__name__}: "
            f"{exc}); the on-disk log is preserved and the load was aborted"
        ) from exc
    candidate.messages = log_data.get("messages", [])
 
 
def _build_load_candidate(
    ctx: Any, name: str,
) -> tuple["ChatSession", Optional["PreparedRecordBinding"], Optional[str]]:
    """Build a fully-hydrated load/switch candidate (PREPARE; no live effects).
 
    Every fallible hydration step (metadata resolve, chat-log read,
    ``TokenStats.from_dict``, ``set_model``, retired-model migration, legacy /
    tool-reference migration, variable-context refresh) runs HERE against the
    detached candidate, so a failure leaves the live session untouched and the
    lease is released by the caller's context. Returns the candidate, the
    prepared canonical-record binding, and any captured (deferred) migration
    notice.
    """
    from oc_chat.core.external_content_gate import prepare_record_binding
    from oc_chat.core.session import (
        ChatSession, TokenStats, apply_session_metadata_fields,
        ensure_system_includes_tool_reference, _migrate_legacy_system_prompt,
    )
    from oc_chat.core.startup import migrate_session_model, refresh_variable_context
 
    resolved = _resolve_activation_pure(name)
    meta = resolved.metadata
 
    candidate = ChatSession(name=meta.get("session_name", name) or name, messages=[])
    _hydrate_candidate_messages(candidate, name)
 
    candidate.reasoning_effort = meta.get("reasoning_effort", candidate.reasoning_effort)
    candidate.chat_logs = meta.get("chat_logs", [])
    candidate.stats = TokenStats.from_dict(meta.get("stats", {}))
    candidate.ui_prefs = dict(meta.get("ui", {}) or {})
    # PURE hydration: fields land on the candidate; the two live effects
    # (trust reset, sandbox global) are deferred to the seam's commit.
    apply_session_metadata_fields(candidate, meta, pure=True)
 
    record_binding = prepare_record_binding(candidate.session_id, resolved.locator)
 
    candidate.set_model(meta.get("model", candidate.model))
    migration_notice = migrate_session_model(
        candidate, ctx.model_manager.provider, capture=True
    )
    candidate.dirty = False
    _migrate_legacy_system_prompt(candidate, getattr(ctx, "tool_schemas", None))
    ensure_system_includes_tool_reference(candidate, getattr(ctx, "tool_schemas", None))
    refresh_variable_context(candidate)
    return candidate, record_binding, migration_notice
 
 
def _save_outgoing_snapshot(outgoing: "ChatSession") -> bool:
    """Save the outgoing session from an IMMUTABLE snapshot (plan 152-b D2 / Sol P1-1).
 
    The pre-152-b live paths auto-saved the outgoing session in place, mutating
    its ``chat_logs`` / ``filepath`` / ``dirty`` before the incoming candidate
    could commit — so a later failure could not leave the old tuple byte-intact.
    Here a shallow ``copy.copy`` absorbs those writes (``save_*`` only REBIND
    those attributes and READ the rest), so the LIVE outgoing object — including
    its ``dirty`` flag, honestly still ``True`` on abort — is never touched.
 
    Returns ``True`` when a save ran (the outgoing was dirty). Raises on save
    failure, which aborts the transition with the old tuple intact.
 
    Pinned boundary (a named residual, NOT a closure claim): this protects the
    in-memory live tuple. It does NOT provide cross-file all-or-nothing
    durability — a failure between the chat-log write and the metadata write can
    still leave one file newer than the other, exactly as today. See report.md.
    """
    if not getattr(outgoing, "dirty", False):
        return False
    import copy
    from oc_chat.core.persistence import save_chat_log, save_session_metadata
    snapshot = copy.copy(outgoing)
    save_chat_log(snapshot)
    save_session_metadata(snapshot)
    return True
 
 
@dataclass(frozen=True)
class LoadTransitionOutcome:
    """What the load/switch handler needs to finish AFTER an atomic transition.
 
    The seam owns the atomic state move; the command layer owns presenter
    messaging. This carries the deferred, best-effort, presenter-facing residue
    (the captured migration notice + whether the outgoing autosave ran) back to
    the handler so it can emit the notices post-commit.
    """
 
    candidate: "ChatSession"
    migration_notice: Optional[str]
    outgoing_saved: bool
 
 
def load_session_into(ctx: Any, name: str) -> LoadTransitionOutcome:
    """Load persisted session *name* into *ctx* atomically (plan 152-b D2). ENTRY POINT.
 
    The load/switch analogue of :func:`replace_session`: it owns the whole
    lifecycle so handlers never hand-assemble the lease or hydrate in place.
 
    1. LEASE first — a rejected call has zero effects.
    2. RESOLVE + build a fully-hydrated candidate (pure; no live mutation).
    3. PREPARE the candidate's owned resources (endpoint/client/cache/sandbox/
       gen-opts/F67/**writer keyed to the LOADED id** — the audit-plane re-key).
    4. Save the OUTGOING session from an immutable snapshot (pre-commit external
       effect); a failure aborts with the old tuple intact.
    5. COMMIT the 152 critical section (+ the D4 record-binding step); identity
       publishes LAST.
 
    Returns a :class:`LoadTransitionOutcome`; the handler emits the deferred
    notices. On ANY failure before commit the live session, its trust, the
    sandbox global, the event writer and ``_id_to_locator`` all stand unchanged.
    """
    func = "load_session_into"
    with transition_entry_lease():
        candidate, record_binding, migration_notice = _build_load_candidate(ctx, name)
        with prepare_transition(ctx, candidate) as prepared:
            prepared.record_binding = record_binding
            # Candidate PREPARE has succeeded here; save the outgoing snapshot as
            # the last pre-commit step, then commit. A save failure raises out of
            # this block -> prepare_transition aborts -> old tuple intact.
            outgoing_saved = _save_outgoing_snapshot(ctx.session)
            commit_transition(ctx, prepared)
    _dbg("OC_CHAT.CORE", func,
         f"loaded session {getattr(candidate, 'name', '?')!r} "
         f"(outgoing_saved={outgoing_saved})", 2)
    return LoadTransitionOutcome(
        candidate=candidate,
        migration_notice=migration_notice,
        outgoing_saved=outgoing_saved,
    )
 
 
def replace_session(ctx: Any, candidate: "ChatSession") -> None:
    """Replace the active session with *candidate*, atomically. THE entry point.
 
    Used by every replacement path (``/session new``, ``/session clear``
    option 2, ``/ask`` option 4) and by the post-attach ``Agent.session``
    setter. Handlers call THIS — a bare ``ctx.session =`` invokes no seam and
    reintroduces the desync.
 
    Shape is pinned: the lease precedes PREPARE, PREPARE owns its resources, and
    COMMIT transfers ownership.
    """
    with transition_entry_lease():
        with prepare_transition(ctx, candidate) as prepared:
            commit_transition(ctx, prepared)
 
 
# ── Shared endpoint coordinator (R10-03) ────────────────────────────────────
 
def commit_endpoint_activation(ctx: Any, prepared: PreparedEndpointActivation) -> None:
    """Commit an ``/endpoint use`` activation inside the SAME critical section.
 
    Acquires the same ``transition_in_progress`` guard so a Ctrl+C between any
    two installs is deferred rather than splitting the tuple. Order: client and
    cache FIRST, then the committed endpoint OBJECT + ``active_name`` +
    ``session.endpoint_name`` + ``dirty`` LAST.
 
    Here — and ONLY here — the fields are written on the LIVE session, because
    ``/endpoint use`` keeps the session it has. The session seam's step (7)
    writes the CANDIDATE's.
    """
    func = "commit_endpoint_activation"
    global _transition_in_progress
    _transition_in_progress = True
    try:
        ctx.client.install_prepared(prepared.client)
        ctx.model_manager.apply_prepared(prepared.model_cache)
        registry = getattr(ctx, "endpoint_registry", None)
        if registry is not None:
            registry.commit_endpoint_activation(prepared.endpoint)
        prepared.session.endpoint_name = prepared.endpoint.name
        # `dirty` so a runtime endpoint switch survives the dirty-gated autosave
        # that `/session new` performs on the outgoing session.
        prepared.session.dirty = True
    finally:
        _transition_in_progress = False
    _dbg("OC_CHAT.CORE", func, f"endpoint committed: {prepared.endpoint.name}", 2)
 
 
def prepare_endpoint_activation(ctx: Any, name: str) -> PreparedEndpointActivation:
    """Resolve + build + prepare an ``/endpoint use`` target. PREPARE, fallible.
 
    A failure here cannot leave registry, client and cache disagreeing, because
    none of them has been touched.
    """
    registry = ctx.endpoint_registry
    endpoint = registry.resolve(name)
    client = ctx.client.build_candidate(endpoint)
    model_cache = ctx.model_manager.prepare_cache(getattr(ctx.session, "model", "") or "")
    return PreparedEndpointActivation(
        endpoint=endpoint,
        client=client,
        model_cache=model_cache,
        session=ctx.session,
    )
 
 
__all__ = [
    "BETWEEN_TURNS",
    "IN_TURN",
    "ChatLogLoadError",
    "IterationStatsSink",
    "LoadTransitionOutcome",
    "PreparedEndpointActivation",
    "PreparedSessionTransition",
    "STATS_CSV_HEADER",
    "TransitionRejectedError",
    "commit_endpoint_activation",
    "commit_transition",
    "consume_pending_sigint",
    "current_phase",
    "load_session_into",
    "mark_between_turns",
    "mark_in_turn",
    "note_pending_sigint",
    "prepare_endpoint_activation",
    "prepare_transition",
    "replace_session",
    "transition_entry_lease",
    "transition_in_progress",
]