session.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/session.py
 
"""
Session and token-stats data models.
 
Purpose
-------
Defines :class:`ChatSession` (the mutable conversation state) and
:class:`TokenStats` (accumulated per-session usage) as dataclasses,
plus helpers for auto-generating session names and for building a
system prompt that embeds tool schemas in a prefix-cache-friendly way.
 
Responsibilities
----------------
- :class:`ChatSession` — holds messages, model, reasoning effort,
  token-profile, name, dirty flag, last-turn prompt-token count, and
  per-session usage stats.
- :class:`TokenStats` — running totals + cache-hit/miss derived via
  :func:`oc_chat.core.cache_usage.usage_prompt_cache_hit_miss`.
- :func:`build_system_prompt` — compose the base policy + a conditional
  tool-use section + the compact tool map (plan 135 §3.3/§3.6); the full
  JSON schema travels only via the API ``tools`` parameter, never here.
- :func:`ensure_system_includes_tool_reference` — idempotent re-attach
  of the compact tool reference when a session is loaded from disk
  (recognises both the compact and the legacy sentinel).
- :func:`apply_session_metadata_fields` — update mutable session
  fields from a metadata dict (used by ``/session load``).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``ChatSession`` is mutable and not thread-safe; one instance per
  REPL.
- ``ensure_system_includes_tool_reference`` is idempotent via the
  compact-tool-map sentinel.
- ``record_usage`` accepts arbitrary dict shapes; missing keys do not
  raise.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.cache_usage (DeepSeek prompt-cache hit ratio reporting)
- oc_chat.core.config (shared config constants and ANSI styles)
- oc_chat.core.git_helpers (git_head_hash for the project-tree cache key)
- oc_chat.core.policy (canonical policy text and the tools and no-tools preambles)
- oc_chat.core.tool_prompt (compact tool map and its sentinel for the system prompt)
- oc_chat.core.deployment_shape (lazy; check_deployment_shape on a resumed sandbox-root override)
- oc_chat.core.event_bootstrap (lazy; _valid_alias to validate the event slugs)
- oc_chat.core.external_content_gate (lazy; session-trust reset and the canonical session-id check)
- oc_chat.core.persistence (lazy; list_sessions to avoid session-name collisions)
- oc_chat.core.provider_plugin (lazy; provider registry for the default model)
- oc_chat.core.risk_profile (lazy; resolve_risk_profile for the deployment-shape check)
- oc_chat.core.session_checkpoints (lazy; reset of the cost-checkpoint latches)
- oc_chat.core.session_mode (lazy; reset of the confirmation mode and remembered answers)
- oc_chat.core.settings (lazy; DEFAULT_SETTINGS project-tree defaults)
- oc_chat.core.startup (lazy; refresh_variable_context after a model switch)
- oc_chat.core.tier2_session_grant (lazy; revoke the session Tier-2 grants)
- oc_chat.core.wordgen (lazy; random_name for the default session name)
- oc_chat.tools._sandbox (lazy; install the session sandbox-root override)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §1.3 TokenStats, §1.4 ChatSession, §7 Session
message dict shape — owners of the per-session state contracts.
"""
import json
import subprocess
import uuid
from dataclasses import dataclass, field
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_chat.core.cache_usage import usage_prompt_cache_hit_miss
from oc_chat.core.config import (
    DEFAULT_MAX_REACT_ITERATIONS,
    DEFAULT_REASONING_EFFORT,
    TOKEN_PROFILES,
)
# Plan 135 §3.5: git logic relocated to git_helpers.py (kept cohesive). The
# private alias preserves the existing internal call site (project-tree cache key).
from oc_chat.core.git_helpers import git_head_hash as _git_head_hash
# Plan 135 §3.6/§3.3: shared policy text + compact tool map (no in-prompt schema dump).
from oc_chat.core.policy import CANONICAL_POLICY, NO_TOOLS_POLICY, TOOLS_POLICY_PREAMBLE
from oc_chat.core.tool_prompt import COMPACT_TOOL_MAP_SENTINEL, compact_tool_map
 
 
def _default_model() -> str:
    """Return the active provider's default model id, or empty if no
    provider has registered yet (rare — only happens in tests that
    construct a ``ChatSession`` without first importing a provider bundle).
    """
    func = "_default_model"
    import os
    from oc_chat.core.provider_plugin import registry
    name = os.environ.get("OC_CHAT_PROVIDER", "deepseek")
    try:
        return registry.get(name).default_model
    except (KeyError, AttributeError) as e:
        # VALIDATION_ERROR — provider not yet registered, expected in early test setup
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
        return ""
 
# Plan 135 §4.3: the default system persona is the canonical Option C policy.
# ``_DEFAULT_SYSTEM`` survives ONLY as this deliberate one-line alias for import
# stability; every other reference was swapped to ``CANONICAL_POLICY`` (the §4.7
# grep gate enforces this).
_DEFAULT_SYSTEM = CANONICAL_POLICY
 
# Appended to the system prompt when tool schemas are embedded (Phase 2 cache prefix).
SESSION_TOOL_SCHEMA_SENTINEL = "<!--dc:embedded-tool-schemas-->"
 
# Marks the user message that holds the per-session variable context
# (project tree + AGENT_CONTEXT.md).  Used by startup helpers to locate
# and refresh the message on session load without touching conversation history.
VARIABLE_CONTEXT_SENTINEL = "<!--oc:variable-context-->"
 
 
# Header that opened the legacy embedded-schema block (pre-135). Migration locates
# the block by THIS header (the JSON dump precedes the sentinel — §4.3), never by
# the sentinel itself.
_LEGACY_TOOL_BLOCK_HEADER = (
    # OCT-LINT: disable=no_hardcoded_secrets reason="Config keys and style strings, not secrets"
    "\n\n--\nOC Chat tool definitions (JSON Schema; duplicated from the API "
    "tools parameter for cache stability):"
)
 
 
def build_system_prompt(
    base_instruction: str, tool_schemas: list | None, *, reference: str = "compact"
) -> str:
    """Compose ``messages[0]``: base policy + a conditional tool-use section.
 
    Plan 135 §3.3/§3.6 — emits the persona/policy, then either the tool-use
    preamble + the compact tool map (when ≥1 tool is exposed) or the no-tools
    variant (no table). The full JSON Schema travels ONLY via the API ``tools``
    parameter; it is never embedded here. ``reference`` is reserved for future
    map variants (``"compact"`` today). Byte-stable for a given
    ``(base_instruction, tool_schemas)``.
    """
    text = (base_instruction or "").rstrip()
    if tool_schemas:
        return f"{text}\n\n{TOOLS_POLICY_PREAMBLE}\n\n{compact_tool_map(tool_schemas)}"
    return f"{text}\n\n{NO_TOOLS_POLICY}"
 
 
def ensure_system_includes_tool_reference(session: "ChatSession", tool_schemas: list | None) -> None:
    """Idempotently ensure ``messages[0]`` carries the tool reference.
 
    No-op when the system message already has the compact-map sentinel OR the
    legacy sentinel (a legacy session is converted by
    :func:`_migrate_legacy_system_prompt` on load, not re-dumped here). Only
    ``messages[0]`` (the immutable system prompt) is touched — never
    ``messages[1]`` (variable context).
    """
    if not tool_schemas or not session.messages:
        return
    first = session.messages[0]
    if first.get("role") != "system":
        return
    content = first.get("content") or ""
    if COMPACT_TOOL_MAP_SENTINEL in content or SESSION_TOOL_SCHEMA_SENTINEL in content:
        return
    first["content"] = build_system_prompt(content, tool_schemas)
    session.dirty = True
 
 
def _migrate_legacy_system_prompt(session: "ChatSession", tool_schemas: list | None) -> bool:
    """Convert a legacy embedded-schema ``messages[0]`` to the compact reference.
 
    Plan 135 §4.3: when the legacy sentinel is present, strip from the legacy
    block **header** (the JSON dump precedes the sentinel) through the
    quick-reference tail, keep everything before the header as the base prompt,
    and rebuild with the compact reference. Fail conservatively — if the sentinel
    is present but the header cannot be found (hand-edited/older layout), warn and
    leave ``messages[0]`` untouched rather than guess a boundary. Returns ``True``
    when a migration was applied.
    """
    func = "_migrate_legacy_system_prompt"
    if not session.messages:
        return False
    first = session.messages[0]
    if first.get("role") != "system":
        return False
    content = first.get("content") or ""
    if SESSION_TOOL_SCHEMA_SENTINEL not in content:
        return False
    idx = content.rfind(_LEGACY_TOOL_BLOCK_HEADER)
    if idx == -1:
        _dbg("OC_CHAT.MAIN", func,
             # OCT-LINT: disable=no_hardcoded_secrets reason="Config keys and style strings, not secrets"
             "WARNING: legacy schema sentinel present but block header not found; "
             "leaving messages[0] untouched (conservative-fail)", 1, override=True)
        return False
    base = content[:idx].rstrip()
    first["content"] = build_system_prompt(base, tool_schemas)
    session.dirty = True
    _dbg("OC_CHAT.MAIN", func, "migrated legacy embedded-schema prompt to compact reference", 2)
    return True
 
 
def _auto_name() -> str:
    """Generate a default session name from random words.
 
    Returns kebab-case names like ``sunny-weaving-honey``.
    Checks existing sessions to avoid collisions.
    """
    from oc_chat.core.wordgen import random_name
    from oc_chat.core.persistence import list_sessions
    existing = {s["name"] for s in list_sessions()}
    return random_name(exists_fn=lambda n: n in existing)
 
 
@dataclass
class SessionCost:
    """Session-scoped USD spend, folded honestly (CAPS-01, plan 237 P2).
 
    One fold over the SAME samples the ``provider_budget_consumed`` events
    carry, priced by the one authoritative source (:mod:`oc_pricing`, through
    ``react_loop._publish_provider_budget``). It is never re-derived from
    tokens by a second formula: two formulas would disagree, and the number
    that stops a session must not be the one nobody can reproduce.
 
    The honesty rule (plan 188 T4, inherited): a sample that cannot be priced
    is COUNTED, never treated as zero. ``priced_usd`` is therefore a
    lower bound on real spend, and ``unpriced_samples`` is how far it may be
    wrong — which is why both travel together into the status line, the mini
    report, the alert events and the hard-cap refusal. The residual is stated
    rather than hidden: unpriced spend cannot trip the hard cap.
 
    ``thresholds_fired_usd`` records which checkpoints this session has already
    announced, so a reload resumes rather than re-announcing them.
    """
 
    priced_usd: float = 0.0
    unpriced_samples: int = 0
    thresholds_fired_usd: list = field(default_factory=list)
 
    def add_sample(self, cost_usd: "float | None") -> None:
        """Fold ONE priced sample, or count one unpriceable sample."""
        if cost_usd is None:
            self.unpriced_samples += 1
            return
        self.priced_usd += float(cost_usd)
 
    def mark_threshold_fired(self, threshold: float) -> None:
        """Record that *threshold* has been announced for this session."""
        if threshold not in self.thresholds_fired_usd:
            self.thresholds_fired_usd.append(float(threshold))
 
    def has_fired(self, threshold: float) -> bool:
        """True when *threshold* was already announced this session."""
        return any(abs(float(t) - float(threshold)) < 1e-9
                   for t in self.thresholds_fired_usd)
 
    def to_dict(self) -> dict:
        """Return a JSON-serialisable dict of the session cost fold."""
        return {
            "priced_usd": float(self.priced_usd),
            "unpriced_samples": int(self.unpriced_samples),
            "thresholds_fired_usd": [float(t) for t in self.thresholds_fired_usd],
        }
 
    @classmethod
    def from_dict(cls, d: object) -> "SessionCost":
        """Reconstruct from a persisted dict; anything unusable yields a fresh zero.
 
        Read tolerantly: a pre-237 ``meta.json`` carries no cost block at all,
        and a corrupt one must start the session at zero rather than refuse to
        load it. Starting at zero is the SAFE direction — the hard cap and the
        checkpoints then fire sooner, never later."""
        if isinstance(d, cls):
            return d
        if not isinstance(d, dict):
            return cls()
        try:
            fired = [float(t) for t in (d.get("thresholds_fired_usd") or [])]
        except (TypeError, ValueError):
            fired = []
        try:
            priced = float(d.get("priced_usd", 0.0) or 0.0)
        except (TypeError, ValueError):
            priced = 0.0
        try:
            unpriced = int(d.get("unpriced_samples", 0) or 0)
        except (TypeError, ValueError):
            unpriced = 0
        return cls(priced_usd=max(0.0, priced),
                   unpriced_samples=max(0, unpriced),
                   thresholds_fired_usd=fired)
 
 
@dataclass
class TokenStats:
    """Per-session aggregated token usage.
 
    Tracks prompt, completion, and total tokens alongside message and
    tool-call counts.  Can be serialised via :meth:`to_dict` and
    reconstructed via :meth:`from_dict`.
    """
    total_prompt_tokens: int = 0
    total_completion_tokens: int = 0
    total_tokens: int = 0
    message_count: int = 0
    tool_call_count: int = 0
    prompt_cache_hit_tokens: int = 0
    prompt_cache_miss_tokens: int = 0
    # CAPS-01 (plan 237 P2): the session USD fold rides HERE rather than as a
    # sibling field on ChatSession because ``TokenStats.to_dict``/``from_dict``
    # is already the one persistence seam every restore path goes through
    # (startup, the 152-b transition seam, batch_runner, list_sessions). A
    # sibling would have needed four call-site edits, three of them in modules
    # this plan's isolation table does not open.
    cost: SessionCost = field(default_factory=SessionCost)
 
    def add_cost_sample(self, cost_usd: "float | None") -> None:
        """Fold one provider usage sample's USD cost (``None`` = unpriceable)."""
        self.cost.add_sample(cost_usd)
 
    def add_usage(self, usage: dict) -> None:
        """Accumulate token usage from a single API response."""
        self.total_prompt_tokens += usage.get("prompt_tokens", 0)
        self.total_completion_tokens += usage.get("completion_tokens", 0)
        self.total_tokens += usage.get("total_tokens", 0)
        hit, miss = usage_prompt_cache_hit_miss(usage)
        self.prompt_cache_hit_tokens += hit
        self.prompt_cache_miss_tokens += miss
 
    def add_message(self) -> None:
        """Increment the user/assistant message counter."""
        self.message_count += 1
 
    def add_tool_call(self) -> None:
        """Increment the tool-call counter."""
        self.tool_call_count += 1
 
    def to_dict(self) -> dict:
        """Return a JSON-serialisable dict of the stats."""
        return {
            "message_count": self.message_count,
            "tool_call_count": self.tool_call_count,
            "total_prompt_tokens": self.total_prompt_tokens,
            "total_completion_tokens": self.total_completion_tokens,
            "total_tokens": self.total_tokens,
            "prompt_cache_hit_tokens": self.prompt_cache_hit_tokens,
            "prompt_cache_miss_tokens": self.prompt_cache_miss_tokens,
            # CAPS-01: additive key. A pre-237 reader ignores it; a 237 reader
            # given a pre-237 dict starts the fold at zero (see SessionCost).
            "session_cost": self.cost.to_dict(),
        }
 
    @classmethod
    def from_dict(cls, d: dict) -> "TokenStats":
        """Reconstruct a :class:`TokenStats` from a dict (e.g. loaded JSON).
 
        If *d* is already a ``TokenStats`` instance it is returned unchanged,
        so callers that receive either a raw dict or a live object are safe.
        """
        if isinstance(d, cls):
            return d
        return cls(
            total_prompt_tokens=d.get("total_prompt_tokens", 0),
            total_completion_tokens=d.get("total_completion_tokens", 0),
            total_tokens=d.get("total_tokens", 0),
            message_count=d.get("message_count", 0),
            tool_call_count=d.get("tool_call_count", 0),
            prompt_cache_hit_tokens=d.get("prompt_cache_hit_tokens", 0),
            prompt_cache_miss_tokens=d.get("prompt_cache_miss_tokens", 0),
            cost=SessionCost.from_dict(d.get("session_cost")),
        )
 
 
@dataclass
class ChatSession:
    """Mutable state for a single chat conversation.
 
    Tracks the message list, model selection, reasoning effort, token
    stats, associated chat logs, and a dirty flag for auto-save logic.
    """
 
    name: str = field(default_factory=_auto_name)
    # M2.5 — stable per-session identity. Minted once at construction, persisted
    # in meta.json, and is the canonical source for ``TurnEvent.session_id`` on
    # every M2 event (retires the M2.3 ``_session_id_for`` shim). Pre-M2.5
    # sessions mint a fresh id on first load under M2.5 and keep it thereafter.
    session_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    model: str = field(default_factory=_default_model)
    reasoning_effort: str = field(default_factory=lambda: DEFAULT_REASONING_EFFORT)
    messages: list = field(default_factory=lambda: [
        {"role": "system", "content": CANONICAL_POLICY}
    ])
    dirty: bool = False
    filepath: str | None = None
    stats: TokenStats = field(default_factory=TokenStats)
    chat_logs: list = field(default_factory=list)  # informational: filename of current chat log
    load_failed: bool = False  # set when a previous load_chat_log raised — guards against overwrite
    ui_prefs: dict = field(default_factory=dict)
    # OCT-LINT: disable=no_hardcoded_secrets reason="Config keys and style strings, not secrets"
    """Per-session UI preferences (theme, show_reasoning, show_stats, …).
 
    Persisted under the ``ui`` key in ``meta.json``.  See ``/ui`` command.
    Phase 0 stores these but only :class:`RichPresenter` (Phase 1+) will
    visibly honour them.
    """
 
    # Phase 3 — largest ``prompt_tokens`` seen on the last completed chat turn (API truth).
    last_turn_max_prompt_tokens: int = 0
    # One-time soft nudge after many user turns (session rollover UX).
    rollover_hint_shown: bool = False
 
    # Phase 4 — token budget profile for completion scaling (``/budget``).
    # OCT-LINT: disable=no_hardcoded_secrets reason="Config keys and style strings, not secrets"
    token_profile: str = 'balanced'
    # Optional override for ReAct max iterations (``/react N``); ``None`` = use default.
    max_react_iterations: int | None = None
 
    # Phase J — id of the plan this session is currently focused on.
    # Value = filename stem without suffix or ``.md`` (e.g. ``032-sunny-weaving-honey``).
    # ``None`` when no plan is selected. Persisted in ``meta.json``.
    active_plan_id: str | None = None
 
    # Phase O4 — per-session generation options (temperature, top_k, etc.).
    # Stored as a plain dict to decouple from the Pydantic model at serialisation time.
    generation_options: dict = field(default_factory=dict)
 
    # Phase O2 — name of the EndpointRegistry entry the session last
    # used (e.g. ``"laptop-gpu"``). Empty string means "use the
    # registry's default active entry on session load".
    endpoint_name: str = ""
 
    # Per-iteration stats file handle (--stats-file PATH).  Opened by main,
    # used by the ReAct loop, closed on session save/exit.  Not serialised.
    # Gated by --no-iteration-stats CLI flag.
    show_iteration_stats: bool = True
    iteration_stats_file: object = field(default=None, repr=False, compare=False)
    # Recent per-turn cache snapshots for ``/cache`` (max length trimmed on append).
    cache_turn_snapshots: list = field(default_factory=list)
 
    # M0.4 — optional per-session filesystem sandbox override.
    sandbox_root_override: str | None = None
 
    # Plan 152 D8 — event-dir rename continuity. Event dirs key on the immutable
    # ``session_id``; these are the name-SLUGS this session wrote under before
    # earlier renames, so ``/trace`` and ``/packet`` can still reach that
    # history. Ordered (multi-epoch A -> B -> C renames are real), validated
    # against the canonical slug grammar, and deduplicated. Written ONLY by the
    # rename transition's own commit — event-dir discovery is read-only.
    legacy_event_slugs: list = field(default_factory=list)
 
    # Plan 152 D8/R11-09 — set on the OLD record when a rename supersedes it, so
    # the on-disk same-session_id sibling every rename creates is deterministically
    # non-canonical from birth instead of being adjudicated by load order.
    superseded_by: str | None = None
 
    def add_user(self, text: str) -> None:
        """Append a user message and mark the session dirty."""
        func = "add_user"
        _dbg("OC_CHAT.MAIN", func, f"chars={len(text)} session={self.name}", 3)
        self.messages.append({"role": "user", "content": text})
        self.dirty = True
 
    def add_assistant(self, text: str) -> None:
        """Append an assistant message, mark dirty, and bump the message count."""
        self.messages.append({"role": "assistant", "content": text})
        self.dirty = True
        self.stats.add_message()
 
    def add_assistant_with_tool_calls(self, msg: dict) -> None:
        """Append an assistant message that contains ``tool_calls`` (added after streaming)."""
        self.messages.append(msg)
        self.dirty = True
        self.stats.add_message()
 
    def add_tool_result(self, result: dict) -> None:
        """Append a tool-result message, mark dirty, and count the tool call."""
        self.messages.append(result)
        self.dirty = True
        self.stats.add_tool_call()
 
    def record_usage(self, usage: dict) -> None:
        """Record token usage from an API response chunk."""
        self.stats.add_usage(usage)
 
    def clear(self) -> None:
        """Reset the conversation, keeping only the system prompt."""
        self.messages = [{"role": "system", "content": self.messages[0]["content"]}]
        self.last_turn_max_prompt_tokens = 0
        self.rollover_hint_shown = False
        self.cache_turn_snapshots = []
        self.max_react_iterations = None
        self.dirty = True
        # M8.0 F67: clearing the conversation also drops session-trusted external
        # sources (an in-place /session clear keeps the same session object, so the
        # per-turn session-change reset would not catch it).
        from oc_chat.core.external_content_gate import reset_session_trust
        reset_session_trust()
        # Plan 235 (SHELL-02): the session-long Tier-2 python grant is session-
        # scoped for the same reason and drops at the same seam.
        from oc_chat.core.tier2_session_grant import revoke_session_grants
        revoke_session_grants()
        # Plan 237 (CAPS-01): a cleared conversation is a fresh session, so its
        # money starts at zero and every checkpoint may fire again. BOTH halves
        # drop here — the persisted fold (this object) and the process-scoped
        # "don't ask again" latches, which are the exact analogue of the F67
        # trust set above.
        from oc_chat.core.session_checkpoints import reset_session_checkpoints
        self.stats.cost = SessionCost()
        reset_session_checkpoints()
        # Plan 283 (SHELL-04): the confirmation mode and the per-program answers
        # this session was told to remember are the same kind of state as the
        # latches above — a standing answer nobody granted twice must not survive
        # into what is, from here on, a different conversation.
        from oc_chat.core.session_mode import reset_session_mode
        reset_session_mode()
 
    def set_system_prompt(self, prompt: str) -> None:
        """Replace the system prompt and mark the session dirty."""
        self.messages[0] = {"role": "system", "content": prompt}
        self.dirty = True
 
    def set_model(self, new_model: str) -> None:
        """Switch the active model through the single choke-point (plan 135 §3.5).
 
        Contract: (1) no-op if ``new_model == self.model``. (2) Otherwise set the
        model and mark the session dirty FIRST — a user-initiated switch must
        persist even when :func:`refresh_variable_context` finds no sentinel
        (e.g. an ``include_env:false`` / legacy / context-free session) — THEN
        refresh the ``messages[1]`` environment block in place so ``Model:`` (and
        the git/dirty facts) stay accurate. Only ever called outside the turn loop
        (the inner loop holds ``session.model`` fixed per turn — inner-loop.md),
        so it cannot perturb a live turn. Load paths call this AFTER ``messages``
        hydration and may clear ``dirty`` afterwards (loading is not a user edit).
        """
        if new_model == self.model:
            return
        self.model = new_model
        self.dirty = True
        # Deferred import: startup imports session at module load (avoid a cycle).
        from oc_chat.core.startup import refresh_variable_context
 
        refresh_variable_context(self)
 
    def mark_clean(self) -> None:
        """Clear the dirty flag (e.g. after a successful save)."""
        self.dirty = False
 
    @property
    def system_prompt(self) -> str:
        """Return the current system prompt text."""
        return self.messages[0]["content"]
 
    @property
    def message_count(self) -> int:
        """Return the number of user + assistant messages (not system/tool)."""
        return sum(1 for m in self.messages if m["role"] in ("user", "assistant"))
 
    def react_cap(self) -> int:
        """Effective ReAct iteration cap (``/react`` override or default).
 
        Returns ``self.max_react_iterations`` when it is an int in
        ``[1, 99]``; otherwise falls back to
        :data:`DEFAULT_MAX_REACT_ITERATIONS`.
        """
        v = self.max_react_iterations
        if isinstance(v, int) and 1 <= v <= 99:
            return v
        return DEFAULT_MAX_REACT_ITERATIONS
 
    def add_chat_log(self, filename: str) -> None:
        """Record a chat log filename associated with this session."""
        if filename not in self.chat_logs:
            self.chat_logs.append(filename)
            self.dirty = True
 
 
 
# ── Project tree cache (keyed by git HEAD hash) ─────────────────────────
 
_GEN_TREE_CACHE: tuple[str, str] | None = None
 
 
def _gitignored_relpaths(root: Path) -> frozenset[str]:
    """Relative paths git ignores under *root* (best-effort; '' on any error).
 
    Uses git's own ignore logic via ``ls-files --others --ignored
    --exclude-standard --directory`` so the tree stays self-maintaining when
    the operator gitignores a dir. Returns forward-slash relpaths, wholly-
    ignored dirs without a trailing slash."""
    try:
        out = subprocess.run(
            ["git", "-C", str(root), "ls-files", "--others", "--ignored",
             "--exclude-standard", "--directory"],
            capture_output=True, text=True, timeout=5, check=False,
        )
    except (OSError, subprocess.SubprocessError):
        return frozenset()
    return frozenset(
        ln.strip().rstrip("/").replace("\\", "/")
        for ln in out.stdout.splitlines() if ln.strip()
    )
 
 
def generate_project_tree(
    max_depth: int = 2,
    *,
    exclude_dirs: "list[str] | None" = None,
    exclude_suffixes: "list[str] | None" = None,
    stub_dirs: "list[str] | None" = None,
    honor_gitignore: bool | None = None,
    work_dir: str | None = None,
) -> str:
    """Build a compact project tree for injection into the system prompt.
 
    Cached by (git HEAD hash + config signature) so repeated session starts
    don't re-scan the filesystem.  Excludes hidden dirs (dot-prefixed), the
    configured ``exclude_dirs``, names ending in any ``exclude_suffixes``
    (e.g. ``.bk``), the ``work_dir`` scratch area, and anything git ignores
    when ``honor_gitignore``.  Dirs in ``stub_dirs`` (e.g. ``sessions``,
    ``plans``) are shown as existence-only with a ``list_dir`` hint so their
    high-churn contents don't bloat the prompt.
 
    Config defaults come from :data:`oc_chat.core.settings.DEFAULT_SETTINGS`
    so a no-arg call matches a settings-default call (plan 133).
    """
    global _GEN_TREE_CACHE
    from oc_chat.core.settings import DEFAULT_SETTINGS
    _pt = DEFAULT_SETTINGS["project_tree"]
 
    excl_dirs = frozenset(exclude_dirs if exclude_dirs is not None else _pt["exclude_dirs"])
    excl_suffixes = tuple(exclude_suffixes if exclude_suffixes is not None else _pt["exclude_suffixes"])
    excl_suffixes = excl_suffixes + (".pyc", ".pyo")
    stubs = frozenset(stub_dirs if stub_dirs is not None else _pt.get("stub_dirs", []))
    honor_gi = _pt["honor_gitignore"] if honor_gitignore is None else honor_gitignore
    wdir = (_pt["work_dir"] if work_dir is None else work_dir) or None
 
    root = Path.cwd().resolve()
    gitignored = _gitignored_relpaths(root) if honor_gi else frozenset()
 
    cfg_sig = repr((max_depth, sorted(excl_dirs), excl_suffixes, sorted(stubs),
                    honor_gi, wdir, sorted(gitignored)))
    cache_key = (_git_head_hash() or "no-git", cfg_sig)
    if _GEN_TREE_CACHE is not None and _GEN_TREE_CACHE[0] == cache_key:
        return _GEN_TREE_CACHE[1]
 
    lines: list[str] = []
    lines.append("## Project Map (max depth {})".format(max_depth))
 
    def _walk(parent: Path, rel_parent: str, depth: int) -> None:
        if depth > max_depth:
            return
        try:
            entries = sorted(parent.iterdir(), key=lambda e: (not e.is_dir(), e.name.lower()))
        except PermissionError:
            return
        for entry in entries:
            name = entry.name
            if name.startswith("."):
                continue
            if name.endswith(excl_suffixes):
                continue
            rel = f"{rel_parent}/{name}" if rel_parent else name
            if rel in gitignored:
                continue
            if entry.is_dir():
                if name in excl_dirs or (wdir and name == wdir):
                    continue
                if name in stubs:
                    lines.append(
                        "    " * depth + "--- " + name
                        + f'/  (contents omitted; use list_dir("{name}/") to enumerate)'
                    )
                    continue
                lines.append("    " * depth + "--- " + name + "/")
                if depth < max_depth:
                    _walk(entry, name, depth + 1)
            else:
                lines.append("    " * depth + "--- " + name)
 
    _walk(root, "", 0)
    tree = "\n".join(lines)
    _GEN_TREE_CACHE = (cache_key, tree)
    return tree
 
 
def _reset_trust_before_exposure() -> None:
    """Clear F67 external-content trust BEFORE a regenerated identity is exposed.
 
    A regenerated ``session_id`` means we could not establish which record this
    is. Whatever trust the process had accumulated belonged to a different
    identity, so it must not carry over. Best-effort by design: a diagnostics
    failure must never block a session load.
    """
    func = "_reset_trust_before_exposure"
    try:
        from oc_chat.core.external_content_gate import reset_session_trust
        reset_session_trust()
        # Plan 235 (SHELL-02): the Tier-2 python grant belonged to the identity we
        # could not establish either, so it drops before the new one is exposed.
        from oc_chat.core.tier2_session_grant import revoke_session_grants
        revoke_session_grants()
        # Plan 237 (CAPS-01): the "don't ask again this session" latches belonged
        # to that identity too, so they drop here as well. The PERSISTED cost fold
        # deliberately does NOT reset at this seam: every caller restores
        # ``session.stats`` from the record's own meta.json BEFORE reaching here
        # (startup._load_session_by_meta, session_transition), so zeroing it here
        # would erase the total the record itself carries and silently hand a
        # resumed session a fresh $50 of headroom. It resets in
        # ``ChatSession.clear``, which is where a session genuinely starts over.
        from oc_chat.core.session_checkpoints import reset_session_checkpoints
        reset_session_checkpoints()
        # Plan 283 (SHELL-04): the Auto mode and the remembered answers belonged
        # to the identity we could not establish either, so they drop before the
        # new one is exposed.
        from oc_chat.core.session_mode import reset_session_mode
        reset_session_mode()
    except Exception as exc:                          # pragma: no cover - defensive
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: trust reset failed: {exc}", 2)
 
 
def _validated_event_slugs(raw: object) -> list:
    """Return the persisted rename aliases that pass the canonical slug grammar.
 
    Untrusted persisted path input: accepted values are EXACTLY canonical
    :func:`~oc_chat.core.event_bootstrap._slug` outputs, which rejects
    separators, ``.``/``..``, absolute / drive / UNC forms and every
    non-canonical spelling in one predicate. Order is preserved and duplicates
    are dropped — a multi-epoch A -> B -> C rename needs the whole ordered
    collection, and one alias cannot represent it.
 
    Invalid entries are dropped with a COUNTED diagnostic that never echoes the
    value.
    """
    func = "_validated_event_slugs"
    if not isinstance(raw, (list, tuple)):
        return []
    from oc_chat.core.event_bootstrap import _valid_alias
    out: list = []
    dropped = 0
    for item in raw:
        if _valid_alias(item) and item not in out:
            out.append(item)
        elif not _valid_alias(item):
            dropped += 1
    if dropped:
        _dbg("OC_CHAT.MAIN", func,
             f"VALIDATION_ERROR: dropped {dropped} invalid legacy_event_slugs entr(ies)", 2)
    return out
 
 
def apply_session_metadata_fields(
    session: "ChatSession", meta: dict, *, pure: bool = False
) -> None:
    """Restore persisted session fields from a ``meta.json`` dict (Phase 3–4).
 
    ``pure`` (plan 152-b D1) is passed ONLY by the load/switch transition seam,
    which hydrates a detached CANDIDATE and defers every live effect to the
    commit critical section. It suppresses EXACTLY the two live-state mutations
    this function otherwise performs directly:
 
    - it does NOT call :func:`_reset_trust_before_exposure` — F67 trust rotation
      is prepared and installed by the seam (``prepare_session_binding`` /
      ``install_prepared_session_binding``), keyed on the candidate's final id;
    - it does NOT call ``set_session_sandbox_root_override`` — the seam resolves
      the override in PREPARE and installs the process-global at commit.
 
    Every FIELD assignment (including ``session.sandbox_root_override``, which the
    seam's ``prepare_transition`` reads back off the candidate) still lands, so a
    pure call fully hydrates the candidate object and returns ``None``. The
    default (``pure=False``) is byte-compatible legacy behaviour for the three
    non-seam callers (startup resume, main startup, batch).
    """
    func = "apply_session_metadata_fields"
    # M2.5 — restore the persisted session_id; keep the construction-minted UUID
    # when the meta predates M2.5 (back-compat) so the id stays stable thereafter.
    #
    # Plan 152 D9 — VALIDATE AT THE PERSISTENCE BOUNDARY. This used to accept any
    # non-empty string, so an arbitrary persisted value became the F67 external
    # -content trust key. A trust key derived from an unvalidated string is not a
    # complete gate, and every neighbouring field here is already validated.
    #
    # On a malformed OR missing id the session ends up with a fresh canonical
    # uuid4 rather than the persisted value — never trusted as-is. Rejecting the
    # load instead would make legacy pre-id artifacts unloadable (anti-lock-in:
    # the user owns their sessions).
    #
    # "Fresh" does NOT mean "always mint a new one". ``ChatSession.__init__``
    # already minted a canonical uuid4 that has never been on disk and has never
    # been trusted, so when the persisted id is unusable the minted one already
    # satisfies the property — keeping it is what the pre-existing MISSING-id
    # back-compat contract guarantees, and minting a second uuid4 would break
    # that for no security gain. A new one is minted only if the in-memory id is
    # itself not canonical.
    from oc_chat.core.external_content_gate import is_canonical_session_id
    sid = meta.get("session_id")
    if is_canonical_session_id(sid):
        session.session_id = sid  # type: ignore[assignment]
    else:
        if not is_canonical_session_id(getattr(session, "session_id", None)):
            session.session_id = str(uuid.uuid4())
        if sid is not None:
            _dbg("OC_CHAT.MAIN", func,
                 # OCT-LINT: disable=no_hardcoded_secrets reason="Config keys and style strings, not secrets"
                 f"VALIDATION_ERROR: persisted session_id is not a canonical uuid4 "
                 f"(type={type(sid).__name__}); kept a freshly-minted id instead", 3)
        # The persisted identity could not be established, so whatever trust this
        # process accumulated belonged to a different identity. Under ``pure`` the
        # seam owns rotation (keyed on the candidate's final id) and installs it
        # at commit BEFORE identity publication — so the live trust set must NOT
        # be cleared here, where the candidate is not yet the active session.
        if not pure:
            _reset_trust_before_exposure()
 
    # Plan 152 D8/R11-09 — the ordered, validated, deduplicated rename-alias
    # collection. Re-validated AT USE by ``event_read_dirs`` too, because
    # ``load_session_metadata`` returns the RAW dict on a validation error, which
    # makes model-level validation alone bypassable.
    session.legacy_event_slugs = _validated_event_slugs(meta.get("legacy_event_slugs"))
    session.last_turn_max_prompt_tokens = int(meta.get("last_turn_max_prompt_tokens", 0) or 0)
    session.rollover_hint_shown = bool(meta.get("rollover_hint_shown", False))
    tp = meta.get("token_profile", "balanced")
    session.token_profile = tp if tp in TOKEN_PROFILES else "balanced"
    mr = meta.get("max_react_iterations")
    session.max_react_iterations = mr if isinstance(mr, int) and 1 <= mr <= 99 else None
    session.cache_turn_snapshots = list(meta.get("cache_turn_snapshots") or [])[-30:]
    apv = meta.get("active_plan_id")
    session.active_plan_id = apv if isinstance(apv, str) and apv else None
    raw_gen = meta.get("generation_options")
    session.generation_options = dict(raw_gen) if isinstance(raw_gen, dict) else {}
    raw_endpoint = meta.get("endpoint_name", "")
    session.endpoint_name = raw_endpoint if isinstance(raw_endpoint, str) else ""
    raw_sandbox = meta.get("sandbox_root_override")
    candidate = raw_sandbox if isinstance(raw_sandbox, str) and raw_sandbox else None
    # AC-V09-5 (M8.0 WU-1): validate the persisted override at apply-time and fail
    # closed to the launch-time root if it does not resolve to an existing
    # directory -- mirroring the existence check ``/sandbox set`` already enforces,
    # so a tampered ``meta.json`` cannot relocate the write sandbox to a planted /
    # non-existent path on resume (M8 F48/F69). Authenticity (operator-signed vs
    # injection-planted-but-valid) is the M8.3 persistence keyed MAC (F48a); this
    # is the deterministic floor.
    if candidate is not None:
        try:
            valid = Path(candidate).expanduser().resolve(strict=False).is_dir()
        except (OSError, ValueError):
            valid = False
        if not valid:
            _dbg(
                "OC_CHAT.MAIN",
                func,
                f"rejected sandbox_root_override {candidate!r} (not an existing "
                f"directory); failing closed to launch-time root",
                2,
            )
            candidate = None
    if candidate is not None:
        # WU-7 / G-1: under guarded/isolated, fail closed if the persisted override
        # would put the write sandbox in overlap with a harness importable root
        # (write->import in-process ACE / F74 MAC preimage). The launch-time root
        # already passed this at startup; this guards the resume apply-path (F48/F57).
        try:
            from oc_chat.core.risk_profile import resolve_risk_profile
            from oc_chat.core.deployment_shape import check_deployment_shape, DeploymentShapeError
            check_deployment_shape(candidate, resolve_risk_profile())
        except DeploymentShapeError:
            _dbg("OC_CHAT.MAIN", func,
                 f"rejected sandbox_root_override {candidate!r} (deployment-shape "
                 f"overlap under guarded/isolated); failing closed to launch-time root", 2)
            candidate = None
    session.sandbox_root_override = candidate
    # Under ``pure`` the process-global is a LIVE effect the seam owns: it
    # resolves the override in PREPARE (off ``session.sandbox_root_override``,
    # just assigned above) and installs it inside the commit critical section.
    # Touching the global here would mutate live state from a non-active
    # candidate — exactly what the pure hydration flag exists to prevent.
    if not pure:
        try:
            from oc_chat.tools._sandbox import set_session_sandbox_root_override
            set_session_sandbox_root_override(session.sandbox_root_override)
        except Exception as exc:
            _dbg(
                "OC_CHAT.MAIN",
                func,
                # OCT-LINT: disable=no_hardcoded_secrets reason="Config keys and style strings, not secrets"
                f"VALIDATION_ERROR: sandbox override sync failed: {exc}",
                3,
            )