runner.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/source_migrations/runner.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_source_migrate_runner.py"
"""
The destructive ``roadmap source migrate`` runner (Plan 142m) — pipeline steps
3-13.
 
Purpose
-------
Execute the pinned, ordered migrate pipeline over a discovered workspace's own
source tree: git-worktree + clean-tree preconditions, the root-level unresolved-
journal refusal, the lstat-level input path-integrity guard, whole-tree version
classification, migration-module discovery + ABI validation + chain resolution +
the whole-chain determinism self-check + terminal-stamp validation, the output
path-integrity pass (path-stability), the dual-volume disk preflight, the per-run
JSONL preimage journal published BEFORE the first source write, per-file atomic
temp+``os.replace`` with a TOCTOU re-compare + ``planned_postimage`` anchor, and
the ``planned_postimage``-guarded auto-rollback folded from the DISK journal. Every
refusal fails closed with ZERO writes; a caught post-write failure auto-rolls-back.
 
Steps 0-2 (semver-shape validation, root resolution, lock acquire/release) live in
the CLI handler (``cli/source.py``); this runner owns steps 3-13 and raises the
``MigrateError`` refusal class (routed to exit 2 via ``_EXIT_BY_CODE``) — or, for a
genuinely unexpected non-IO bug, re-raises after attempting rollback (the disclosed
backstop residual).
 
Responsibilities
----------------
- Run the ordered pipeline (steps 3-13) with the pinned first-refusal semantics.
- Discover + ABI-validate migration modules; resolve + execute the migration chain.
- Journal every write-set preimage, write atomically, and auto-roll-back on failure.
 
Contracts
---------
- Every refusal fails closed with ZERO source writes (fail-closed everywhere).
- The runner NEVER hand-edits a generated view and NEVER writes under
  ``registry/source_migrations/`` (the journal is gitignored runtime state).
- Every expected ``OSError``/``PermissionError`` is wrapped at-site to a durable
  ``SRC_*`` code; only a genuinely unexpected non-IO bug reaches the backstop.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L1 - SYSTEM/SECURITY-class refusals + the backstop (override=True)
    L2 - journal open/close, migration start/end, rollback start/end
    L3 - per-record classification, per-path write/rollback, VALIDATION refusals
    L4 - module load
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.source.scanner (scan_source_tree - the candidate records)
- oc_chat.registry.source.canonical (load_record, format_record_text and format_source_tree - parse, re-emit and canonical check)
- oc_chat.registry.source.hashing (raw_source_tree_hash and source_hash - the tree hashes before the run and after rollback)
- oc_chat.registry.source_migrations.abi (SourceMigration and MigrationResult - validating migrations and their output)
- oc_chat.registry.source_migrations.journal (the per-run preimage journal writer and reader)
- oc_chat.registry.source_migrations.errors (MigrateError - the refusal class)
- oc_chat.registry.clock (RealClock - the timestamp in the run id)
- oc_chat.registry.regenerate.envelope (the SRC_* refusal codes)
- oc_chat.registry.source.errors (SourceError - a parse or canonical failure during classification)
- oc_chat.registry.source.promoted_item (lazy; PromotedItem - the expected model class for promoted records)
- option_c_registry_contracts (lazy; SOURCE_ENTITIES - the expected model class for every other record kind)
"""
from __future__ import annotations
 
import hashlib
import importlib
import os
import re
import shutil
import stat
import subprocess
import uuid
from pathlib import Path
from typing import Any, Callable, Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.clock import RealClock
from oc_chat.registry.regenerate.envelope import (
    SRC_JOURNAL_UNRESOLVED,
    SRC_JOURNAL_WRITE_FAILED,
    SRC_MIGRATION_INVALID,
    SRC_MIGRATION_OUTPUT_INVALID,
    SRC_MIGRATION_TARGET_INVALID,
    SRC_MIGRATION_WRITE_FAILED,
    SRC_NO_MIGRATION,
    SRC_SOURCE_MUTATED,
    SRC_TREE_DIRTY,
    SRC_TREE_NOT_VERSIONED,
    SRC_UNSAFE_PATH,
)
from oc_chat.registry.source.canonical import format_record_text, load_record
from oc_chat.registry.source.errors import SourceError
from oc_chat.registry.source.hashing import raw_source_tree_hash, source_hash
from oc_chat.registry.source.scanner import scan_source_tree
from oc_chat.registry.source_migrations import journal as journal_mod
from oc_chat.registry.source_migrations.abi import (
    MigrationResult,
    SourceMigration,
)
from oc_chat.registry.source_migrations.errors import MigrateError
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: Canonical SemVer core, ``re.fullmatch`` (NEVER ``re.match`` — prefix semantics
#: accept ``1.2.03``/``1.2.3-alpha``): each component ``0|[1-9][0-9]*`` — no
#: leading zeroes/signs/whitespace/prerelease/build (step 0).
_TARGET_SEMVER_RE = re.compile(r"(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)")
#: The migration-module discovery allowlist (§8a): support/``_private``/dunder
#: stems never match, so an unfiltered import can never pull ``abi``/``journal``/…
#: (which expose no ``MIGRATION``) into discovery.
DISCOVERY_STEM_RE = re.compile(
    r"v(0|[1-9][0-9]*)_(0|[1-9][0-9]*)_(0|[1-9][0-9]*)_[A-Za-z0-9_]+"
)
_JOURNAL_RELATIVE = Path(".option_c") / "orchestration" / "source_migrations" / ".journal"
_DISK_HEADROOM = 1.5
 
 
# --- IO seams (monkeypatched by tests for ordering / failure injection) -------
def _replace(src: Path, dst: Path) -> None:
    """Atomic ``os.replace`` seam (overridable for ordering/failure injection)."""
    os.replace(src, dst)
 
 
def _fsync(fileobj: "object") -> None:
    """fsync a file object (overridable seam)."""
    os.fsync(fileobj.fileno())  # type: ignore[attr-defined]
 
 
def _disk_usage(path: Path):
    """``shutil.disk_usage`` seam (per-path monkeypatchable for the preflight);
    returns the ``(total, used, free)`` named tuple."""
    return shutil.disk_usage(str(path))
 
 
def _new_run_id() -> str:
    """A unique per-run journal id (timestamp + uuid) — fixed-injected in tests so
    a version-only filename can never overwrite an unresolved recovery journal."""
    stamp = RealClock().now_wall()
    safe = re.sub(r"[^0-9A-Za-z]", "", stamp)
    return f"{safe}-{uuid.uuid4().hex}"
 
 
# --- semver -------------------------------------------------------------------
def validate_target_semver(raw: Any) -> str:
    """Step 0: validate the ``--to-schema-version`` SHAPE at handler entry —
    canonical SemVer core via ``fullmatch``. Refuse a non-semver argument with
    ``SRC_MIGRATION_TARGET_INVALID`` (all-null hashes; nothing examined)."""
    func = "validate_target_semver"
    if not isinstance(raw, str) or _TARGET_SEMVER_RE.fullmatch(raw) is None:
        _dbg("OC_CHAT.MAIN", func, f"invalid target {raw!r}", 3)
        raise MigrateError(
            f"--to-schema-version {raw!r} is not a valid source-record semver",
            code=SRC_MIGRATION_TARGET_INVALID,
            # OCT-LINT: disable=no_hardcoded_secrets reason="SRC_* codes, semver regex, and hints are the spec contract"
            hint="not a valid source-record semver (MAJOR.MINOR.PATCH)",
        )
    return raw
 
 
def _version_tuple(version: str) -> tuple[int, ...]:
    """Numeric per-field tuple comparison — never string compare (``0.10.0`` vs
    ``0.2.0``). Pads to three components."""
    parts = [int(component) for component in version.split(".")]
    while len(parts) < 3:
        parts.append(0)
    return tuple(parts)
 
 
# --- git precondition (steps 3-4) ---------------------------------------------
def _git_status_porcelain(root: Path) -> Optional[str]:
    """``git status --porcelain .`` under ``root``, or ``None`` if ``root`` is not
    inside a git worktree / ``git`` is missing / ``git status`` failed. UNLIKE
    ``format``, migrate treats EVERY ``None`` as fail-closed
    ``SRC_TREE_NOT_VERSIONED`` (a rewrite whose rollback point cannot be PROVEN has
    none)."""
    try:
        inside = subprocess.run(
            ["git", "rev-parse", "--is-inside-work-tree"],
            cwd=str(root), capture_output=True, text=True, check=False,
        )
    except (OSError, ValueError):
        return None
    if inside.returncode != 0 or inside.stdout.strip() != "true":
        return None
    result = subprocess.run(
        ["git", "status", "--porcelain", "."],
        cwd=str(root), capture_output=True, text=True, check=False,
    )
    if result.returncode != 0:
        return None
    return result.stdout
 
 
# --- path integrity (step 6b) -------------------------------------------------
def _is_reparse_point(lstat_result: os.stat_result) -> bool:
    file_attribute_reparse_point = 0x400
    attrs = getattr(lstat_result, "st_file_attributes", 0)
    return bool(attrs & file_attribute_reparse_point)
 
 
def _path_integrity_guard(root: Path, records: list) -> None:
    """Step 6b: an ``lstat``-level pass (no content read) over every candidate —
    fail closed ``SRC_UNSAFE_PATH`` on a symlink / junction / reparse-point /
    hardlink / resolved-outside-root entry (a destructive writer requirement)."""
    func = "_path_integrity_guard"
    root_resolved = Path(root).resolve()
    for scanned in records:
        path = scanned.path
        lstat_result = path.lstat()
        reason: Optional[str] = None
        if stat.S_ISLNK(lstat_result.st_mode):
            reason = "symlink"
        elif _is_reparse_point(lstat_result):
            reason = "reparse-point/junction"
        elif stat.S_ISREG(lstat_result.st_mode) and lstat_result.st_nlink > 1:
            reason = "hardlink"
        elif not path.resolve().is_relative_to(root_resolved):
            reason = "resolves outside the source root"
        if reason is not None:
            _dbg("OC_CHAT.MAIN", func, f"unsafe path {scanned.rel_path}: {reason}", 1, override=True)
            raise MigrateError(
                f"unsafe candidate path {scanned.rel_path}: {reason}",
                code=SRC_UNSAFE_PATH,
                records=[scanned.rel_path],
                hint="remove the symlink/junction/hardlink under source/ before migrate",
            )
 
 
# --- migration discovery (production default) ---------------------------------
def _production_package_dir() -> Path:
    """The directory production discovery scans — ``registry/source_migrations/``.
    A seam (NOT the factory) so a production-namespace test can point discovery at
    an isolated dir without polluting the shipped package."""
    return Path(__file__).resolve().parent
 
 
def _import_migration_module(path: Path) -> Any:
    """Import a discovered migration file by spec (decoupled from the package
    import machinery); any import-time exception propagates to the 8a trust
    boundary and wraps to ``SRC_MIGRATION_INVALID``."""
    import importlib.util
 
    spec = importlib.util.spec_from_file_location(f"_oc_source_migration_{path.stem}", path)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load migration module {path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module
 
 
def production_migration_provider() -> list[Any]:
    """The default discovery seam: import ONLY module stems matching the allowlist
    regex and collect their ``MIGRATION`` attribute. Support modules never match
    the regex, so the shipped empty-scaffolding path discovers ZERO migrations."""
    func = "production_migration_provider"
    package_dir = _production_package_dir()
    raw_migrations: list[Any] = []
    for entry in sorted(package_dir.glob("*.py")):
        match = DISCOVERY_STEM_RE.fullmatch(entry.stem)
        if match is None:
            continue
        module = _import_migration_module(entry)
        migration = getattr(module, "MIGRATION")
        # Stem-consistency: a shipped module's declared target/domain MUST match
        # its filename (a contradiction is a latent bug worth refusing — wrapped to
        # SRC_MIGRATION_INVALID by the 8a trust boundary).
        stem_version = ".".join(match.group(1, 2, 3))
        stem_domain = entry.stem.split("_", 3)[3]
        declared_target = getattr(migration, "target_schema_version", None)
        declared_domain = getattr(migration, "domain", None)
        if declared_target != stem_version or declared_domain != stem_domain:
            raise ValueError(
                f"migration module {entry.stem} declares target/domain "
                f"({declared_target}/{declared_domain}) contradicting its filename "
                f"({stem_version}/{stem_domain})"
            )
        raw_migrations.append(migration)
    _dbg("OC_CHAT.MAIN", func, f"discovered {len(raw_migrations)} migration module(s)", 2)
    return raw_migrations
 
 
# --- chain resolution (step 8b) -----------------------------------------------
def _find_chains(
    start_version: str, domain: str, target: str, migrations: list[SourceMigration]
) -> list[list[SourceMigration]]:
    """Every simple chain (strictly increasing version) from ``start_version`` to
    ``target`` over the domain-matching validated migrations. Zero ⇒ no path; more
    than one ⇒ ambiguous."""
    start = _version_tuple(start_version)
    goal = _version_tuple(target)
    applicable = [m for m in migrations if m.domain == domain]
    chains: list[list[SourceMigration]] = []
 
    def _walk(current: tuple, path: list, visited: set) -> None:
        if current == goal:
            chains.append(list(path))
            return
        for migration in applicable:
            produces = _version_tuple(migration.target_schema_version)
            low = _version_tuple(migration.min_source_schema_version)
            high = _version_tuple(migration.max_source_schema_version)
            if low <= current <= high and produces > current and produces not in visited:
                _walk(produces, path + [migration], visited | {produces})
 
    _walk(start, [], {start})
    return chains
 
 
def _run_chain(chain: list[SourceMigration], model: Any) -> Any:
    """Execute the resolved chain over ``model`` (value-level threading — hop N's
    output model feeds hop N+1 directly, no re-serialize). Enforces the 142m
    replace-only / single-entry / ``old_path is None`` ABI guard per hop
    (``SRC_MIGRATION_INVALID``). Returns the terminal model. Migration-module
    exceptions are wrapped by the 8c call site (the trust boundary)."""
    current = model
    for migration in chain:
        result = migration.upgrade(current)
        validated = MigrationResult.model_validate(result)
        if len(validated.entries) != 1:
            raise MigrateError(
                f"replace-only ABI: expected exactly one entry, got {len(validated.entries)}",
                code=SRC_MIGRATION_INVALID,
            )
        entry = validated.entries[0]
        if entry.action != "replace":
            raise MigrateError(
                f"replace-only ABI: unsupported action {entry.action!r} (142m ships replace only)",
                code=SRC_MIGRATION_INVALID,
            )
        if entry.old_path is not None:
            raise MigrateError(
                "replace-only ABI: old_path must be None for a replace",
                code=SRC_MIGRATION_INVALID,
            )
        if entry.model is None:
            raise MigrateError(
                "replace-only ABI: a replace entry must carry a model",
                code=SRC_MIGRATION_INVALID,
            )
        current = entry.model
    return current
 
 
def _terminal_entry(chain: list[SourceMigration], model: Any) -> Any:
    """Re-run the chain capturing the TERMINAL hop's single entry (for the step-9
    output path-integrity pass). Only the terminal output is written — intermediate
    hop outputs are consumed via ``.model`` only (no re-serialize)."""
    current = model
    entry = None
    for migration in chain:
        validated = MigrationResult.model_validate(migration.upgrade(current))
        entry = validated.entries[0]
        current = entry.model
    return entry
 
 
def _model_class_for(scanned: Any) -> type:
    """The record's KIND's current parsed model class (kind-stable check)."""
    from option_c_registry_contracts import SOURCE_ENTITIES
 
    from oc_chat.registry.source.promoted_item import PromotedItem
 
    if scanned.kind == "promoted_item":
        return PromotedItem
    return SOURCE_ENTITIES[scanned.kind]
 
 
# --- helpers ------------------------------------------------------------------
def _sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()
 
 
def _record_id(model: Any) -> str:
    return (
        getattr(model, "id", None)
        or getattr(model, "slug", None)
        or getattr(model, "track_code", None)
        or "?"
    )
 
 
def _safe_unlink(path: Path) -> None:
    try:
        path.unlink(missing_ok=True)
    except OSError:
        pass
 
 
def _wrap_module_failure(exc: Exception, records: list[str], raw: Optional[str],
                         src: Optional[str]) -> MigrateError:
    """The migration-module trust boundary: ANY exception escaping module code
    (import / ABI validation / determinism / execution) wraps to
    ``SRC_MIGRATION_INVALID`` (never a runner traceback)."""
    func = "_wrap_module_failure"
    # OCT-LINT: disable=no_hardcoded_secrets reason="SRC_* codes, semver regex, and hints are the spec contract"
    _dbg("OC_CHAT.MAIN", func, f"module failure -> SRC_MIGRATION_INVALID: {exc}", 3)
    return MigrateError(
        f"invalid migration module: {type(exc).__name__}: {exc}",
        code=SRC_MIGRATION_INVALID,
        records=records,
        raw_source_tree_hash=raw,
        source_hash=src,
        hint="fix the discovered migration module (ABI / determinism / execution)",
    )
 
 
# --- the pipeline (steps 3-13) ------------------------------------------------
def run_migration(
    *,
    source_root: Path,
    workspace_root: Path,
    target: str,
    provider: Callable[[], list[Any]],
    quiet: bool = False,
) -> dict:
    """Run pipeline steps 3-13 over the discovered workspace's source tree. Returns
    the SUCCESS payload; raises ``MigrateError`` (routed to exit 2) on any refusal.
    The caller (handler) owns steps 0-2 (semver shape, root resolution, the lock)."""
    func = "run_migration"
    source_root = Path(source_root).resolve()
    journal_root = Path(workspace_root) / _JOURNAL_RELATIVE
    target_journal_dir = journal_root / target
    target_tuple = _version_tuple(target)
    _dbg("OC_CHAT.MAIN", func, f"migrate -> {target} at {source_root}", 2)
 
    # Steps 3-4: git worktree? clean tree?
    status = _git_status_porcelain(source_root)
    if status is None:
        _dbg("OC_CHAT.MAIN", func, "not a git worktree", 2)
        raise MigrateError(
            f"{source_root} is not inside a git worktree; a schema rewrite needs a "
            "rollback point",
            code=SRC_TREE_NOT_VERSIONED,
            hint="run migrate inside a git worktree so a rollback point exists",
        )
    if status.strip():
        dirty = sorted(line[3:].split(" -> ")[-1] for line in status.splitlines() if line.strip())
        _dbg("OC_CHAT.MAIN", func, f"dirty tree ({len(dirty)} paths)", 3)
        raise MigrateError(
            f"resolved source root {source_root} has uncommitted changes; refusing a "
            "destructive migrate (no rollback point)",
            code=SRC_TREE_DIRTY,
            records=dirty,
            hint="commit or stash changes under the resolved source root before 'migrate'",
        )
 
    # Step 5: no unresolved journal for this ROOT (any target, any header identity).
    _refuse_on_unresolved_journal(journal_root, source_root)
 
    # Step 6a: enumerate candidates (walk + classify, no content reads).
    records = scan_source_tree(source_root)
    # Step 6b: candidate path-integrity (lstat-level; raw tree never read yet).
    _path_integrity_guard(source_root, records)
 
    # Step 6c: raw hash + parse + canonical + per-candidate content sha.
    raw_hash: Optional[str] = None
    src_hash: Optional[str] = None
    candidate_sha: dict[str, str] = {}
    candidate_len: dict[str, int] = {}
    candidate_mode: dict[str, Optional[int]] = {}
    loaded: list[tuple] = []
    try:
        raw_hash = raw_source_tree_hash(source_root)
        for scanned in records:
            model, body = load_record(scanned)
            raw_bytes = scanned.path.read_bytes()
            candidate_sha[scanned.rel_path] = _sha256(raw_bytes)
            candidate_len[scanned.rel_path] = len(raw_bytes)
            candidate_mode[scanned.rel_path] = _file_mode(scanned.path)
            loaded.append((scanned, model, body))
        from oc_chat.registry.source.canonical import format_source_tree
        format_source_tree(records, check=True)
        src_hash = source_hash(records)
    except SourceError as error:
        # 6c parse/canonical failure: raw present, source null (existing nullability).
        setattr(error, "raw_source_tree_hash", raw_hash)
        setattr(error, "source_hash", None)
        raise
 
    # Step 7: VERSION classification only (no discovery). all==target -> no-op
    # RIGHT HERE (structural idempotence, before any disk check).
    write_set: list[tuple] = []
    unchanged = 0
    for scanned, model, body in loaded:
        record_version = _version_tuple(model.schema_version)
        if record_version == target_tuple:
            unchanged += 1
            continue
        if record_version > target_tuple:
            _dbg("OC_CHAT.MAIN", func, f"downgrade request {scanned.rel_path}", 3)
            raise MigrateError(
                f"record {scanned.rel_path} is at {model.schema_version} > target {target}; "
                "source migrations are one-way",
                code=SRC_NO_MIGRATION,
                records=[scanned.rel_path],
                hint="source migrations are one-way; use git to reverse",
                raw_source_tree_hash=raw_hash,
                source_hash=src_hash,
            )
        write_set.append((scanned, model, body))
    if not write_set:
        _dbg("OC_CHAT.MAIN", func, "no-op (all records == target)", 2)
        return _success_payload(target, [], unchanged)
 
    # Step 8a: discovery + per-module ABI validation (trust boundary).
    try:
        raw_migrations = provider()
        migrations = [SourceMigration.model_validate(raw) for raw in raw_migrations]
    except Exception as exc:  # noqa: BLE001 — the trust boundary wraps EVERYTHING
        raise _wrap_module_failure(exc, [], raw_hash, src_hash) from exc
 
    # Step 8b: chain-resolve per write-set record over the VALIDATED module set.
    resolved: dict[str, list[SourceMigration]] = {}
    for scanned, model, body in write_set:
        chains = _find_chains(model.schema_version, scanned.kind, target, migrations)
        if not chains:
            _dbg("OC_CHAT.MAIN", func, f"no chain for {scanned.rel_path}", 3)
            raise MigrateError(
                f"no complete migration chain to {target} for {scanned.rel_path}",
                code=SRC_NO_MIGRATION,
                records=[scanned.rel_path],
                hint="no source-migration path exists for the requested target",
                raw_source_tree_hash=raw_hash,
                source_hash=src_hash,
            )
        if len(chains) > 1:
            _dbg("OC_CHAT.MAIN", func, f"ambiguous chain for {scanned.rel_path}", 3)
            raise MigrateError(
                f"ambiguous migration namespace: {len(chains)} complete chains to "
                f"{target} for {scanned.rel_path}",
                code=SRC_MIGRATION_INVALID,
                records=[scanned.rel_path],
                hint="two migration paths reach the target; the namespace is ambiguous",
                raw_source_tree_hash=raw_hash,
                source_hash=src_hash,
            )
        resolved[scanned.rel_path] = chains[0]
 
    # Step 8c: whole-chain determinism self-check + terminal-stamp validation.
    outputs: dict[str, tuple] = {}          # rel_path -> (model, canon_bytes, postimage_sha)
    entries_by_path: dict[str, Any] = {}    # rel_path -> the TERMINAL entry
    for scanned, model, body in write_set:
        chain = resolved[scanned.rel_path]
        try:
            first = _run_chain(chain, load_record(scanned)[0])
            second = _run_chain(chain, load_record(scanned)[0])
            canon_first = format_record_text(first, body).encode("utf-8")
            canon_second = format_record_text(second, body).encode("utf-8")
            entries_by_path[scanned.rel_path] = _terminal_entry(chain, load_record(scanned)[0])
        except MigrateError as error:
            setattr(error, "raw_source_tree_hash", raw_hash)
            setattr(error, "source_hash", src_hash)
            raise
        except Exception as exc:  # noqa: BLE001 — trust boundary (module execution)
            raise _wrap_module_failure(exc, [scanned.rel_path], raw_hash, src_hash) from exc
        if canon_first != canon_second:
            raise MigrateError(
                f"migration determinism self-check failed for {scanned.rel_path}",
                code=SRC_MIGRATION_INVALID, records=[scanned.rel_path],
                hint="the migration is not deterministic for identical input",
                raw_source_tree_hash=raw_hash, source_hash=src_hash,
            )
        if getattr(first, "schema_version", None) != target:
            raise MigrateError(
                f"migration does not stamp its declared target on {scanned.rel_path}",
                code=SRC_MIGRATION_INVALID, records=[scanned.rel_path],
                hint="the terminal output must carry schema_version == the target",
                raw_source_tree_hash=raw_hash, source_hash=src_hash,
            )
        expected_cls = _model_class_for(scanned)
        if not isinstance(first, expected_cls):
            raise MigrateError(
                f"terminal output for {scanned.rel_path} is not a {expected_cls.__name__}",
                code=SRC_MIGRATION_INVALID, records=[scanned.rel_path],
                hint="the terminal output must be the record's kind's current model",
                raw_source_tree_hash=raw_hash, source_hash=src_hash,
            )
        outputs[scanned.rel_path] = (first, canon_first, _sha256(canon_first))
 
    # Step 9: output path-integrity pass (path-stability).
    _validate_output_paths(source_root, write_set, entries_by_path, records, raw_hash, src_hash)
 
    # Step 10: dual-volume disk preflight (>=1 write only).
    run_id = _new_run_id()
    header = journal_mod.header_dict(target, str(source_root), run_id)
    _disk_preflight(source_root, journal_root, header, write_set, candidate_len,
                    candidate_mode, outputs, raw_hash, src_hash)
 
    # Steps 11-13: journal + writes + rollback/resolve.
    return _write_phase(
        source_root=source_root, target=target, target_journal_dir=target_journal_dir,
        header=header, write_set=write_set, outputs=outputs, candidate_sha=candidate_sha,
        candidate_mode=candidate_mode, unchanged=unchanged, raw_hash=raw_hash, src_hash=src_hash,
    )
 
 
def _refuse_on_unresolved_journal(journal_root: Path, source_root: Path) -> None:
    """Step 5: refuse to start on ANY unresolved published journal in this
    workspace's ``.journal/`` — ANY target, ANY header identity (a mismatch means a
    MOVED workspace; the hint names both identities)."""
    func = "_refuse_on_unresolved_journal"
    unresolved = journal_mod.scan_unresolved(journal_root)
    if not unresolved:
        return
    offending = unresolved[0]
    recorded_root = offending.source_root
    _dbg("OC_CHAT.MAIN", func, f"unresolved journal {offending.path.name}", 1, override=True)
    if recorded_root and Path(recorded_root) != source_root:
        hint = (
            f"journal was recorded for {recorded_root}; workspace appears moved — "
            "inspect before archiving"
        )
    else:
        hint = "resolve or archive the recovery journal (inspect it) before migrate"
    raise MigrateError(
        f"an unresolved recovery journal exists ({offending.path.name}); refusing to start",
        code=SRC_JOURNAL_UNRESOLVED,
        records=[str(offending.path)],
        hint=hint,
        journal_path=str(offending.path),
    )
 
 
def _validate_output_paths(source_root, write_set, entries_by_path, records,
                           raw_hash, src_hash) -> None:
    """Step 9: every OUTPUT path is relative + normalized + under the source root +
    path-stable (== the originating record's ``rel_path``) + non-duplicate + not an
    overwrite of a non-candidate. Any violation ⇒ ``SRC_MIGRATION_OUTPUT_INVALID``,
    ZERO writes (a distinct code from the input-tree ``SRC_UNSAFE_PATH``)."""
    func = "_validate_output_paths"
    root_resolved = Path(source_root).resolve()
    candidate_rels = {scanned.rel_path for scanned in records}
    seen: set[str] = set()
    for scanned, model, body in write_set:
        entry = entries_by_path[scanned.rel_path]  # the TERMINAL (written) entry
        output_path = entry.output_path
        reason = _output_path_reason(output_path, scanned.rel_path, root_resolved,
                                     candidate_rels, seen)
        if reason is not None:
            _dbg("OC_CHAT.MAIN", func, f"invalid output {output_path!r}: {reason}", 3)
            raise MigrateError(
                f"migration output path {output_path!r} is invalid: {reason}",
                code=SRC_MIGRATION_OUTPUT_INVALID,
                records=[scanned.rel_path],
                hint="a migration output path must equal the record's own rel_path (142m)",
                raw_source_tree_hash=raw_hash,
                source_hash=src_hash,
            )
        seen.add(output_path)
 
 
def _output_path_reason(output_path, rel_path, root_resolved, candidate_rels, seen):
    if not isinstance(output_path, str) or output_path == "":
        return "empty or non-string path"
    if "\\" in output_path or output_path.startswith("/") or _is_absolute(output_path):
        return "absolute or alternate-separator path"
    parts = output_path.split("/")
    if ".." in parts or "." in parts:
        return "contains a '..'/'.' segment"
    resolved = (root_resolved / output_path).resolve()
    if not resolved.is_relative_to(root_resolved):
        return "resolves outside the source root"
    if output_path in seen:
        return "duplicate output path"
    if output_path != rel_path:
        # A path-changing replace is a rename/create in disguise (leaves the old
        # record behind / overwrites another candidate / escapes the allowlist).
        return "path-changing replace (must equal the record's rel_path — path-stable)"
    return None
 
 
def _is_absolute(output_path: str) -> bool:
    import ntpath
    import posixpath
 
    return ntpath.isabs(output_path) or posixpath.isabs(output_path)
 
 
def _file_mode(path: Path) -> Optional[int]:
    try:
        return stat.S_IMODE(os.stat(path).st_mode)
    except OSError:
        return None
 
 
def _disk_preflight(source_root, journal_root, header, write_set, candidate_len,
                    candidate_mode, outputs, raw_hash, src_hash) -> None:
    """Step 10: EXACT projected-journal bytes (not a scalar multiplier) + the
    per-volume free-space bounds. Reached only with >=1 write; a low-disk no-op or
    fail-closed classification never reaches here."""
    func = "_disk_preflight"
    preimages = [
        journal_mod.PreimageRecord(
            rel_path=scanned.rel_path,
            preimage_bytes=b"\0" * candidate_len[scanned.rel_path],
            file_mode=candidate_mode[scanned.rel_path],
        )
        for scanned, _model, _body in write_set
    ]
    postimage_shas = [outputs[scanned.rel_path][2] for scanned, _m, _b in write_set]
    projected_journal = journal_mod.project_journal_bytes(header, preimages, postimage_shas)
    journal_required = int(projected_journal * _DISK_HEADROOM)
 
    largest_temp = max(len(outputs[scanned.rel_path][1]) for scanned, _m, _b in write_set)
    net_growth = sum(
        max(0, len(outputs[scanned.rel_path][1]) - candidate_len[scanned.rel_path])
        for scanned, _m, _b in write_set
    )
    source_required = int((largest_temp + net_growth) * _DISK_HEADROOM)
 
    journal_root.mkdir(parents=True, exist_ok=True)
    journal_free = _disk_usage(journal_root).free
    source_free = _disk_usage(source_root).free
    same_volume = _same_volume(journal_root, source_root)
 
    failed = (
        journal_free < journal_required
        or source_free < source_required
        or (same_volume and source_free < journal_required + source_required)
    )
    if failed:
        _dbg("OC_CHAT.MAIN", func, "disk preflight failed", 1, override=True)
        raise MigrateError(
            "insufficient free disk space for the migrate journal + source rewrite",
            code=SRC_JOURNAL_WRITE_FAILED,
            hint="no source writes were made; free disk space / fix permissions and retry",
            raw_source_tree_hash=raw_hash,
            source_hash=src_hash,
        )
 
 
def _same_volume(path_a: Path, path_b: Path) -> bool:
    try:
        return os.stat(path_a).st_dev == os.stat(path_b).st_dev
    except OSError:
        return False
 
 
def _success_payload(target, migrated, unchanged, journal_path=None,
                     journal_unresolved=False, hint=None) -> dict:
    payload: dict = {
        "ok": True,
        "target_schema_version": target,
        "records_migrated": migrated,
        "records_unchanged": unchanged,
    }
    if journal_path is not None:
        payload["journal_path"] = journal_path
    if journal_unresolved:
        payload["journal_unresolved"] = True
        if hint is not None:
            payload["hint"] = hint
    return payload
 
 
def _migrated_list(write_set, outputs, target) -> list[dict]:
    """The SUCCESS payload ``records_migrated`` — ``from`` = the 6c scanned
    ``schema_version``, ``to`` = the VALIDATED terminal output field (never the CLI
    argument echoed back). Sorted by rel_path for determinism."""
    migrated = []
    for scanned, model, _body in sorted(write_set, key=lambda item: item[0].rel_path):
        terminal = outputs[scanned.rel_path][0]
        migrated.append({
            "id": _record_id(model),
            "from": model.schema_version,
            "to": terminal.schema_version,
        })
    return migrated
 
 
# --- steps 11-13: journal + writes + rollback/resolve -------------------------
def _write_phase(*, source_root, target, target_journal_dir, header, write_set,
                 outputs, candidate_sha, candidate_mode, unchanged, raw_hash, src_hash) -> dict:
    func = "_write_phase"
    run_id = header["run_id"]
    writer = journal_mod.JournalWriter(target_journal_dir, run_id, header)
 
    # Step 11: build preimages (re-read WRITE-SET + TOCTOU vs 6c sha) + publish
    # BEFORE the first source write.
    preimages = []
    try:
        for scanned, _model, _body in write_set:
            rel = scanned.rel_path
            current = (source_root / rel).read_bytes()
            if _sha256(current) != candidate_sha[rel]:
                raise MigrateError(
                    f"{rel} was mutated between the scan and the pre-journal recompare",
                    code=SRC_SOURCE_MUTATED, records=[rel],
                    hint="a source file changed after the clean-tree check; re-run migrate",
                )
            preimages.append(journal_mod.build_preimage(rel, current, candidate_mode[rel]))
        writer.build(preimages)
        writer.publish()
    except MigrateError:
        writer.cleanup_partial()  # S5-R11-P3-2: cleanup on the mutation exit too
        raise
    except OSError as exc:
        writer.cleanup_partial()
        raise MigrateError(
            "the migration journal could not be built/published before the first write",
            code=SRC_JOURNAL_WRITE_FAILED,
            hint="no source writes were made; free disk space / fix permissions and retry",
            raw_source_tree_hash=raw_hash, source_hash=src_hash,
        ) from exc
 
    _dbg("OC_CHAT.MAIN", func, f"journal published; migrating {len(write_set)} record(s)", 2)
 
    # Step 12: per-file temp-write -> planned_postimage -> TOCTOU -> replace -> applied.
    written: list[str] = []
    in_flight: Optional[tuple] = None
    try:
        for scanned, _model, _body in write_set:
            rel = scanned.rel_path
            path = source_root / rel
            _canon_bytes, postimage_sha = outputs[rel][1], outputs[rel][2]
            current = path.read_bytes()
            if _sha256(current) != candidate_sha[rel]:
                raise MigrateError(
                    f"{rel} was mutated before its os.replace (per-file TOCTOU)",
                    code=SRC_SOURCE_MUTATED, records=[rel],
                    hint="a source file changed after the clean-tree check; re-run migrate",
                )
            tmp = path.parent / f"{path.name}.{uuid.uuid4().hex}.tmp"
            in_flight = (rel, postimage_sha)
            try:
                with open(tmp, "wb") as fileobj:
                    fileobj.write(_canon_bytes)
                    fileobj.flush()
                    _fsync(fileobj)
                writer.planned_postimage(rel, postimage_sha)  # durable anchor BEFORE replace
                _replace(tmp, path)
                written.append(rel)
                _dbg("OC_CHAT.MAIN", func, f"wrote {rel}", 3)
                writer.applied(rel)
            except OSError as exc:
                _safe_unlink(tmp)
                raise MigrateError(
                    f"source write failed for {rel}",
                    code=SRC_MIGRATION_WRITE_FAILED, records=[rel],
                ) from exc
            finally:
                _safe_unlink(tmp)
            in_flight = None
    except MigrateError as error:
        _finalize_post_write_failure(
            error, writer, source_root, written, candidate_sha, in_flight, raw_hash, src_hash,
        )
        raise
    except Exception:  # noqa: BLE001 — the backstop: rollback, log L1, re-raise
        _dbg("OC_CHAT.MAIN", func, "unexpected backstop failure — attempting rollback", 1,
             override=True)
        try:
            outcome, _remaining = _rollback(writer, source_root, written, candidate_sha, in_flight)
            if outcome == "complete":
                writer.resolve(journal_mod.OUTCOME_ROLLED_BACK)
        except Exception:  # noqa: BLE001 — best-effort; the original error re-raises
            pass
        raise
 
    # Step 13: clean completion -> terminal resolved event (binding) + archive.
    migrated = _migrated_list(write_set, outputs, target)
    journal_unresolved = False
    hint = None
    journal_path = None
    try:
        writer.resolve(journal_mod.OUTCOME_SUCCESS)
        journal_path = str(writer.archive())
    except OSError:
        journal_unresolved = True
        journal_path = str(writer.final_path)
        hint = ("journal could not be marked resolved; the next migrate for this source "
                "root (any target) will refuse until it is manually archived")
        _dbg("OC_CHAT.MAIN", func, "resolved-append failed (outcome unchanged)", 1, override=True)
    _dbg("OC_CHAT.MAIN", func, f"migrate ok ({len(migrated)} migrated, {unchanged} unchanged)", 2)
    return _success_payload(target, migrated, unchanged, journal_path, journal_unresolved, hint)
 
 
def _finalize_post_write_failure(error, writer, source_root, written, candidate_sha,
                                 in_flight, raw_hash, src_hash) -> None:
    """Populate the failure envelope's recovery fields after a caught post-first-
    write failure: auto-rollback folded from the DISK journal, then set
    ``rollback``/``journal_path``/``remaining_paths``/hashes per the failure-stage
    table row for the primary code."""
    func = "_finalize_post_write_failure"
    outcome, remaining = _rollback(writer, source_root, written, candidate_sha, in_flight)
    remaining_paths = [rel for rel, _observed in remaining]
    error.rollback = outcome
    if outcome == "failed":
        # Recovery state: tree is partially migrated; hashes would mislead.
        error.raw_source_tree_hash = None
        error.source_hash = None
        error.remaining_paths = remaining_paths
        error.journal_path = str(writer.final_path)
        _dbg("OC_CHAT.MAIN", func, f"rollback FAILED; {len(remaining_paths)} remaining", 1,
             override=True)
        return
    # outcome == "complete": resolve the journal (rolled_back) then decide hashes.
    if error.code == SRC_MIGRATION_WRITE_FAILED:
        restored_raw = raw_source_tree_hash(source_root)
        restored_src = source_hash(scan_source_tree(source_root))
        if restored_raw == raw_hash and restored_src == src_hash:
            error.raw_source_tree_hash = raw_hash  # rollback receipt: bytes restored
            error.source_hash = src_hash
        else:
            # The recompute-and-assert failed: downgrade to rollback: failed.
            error.rollback = "failed"
            error.raw_source_tree_hash = None
            error.source_hash = None
            error.remaining_paths = written
            error.journal_path = str(writer.final_path)
            _dbg("OC_CHAT.MAIN", func, "post-rollback hash mismatch -> failed", 1, override=True)
            return
    else:
        # SRC_SOURCE_MUTATED (or any other): a foreign edit persists past rollback,
        # so the tree is original-except-that-edit — NOT any hashable state.
        error.raw_source_tree_hash = None
        error.source_hash = None
    _resolve_after_rollback(writer, error)
 
 
def _resolve_after_rollback(writer, error) -> None:
    """Append the terminal ``resolved`` (rolled_back) event; a resolve-append
    failure NEVER changes the outcome — it surfaces as ``journal_unresolved``."""
    try:
        writer.resolve(journal_mod.OUTCOME_ROLLED_BACK)
        error.journal_path = str(writer.archive())
    except OSError:
        error.journal_unresolved = True
        error.journal_path = str(writer.final_path)
        error.hint = (
            (error.hint + " ; " if error.hint else "")
            + "journal could not be marked resolved; the next migrate will refuse until "
            "it is manually archived"
        )
 
 
def _rollback(writer, source_root, written, candidate_sha, in_flight) -> tuple:
    """``planned_postimage``-guarded auto-rollback, folded from the DISK journal
    (a manual crash-recovery round-trip). For each written path: restore the
    preimage ONLY if the on-disk bytes hash to the journalled ``planned_postimage``;
    current == preimage ⇒ already-restored no-op; neither ⇒ a foreign edit — do NOT
    overwrite (mark failed, keep the journal). Restores replay net-space-freeing-
    first. Returns ``(outcome, remaining)`` where remaining = ``[(rel, observed)]``."""
    func = "_rollback"
    _dbg("OC_CHAT.MAIN", func, "rollback start", 2)
    parsed = journal_mod.read_journal(writer.final_path)
    targets: list[tuple] = []
    for entry in parsed.entries:
        planned = entry.planned_postimage_sha
        if planned is None and in_flight is not None and entry.rel_path == in_flight[0]:
            planned = in_flight[1]  # the in-flight path whose event lines are missing
        if planned is None:
            continue  # this path was never (about to be) replaced
        targets.append((entry, planned))
 
    def _freed(item: tuple) -> int:
        entry, _planned = item
        path = source_root / entry.rel_path
        try:
            current_len = len(path.read_bytes())
        except OSError:
            current_len = 0
        return len(journal_mod.preimage_bytes_of(entry)) - current_len  # frees most first
 
    targets.sort(key=_freed)
    remaining: list[tuple] = []
    for entry, planned in targets:
        rel = entry.rel_path
        path = source_root / rel
        preimage = journal_mod.preimage_bytes_of(entry)
        preimage_sha = _sha256(preimage)
        try:
            current = path.read_bytes()
        except OSError:
            remaining.append((rel, None))
            continue
        current_sha = _sha256(current)
        if current_sha == preimage_sha:
            continue  # already restored / crash-before-replace no-op (idempotent)
        if current_sha != planned:
            _dbg("OC_CHAT.MAIN", func, f"foreign edit on {rel} — refusing overwrite", 1,
                 override=True)
            remaining.append((rel, current_sha))
            continue
        tmp = path.parent / f"{path.name}.{uuid.uuid4().hex}.rb.tmp"
        try:
            reverify = path.read_bytes()  # rollback's own pre-replace TOCTOU re-verify
            if _sha256(reverify) != planned:
                remaining.append((rel, _sha256(reverify)))
                _safe_unlink(tmp)
                continue
            with open(tmp, "wb") as fileobj:
                fileobj.write(preimage)
                fileobj.flush()
                _fsync(fileobj)
            _replace(tmp, path)
            file_mode = entry.preimage.get("file_mode")
            if file_mode is not None:
                try:
                    os.chmod(path, file_mode)  # best-effort (Windows: partial)
                except OSError:
                    pass
            _dbg("OC_CHAT.MAIN", func, f"restored {rel}", 3)
        except OSError:
            remaining.append((rel, current_sha))
        finally:
            _safe_unlink(tmp)
    outcome = "failed" if remaining else "complete"
    _dbg("OC_CHAT.MAIN", func, f"rollback {outcome}", 2)
    return outcome, remaining