source.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/source.py
 
# OCT-LINT: disable=no_hardcoded_secrets reason="CLI help strings and projection names are not secrets"
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/source/test_cli.py"
"""
CLI commands for the registry AuthoredSource format (Plan 142 P1.3 + 142j R5).
 
Purpose
-------
Expose the §10 read-only ``source`` surface: ``format [--check]`` (the §3.2
idempotent canonicaliser, with the 142j write-mode clean-tree precondition),
``validate`` (the §5.3 steps 1–6 chain, fail-fast), ``lint`` (non-failing style
warnings), ``diff`` (a semantic diff over parsed/canonical models), and
``hash-preimage`` (the §3.2 debug preimage). The destructive ``migrate`` command
is Plan 142m.
 
Responsibilities
----------------
- Resolve the source tree (default ``<workspace>/oc_chat/roadmap/source``).
- Run each read-only command, emitting the §8 ``build_envelope`` on failure and
  each command's own success payload otherwise.
- Enforce the write-mode clean-tree precondition + the worker discovery guard.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - CLI lifecycle
    L3 - command argument summaries
    L4 - subcommand dispatch details
 
Contracts
---------
- Read-only except ``source format`` (without ``--check``) which rewrites
  non-canonical records in place (refused on a dirty resolved-source tree).
- Non-zero exit emits the §10 envelope; ``diff`` owns 0/1/2 (exit 1 = "differs").
 
Dependencies
------------
- argparse (CLI parsing)
- oc_chat.registry.discovery (workspace/registry resolution + the worker guard)
- oc_diagnostics (_dbg structured logger)
- oc_chat.registry.backends.sqlite (SQLiteBackend connection for the logical-state hash preimage)
- oc_chat.registry.regenerate.cross_record (validate_cross_record, the cross-record step of validate)
- oc_chat.registry.regenerate.envelope (RegenerateError, SRC_TREE_DIRTY, build_envelope and exit_code_for)
- oc_chat.registry.regenerate.lock (acquire and release the shared regenerate lock around migrate)
- oc_chat.registry.regenerate.pipeline (_categorize, the record load and schema steps of validate and lint)
- oc_chat.registry.source.canonical (format_source_tree canonicaliser and load_record for diff and hash-preimage)
- oc_chat.registry.source.errors (SourceError and SourceScanError for the error envelopes and exit codes)
- oc_chat.registry.source.hashing (raw-source-tree and source hashes plus the hash-preimage projections)
- oc_chat.registry.source.scanner (scan_source_tree over the resolved source tree)
- oc_chat.registry.source_migrations.errors (MigrateError for the target schema version refusal)
- oc_chat.registry.source_migrations.runner (production migration provider, run_migration and target semver validation)
- yaml (YAML output for hash-preimage)
- git (not imported; run as a subprocess for rev-parse and status --porcelain, the clean-tree precondition of write-mode source format)
"""
from __future__ import annotations
 
import argparse
import json
from pathlib import Path
import subprocess
import sys
 
import yaml
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.discovery import discover_registry_location, WorkerRegistryError
from oc_chat.registry.regenerate import lock
from oc_chat.registry.regenerate.cross_record import validate_cross_record
from oc_chat.registry.regenerate.envelope import (
    RegenerateError,
    SRC_TREE_DIRTY,
    build_envelope,
    exit_code_for,
)
from oc_chat.registry.regenerate.pipeline import _categorize
from oc_chat.registry.source import hashing
from oc_chat.registry.source.canonical import format_source_tree, load_record
from oc_chat.registry.source.errors import (
    SourceError,
    SourceScanError,
    SourceValidationError,
)
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.errors import MigrateError
from oc_chat.registry.source_migrations.runner import (
    production_migration_provider,
    run_migration,
    validate_target_semver,
)
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: scanner-kind -> the diff pseudo-record kind label (promoted/inline sub_phase
#: both normalize to "sub_phase"; §5.6 inline↔promoted moves are path-changes).
_DIFF_KIND = {
    "manifest": "manifest", "era": "era", "track": "track", "phase": "phase",
    "roadmap_item": "item", "version": "version", "framing": "framing", "note": "note",
}
 
 
def add_source_subcommands(roadmap_commands: argparse._SubParsersAction) -> None:
    """Attach ``source`` (format/validate/lint/diff) + ``hash-preimage`` (§10)."""
 
    source_parser = roadmap_commands.add_parser(
        "source", help="Validate/format/lint/diff the AuthoredSource roadmap tree."
    )
    source_commands = source_parser.add_subparsers(dest="source_command")
 
    format_parser = source_commands.add_parser(
        "format", help="Canonicalise source records (§3.2); --check to verify only."
    )
    format_parser.add_argument("registry", nargs="?")
    format_parser.add_argument("--source-dir", dest="source_dir")
    format_parser.add_argument("--check", action="store_true")
    format_parser.add_argument("--json", action="store_true")
    format_parser.set_defaults(handler=_cmd_source_format)
 
    validate_parser = source_commands.add_parser(
        "validate", help="Run the §5.3 steps 1–6 validation chain (fail-fast, exit 1)."
    )
    validate_parser.add_argument("registry", nargs="?")
    validate_parser.add_argument("--source-dir", dest="source_dir")
    validate_parser.add_argument("--json", action="store_true")
    validate_parser.add_argument("--quiet", action="store_true")
    validate_parser.set_defaults(handler=_cmd_source_validate)
 
    lint_parser = source_commands.add_parser(
        "lint", help="Non-failing style warnings (exit 0 even with warnings)."
    )
    lint_parser.add_argument("registry", nargs="?")
    lint_parser.add_argument("--source-dir", dest="source_dir")
    lint_parser.add_argument("--json", action="store_true")
    lint_parser.add_argument("--quiet", action="store_true")
    lint_parser.set_defaults(handler=_cmd_source_lint)
 
    diff_parser = source_commands.add_parser(
        "diff", help="Semantic diff of two source trees (exit 0 same / 1 differs / 2 error)."
    )
    diff_parser.add_argument("a")
    diff_parser.add_argument("b")
    diff_parser.add_argument("--json", action="store_true")
    diff_parser.add_argument("--quiet", action="store_true")
    diff_parser.set_defaults(handler=_cmd_source_diff)
 
    # Plan 142m: the DESTRUCTIVE migrate leaf. Deliberately takes NO --source-dir
    # (Sonnet5-R07-P1-2) — it operates ONLY on the discovered workspace's own
    # source tree, so the lock + journal never split from the tree being rewritten.
    migrate_parser = source_commands.add_parser(
        "migrate",
        help="Destructively migrate source records to --to-schema-version (§4): "
             "requires a clean git worktree; journalled + auto-rollback; no --source-dir.",
    )
    migrate_parser.add_argument("registry", nargs="?")
    migrate_parser.add_argument(
        "--to-schema-version", dest="to_schema_version", required=True,
        help="Target source-record semver (MAJOR.MINOR.PATCH).",
    )
    migrate_parser.add_argument("--json", action="store_true")
    migrate_parser.add_argument("--quiet", action="store_true")
    migrate_parser.set_defaults(handler=_cmd_source_migrate)
 
    preimage_parser = roadmap_commands.add_parser(
        "hash-preimage", help="Print the canonical preimage a §3.2 hash is taken over."
    )
    preimage_parser.add_argument("registry", nargs="?")
    preimage_parser.add_argument("--source-dir", dest="source_dir")
    preimage_parser.add_argument(
        "--projection",
        choices=["source-tree", "raw-source-tree", "semantic", "logical-state"],
        default="source-tree",
    )
    preimage_parser.add_argument("--format", choices=["json", "yaml"], default="json")
    preimage_parser.set_defaults(handler=_cmd_hash_preimage)
 
 
# --- shared helpers ---------------------------------------------------------
def _source_dir(args: argparse.Namespace) -> Path:
    if args.source_dir:
        return Path(args.source_dir)
    location = discover_registry_location(registry_path=args.registry)
    return Path(location.workspace_root) / "oc_chat" / "roadmap" / "source"
 
 
def _resolve_root_or_exit(args: argparse.Namespace) -> "tuple[Path | None, int | None]":
    """Resolve the source dir; on the R7 worker-guard failure (a registry-discovery
    error, NOT a §8 envelope) print the stderr token and return exit 2."""
    try:
        return _source_dir(args), None
    except WorkerRegistryError as error:
        print(str(error), file=sys.stderr)
        return None, 2
 
 
def _emit_envelope(error: SourceError, as_json: bool) -> None:
    """The pre-142j compact envelope (no §8 hash keys) — kept for the EXISTING
    ``format``/``hash-preimage`` error paths (unchanged; only the new SRC_TREE_DIRTY
    path uses ``build_envelope``, R7 narrow-scope)."""
    envelope = {
        "ok": False,
        "code": error.code,
        "message": str(error),
        "records": error.records,
        "hint": error.hint,
    }
    if isinstance(error, SourceScanError):
        envelope["scan_error"] = error.scan_error
    if as_json:
        print(json.dumps(envelope, indent=2, sort_keys=True))
    else:
        print(f"source failed [{error.code}]: {error}")
 
 
def _print_error_envelope(envelope: dict, as_json: bool) -> None:
    """Print the full §8 ``build_envelope`` (with the hash-nullability keys)."""
    if as_json:
        print(json.dumps(envelope, indent=2, sort_keys=True))
    else:
        print(f"source failed [{envelope.get('code')}]: {envelope.get('message')}")
 
 
def _git_status_porcelain(root: Path) -> "str | None":
    """Return ``git status --porcelain`` limited to ``root``'s subtree, or ``None``
    if ``root`` is not inside a git worktree (R7 write-mode clean-tree precondition;
    scoped to the resolved root, so unrelated edits elsewhere never block)."""
    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
 
 
# --- format (write-mode clean-tree precondition + narrow SRC_TREE_DIRTY) -----
def _cmd_source_format(args: argparse.Namespace) -> int:
    root, exit_code = _resolve_root_or_exit(args)
    if exit_code is not None:
        return exit_code
    format_warnings: list[dict] = []
    if not args.check:  # R7: the precondition gates write-mode only, never --check
        status = _git_status_porcelain(root)
        if status is None:
            warning = {
                "rule": "clean_tree_precondition",
                "message": f"{root} is not inside a git worktree; skipping the clean-tree "
                           "precondition (no rollback point is enforced for this format)",
                "severity": "warning",
            }
            format_warnings.append(warning)
            if not args.json:
                print(f"warning: {warning['message']}", file=sys.stderr)
        elif status.strip():
            # R7 narrow-scope: the SRC_TREE_DIRTY refusal uses build_envelope (full
            # §8 key set) + an EXPLICIT exit 2 (NOT exit_code_for — a SRC_ code that
            # exits 2, in _EXIT_BY_CODE; the OTHER format error paths stay as-is).
            dirty = [line[3:].split(" -> ")[-1] for line in status.splitlines() if line.strip()]
            error = SourceError(
                f"resolved source root {root} has uncommitted changes; refusing a "
                "destructive format (no rollback point)",
                code=SRC_TREE_DIRTY,
                records=sorted(dirty),
                hint="commit or stash changes under the resolved source root before 'format'",
            )
            _print_error_envelope(build_envelope(error), args.json)
            return 2
    try:
        records = scan_source_tree(root)
        changed = format_source_tree(records, check=args.check)
    except SourceError as error:
        # EXISTING paths, unchanged (do NOT convert to exit_code_for — would flip
        # SourceScanError's exit 2→1, DeepSeek-R11-NP1-1).
        _emit_envelope(error, args.json)
        return 2 if isinstance(error, SourceScanError) else 1
    payload = {"ok": True, "checked": args.check, "changed": changed, "count": len(records)}
    if format_warnings:
        payload["warnings"] = format_warnings
    if args.json:
        print(json.dumps(payload, indent=2, sort_keys=True))
    else:
        verb = "non-canonical" if args.check else "reformatted"
        print(f"source format ok ({len(changed)} {verb}, {len(records)} records)")
    return 0
 
 
# --- validate (§5.3 steps 1–6, fail-fast) -----------------------------------
def _cmd_source_validate(args: argparse.Namespace) -> int:
    root, exit_code = _resolve_root_or_exit(args)
    if exit_code is not None:
        return exit_code
    raw_hash: "str | None" = None
    src_hash: "str | None" = None
    try:
        records = scan_source_tree(root)                 # step 1 (+ scanner casefold/notes)
        raw_hash = raw_source_tree_hash(root)
        categorized = _categorize(records)               # steps 2–3 (load + local schema + device)
        format_source_tree(records, check=True)          # step 4 canonicalisation check
        src_hash = source_hash(records)                  # step 5
        validate_cross_record(categorized)               # step 6 (incl. framing + item collision)
    except SourceError as error:
        # §10 hash-nullability falls out of the variable state (each hash is set
        # only after its step succeeds): scan → both null; parse/canon → raw only;
        # cross-record → raw + source.
        envelope = build_envelope(error, raw_source_tree_hash=raw_hash, source_hash=src_hash)
        _print_error_envelope(envelope, args.json)
        return exit_code_for(error)
    if args.json:
        print(json.dumps({"ok": True}, indent=2, sort_keys=True))
    elif not args.quiet:
        print(f"source validate ok ({len(records)} records)")
    return 0
 
 
# --- lint (non-failing style warnings) --------------------------------------
def _cmd_source_lint(args: argparse.Namespace) -> int:
    root, exit_code = _resolve_root_or_exit(args)
    if exit_code is not None:
        return exit_code
    raw_hash: "str | None" = None
    try:
        records = scan_source_tree(root)
        raw_hash = raw_source_tree_hash(root)
        categorized = _categorize(records)
    except SourceError as error:
        envelope = build_envelope(error, raw_source_tree_hash=raw_hash)
        _print_error_envelope(envelope, args.json)
        return exit_code_for(error)
    warnings = _lint_inactive_tracks(categorized)
    if args.json:
        print(json.dumps({"ok": True, "warnings": warnings}, indent=2, sort_keys=True))
    else:
        for warning in warnings:  # warnings are diagnostics, never --quiet-suppressed
            print(f"{warning['rule']}: {warning['path']}: {warning['message']}")
        if not warnings and not args.quiet:
            print(f"source lint ok (0 warnings, {len(records)} records)")
    return 0
 
 
def _lint_inactive_tracks(categorized: dict) -> list[dict]:
    """The first required lint (§3.2): a record whose ``track_codes`` reference an
    inactive track. Non-failing; deterministically ordered."""
    inactive = {
        track.track_code for track in categorized.get("tracks", [])
        if not getattr(track, "active", True)
    }
    owners: list[tuple[str, list, str]] = []
    for phase, _body in categorized.get("phases", []):
        owners.append((phase.id, getattr(phase, "track_codes", []) or [], f"phases/{phase.id}.md"))
    for item, _body in categorized.get("items", []):
        owners.append((item.id, getattr(item, "track_codes", []) or [],
                       f"items/{item.track_kind}/{item.id}.md"))
    for scanned, item, _body in categorized.get("promoted", []):
        owners.append((item.id, getattr(item, "track_codes", []) or [], scanned.rel_path))
    warnings: list[dict] = []
    for owner_id, codes, location in owners:
        for code in codes:
            if code in inactive:
                warnings.append({
                    "rule": "inactive_track_membership",
                    "path": location,
                    "message": f"{owner_id!r} references inactive track {code!r}",
                    "severity": "warning",
                })
    warnings.sort(key=lambda warning: (warning["path"], warning["message"]))
    return warnings
 
 
# --- diff (semantic diff over parsed/canonical models) ----------------------
def _cmd_source_diff(args: argparse.Namespace) -> int:
    # diff is EXEMPT from exit_code_for — exit 1 means "differs" (normal), not
    # "error"; a scan/validation error returns 2 (usage/error class).
    try:
        units_a = _diff_units(Path(args.a))
        units_b = _diff_units(Path(args.b))
    except SourceError as error:
        _print_error_envelope(build_envelope(error), args.json)
        return 2
    result = _compare_units(units_a, units_b)
    differs = bool(result["added"] or result["removed"] or result["changed"])
    if args.json:
        print(json.dumps(result, indent=2, sort_keys=True))
    elif differs:
        _print_diff_human(result)
    elif not args.quiet:
        print("source diff: no differences")
    return 1 if differs else 0
 
 
def _diff_units(root: Path) -> dict:
    """Return ``{full_key_tuple: unit}`` over the parsed models — every entity plus
    each inline sub_phase / promoted item as a normalized ``sub_phase`` pseudo-record
    (same key ``{identity, parent_phase}``), so an inline↔promoted move is a path
    change, not add+remove (§5.6)."""
    units: dict[tuple, dict] = {}
    for scanned in scan_source_tree(root):
        model, _body = load_record(scanned)
        if scanned.kind == "phase":
            full, keyobj = _logical_key("phase", scanned, model)
            units[full] = _unit("phase", keyobj, scanned.rel_path, model)
            for sub in getattr(model, "sub_phases", []) or []:
                full_s = ("sub_phase", sub.id, None, model.id)
                keyobj_s = {"identity": sub.id, "parent_phase": model.id}
                units[full_s] = _unit("sub_phase", keyobj_s, f"{scanned.rel_path}#{sub.id}", sub)
        elif scanned.kind == "promoted_item":
            full_s = ("sub_phase", model.id, None, scanned.phase_code)
            keyobj_s = {"identity": model.id, "parent_phase": scanned.phase_code}
            units[full_s] = _unit("sub_phase", keyobj_s, scanned.rel_path, model)
        else:
            full, keyobj = _logical_key(scanned.kind, scanned, model)
            units[full] = _unit(_DIFF_KIND[scanned.kind], keyobj, scanned.rel_path, model)
    return units
 
 
def _logical_key(kind: str, scanned, model) -> "tuple[tuple, dict]":
    """Per-kind logical key (Codex-R10-P1): track→track_code; version→(route, slug);
    manifest→singleton; framing/note/era/item/phase→id. Returns
    ``(hashable_full_key, emitted_key_object)`` (absent fields omitted)."""
    if kind == "track":
        return ("track", model.track_code, None, None), {"identity": model.track_code}
    if kind == "version":
        route = scanned.changelog_kind
        return ("version", model.slug, route, None), {"identity": model.slug, "route": route}
    if kind == "manifest":
        return ("manifest", "manifest", None, None), {"identity": "manifest"}
    return (kind, model.id, None, None), {"identity": model.id}
 
 
def _unit(kind: str, keyobj: dict, path: str, model) -> dict:
    return {"kind": kind, "key": keyobj, "path": path, "fields": model.model_dump(mode="json")}
 
 
def _compare_units(units_a: dict, units_b: dict) -> dict:
    keys_a, keys_b = set(units_a), set(units_b)
    added = [_locate(units_b[key]) for key in keys_b - keys_a]
    removed = [_locate(units_a[key]) for key in keys_a - keys_b]
    changed: list[dict] = []
    for key in keys_a & keys_b:
        unit_a, unit_b = units_a[key], units_b[key]
        field_names = set(unit_a["fields"]) | set(unit_b["fields"])
        fields = sorted(
            name for name in field_names
            if unit_a["fields"].get(name) != unit_b["fields"].get(name)
        )
        if fields or unit_a["path"] != unit_b["path"]:
            changed.append({
                "kind": unit_b["kind"], "key": unit_b["key"],
                "old_path": unit_a["path"], "new_path": unit_b["path"], "fields": fields,
            })
    added.sort(key=_diff_sort_key)
    removed.sort(key=_diff_sort_key)
    changed.sort(key=_diff_sort_key)
    return {"added": added, "removed": removed, "changed": changed}
 
 
def _locate(unit: dict) -> dict:
    return {"kind": unit["kind"], "key": unit["key"], "path": unit["path"]}
 
 
def _diff_sort_key(record: dict) -> tuple:
    key = record["key"]
    return (
        record["kind"], key["identity"], key.get("route") or "", key.get("parent_phase") or "",
        record.get("path", record.get("old_path", "")),
    )
 
 
def _print_diff_human(result: dict) -> None:
    for record in result["removed"]:
        print(f"- {record['kind']} {record['key']} ({record['path']})")
    for record in result["added"]:
        print(f"+ {record['kind']} {record['key']} ({record['path']})")
    for record in result["changed"]:
        moved = "" if record["old_path"] == record["new_path"] else \
            f" {record['old_path']} -> {record['new_path']}"
        fields = f" fields={record['fields']}" if record["fields"] else ""
        print(f"~ {record['kind']} {record['key']}{moved}{fields}")
 
 
# --- migrate (the destructive Plan 142m engine) -----------------------------
def _migration_registry_factory():
    """The migration-discovery seam (private, non-CLI). Default = PRODUCTION
    discovery over ``registry/source_migrations/``. Tests monkeypatch THIS symbol
    (the consuming namespace) to inject fakes; there is NO ``--migrations-package``
    flag — a visible flag pointing migrate at arbitrary code is a production
    footgun. Production CLI always uses the default."""
    return production_migration_provider
 
 
def _resolve_migrate_paths(args: argparse.Namespace) -> "tuple[Path, Path]":
    """Resolve (source_root, workspace_root) through THIS module's binding of
    ``discover_registry_location`` (the consuming-namespace seam the tests patch).
    migrate takes NO ``--source-dir``: root = the discovered workspace's own source
    tree, and the shared lock + journal key to the SAME workspace root."""
    location = discover_registry_location(registry_path=args.registry)
    workspace_root = Path(location.workspace_root)
    source_root = (workspace_root / "oc_chat" / "roadmap" / "source").resolve()
    return source_root, workspace_root
 
 
def _cmd_source_migrate(args: argparse.Namespace) -> int:
    """The ``roadmap source migrate`` handler — pipeline steps 0-2 (semver shape,
    root resolution, the shared lock) + the runner (steps 3-13). Mirrors
    ``regenerate/pipeline`` lock choreography: step 0 in its OWN pre-lock
    try/except; a separate try wrapping the runner whose
    ``except (SourceError, RegenerateError)`` catches BOTH the migrate refusals and
    the lock's ``REGEN_LOCK_HELD``, with a ``finally`` that releases the lock
    UNCONDITIONALLY."""
    func = "_cmd_source_migrate"
    # Step 0: validate the --to-schema-version SHAPE (own pre-lock try/except;
    # all-null hashes; never argparse usage-exit).
    try:
        target = validate_target_semver(args.to_schema_version)
    except MigrateError as error:
        _print_error_envelope(build_envelope(error), args.json)
        return 2
    # Step 1: resolve the source root (worker-guard -> exit 2 + stderr, NO envelope).
    try:
        source_root, workspace_root = _resolve_migrate_paths(args)
    except WorkerRegistryError as error:
        print(str(error), file=sys.stderr)
        return 2
    lock_path = workspace_root / ".option_c" / "orchestration" / "regenerate.lock"
    # Step 2: acquire the SHARED regenerate lock (its own try/except — the
    # REGEN_LOCK_HELD RegenerateError is raised by lock.acquire itself).
    try:
        handle = lock.acquire(lock_path)
    except RegenerateError as error:
        _print_error_envelope(build_envelope(error), args.json)
        return exit_code_for(error)
    try:
        provider = _migration_registry_factory()
        payload = run_migration(
            source_root=source_root, workspace_root=workspace_root, target=target,
            provider=provider, quiet=args.quiet,
        )
    except (SourceError, RegenerateError) as error:
        envelope = build_envelope(
            error,
            raw_source_tree_hash=getattr(error, "raw_source_tree_hash", None),
            source_hash=getattr(error, "source_hash", None),
            rollback=getattr(error, "rollback", None),
            journal_path=getattr(error, "journal_path", None),
            remaining_paths=getattr(error, "remaining_paths", None),
            journal_unresolved=getattr(error, "journal_unresolved", None),
        )
        _print_error_envelope(envelope, args.json)
        return exit_code_for(error)
    finally:
        lock.release(handle)
    _print_migrate_success(payload, args.json, args.quiet)
    _dbg("OC_CHAT.MAIN", func, "migrate ok", 2)
    return 0
 
 
def _print_migrate_success(payload: dict, as_json: bool, quiet: bool) -> None:
    """Emit the migrate SUCCESS payload. ``--quiet`` suppresses ONLY the non-JSON
    human success line; it NEVER suppresses the ``--json`` payload or the
    journal-unresolved stderr warning."""
    if payload.get("journal_unresolved"):
        print(f"warning: {payload.get('hint')}", file=sys.stderr)  # never --quiet-suppressed
    if as_json:
        print(json.dumps(payload, indent=2, sort_keys=True))
    elif not quiet:
        migrated = payload["records_migrated"]
        print(
            f"source migrate ok ({len(migrated)} migrated, "
            f"{payload['records_unchanged']} unchanged, "
            f"target {payload['target_schema_version']})"
        )
 
 
# --- hash-preimage (unchanged §3.2 debug surface) ---------------------------
def _cmd_hash_preimage(args: argparse.Namespace) -> int:
    root, exit_code = _resolve_root_or_exit(args)
    if exit_code is not None:
        return exit_code
    try:
        if args.projection == "raw-source-tree":
            preimage = hashing.hash_preimage("raw-source-tree", root=root)
        elif args.projection == "logical-state":
            location = discover_registry_location(registry_path=args.registry)
            backend = SQLiteBackend(location.registry_path)
            with backend.connect() as conn:
                preimage = hashing.hash_preimage("logical-state", conn=conn)
        else:
            records = scan_source_tree(root)
            if args.projection == "semantic":
                # [OP-5] promoted parity: include promoted records in the
                # semantic projection (the earlier model-gap filter is removed).
                loaded = [
                    (record, load_record(record)[0])
                    for record in records
                ]
                preimage = hashing.hash_preimage("semantic", loaded=loaded)
            else:
                preimage = hashing.hash_preimage("source-tree", records=records)
    except SourceError as error:
        _emit_envelope(error, args.format == "json")
        return 2 if isinstance(error, SourceScanError) else 1
    if args.format == "yaml":
        print(yaml.safe_dump(preimage, sort_keys=True, allow_unicode=True), end="")
    else:
        print(json.dumps(preimage, indent=2, sort_keys=True))
    return 0