persistence.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/persistence.py
 
"""
Session persistence layer.
 
Purpose
-------
Read and write session metadata, chat logs, exports, and tool
artifacts to disk; preserve prior versions in an ``archive/``
subdirectory so the user never silently loses history.
 
Storage layout (per-session directory, with timestamped filenames)::
 
    oc_chat/sessions/
      archive/{name}/                         # archived (pre-overwrite) snapshots
        {name}_{YYYY-MM-DD--HH-MM-SS}_meta.json
        {name}_{YYYY-MM-DD--HH-MM-SS}_chat_log.json
      legacy/                                 # pre-session-system orphan logs
        chat_log_*.json
      {session_name}/                         # one directory per live session
        {name}_{YYYY-MM-DD--HH-MM-SS}_meta.json       # exactly one current
        {name}_{YYYY-MM-DD--HH-MM-SS}_chat_log.json   # exactly one current
        {name}_{YYYY-MM-DD--HH-MM-SS}_export.md       # only if /session export run
 
Every save first archives the existing same-kind file (preserving its
original timestamped filename) before writing the new one with a
fresh timestamp. Identical-content moves are deduplicated at the
archive level to avoid pointless growth.
 
Session names are filesystem-safe by construction: :func:`fs_safe_name`
maps a name to a single legal path component at the persistence
boundary, so ``{session_name}`` above is always exactly one directory
level deep and always a legal filename prefix (plan 225 / LANES-04).
 
Responsibilities
----------------
- :func:`fs_safe_name` — the one name → path-component mapping, applied
  wherever a session name reaches the filesystem.
- :func:`save_session_metadata`, :func:`save_chat_log` — write the
  two canonical session files with archive-on-overwrite semantics.
- :func:`detect_incomplete_tail` — identify crash-shaped incomplete
  message tails before a loaded session is resumed.
- :func:`save_tool_artifact` — persist large tool outputs out-of-band
  so the API context can keep a truncated stub.
- :func:`list_sessions` — enumerate live session directories with
  unreadable files logged at L2 and skipped (not raised).
- :func:`migrate_legacy_layout` — one-shot migration from the pre-
  session-system layout where chat logs lived at the package root.
 
Diagnostics
-----------
Domain: OC_CHAT.CTX
Levels:
    L2 — skip-unreadable warnings; archive transitions
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- All public save_* functions are atomic w.r.t. partial writes (write
  to a temp path then ``os.replace``).
- Hashes are computed with ``hashlib.sha256`` over canonical JSON to
  detect identical-content archive collisions.
- Functions never raise on a single unreadable session — they log and
  skip so ``/session list`` keeps working.
- A stored record satisfies ``metadata["session_name"] == <directory
  name> == <stamped filename prefix>``; the save boundary enforces it
  via :func:`fs_safe_name` and normalises the live session to match.
- :func:`fs_safe_name` is idempotent and is the identity on any name
  that is already a legal path component, so it is a no-op for every
  session saved before plan 225.
 
Environment variables
---------------------
- ``OC_CHAT_SESSIONS_DIR`` — override :data:`DEFAULT_SESSIONS_DIR` at
  import time (default: ``"oc_chat/sessions"``). Lets users relocate
  the sessions store outside the repo so plain filesystem archive
  copies of the working tree don't drag the ~500MB session history
  along. An explicit ``sessions_dir=`` argument to any persistence
  function still wins over the env-var default.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.session (ChatSession dataclass and TokenStats)
- oc_chat.core.session_models (SessionMetadata and ChatLog validation of loaded sidecars)
- pydantic (ValidationError when a loaded sidecar fails its model)
- rich (lazy; Console, Markdown, Panel, Rule and Text for the styled HTML export)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §7 Session message dict shape — owner of the
JSON sidecar shapes for meta.json and chat_log.json (see also gap
analysis §9 for the SessionMetadata / ChatLog Pydantic candidates).
"""
 
from __future__ import annotations
 
import hashlib
import json
import os
import shutil
import sys
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Literal
from uuid import uuid4
 
from oc_diagnostics import _dbg
from pydantic import ValidationError
 
from oc_chat.core.session import TokenStats, VARIABLE_CONTEXT_SENTINEL
from oc_chat.core.session_models import ChatLog, SessionMetadata
 
if TYPE_CHECKING:
    from oc_chat.core.session import ChatSession
 
DEFAULT_SESSIONS_DIR = os.environ.get("OC_CHAT_SESSIONS_DIR", "oc_chat/sessions")
ARCHIVE_DIR_NAME = "archive"
LEGACY_DIR_NAME = "legacy"
CLEAN_EXIT_MARKER_NAME = ".oc_chat_clean_exit"
 
# ─── Filesystem-safe session names (plan 225 / LANES-04) ─────────────────────
 
#: What a name that maps to nothing usable degrades to.  Chosen to be
#: recognisable in a session list rather than cryptic.
FS_SAFE_PLACEHOLDER = "unnamed-session"
 
#: Path separators.  Mapped to a DOUBLE hyphen rather than a single one so the
#: vendor boundary of a model id stays readable (``upstage/solar-pro4`` reads as
#: ``upstage--solar-pro4``) and so the mapping is unlikely to collide with a
#: hand-typed name that already contains a hyphen.
_FS_SEPARATORS = ("/", "\\")
 
#: Characters Windows rejects outright in a path component.
_FS_RESERVED_CHARS = ':*?"<>|'
 
#: Windows reserved device names.  A directory called ``CON`` cannot be created
#: at all, which is the same failure class as the embedded separator.
_FS_RESERVED_DEVICE_NAMES = frozenset(
    {"con", "prn", "aux", "nul"}
    | {f"com{i}" for i in range(1, 10)}
    | {f"lpt{i}" for i in range(1, 10)}
)
 
 
def fs_safe_name(name: str) -> str:
    """Map *name* to a string that is always usable as ONE path component.
 
    A session name reaches the filesystem three times — as the session
    directory, as the archive directory, and as the prefix of every stamped
    filename.  Before plan 225 the name was passed through verbatim, so a
    vendor-prefixed model id used as a session name (``upstage/solar-pro4``)
    silently created TWO directory levels for the directory while embedding the
    same separator in the FILENAME, producing a path one uncreated directory
    deeper and an ``[Errno 2]`` on every auto-save (LANES-04).
 
    The mapping is:
 
    - ``/`` and ``\\`` → ``--`` (see :data:`_FS_SEPARATORS`);
    - :data:`_FS_RESERVED_CHARS` and ASCII control characters → ``_``;
    - leading/trailing whitespace and trailing dots removed (Windows cannot
      create a component ending in either, and this is what collapses ``.``
      and ``..`` — a name that would otherwise escape the sessions root);
    - a Windows reserved device name gains a ``_`` on its stem (``con.log`` →
      ``con_.log``), appended to the STEM rather than the whole string so the
      result is not itself reserved;
    - an empty result degrades to :data:`FS_SAFE_PLACEHOLDER`.
 
    The mapping is deterministic and idempotent — ``fs_safe_name`` of its own
    output is that output — which is what lets it be applied at every boundary
    without a "has this already been sanitised?" flag.  It is the IDENTITY on
    every name that is already a legal path component, so it is a strict no-op
    for the existing saved sessions (proven by the plan-225 W3 witness).
    """
    text = name or ""
    for separator in _FS_SEPARATORS:
        text = text.replace(separator, "--")
    text = "".join(
        "_" if (ch in _FS_RESERVED_CHARS or ord(ch) < 32) else ch
        for ch in text
    )
    # Order matters: strip whitespace both ends first, then trailing dots, so
    # "name. " and " ..name" both settle in one pass (keeping the function
    # idempotent — a second call finds nothing left to strip).
    text = text.strip().rstrip(". ")
    if not text:
        return FS_SAFE_PLACEHOLDER
    stem, dot, rest = text.partition(".")
    if stem.lower() in _FS_RESERVED_DEVICE_NAMES:
        text = f"{stem}_{dot}{rest}"
    return text
 
 
# ─── Path / time / hash helpers ──────────────────────────────────────────────
 
def _ensure_dir(path: Path) -> Path:
    path.mkdir(parents=True, exist_ok=True)
    return path
 
 
def _sessions_root(base: str = DEFAULT_SESSIONS_DIR) -> Path:
    return _ensure_dir(Path(base))
 
 
def _named_session_dir(name: str, base: str = DEFAULT_SESSIONS_DIR) -> Path:
    """Where *name*'s session directory lives — WITHOUT creating anything.
 
    Split out from :func:`_session_dir` because the read paths (listing,
    loading, existence checks) must never bring a directory into being as a
    side effect of asking about it.
    """
    return Path(base) / fs_safe_name(name)
 
 
def _name_lookup_dirs(name: str, base: str = DEFAULT_SESSIONS_DIR) -> list[Path]:
    """Candidate directories for *name*, most-faithful first.
 
    The VERBATIM directory is tried before the sanitised one so a record that
    predates plan 225 always wins over a name that merely maps onto it.  On
    Windows the two are identical for every legacy name (the characters
    :func:`fs_safe_name` maps cannot occur in a Windows path component), so
    lookup is unchanged there; on POSIX, where ``:`` is a legal filename
    character, the verbatim candidate keeps such a record reachable.
    """
    verbatim = Path(base) / name
    sanitised = _named_session_dir(name, base)
    if verbatim == sanitised:
        return [sanitised]
    return [verbatim, sanitised]
 
 
def _session_dir(name: str, base: str = DEFAULT_SESSIONS_DIR) -> Path:
    return _ensure_dir(_named_session_dir(name, base))
 
 
def _archive_dir(name: str, base: str = DEFAULT_SESSIONS_DIR) -> Path:
    return _ensure_dir(Path(base) / ARCHIVE_DIR_NAME / fs_safe_name(name))
 
 
def _legacy_dir(base: str = DEFAULT_SESSIONS_DIR) -> Path:
    return _ensure_dir(Path(base) / LEGACY_DIR_NAME)
 
 
def _now_stamp() -> str:
    """Filename-safe timestamp: YYYY-MM-DD--HH-MM-SS."""
    return datetime.now().strftime("%Y-%m-%d--%H-%M-%S")
 
 
def _now_stamp_compact() -> str:
    """Compact timestamp used inside JSON metadata fields: YYYYMMDD_HHMMSS."""
    return datetime.now().strftime("%Y%m%d_%H%M%S")
 
 
def _stamp_from_mtime(p: Path) -> str:
    """Filename-safe timestamp derived from a file's modification time."""
    return datetime.fromtimestamp(p.stat().st_mtime).strftime("%Y-%m-%d--%H-%M-%S")
 
 
def _stamped_name(session_name: str, kind: str, ext: str = "json", stamp: str | None = None) -> str:
    """Build ``{name}_{stamp}_{kind}.{ext}`` for storage in the session dir.
 
    The name goes through :func:`fs_safe_name` for the same reason the
    directory does: this string IS a path component, and an embedded separator
    here pointed the write at a directory nobody had created (LANES-04).
    """
    return f"{fs_safe_name(session_name)}_{stamp or _now_stamp()}_{kind}.{ext}"
 
 
def _hash_file(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()
 
 
def _strip_bom(data: bytes) -> bytes:
    if data.startswith(b"\xef\xbb\xbf"):
        return data[3:]
    return data
 
 
def _read_json_strip_bom(path: Path) -> dict:
    """Read a JSON file tolerating an optional UTF-8 BOM."""
    raw = path.read_bytes()
    return json.loads(_strip_bom(raw).decode("utf-8"))
 
 
def _glob_latest(session_dir: Path, kind: str, ext: str = "json") -> Path:
    """Return the most-recently-modified ``*_{kind}.{ext}`` file in *session_dir*, or raise."""
    matches = sorted(
        session_dir.glob(f"*_{kind}.{ext}"),
        key=lambda p: p.stat().st_mtime,
    )
    if not matches:
        raise FileNotFoundError(
            f"No {kind} file found in {session_dir}"
        )
    return matches[-1]
 
 
# ─── Archive-before-overwrite ────────────────────────────────────────────────
 
def _archive_kind(session_dir: Path, session_name: str, kind: str,
                  sessions_dir: str = DEFAULT_SESSIONS_DIR) -> None:
    """Move every existing ``*_{kind}.*`` in *session_dir* to ``archive/{name}/``.
 
    Preserves the original timestamped filename (so meta/chat_log pairs
    can be matched by the timestamp in the name).  If a byte-identical
    copy already exists in the archive, the source file is just deleted
    rather than moved, to avoid duplicate snapshots.
    """
    existing = list(session_dir.glob(f"*_{kind}.*"))
    if not existing:
        return
    archive = _archive_dir(session_name, sessions_dir)
    for old in existing:
        target = archive / old.name
        if target.exists() and _hash_file(target) == _hash_file(old):
            old.unlink()
            continue
        # If a non-identical file already lives at target (rare: same name,
        # different content — e.g. clock skew), avoid clobber by appending
        # a counter.
        final_target = target
        counter = 1
        while final_target.exists():
            final_target = archive / f"{old.stem}__{counter}{old.suffix}"
            counter += 1
        shutil.move(str(old), str(final_target))
 
 
# ─── Session metadata (lightweight) ───────────────────────────────────────────
 
def _canonical_session_name(session: ChatSession) -> str:
    """Return *session*'s filesystem-safe name, normalising the live object.
 
    Every record in the store satisfies ``metadata["session_name"] ==
    <directory name> == <stamped filename prefix>``; that invariant is what
    lets :func:`list_sessions` hand a name straight back to the loader and to
    :func:`session_exists`.  A name the filesystem cannot hold breaks the
    invariant at the moment of writing, so the sanitised form is what gets
    persisted — and the LIVE object is normalised to match, so the REPL header,
    ``/session list`` highlighting, and a later rename all name the same record
    as the one on disk.
 
    Mutating the session here is consistent with what these save functions
    already do (``filepath``, ``chat_logs``, ``dirty``/``mark_clean``), and is a
    no-op for every name that is already a legal path component.
    """
    func = "_canonical_session_name"
    raw = getattr(session, "name", "") or ""
    safe = fs_safe_name(raw)
    if safe != raw:
        session.name = safe
        _dbg("OC_CHAT.CTX", func,
             f"session name {raw!r} is not a legal path component; "
             f"normalised to {safe!r}", 2)
    return safe
 
 
def save_session_metadata(session: ChatSession, sessions_dir: str = DEFAULT_SESSIONS_DIR) -> str:
    """Save session metadata.  Returns the absolute path written."""
    name = _canonical_session_name(session)
    sdir = _session_dir(name, sessions_dir)
 
    # Archive any existing meta file before writing the new one.
    _archive_kind(sdir, name, "meta", sessions_dir)
 
    fname = _stamped_name(name, "meta")
    path = sdir / fname
 
    metadata = {
        "session_name": name,
        # M2.5 — persist the stable per-session identity used for TurnEvent.session_id.
        "session_id": getattr(session, "session_id", ""),
        "created": getattr(session, "created", _now_stamp_compact()),
        "last_modified": _now_stamp_compact(),
        "model": session.model,
        "reasoning_effort": session.reasoning_effort,
        "system_prompt": session.system_prompt,
        "stats": session.stats.to_dict(),
        # chat_logs is informational only — actual file is found via glob.
        "chat_logs": list(session.chat_logs),
        # UI preferences set via /ui <subcommand>.  Phase 0 just stores them;
        # Phase 1+ presenters honour them (e.g. show_reasoning).
        "ui": dict(getattr(session, "ui_prefs", {}) or {}),
        "last_turn_max_prompt_tokens": int(getattr(session, "last_turn_max_prompt_tokens", 0) or 0),
        "rollover_hint_shown": bool(getattr(session, "rollover_hint_shown", False)),
        "token_profile": str(getattr(session, "token_profile", "balanced") or "balanced"),
        "max_react_iterations": getattr(session, "max_react_iterations", None),
        "cache_turn_snapshots": list(getattr(session, "cache_turn_snapshots", []) or [])[-30:],
        "active_plan_id": getattr(session, "active_plan_id", None),
        "generation_options": dict(getattr(session, "generation_options", None) or {}),
        "endpoint_name": str(getattr(session, "endpoint_name", "") or ""),
        "sandbox_root_override": getattr(session, "sandbox_root_override", None),
        # Plan 152 D8 — the two additive rename-continuity keys. Read tolerantly:
        # an absent key means "no aliases" / "not superseded".
        "legacy_event_slugs": list(getattr(session, "legacy_event_slugs", None) or []),
        "superseded_by": getattr(session, "superseded_by", None),
    }
 
    with open(path, "w", encoding="utf-8") as f:
        json.dump(metadata, f, indent=2, ensure_ascii=False)
 
    session.filepath = str(path)
    session.mark_clean()
    return str(path)
 
 
def _resolve_record_path(name_or_path: str, kind: str,
                         sessions_dir: str = DEFAULT_SESSIONS_DIR) -> Path:
    """Resolve a name-or-path argument to a concrete ``*_{kind}.*`` file.
 
    The two loaders accept a UNION — an absolute path, a ``*.json`` path, a
    relative path, or a bare session NAME — and this is where that union is
    decided, in one place, for both.
 
    Resolution order:
 
    1. absolute, or ending in ``.json`` — honoured verbatim (unchanged);
    2. separator-bearing — tried as a filesystem path, then relative to the
       store (unchanged);
    3. otherwise, and as the LAST resort for 2, treated as a session NAME and
       resolved through :func:`_name_lookup_dirs`.
 
    Step 3 is the plan-225 addition, and it is strictly additive: it runs only
    where the pre-225 code had already run out of candidates and was about to
    raise :class:`FileNotFoundError`.  No argument that used to resolve
    successfully resolves differently — which is what makes the existing saved
    sessions provably unaffected (the W3 witness).  It matters because a
    separator-bearing argument is BOTH a plausible relative path AND, since the
    persistence boundary started mapping names, a plausible session name.
 
    Raises :class:`FileNotFoundError` when no candidate holds a *kind* file.
    """
    arg = name_or_path.strip()
    p = Path(arg)
    if p.is_absolute() or arg.endswith(".json"):
        return p
    if "/" in arg or "\\" in arg:
        if p.exists():
            return p
        direct = Path(sessions_dir) / arg
        if direct.is_file():
            return direct
    for candidate in _name_lookup_dirs(arg, sessions_dir):
        try:
            return _glob_latest(candidate, kind)
        except FileNotFoundError:
            continue
    raise FileNotFoundError(
        f"No {kind} file found for session {arg!r} under {sessions_dir}"
    )
 
 
@dataclass(frozen=True)
class LoadedSessionMetadata:
    """A loaded metadata dict PLUS the canonical record it came from (plan 152 R11-10).
 
    The duplicate-``session_id`` rule needs to know WHICH record presented an id,
    and the old return shape made that unimplementable: it returned a bare dict
    and DISCARDED the resolved path.
 
    ``canonical_record_locator`` is the session DIRECTORY under the store root,
    never a timestamped sidecar filename — meta files are stamped and archived on
    every save, so a filename would make a same-record re-save look like a
    duplicate. Comparison is on ``resolve()``d real paths, case-folded, which
    makes it correct on Windows and immune to symlink spoofing.
    """
 
    metadata: dict
    canonical_record_locator: str
 
    def __iter__(self):
        """Allow ``meta, locator = load_session_metadata_located(...)`` unpacking."""
        yield self.metadata
        yield self.canonical_record_locator
 
 
def canonical_record_locator(path: Path) -> str:
    """Normalise *path*'s owning session DIRECTORY into a comparable locator.
 
    Case-folded and fully resolved so two spellings of the same directory — and a
    symlinked alias — compare equal, while genuinely different records do not.
    """
    try:
        directory = path if path.is_dir() else path.parent
        return str(directory.resolve(strict=False)).casefold()
    except Exception:
        return str(path).casefold()
 
 
def load_session_metadata_located(
    name_or_path: str, sessions_dir: str = DEFAULT_SESSIONS_DIR
) -> LoadedSessionMetadata:
    """Load session metadata AND report which record it came from.
 
    Same semantics as :func:`load_session_metadata` (which delegates here and
    drops the locator for its existing callers), plus the
    ``canonical_record_locator`` the live load/switch boundary needs to detect a
    duplicate ``session_id`` presented by a NON-canonical record.
    """
    func = "load_session_metadata_located"
    path = _resolve_record_path(name_or_path, "meta", sessions_dir)
    locator = canonical_record_locator(path)
    raw = _read_json_strip_bom(path)
    try:
        return LoadedSessionMetadata(
            metadata=SessionMetadata.model_validate(raw).model_dump(),
            canonical_record_locator=locator,
        )
    except ValidationError as exc:
        _dbg(
            "OC_CHAT.CTX",
            func,
            f"SessionMetadata validation failed for {path}: {exc.error_count()} errors; returning raw dict",
            2,
            override=True,
        )
        return LoadedSessionMetadata(metadata=raw, canonical_record_locator=locator)
 
 
def mark_record_superseded(
    old_name: str, new_name: str, sessions_dir: str = DEFAULT_SESSIONS_DIR
) -> bool:
    """Write the additive ``superseded_by`` mark into *old_name*'s record.
 
    A rename writes both files under the NEW name while the OLD record stays on
    disk carrying the SAME ``session_id`` — store directories key the RAW name
    and nothing ever deletes the old one. Without a mark, which of the two
    same-id records is canonical would be decided by LOAD ORDER, and the current
    record could lose its identity (and its whole event history) to its own
    stale snapshot.
 
    Marking makes the doppelganger non-canonical BY CONSTRUCTION. The mark is
    purely additive — the house never-delete rule stands, the old record remains
    fully loadable.
 
    Returns ``True`` when the mark landed. Best-effort by design: a failure
    degrades to the deterministic newest-meta-timestamp rule, which resolves to
    the same record, so it is never fatal to the rename.
    """
    func = "mark_record_superseded"
    try:
        try:
            path = _resolve_record_path(old_name, "meta", sessions_dir)
        except FileNotFoundError:
            # No such record — the ordinary "renamed a never-saved session"
            # case.  Quiet and False, exactly as before plan 225.
            return False
        raw = _read_json_strip_bom(path)
        # The mark POINTS AT a record, so it stores the canonical (on-disk)
        # name — a raw name here would name a record that does not exist.
        raw["superseded_by"] = fs_safe_name(new_name)
        with open(path, "w", encoding="utf-8") as fh:
            json.dump(raw, fh, indent=2, ensure_ascii=False)
        _dbg("OC_CHAT.CTX", func, f"marked {old_name!r} superseded_by {new_name!r}", 2)
        return True
    except Exception as exc:
        _dbg("OC_CHAT.CTX", func,
             f"EXTERNAL_ERROR: could not mark {old_name!r} superseded: "
             f"{type(exc).__name__}: {exc}; falling back to newest-timestamp adjudication",
             2, override=True)
        return False
 
 
def load_session_metadata(name_or_path: str, sessions_dir: str = DEFAULT_SESSIONS_DIR) -> dict:
    """Load a session metadata file.
 
    Accepts a session name (resolved to ``{base}/{name}/*_meta.json``) or
    a direct path.  Returns the raw dict, validated against
    :class:`~oc_chat.core.session_models.SessionMetadata`.  On
    :class:`pydantic.ValidationError` (e.g. a hard type mismatch from
    an external edit) the raw on-disk dict is returned unchanged and
    the failure is logged — no exception escapes this boundary.
 
    Callers that need to know WHICH record the metadata came from (the live
    load/switch duplicate-id check) should use
    :func:`load_session_metadata_located` instead.
    """
    return load_session_metadata_located(name_or_path, sessions_dir).metadata
 
 
# ─── Chat logs (full conversations) ───────────────────────────────────────────
 
def save_chat_log(session: ChatSession, sessions_dir: str = DEFAULT_SESSIONS_DIR) -> str:
    """Save the current conversation as a JSON chat log.  Returns the path written.
 
    If ``session.load_failed`` is True and the messages list looks empty
    (only a system prompt or one user message), the save is **refused**
    to avoid overwriting a chat log we previously failed to read.
    """
    name = _canonical_session_name(session)
 
    # Defensive guard: never overwrite a chat log we failed to load.
    if getattr(session, "load_failed", False) and len(session.messages) <= 2:
        raise RuntimeError(
            f"Refusing to save chat log for '{name}': previous load failed and "
            f"the message buffer is empty/near-empty.  The original chat log on "
            f"disk is being preserved."
        )
 
    sdir = _session_dir(name, sessions_dir)
    _archive_kind(sdir, name, "chat_log", sessions_dir)
 
    fname = _stamped_name(name, "chat_log")
    path = sdir / fname
 
    log_data = {
        "session_name": name,
        "timestamp": _now_stamp_compact(),
        "model": session.model,
        "reasoning_effort": session.reasoning_effort,
        "messages": session.messages,
        "usage": {
            "prompt_tokens": session.stats.total_prompt_tokens,
            "completion_tokens": session.stats.total_completion_tokens,
            "total_tokens": session.stats.total_tokens,
            "prompt_cache_hit_tokens": session.stats.prompt_cache_hit_tokens,
            "prompt_cache_miss_tokens": session.stats.prompt_cache_miss_tokens,
        },
    }
 
    with open(path, "w", encoding="utf-8") as f:
        json.dump(log_data, f, indent=2, ensure_ascii=False)
 
    # Update the (informational) chat_logs list to point at this current file.
    session.chat_logs = [fname]
    session.dirty = False
    return str(path)
 
 
def load_chat_log(name_or_path: str, sessions_dir: str = DEFAULT_SESSIONS_DIR) -> dict:
    """Load a chat log.
 
    Accepts a session name (resolved to the most-recent
    ``*_chat_log.json`` inside ``{base}/{name}/``) or a direct path.
    The on-disk dict is validated against
    :class:`~oc_chat.core.session_models.ChatLog`. On
    :class:`pydantic.ValidationError` the raw dict is returned
    unchanged and the failure is logged.
    """
    func = "load_chat_log"
    path = _resolve_record_path(name_or_path, "chat_log", sessions_dir)
    raw = _read_json_strip_bom(path)
    try:
        return ChatLog.model_validate(raw).model_dump()
    except ValidationError as exc:
        _dbg(
            "OC_CHAT.CTX",
            func,
            f"ChatLog validation failed for {path}: {exc.error_count()} errors; returning raw dict",
            2,
            override=True,
        )
        return raw
 
 
def save_all(session: ChatSession) -> None:
    """Save both session metadata (lightweight) and chat log (full conversation).
 
    If ``session.load_failed`` is set, :func:`save_chat_log` will refuse
    to overwrite the existing log on disk — we still save metadata so
    stats are persisted. Caller-friendly wrapper used at REPL exit
    (and historically by ``main._save_all``); per-turn auto-save
    inside the ReAct loop calls the two saves directly so it can
    react differently on a ``RuntimeError`` from ``save_chat_log``.
    """
    func = "save_all"
    log_path = None
    try:
        log_path = save_chat_log(session)
    except RuntimeError as exc:
        # VALIDATION_ERROR — load_failed guard tripped; preserve on-disk log
        _dbg("OC_CHAT.CTX", func, f"VALIDATION_ERROR: load_failed: {exc}", 3)
        print(f"\n[Chat log NOT saved: {exc}]")
    meta_path = save_session_metadata(session)
    if log_path:
        print(f"\n[Chat log saved to {log_path}]")
    print(f"[Session metadata saved to {meta_path}]")
 
 
# ─── Markdown export ──────────────────────────────────────────────────────────
 
def export_markdown(session: ChatSession, filepath: str | None = None, sessions_dir: str = DEFAULT_SESSIONS_DIR) -> str:
    """Export the conversation as a Markdown file.  Returns the path written."""
    name = session.name
    sdir = _session_dir(name, sessions_dir)
 
    if filepath is None:
        # _archive_kind uses ``*_{kind}.*`` so it matches *_export.md too.
        _archive_kind(sdir, name, "export", sessions_dir)
        fname = _stamped_name(name, "export", ext="md")
        filepath = str(sdir / fname)
 
    timestamp = _now_stamp_compact()
    lines = [
        f"# Chat Export — {name}",
        "",
        f"**Exported:** {timestamp}  ",
        f"**Model:** {session.model}  ",
        f"**Reasoning effort:** {session.reasoning_effort}",
        f"**Total tokens:** {session.stats.total_tokens} "
        f"(prompt: {session.stats.total_prompt_tokens}, "
        f"completion: {session.stats.total_completion_tokens})",
        "",
    ]
    for msg in session.messages:
        role    = msg.get("role", "")
        content = msg.get("content") or ""
        if role == "system":
            lines += ["---", "**System**", "", content, ""]
        elif role == "user":
            lines += ["---", "**You**", "", content, ""]
        elif role == "assistant" and content:
            usage = msg.get("usage", {})
            usage_str = ""
            if usage:
                usage_str = f"  *(tokens: {usage.get('total_tokens', '?')})*"
            lines += ["---", f"**Assistant**{usage_str}", "", content, ""]
    with open(filepath, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))
    return filepath
 
 
# ─── HTML export (PR 1.D) ─────────────────────────────────────────────────────
 
 
def export_html(session: ChatSession, filepath: str | None = None,
                sessions_dir: str | None = None,
                code_theme: str = "monokai") -> str:
    """Export the conversation as a styled, single-file HTML document.
 
    Uses ``rich.console.Console(record=True)`` to render the same
    Markdown / styled output the user sees in the terminal, then writes
    it via :meth:`rich.console.Console.save_html` to a self-contained
    ``.html`` file (inline CSS, no external assets).
 
    *filepath* defaults to ``oc_chat/sessions/<name>/<name>_<ts>_export.html``.
    Raises :class:`ImportError` if ``rich`` is not installed — the
    caller (``cmd_session``) is responsible for surfacing this as a
    user-friendly message and pointing at ``pip install -e .[rich]``.
    """
    func = "export_html"
    # Lazy-import so the plain console doesn't require rich.
    from rich.console import Console
    from rich.markdown import Markdown
    from rich.panel import Panel
    from rich.rule import Rule
    from rich.text import Text
 
    if sessions_dir is None:
        sessions_dir = DEFAULT_SESSIONS_DIR
 
    name = session.name
    sdir = _session_dir(name, sessions_dir)
    if filepath is None:
        # Archive any existing *_export.html before writing a fresh one,
        # mirroring the Markdown export path's behaviour.
        for old_html in list(sdir.glob("*_export.html")):
            archive = _archive_dir(name, sessions_dir)
            tgt = archive / old_html.name
            if tgt.exists() and _hash_file(tgt) == _hash_file(old_html):
                old_html.unlink()
            else:
                shutil.move(str(old_html), str(tgt))
        fname = _stamped_name(name, "export", ext="html")
        filepath = str(sdir / fname)
 
    # ``record=True`` makes Console accumulate every styled segment so we
    # can export them as HTML at the end.  ``width`` is fixed at 100 for
    # consistent rendering regardless of terminal size.
    console = Console(record=True, width=100, file=open(os.devnull, "w"))
 
    console.print(Rule(title=f"OC Chat — {name}", style="cyan"))
    console.print(Text.assemble(
        ("Model: ", "bold"), (session.model, "cyan"), "\n",
        ("Reasoning effort: ", "bold"),
            (session.reasoning_effort, "cyan"), "\n",
        ("Tokens: ", "bold"),
            (f"{session.stats.total_tokens:,}", "cyan"),
            (f"  (prompt: {session.stats.total_prompt_tokens:,}, "
             f"completion: {session.stats.total_completion_tokens:,})", "dim"),
    ))
    console.print()
 
    for msg in session.messages:
        role    = msg.get("role", "")
        content = (msg.get("content") or "").strip()
        if not content:
            continue
        if role == "system":
            console.print(Panel(content, title="System", border_style="dim",
                                title_align="left"))
        elif role == "user":
            console.print(Panel(content, title="You", border_style="blue",
                                title_align="left"))
        elif role == "assistant":
            usage = msg.get("usage") or {}
            tok = usage.get("total_tokens")
            title = "Assistant" if tok is None else f"Assistant — {tok} tok"
            try:
                console.print(Panel(
                    Markdown(content, code_theme=code_theme),
                    title=title, border_style="green", title_align="left",
                ))
            except Exception as e:
                # SYSTEM_ERROR — Markdown render failed; fall back to plain panel
                _dbg("OC_CHAT.CTX", func, f"SYSTEM_ERROR: {type(e).__name__}: {e}", 1, override=True)
                console.print(Panel(content, title=title,
                                    border_style="green", title_align="left"))
 
    Path(filepath).parent.mkdir(parents=True, exist_ok=True)
    # ``inline_styles=True`` makes the output a self-contained file —
    # no external CSS, suitable for emailing or pasting into a wiki.
    console.save_html(filepath, inline_styles=True)
    return filepath
 
 
# ─── Tool artifacts (PR 1.C) ──────────────────────────────────────────────────
 
 
def save_tool_artifact(session_name: str, tool_name: str, text: str,
                       sessions_dir: str | None = None) -> Path:
    """Spill a long tool-result payload to ``sessions/{name}/artifacts/``.
 
    Returns the absolute path written.  The filename follows the same
    ``{name}_{timestamp}_{kind}.{ext}`` convention used elsewhere in the
    storage layer; ``kind`` is ``tool_<tool_name>`` so artifacts from the
    same tool are easy to find.
 
    *sessions_dir* defaults to the live module-level
    :data:`DEFAULT_SESSIONS_DIR` so tests can monkeypatch it.
 
    Used by the Rich presenter (and in future the Textual presenter) to
    keep long ``read_file`` / ``run_command`` outputs out of the
    scrollback.  The presenter prints a ``↳ saved (N KB) → /open last``
    one-liner instead.
    """
    if sessions_dir is None:
        sessions_dir = DEFAULT_SESSIONS_DIR
    sdir = _session_dir(session_name, sessions_dir) / "artifacts"
    sdir.mkdir(parents=True, exist_ok=True)
    safe_tool = "".join(c if c.isalnum() or c in "-_" else "_" for c in tool_name)
    # Plan 134 B1: the second-precision stamp alone collides when two same-tool
    # spills land in the same second (the captured "two read_file → same file"
    # overwrite). A STATELESS uniquifier — nanosecond time + a random uuid4
    # fragment — keeps filenames distinct without a shared counter, so it is
    # concurrency-safe under parallel agents. Only this filename build changes;
    # other _stamped_name callers are untouched.
    uniq = f"{time.time_ns()}_{uuid4().hex[:6]}"
    fname = _stamped_name(session_name, f"tool_{safe_tool}_{uniq}", ext="txt")
    path = sdir / fname
    path.write_text(text, encoding="utf-8")
    return path
 
 
def find_latest_artifact(session_name: str,
                         sessions_dir: str | None = None) -> Path | None:
    """Return the most-recently-written artifact in this session, or None.
 
    *sessions_dir* defaults to the live module-level
    :data:`DEFAULT_SESSIONS_DIR`.
    """
    if sessions_dir is None:
        sessions_dir = DEFAULT_SESSIONS_DIR
    sdir = _session_dir(session_name, sessions_dir) / "artifacts"
    if not sdir.is_dir():
        return None
    files = list(sdir.glob("*.txt"))
    if not files:
        return None
    return max(files, key=lambda p: p.stat().st_mtime)
 
 
# ─── Listing sessions ─────────────────────────────────────────────────────────
 
def list_sessions(sessions_dir: str = DEFAULT_SESSIONS_DIR) -> list[dict]:
    """Scan ``sessions_dir`` for per-session subdirectories.  Returns one entry per session.
 
    Each entry: ``name``, ``model``, ``stats`` (TokenStats),
    ``last_modified`` (display string), ``filepath`` (path of the
    current ``*_meta.json``), ``chat_logs`` (informational list).
    Corrupt or unreadable metadata files are skipped with a warning.
    """
    func = "list_sessions"
    base = Path(sessions_dir)
    if not base.is_dir():
        return []
 
    results: list[dict] = []
    for child in sorted(base.iterdir()):
        if not child.is_dir():
            continue
        # Skip reserved dirs.
        if child.name in (ARCHIVE_DIR_NAME, LEGACY_DIR_NAME):
            continue
        try:
            meta_path = _glob_latest(child, "meta")
        except FileNotFoundError:
            continue  # session dir without a meta file — skip silently
        try:
            data = _read_json_strip_bom(meta_path)
        except (json.JSONDecodeError, OSError) as exc:
            _dbg("OC_CHAT.CTX", func, f"skipping unreadable session file: {meta_path} ({exc})", 2)
            continue
 
        name  = data.get("session_name", child.name)
        model = data.get("model", "?")
        stats = TokenStats.from_dict(data.get("stats", {}))
        last_modified = datetime.fromtimestamp(meta_path.stat().st_mtime).strftime(
            "%Y-%m-%d %H:%M"
        )
        # Look up the chat log size — useful when token totals are 0
        # (e.g. legacy sessions where usage was never recorded).
        chat_log_size = 0
        try:
            chat_log_size = _glob_latest(child, "chat_log").stat().st_size
        except FileNotFoundError:
            pass
        results.append({
            "name": name,
            "model": model,
            "stats": stats,
            "last_modified": last_modified,
            "filepath": str(meta_path),
            "chat_logs": data.get("chat_logs", []),
            "reasoning_effort": data.get("reasoning_effort", "high"),
            "chat_log_size_bytes": chat_log_size,
            "ui": data.get("ui", {}) or {},
        })
    return results
 
 
def session_exists(name: str, sessions_dir: str = DEFAULT_SESSIONS_DIR) -> bool:
    """Return True if *name*'s session directory holds at least one ``*_meta.json``.
 
    Checks the same candidates the loader resolves (:func:`_name_lookup_dirs`),
    so "exists" and "loadable" cannot disagree — a name the caller is told
    exists is a name the caller can then load.  A bug-era half-created
    directory (``upstage/solar-pro4/``, holding nothing) does not answer True:
    it carries no metadata file.
    """
    for sdir in _name_lookup_dirs(name, sessions_dir):
        if sdir.is_dir() and any(sdir.glob("*_meta.json")):
            return True
    return False
 
 
# ─── One-shot migration: flat layout → per-session-directory ──────────────────
 
# Crash recovery helpers.
 
def _is_variable_context_message(message: dict) -> bool:
    return (
        message.get("role") == "user"
        and isinstance(message.get("content"), str)
        and message["content"].startswith(VARIABLE_CONTEXT_SENTINEL)
    )
 
 
def _conversation_indexes(messages: list[dict]) -> list[int]:
    return [
        index
        for index, message in enumerate(messages)
        if not _is_variable_context_message(message)
    ]
 
 
def _tool_call_ids(message: dict) -> list[str]:
    ids: list[str] = []
    for call in message.get("tool_calls") or []:
        if isinstance(call, dict) and isinstance(call.get("id"), str):
            ids.append(call["id"])
    return ids
 
 
def _message_has_content(message: dict) -> bool:
    content = message.get("content")
    return isinstance(content, str) and bool(content.strip())
 
 
def _tail_report(kind: str, tail_indexes: list[int], summary: str) -> dict:
    return {
        "kind": kind,
        "tail_indexes": tail_indexes,
        "summary": summary,
    }
 
 
def _tool_tail_report(messages: list[dict], indexes: list[int]) -> dict | None:
    last_position = len(indexes) - 1
    tail_indexes: list[int] = []
    seen_tool_ids: set[str] = set()
 
    while last_position >= 0:
        index = indexes[last_position]
        message = messages[index]
        if message.get("role") != "tool":
            break
        tail_indexes.append(index)
        tool_call_id = message.get("tool_call_id")
        if isinstance(tool_call_id, str):
            seen_tool_ids.add(tool_call_id)
        last_position -= 1
 
    if last_position < 0:
        return None
 
    assistant_index = indexes[last_position]
    assistant = messages[assistant_index]
    if assistant.get("role") != "assistant":
        return None
 
    required_ids = set(_tool_call_ids(assistant))
    if not required_ids:
        return None
 
    missing_ids = required_ids - seen_tool_ids
    if not missing_ids:
        return None
 
    ordered_tail = [assistant_index, *reversed(tail_indexes)]
    missing = ", ".join(sorted(missing_ids))
    return _tail_report(
        "orphaned_tool_calls",
        ordered_tail,
        f"Assistant tool call(s) missing tool response(s): {missing}.",
    )
 
 
def detect_incomplete_tail(messages: list[dict]) -> dict | None:
    """Return a structured crash-recovery report for an incomplete tail.
 
    The detector is deliberately conservative. It ignores OC Chat's
    synthetic variable-context user message and reports only the three
    tail shapes that can poison the next model request after a crash.
    """
    indexes = _conversation_indexes(messages)
    if not indexes:
        return None
 
    last_index = indexes[-1]
    last_message = messages[last_index]
    role = last_message.get("role")
 
    if role == "user":
        return _tail_report(
            "orphaned_user",
            [last_index],
            "Session ends with a user message that has no assistant response.",
        )
 
    if role == "tool":
        return _tool_tail_report(messages, indexes)
 
    if role != "assistant":
        return None
 
    tool_call_ids = _tool_call_ids(last_message)
    if tool_call_ids:
        missing = ", ".join(sorted(tool_call_ids))
        return _tail_report(
            "orphaned_tool_calls",
            [last_index],
            f"Assistant tool call(s) missing tool response(s): {missing}.",
        )
 
    if not _message_has_content(last_message):
        return _tail_report(
            "truncated_assistant",
            [last_index],
            "Session ends with an empty assistant message.",
        )
 
    return None
 
 
def apply_tail_recovery(
    messages: list[dict],
    choice: Literal["drop", "keep", "fresh"],
) -> list[dict]:
    """Return messages after applying an incomplete-tail recovery choice."""
    if choice == "keep":
        return list(messages)
    if choice == "fresh":
        if messages and messages[0].get("role") == "system":
            return [messages[0]]
        return []
    if choice != "drop":
        raise ValueError(f"Unsupported recovery choice: {choice}")
 
    report = detect_incomplete_tail(messages)
    if report is None:
        return list(messages)
    tail_indexes = set(report["tail_indexes"])
    return [
        message
        for index, message in enumerate(messages)
        if index not in tail_indexes
    ]
 
 
def write_clean_exit_marker(sessions_dir: str = DEFAULT_SESSIONS_DIR) -> None:
    """Write the marker used to distinguish graceful exit from crash exit."""
    base = _sessions_root(sessions_dir)
    marker_path = base / CLEAN_EXIT_MARKER_NAME
    temp_path = marker_path.with_suffix(".tmp")
    payload = {"timestamp": datetime.now().isoformat(timespec="seconds")}
    with open(temp_path, "w", encoding="utf-8") as f:
        json.dump(payload, f, indent=2)
    os.replace(temp_path, marker_path)
 
 
def read_crash_signal(sessions_dir: str = DEFAULT_SESSIONS_DIR) -> dict | None:
    """Return the latest-session crash signal, or None when startup is clean."""
    sessions = list_sessions(sessions_dir)
    if not sessions:
        return None
 
    latest = max(sessions, key=lambda session: os.path.getmtime(session["filepath"]))
    latest_mtime = Path(latest["filepath"]).stat().st_mtime
    marker_path = Path(sessions_dir) / CLEAN_EXIT_MARKER_NAME
 
    if marker_path.exists() and marker_path.stat().st_mtime >= latest_mtime:
        return None
 
    return {
        "last_session_name": latest["name"],
        "last_modified": latest["last_modified"],
    }
 
 
def migrate_legacy_layout(sessions_dir: str = DEFAULT_SESSIONS_DIR) -> dict:
    """Convert any flat-layout session files to the per-session-directory layout.
 
    Idempotent: already-migrated sessions are skipped.
 
    Returns a dict summarising the migration: ``migrated``, ``recovered``,
    ``orphans_moved``, ``errors``.
    """
    base = _sessions_root(sessions_dir)
    summary = {"migrated": 0, "recovered": 0, "orphans_moved": 0, "errors": []}
 
    # Step 1: per-session migration of any flat ``{name}.json`` files.
    flat_metas = sorted(p for p in base.glob("*.json") if p.is_file())
    for flat_meta in flat_metas:
        name = flat_meta.stem
        try:
            data = _read_json_strip_bom(flat_meta)
        except Exception as exc:
            summary["errors"].append(f"{flat_meta.name}: {exc}")
            continue
 
        sdir = base / name
        sdir.mkdir(parents=True, exist_ok=True)
 
        # Write meta file with mtime-based timestamp.
        meta_stamp = _stamp_from_mtime(flat_meta)
        new_meta_name = _stamped_name(name, "meta", stamp=meta_stamp)
        new_meta_path = sdir / new_meta_name
        if not new_meta_path.exists():
            with open(new_meta_path, "w", encoding="utf-8") as f:
                json.dump(data, f, indent=2, ensure_ascii=False)
 
        # Migrate the matching chat log if one exists.
        old_log = base / "Chat_logs" / f"{name}_log.json"
        if old_log.exists():
            log_stamp = _stamp_from_mtime(old_log)
            new_log_name = _stamped_name(name, "chat_log", stamp=log_stamp)
            new_log_path = sdir / new_log_name
            if not new_log_path.exists():
                try:
                    log_data = _read_json_strip_bom(old_log)
                except Exception as exc:
                    summary["errors"].append(f"{old_log.name}: {exc}")
                else:
                    with open(new_log_path, "w", encoding="utf-8") as f:
                        json.dump(log_data, f, indent=2, ensure_ascii=False)
                    old_log.unlink()
 
        flat_meta.unlink()
        summary["migrated"] += 1
 
    # Step 2: move any orphan chat_log_*.json (no matching session dir) to legacy/.
    old_chat_dir = base / "Chat_logs"
    if old_chat_dir.is_dir():
        legacy = _legacy_dir(sessions_dir)
        for orphan in list(old_chat_dir.iterdir()):
            if orphan.name == ".gitkeep":
                continue
            if orphan.is_file():
                tgt = legacy / orphan.name
                if not tgt.exists():
                    shutil.move(str(orphan), str(tgt))
                    summary["orphans_moved"] += 1
                else:
                    orphan.unlink()  # already in legacy
        # Try to remove the (now-empty) Chat_logs dir.
        try:
            # Remove .gitkeep last so the dir can be removed.
            gitkeep = old_chat_dir / ".gitkeep"
            if gitkeep.exists() and not any(p for p in old_chat_dir.iterdir()
                                            if p.name != ".gitkeep"):
                gitkeep.unlink()
            if not any(old_chat_dir.iterdir()):
                old_chat_dir.rmdir()
        except OSError:
            pass
 
    return summary