#!/usr/bin/env python3
"""Harness verification-ledger inputsHash fingerprint reuse (D6).

Subcommands:
  hash       — compute order-independent inputsHash for a file set
  can-reuse  — decide reuse / rerun / insufficient-evidence
  record     — write validation result + inputsHash/inputsFiles into ledger
  render-report — derive the test report from ledger+events (batch 2 WI-4a)

Python 3.10+, stdlib only. UTF-8 without BOM. Windows path safe.
"""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import re
import subprocess
import sys
import tempfile
import uuid
from pathlib import Path
from typing import Any

SCRIPTS_DIR = Path(__file__).resolve().parent
if str(SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPTS_DIR))

import harness_paths  # noqa: E402
import harness_plan_finalize as hpf  # noqa: E402
import harness_profile  # noqa: E402


if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")
if hasattr(sys.stderr, "reconfigure"):
    sys.stderr.reconfigure(encoding="utf-8")


VERIFICATIONS = frozenset(
    {
        "compile",
        "unitTest",
        "unitTestFull",
        "apiTest",
        "browserTest",
        "install",
        "package",
        "dbCompatibility",
    }
)
STATUS_MAP = {
    "ok": "OK",
    "OK": "OK",
    "fail": "FAIL",
    "FAIL": "FAIL",
    "not_run": "NOT_RUN",
    "NOT_RUN": "NOT_RUN",
}
BROAD_SCOPES = frozenset({"module", "module-am", "full"})

# --- Ledger v2 (cluster 2) ---
LEDGER_VERSION = "harness-ledger-2"
DIFF_HASH_VERSION = "content-changeset-2"
TEST_TRACKING_REL = Path("evidence") / "test-tracking.json"
TEST_TRACKING_REASONS = frozenset({"tdd-created", "stale-test-repair", "test-updated"})
# Coverage lattice: a recorded verification's coverage rank must meet the
# verification's required rank. Prevents incremental evidence from satisfying
# a module/full gate (UT-015 / API-005).
COVERAGE_RANK = {"incremental": 0, "module": 1, "module-am": 2, "full": 3}
REQUIRED_COVERAGE = {
    "unitTest": 0,      # incremental suffices (scope checked separately)
    "unitTestFull": 1,  # module or broader
    "compile": 1,
    "apiTest": 1,
    "browserTest": 1,
    "install": 2,       # module-am or broader
    "package": 2,
    "dbCompatibility": 1,
}
# git empty-tree object id (used as base fallback when no commit exists).
_EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"


def resolve_verification_target(
    verification: str,
    project_root: Path | None,
) -> dict[str, Any] | None:
    """Resolve a built-in or project-declared profile-v3 verification target."""
    if not isinstance(verification, str) or not re.fullmatch(
        r"[A-Za-z][A-Za-z0-9_-]{0,63}",
        verification,
    ):
        return None
    if project_root is not None:
        profile = harness_profile.load_profile(project_root)
        graph = (
            profile.get("verificationGraph")
            if isinstance(profile, dict)
            else None
        )
        targets = graph.get("targets") if isinstance(graph, dict) else None
        target = targets.get(verification) if isinstance(targets, dict) else None
        if isinstance(target, dict):
            required = str(target.get("requiredCoverage") or "module")
            if required not in COVERAGE_RANK:
                return None
            return {**target, "requiredCoverage": required}
    if verification in VERIFICATIONS:
        required_rank = REQUIRED_COVERAGE.get(verification, 1)
        required = next(
            (
                name
                for name, rank in COVERAGE_RANK.items()
                if rank == required_rank
            ),
            "module",
        )
        return {
            "commandKey": verification,
            "dependsOn": [],
            "requiredCoverage": required,
            "candidate": verification == "unitTestFull",
            "requiredCapabilities": [],
        }
    return None


def verification_target_identity(
    verification: str,
    target: dict[str, Any],
) -> str:
    payload = json.dumps(
        {"verification": verification, "target": target},
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(payload).hexdigest()


def _hash_text(value: str) -> str:
    return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()


def command_set_hash(command: str) -> str:
    """Stable identity for the exact command set represented by one target."""
    return _hash_text(canonical_command(command))


def canonical_command(command: str) -> str:
    """Normalize supported launch wrappers without mixing runner notes into identity."""
    value = re.sub(r"\s+", " ", str(command).strip())
    value = re.sub(r"\s*\(safe runner\)\s*$", "", value, flags=re.I)
    value = re.sub(r"^npx(?:\s+--no-install)?\s+", "", value, flags=re.I)
    return value.strip()


def default_environment_hash(project_root: Path | None = None) -> str:
    payload = {
        "osName": os.name,
        "platform": sys.platform,
        "pythonImplementation": sys.implementation.name,
        "projectFilesystem": "windows" if os.name == "nt" else "posix",
    }
    return _hash_text(json.dumps(payload, sort_keys=True, separators=(",", ":")))


def default_toolchain_hash(project_root: Path | None = None) -> str:
    root = Path(project_root).resolve() if project_root is not None else None
    manifests: list[dict[str, str]] = []
    if root is not None:
        for name in (
            "package.json", "pyproject.toml", "pom.xml", "build.gradle",
            "build.gradle.kts", "go.mod", "Cargo.toml",
        ):
            path = root / name
            if path.is_file():
                manifests.append({
                    "name": name,
                    "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
                })
    payload = {
        "python": sys.version.split()[0],
        "implementation": sys.implementation.name,
        "manifests": manifests,
    }
    return _hash_text(json.dumps(payload, sort_keys=True, separators=(",", ":")))


def frozen_ownership_check(change_dir: Path) -> dict[str, Any]:
    """Reject evidence writes after a Run capsule's ownership scope changed."""
    try:
        state_root = Path(
            harness_paths.resolve_state_dir_for_contract(change_dir)
        ).resolve()
        contract = harness_paths.load_change_contract(change_dir)
    except (OSError, ValueError, json.JSONDecodeError):
        return {"ok": True, "code": "OWNERSHIP_FREEZE_NOT_APPLICABLE"}
    capsule_root = state_root / "runtime" / "phase-context"
    if not capsule_root.is_dir():
        return {"ok": True, "code": "OWNERSHIP_FREEZE_NOT_APPLICABLE"}
    capsules: list[dict[str, Any]] = []
    # capsule 文件名是 <phase>-<hash>.json；execute 合并（run/test→execute）后
    # 新旧两批名字都可能躺在在途 change 里，按内容归一判定而不是按文件名。
    for path in capsule_root.glob("*.json"):
        try:
            item = json.loads(path.read_text(encoding="utf-8-sig"))
        except (OSError, json.JSONDecodeError):
            continue
        if (
            isinstance(item, dict)
            and harness_paths.resolve_phase_name(item.get("phase")) == "execute"
            and not item.get("closedAt")
            and _nonempty_str(item.get("ownershipHash"))
        ):
            capsules.append(item)
    if not capsules:
        return {"ok": True, "code": "OWNERSHIP_FREEZE_NOT_APPLICABLE"}
    capsule = sorted(capsules, key=lambda item: str(item.get("createdAt") or ""))[-1]
    current = ownership_hash(contract)
    if capsule.get("ownershipHash") != current:
        return {
            "ok": False,
            "code": "OWNERSHIP_CHANGED_BEFORE_VERIFICATION",
            "message": (
                "产品范围在 Run 启动后发生变化；请先通过受控范围更新重启 Run，"
                "系统将只失效受影响的验证。"
            ),
            "storedOwnershipHash": capsule.get("ownershipHash"),
            "currentOwnershipHash": current,
            "runId": capsule.get("runId"),
        }
    return {"ok": True, "code": "OWNERSHIP_FROZEN"}


def product_tree_hash(project_root: Path | None) -> str | None:
    """Git tree identity is content provenance; a commit SHA is not."""
    if project_root is None:
        return None
    tree = _git_text(project_root, "rev-parse", "--verify", "HEAD^{tree}")
    return "sha256:" + tree if _nonempty_str(tree) else None


def worktree_product_tree_hash(project_root: Path | None) -> str | None:
    """当前工作区产品内容的 git tree 哈希（R4：绑定候选实际内容）。

    与 product_tree_hash（HEAD^{tree}）的区别：本函数经由临时索引
    （GIT_INDEX_FILE，不触碰真实索引）把工作区实际内容写成树对象——
    未提交的被跟踪文件修改与未跟踪的非 .harness 文件都会改变结果；
    .harness/ 状态目录始终排除。无产品变更时与 HEAD^{tree} 同值，
    因此空提交/纯 .harness 漂移不会误拒既有收据。
    """
    if project_root is None:
        return None
    head = _git_text(project_root, "rev-parse", "--verify", "HEAD^{tree}")
    if not _nonempty_str(head):
        return None
    # 调用方可能传入仓库内任意子目录（如 change_dir）；pathspec
    # （.harness）与 add -A 的工作区语义都必须以仓库根为 cwd 解析。
    top = _git_text(project_root, "rev-parse", "--show-toplevel")
    work_dir = Path(top) if _nonempty_str(top) else project_root
    fd, tmp = tempfile.mkstemp(prefix="harness-ledger-idx-")
    os.close(fd)
    env = {**os.environ, "GIT_INDEX_FILE": tmp}

    def _run(*args: str) -> str | None:
        proc = subprocess.run(
            ["git", *args],
            cwd=str(work_dir),
            env=env,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            check=False,
        )
        if proc.returncode != 0:
            return None
        return proc.stdout.strip()

    try:
        if _run("read-tree", "HEAD") is None:
            return None
        if _run("add", "-A") is None:
            return None
        # .harness/ 是 harness 状态目录，不计入产品内容（未被 gitignore
        # 的仓库里它是未跟踪文件，会被 add -A 纳入，必须显式剔除）。
        _run("rm", "-r", "--cached", "-f", "--ignore-unmatch", "--quiet", "--",
             ".harness")
        tree = _run("write-tree")
        return "sha256:" + tree if _nonempty_str(tree) else None
    finally:
        try:
            os.unlink(tmp)
        except OSError:
            pass


def lock_hash(project_root: Path | None) -> str | None:
    """Hash present dependency lockfiles, when the project exposes any."""
    if project_root is None:
        return None
    names = (
        "package-lock.json", "npm-shrinkwrap.json", "pnpm-lock.yaml",
        "yarn.lock", "Cargo.lock", "go.sum", "poetry.lock",
    )
    files = [project_root / name for name in names if (project_root / name).is_file()]
    if not files:
        return None
    digest = hashlib.sha256()
    for path in files:
        digest.update(path.name.encode("utf-8"))
        digest.update(b"\0")
        digest.update(hashlib.sha256(path.read_bytes()).digest())
    return "sha256:" + digest.hexdigest()


def canonical_reuse_key(entry: dict[str, Any]) -> dict[str, str]:
    """Commit-independent verification identity used to authorize reuse."""
    fields = (
        "verification", "productTreeHash", "commandSetHash", "environmentHash",
        "toolchainHash", "lockHash", "dbSchemaHash", "targetIdentity",
    )
    return {
        field: str(entry.get(field)).strip()
        for field in fields
        if _nonempty_str(entry.get(field))
    }


def now_iso() -> str:
    return dt.datetime.now().astimezone().isoformat(timespec="milliseconds")


def emit_json(payload: dict[str, Any], *, as_json: bool) -> None:
    text = json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
    if as_json:
        sys.stdout.write(text)
    else:
        ok = payload.get("ok", True)
        reuse = payload.get("reuse")
        if reuse is not None:
            sys.stdout.write(f"reuse={reuse} reason={payload.get('reason')}\n")
        elif "diffHash" in payload:
            sys.stdout.write(f"{payload['diffHash']}\n")
        elif "inputsHash" in payload:
            sys.stdout.write(f"{payload['inputsHash']}\n")
        else:
            sys.stdout.write(("ok" if ok else "error") + "\n")


def _compact_record_payload(payload: dict[str, Any]) -> dict[str, Any]:
    """C5: record compact — ok/action/verification/status only."""
    return {
        "ok": payload.get("ok", True),
        "action": payload.get("action"),
        "verification": payload.get("verification"),
        "status": payload.get("status"),
    }


def _compact_can_reuse_payload(payload: dict[str, Any]) -> dict[str, Any]:
    """C5: can-reuse compact — ok/reuse/code；拒绝复用时附带可执行的原因。

    只回 ok/reuse/code 时，调用方拿到裸 `{"reuse": false}` 无从判断该怎么办，
    只能再跑一次 `--verbose`——原因（profile 缺 verificationInputs、证据不完整等）
    是单行短文本，值得留在默认输出里。允许复用时保持原样精简。
    """
    compact = {
        "ok": payload.get("ok", True),
        "reuse": payload.get("reuse"),
        "code": payload.get("code"),
    }
    if payload.get("reuse") is not True:
        for key in ("reason", "executionNeed", "detail"):
            value = payload.get(key)
            if value not in (None, "", [], {}):
                compact[key] = value
    return compact


def emit_compact_or_verbose(
    payload: dict[str, Any],
    *,
    as_json: bool,
    verbose: bool,
    compact_fn,
) -> None:
    """Emit compact payload by default; full payload when --verbose."""
    out = payload if verbose else compact_fn(payload)
    emit_json(out, as_json=as_json)


def emit_error(
    message: str,
    *,
    as_json: bool,
    code: int = 1,
    error_code: str | None = None,
    extra: dict[str, Any] | None = None,
) -> int:
    payload: dict[str, Any] = {"ok": False, "error": message}
    if error_code:
        payload["code"] = error_code
    if extra:
        payload.update(extra)
    if as_json:
        sys.stderr.write(json.dumps(payload, ensure_ascii=False) + "\n")
    else:
        sys.stderr.write(f"error: {message}\n")
    return code


def resolve_path(raw: str) -> Path:
    return Path(raw).expanduser().resolve()


def parse_files_arg(raw: str | None) -> list[str]:
    if raw is None or not str(raw).strip():
        return []
    parts = [p.strip() for p in str(raw).split(",")]
    return [p for p in parts if p]


def parse_files_manifest(raw: str | None) -> list[str]:
    if raw is None or not str(raw).strip():
        return []
    path = Path(str(raw)).expanduser().resolve()
    if not path.is_file():
        raise FileNotFoundError(f"files manifest not found: {path}")
    return [
        line.strip()
        for line in path.read_text(encoding="utf-8-sig").splitlines()
        if line.strip() and not line.lstrip().startswith("#")
    ]


class ProjectRootRequiredError(ValueError):
    """Raised when relative verification inputs have no stable project basis."""


def resolve_input_files(
    files: list[str], project_root: Path | None
) -> list[str]:
    if project_root is None:
        return files
    root = project_root.expanduser().resolve()
    resolved: list[str] = []
    for raw in files:
        candidate = Path(raw).expanduser()
        candidate = candidate.resolve() if candidate.is_absolute() else (root / candidate).resolve()
        try:
            candidate.relative_to(root)
        except ValueError as exc:
            raise ValueError(
                f"INPUT_OUTSIDE_PROJECT: {candidate} is outside {root}"
            ) from exc
        resolved.append(str(candidate))
    return resolved


def input_files_from_args(args: argparse.Namespace) -> tuple[list[str], Path | None]:
    project_raw = getattr(args, "project", None)
    project_root = (
        Path(str(project_raw)).expanduser().resolve() if project_raw else None
    )
    files = parse_files_arg(getattr(args, "files", None))
    manifest = parse_files_manifest(getattr(args, "files_from", None))
    if files and manifest:
        raise ValueError("use only one of --files and --files-from")
    selected = files or manifest
    if project_root is None and any(
        not Path(raw).expanduser().is_absolute() for raw in selected
    ):
        raise ProjectRootRequiredError(
            "relative verification inputs require --project so their identity "
            "does not depend on the caller's current directory"
        )
    return resolve_input_files(selected, project_root), project_root


def infer_execution_project_root(change_dir: Path) -> Path | None:
    """Resolve the tree under test from trusted change metadata.

    Worktree changes keep their durable contract under the main checkout while
    product files live in the execution worktree.  Requiring every caller to
    reconstruct that split caused ledger commands to hash the main checkout or
    fail repeatedly.  Prefer the persisted execution root and otherwise leave
    legacy callers unchanged.
    """
    change_dir = change_dir.expanduser().resolve()
    project_root = next(
        (ancestor.parent for ancestor in change_dir.parents if ancestor.name == ".harness"),
        None,
    )
    if project_root is None:
        return None

    def git_common_dir(root: Path) -> Path | None:
        process = subprocess.run(
            ["git", "-C", str(root), "rev-parse", "--path-format=absolute", "--git-common-dir"],
            text=True,
            encoding="utf-8",
            errors="replace",
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )
        if process.returncode != 0 or not process.stdout.strip():
            return None
        raw = Path(process.stdout.strip()).expanduser()
        return raw.resolve() if raw.is_absolute() else (root / raw).resolve()

    project_common_dir = git_common_dir(project_root)

    def trusted_worktree(candidate: Path) -> Path | None:
        if not candidate.is_dir():
            return None
        process = subprocess.run(
            ["git", "-C", str(candidate), "rev-parse", "--show-toplevel"],
            text=True,
            encoding="utf-8",
            errors="replace",
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )
        if process.returncode != 0 or not process.stdout.strip():
            return None
        top_level = Path(process.stdout.strip()).expanduser().resolve()
        if top_level != candidate:
            return None
        # Origin URL + root commit identifies a repository lineage, not a
        # linked checkout. Independent clones can share both. A real linked
        # worktree must share the exact Git common directory with the main tree.
        if project_common_dir is None or git_common_dir(top_level) != project_common_dir:
            return None
        return top_level

    candidates = (
        (
            change_dir / "meta" / "change-context.json",
            ("worktreePath", "path", "worktreeRoot"),
        ),
        (
            change_dir / "meta" / "worktree.json",
            ("worktreePath", "path", "worktreeRoot"),
        ),
    )
    for metadata_path, keys in candidates:
        if not metadata_path.is_file():
            continue
        try:
            payload = json.loads(metadata_path.read_text(encoding="utf-8-sig"))
        except (OSError, json.JSONDecodeError):
            continue
        if not isinstance(payload, dict):
            continue
        for key in keys:
            raw = payload.get(key)
            if not isinstance(raw, str) or not raw.strip():
                continue
            raw_path = Path(raw).expanduser()
            candidate = (
                raw_path.resolve()
                if raw_path.is_absolute()
                else (project_root / raw_path).resolve()
            )
            if key == "worktreeRoot" and candidate.name != change_dir.name:
                nested = (candidate / change_dir.name).resolve()
                trusted_nested = trusted_worktree(nested)
                if trusted_nested is not None:
                    return trusted_nested
                if trusted_worktree(candidate) is None:
                    continue
            trusted_candidate = trusted_worktree(candidate)
            if trusted_candidate is not None:
                return trusted_candidate
    return None


def declares_execution_worktree(change_dir: Path) -> bool:
    """Return whether persisted metadata explicitly selects another checkout."""
    for metadata_path in (
        change_dir / "meta" / "change-context.json",
        change_dir / "meta" / "worktree.json",
    ):
        if not metadata_path.is_file():
            continue
        try:
            payload = json.loads(metadata_path.read_text(encoding="utf-8-sig"))
        except (OSError, json.JSONDecodeError):
            continue
        if isinstance(payload, dict) and any(
            isinstance(payload.get(key), str) and str(payload[key]).strip()
            for key in ("worktreePath", "path", "worktreeRoot")
        ):
            return True
    return False


def apply_inferred_project_root(args: argparse.Namespace, change_dir: Path) -> None:
    """Fill an omitted --project without overriding an explicit caller choice."""
    if getattr(args, "project", None):
        return
    inferred = infer_execution_project_root(change_dir)
    if inferred is not None:
        args.project = str(inferred)


# Per-process sha256 cache keyed by (st_dev, st_ino) file identity +
# (size, mtime_ns) fingerprint — the same discipline as harness_archive's
# sha256_file. compute_inputs_hash is the shared verification-fingerprint
# primitive (state snapshot segments, ledger validation reuse, service
# session fingerprint polling) and repeatedly hashes the same unchanged input
# files within one process. Any write updates mtime so a stat match proves the
# bytes are unchanged; identity survives renames on the same volume. Unreadable
# files are never cached (callers must see the FileNotFoundError).
_LEDGER_SHA256_CACHE_MAX = 131_072
_ledger_sha256_cache: dict[tuple, tuple[int, int, str]] = {}


def _ledger_sha256_cache_key(path: Path, stat: os.stat_result) -> tuple:
    if stat.st_ino:
        return (stat.st_dev, stat.st_ino)
    return ("path", str(path))


def sha256_file(path: Path) -> str:
    try:
        stat = path.stat()
    except OSError:
        stat = None
    if stat is not None:
        cached = _ledger_sha256_cache.get(_ledger_sha256_cache_key(path, stat))
        if (
            cached is not None
            and cached[0] == stat.st_size
            and cached[1] == stat.st_mtime_ns
            and cached[2] == stat.st_ctime_ns
        ):
            return cached[3]
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        while True:
            chunk = handle.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    result = digest.hexdigest()
    try:
        # Stat after the read so the cached fingerprint matches the bytes that
        # were actually hashed.
        stat = path.stat()
    except OSError:
        return result
    if len(_ledger_sha256_cache) >= _LEDGER_SHA256_CACHE_MAX:
        _ledger_sha256_cache.clear()
    # (size, mtime_ns, ctime_ns): mtime + ctime share one clock (~2ms NTFS
    # granularity) - a second sample, not an independent signal; same-tick
    # rewrites remain a documented cache boundary.
    _ledger_sha256_cache[_ledger_sha256_cache_key(path, stat)] = (
        stat.st_size,
        stat.st_mtime_ns,
        stat.st_ctime_ns,
        result,
    )
    return result


def compute_inputs_hash(
    file_paths: list[str], *, project_root: Path | str | None = None
) -> tuple[str, list[str]]:
    """Per-file content sha256 → sort digests → hash again (order-independent)."""
    root = (
        Path(project_root).expanduser().resolve()
        if project_root is not None
        else None
    )
    by_path: dict[str, str] = {}
    for raw in file_paths:
        path = resolve_path(raw)
        if not path.is_file():
            raise FileNotFoundError(f"file not found: {raw}")
        if root is not None:
            try:
                logical_path = path.relative_to(root).as_posix()
            except ValueError as exc:
                raise ValueError(
                    f"INPUT_OUTSIDE_PROJECT: {path} is outside {root}"
                ) from exc
        else:
            logical_path = path.as_posix()
        by_path[logical_path] = sha256_file(path)

    # Stable file list for callers; bind every resolved path to its digest.
    # Hashing only the content multiset let a path swap incorrectly reuse a
    # verification result.
    resolved_files_sorted = sorted(by_path.keys())

    combined = hashlib.sha256()
    for path in resolved_files_sorted:
        combined.update(path.encode("utf-8"))
        combined.update(b"\0")
        combined.update(by_path[path].encode("ascii"))
        combined.update(b"\n")
    return f"sha256:{combined.hexdigest()}", resolved_files_sorted


def _git_bytes(args: list[str], cwd: Path) -> bytes:
    """Run git, return raw stdout bytes. Raises RuntimeError on non-zero exit."""
    proc = subprocess.run(["git", *args], cwd=str(cwd), capture_output=True)
    if proc.returncode != 0:
        msg = proc.stderr.decode("utf-8", "replace").strip()
        raise RuntimeError(f"git {' '.join(args)} failed: {msg}")
    return proc.stdout


def _git_text(cwd: Path, *args: str) -> str | None:
    """Run git, return stripped stdout; None on non-zero exit."""
    proc = subprocess.run(
        ["git", *args],
        cwd=str(cwd),
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        check=False,
    )
    if proc.returncode != 0:
        return None
    return proc.stdout.strip()


def _root_commit(repo_root: Path) -> str | None:
    out = _git_bytes(["rev-list", "--max-parents=0", "HEAD"], repo_root)
    lines = [ln for ln in out.decode("utf-8", "replace").splitlines() if ln.strip()]
    return lines[0] if lines else None


def _changed_paths(repo_root: Path, base: str | None) -> tuple[str, list[str]]:
    """Commit-invariant change set: tracked files differing from base (working
    tree) plus untracked files. Sorted by repo-relative path.

    Working-tree content is unchanged by a checkpoint commit, so the change set
    and its bytes are identical before/after commit (UT-011).
    """
    if not base:
        base = _root_commit(repo_root) or _EMPTY_TREE
    out = _git_bytes(["diff", "--name-only", "--no-renames", "-z", base], repo_root)
    paths: set[str] = set()
    for name in out.split(b"\x00"):
        if name:
            paths.add(name.decode("utf-8"))
    out2 = _git_bytes(["ls-files", "--others", "--exclude-standard", "-z"], repo_root)
    for name in out2.split(b"\x00"):
        if name:
            paths.add(name.decode("utf-8"))
    return base, sorted(paths)


def _inside(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
        return True
    except ValueError:
        return False


def _tracked_test_contents(
    repo_root: Path,
    change_dir: Path | str | None,
) -> tuple[dict[str, bytes], str | None]:
    """Load exact test paths recorded by harness_test_guard.

    The ledger validates the same security-critical manifest fields again so a
    hand-edited manifest cannot silently widen the fingerprint or reuse stale
    evidence. Missing manifests remain backward-compatible and contribute no
    additional paths.
    """
    if change_dir is None:
        return {}, None
    candidate = Path(change_dir)
    if not candidate.is_absolute():
        candidate = repo_root / candidate
    change_root = candidate.resolve()
    main_root = harness_paths.resolve_main_project_root(repo_root)
    if not (_inside(change_root, repo_root) or _inside(change_root, main_root)):
        raise ValueError("TEST_TRACKING_CHANGE_DIR_OUTSIDE_PROJECT")
    state_root = _state_dir(change_root)
    manifest_path = (state_root / TEST_TRACKING_REL).resolve()
    if not _inside(manifest_path, state_root):
        raise ValueError("TEST_TRACKING_MANIFEST_OUTSIDE_CHANGE")
    if not manifest_path.is_file():
        return {}, None
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"TEST_TRACKING_MANIFEST_INVALID: {exc}") from exc
    if isinstance(manifest, dict) and manifest.get("schemaVersion") == 2:
        return _tracked_test_contents_v2(repo_root, manifest, manifest_path)
    if (
        not isinstance(manifest, dict)
        or manifest.get("schemaVersion") != 1
        or manifest.get("mode") != "force-track-touched"
        or manifest.get("projectRoot") != str(repo_root)
    ):
        raise ValueError("TEST_TRACKING_MANIFEST_INVALID")
    entries = manifest.get("files")
    if not isinstance(entries, list) or not entries:
        raise ValueError("TEST_TRACKING_MANIFEST_INVALID: EMPTY_FILES")

    contents: dict[str, bytes] = {}
    for item in entries:
        if not isinstance(item, dict):
            raise ValueError("TEST_TRACKING_MANIFEST_INVALID")
        rel = item.get("path")
        expected_hash = item.get("sha256")
        if (
            not isinstance(rel, str)
            or not rel
            or not isinstance(expected_hash, str)
            or item.get("reason") not in TEST_TRACKING_REASONS
            or type(item.get("ignored")) is not bool
            or type(item.get("trackedBefore")) is not bool
        ):
            raise ValueError("TEST_TRACKING_MANIFEST_INVALID")
        raw_path = Path(rel)
        resolved = (repo_root / raw_path).resolve()
        if raw_path.is_absolute() or not _inside(resolved, repo_root):
            raise ValueError(f"TEST_TRACKING_PATH_OUTSIDE_PROJECT: {rel}")
        normalized = resolved.relative_to(repo_root).as_posix()
        if normalized != rel or not resolved.is_file():
            raise ValueError(f"TEST_TRACKING_FILE_INVALID: {rel}")
        content = resolved.read_bytes()
        actual_hash = "sha256:" + hashlib.sha256(content).hexdigest()
        if actual_hash != expected_hash:
            raise ValueError(f"TEST_TRACKING_HASH_DRIFT: {rel}")
        contents[rel] = content
    return {rel: contents[rel] for rel in sorted(contents)}, str(manifest_path)


def _tracked_test_contents_v2(
    repo_root: Path,
    manifest: dict[str, Any],
    manifest_path: Path,
) -> tuple[dict[str, bytes], str | None]:
    """Validate a schema-2 manifest: repositoryId equality + logical hashes."""
    if manifest.get("mode") != "force-track-touched":
        raise ValueError("TEST_TRACKING_MANIFEST_INVALID")
    repository_id = manifest.get("repositoryId")
    if not isinstance(repository_id, str) or not repository_id.startswith("sha256:"):
        raise ValueError("TEST_TRACKING_MANIFEST_INVALID")
    if repository_id != harness_paths.repository_identity(repo_root):
        raise ValueError("TEST_TRACKING_REPOSITORY_MISMATCH")
    entries = manifest.get("files")
    if not isinstance(entries, list) or not entries:
        raise ValueError("TEST_TRACKING_MANIFEST_INVALID: EMPTY_FILES")

    contents: dict[str, bytes] = {}
    for item in entries:
        if not isinstance(item, dict):
            raise ValueError("TEST_TRACKING_MANIFEST_INVALID")
        rel = item.get("path")
        expected = item.get("logicalHash") or item.get("binaryHash")
        if (
            not isinstance(rel, str)
            or not rel
            or not isinstance(expected, str)
            or not (expected.startswith("gitblob:") or expected.startswith("sha256:"))
            or item.get("reason") not in TEST_TRACKING_REASONS
            or type(item.get("ignored")) is not bool
            or not isinstance(item.get("introducedBy"), str)
            or not isinstance(item.get("touchedBy"), list)
            or item.get("commitScope") not in ("current-change", "foreign-change")
        ):
            raise ValueError("TEST_TRACKING_MANIFEST_INVALID")
        raw_path = Path(rel)
        resolved = (repo_root / raw_path).resolve()
        if raw_path.is_absolute() or not _inside(resolved, repo_root):
            raise ValueError(f"TEST_TRACKING_PATH_OUTSIDE_PROJECT: {rel}")
        normalized = resolved.relative_to(repo_root).as_posix()
        if normalized != rel or not resolved.is_file():
            raise ValueError(f"TEST_TRACKING_FILE_INVALID: {rel}")
        content = resolved.read_bytes()
        actual = _logical_file_hash(repo_root, rel, content)
        if actual != expected:
            raise ValueError(f"TEST_TRACKING_HASH_DRIFT: {rel}")
        contents[rel] = content
    return {rel: contents[rel] for rel in sorted(contents)}, str(manifest_path)


def _logical_file_hash(repo_root: Path, rel: str, content: bytes) -> str:
    """Mirror of harness_test_guard.logical_file_hash for validation."""
    attr = subprocess.run(
        ["git", "check-attr", "text", "--", rel],
        cwd=str(repo_root),
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        check=False,
    )
    attr_out = attr.stdout.strip() if attr.returncode == 0 else ""
    byte_hash = "sha256:" + hashlib.sha256(content).hexdigest()
    if attr_out.endswith(": unset") or b"\x00" in content:
        return byte_hash
    try:
        content.decode("utf-8")
    except UnicodeDecodeError:
        return byte_hash
    proc = subprocess.run(
        ["git", "hash-object", "--path", rel, "--stdin"],
        input=content,
        capture_output=True,
        cwd=str(repo_root),
        check=False,
    )
    if proc.returncode != 0:
        return byte_hash
    return "gitblob:" + proc.stdout.decode("ascii").strip()


def compute_diff_hash(
    repo_root: Path,
    base: str | None = None,
    change_dir: Path | str | None = None,
) -> tuple[str, dict[str, Any]]:
    """Byte-level, commit-invariant diff hash (cluster 2).

    Content-based change set: every file whose working-tree content differs
    from base, plus untracked files. Each entry is length-framed
    (path-len | path | exists-flag | content-len | content) so the payload is
    independent of shell, console encoding, BOM and system newlines. Stable
    across a checkpoint commit because git does not mutate the working tree.
    """
    repo_root = Path(repo_root).resolve()
    base, paths = _changed_paths(repo_root, base)
    tracked_test_contents, manifest_path = _tracked_test_contents(repo_root, change_dir)
    paths = sorted(set(paths).union(tracked_test_contents))
    digest = hashlib.sha256()
    digest.update(DIFF_HASH_VERSION.encode("utf-8"))
    digest.update(b"\x00")
    for rel in paths:
        path_bytes = rel.encode("utf-8")
        digest.update(len(path_bytes).to_bytes(4, "big"))
        digest.update(path_bytes)
        abs_path = repo_root / rel
        if rel in tracked_test_contents:
            content = tracked_test_contents[rel]
            digest.update(b"\x01")
            digest.update(len(content).to_bytes(8, "big"))
            digest.update(content)
        elif abs_path.is_file():
            content = abs_path.read_bytes()
            digest.update(b"\x01")
            digest.update(len(content).to_bytes(8, "big"))
            digest.update(content)
        else:
            # Deleted since base: record absence with no content.
            digest.update(b"\x00")
            digest.update((0).to_bytes(8, "big"))
    for rel, verified_content in tracked_test_contents.items():
        path = repo_root / rel
        try:
            current_content = path.read_bytes()
        except OSError as exc:
            raise ValueError(f"TEST_TRACKING_HASH_DRIFT: {rel}") from exc
        if current_content != verified_content:
            raise ValueError(f"TEST_TRACKING_HASH_DRIFT: {rel}")
    try:
        head = _git_bytes(["rev-parse", "HEAD"], repo_root).decode("utf-8", "replace").strip()
    except RuntimeError:
        head = ""
    meta = {
        "algorithmVersion": DIFF_HASH_VERSION,
        "fileCount": len(paths),
        "base": base,
        "head": head or None,
        "trackedTestFileCount": len(tracked_test_contents),
        "testTrackingManifest": manifest_path,
    }
    return f"sha256:{digest.hexdigest()}", meta


def derive_coverage(verification: str, scope: str | None) -> str:
    """Derive coverage lattice value from verification + scope (cluster 2)."""
    s = str(scope).strip() if scope else ""
    if verification == "unitTest":
        return "module" if s in BROAD_SCOPES else "incremental"
    if verification == "unitTestFull":
        return "full" if s == "full" else "module"
    if verification in ("install", "package"):
        return "module-am"
    return "module"


def expand_profile_input_files(
    project: Path, profile_input: str
) -> tuple[list[str], str | None]:
    """Expand verificationInputs[profile_input] globs from build-profile.json.

    Profile is loaded via ``harness_profile.load_profile`` (C7: common_root then
    execution overlay) so linked worktrees without a local build-profile still
    reuse the main checkout profile. Globs remain relative to the execution
    ``project`` root (not common_root) so inputsHash tracks the tree under test.

    Returns (files, error); error is None on success.
    profile 缺失 / key 缺失 / glob 无匹配 / 结果为空 → 返回 ([], "<reason>")，
    调用方据此返回 insufficient-evidence，执行全量测试但不允许缓存复用。
    """
    profile = harness_profile.load_profile(Path(project))
    if profile is None:
        # load_profile swallows JSON errors; restore actionable unreadable diag
        # when a profile file exists but cannot be parsed (review YELLOW-1).
        project_root = Path(project).resolve()
        common = harness_paths.common_root(project_root)
        for root in (common, project_root):
            candidate = root / ".harness" / "config" / "build-profile.json"
            if not candidate.is_file():
                continue
            try:
                json.loads(candidate.read_text(encoding="utf-8-sig"))
            except (OSError, json.JSONDecodeError) as exc:
                return [], f"build-profile.json unreadable: {exc}"
        return [], "build-profile.json missing; run harness_preflight.py detect"
    if not isinstance(profile, dict):
        return [], "build-profile.json is not an object"
    inputs = profile.get("verificationInputs")
    if not isinstance(inputs, dict) or profile_input not in inputs:
        return [], f"verificationInputs.{profile_input} missing in build-profile.json"
    patterns = inputs[profile_input]
    if not isinstance(patterns, list) or not patterns:
        return [], f"verificationInputs.{profile_input} is empty or invalid"

    base = Path(project).resolve()
    seen: set[str] = set()
    for pat in patterns:
        if not isinstance(pat, str) or not pat.strip():
            continue
        for match in base.glob(pat):
            if not match.is_file():
                continue
            resolved = match.resolve()
            try:
                resolved.relative_to(base)
            except ValueError:
                # 拒绝 project 外部路径，禁止 glob 逃逸。
                continue
            seen.add(resolved.as_posix())
    if not seen:
        return [], f"verificationInputs.{profile_input} matched no files"
    return sorted(seen), None


def ensure_profile_input_target(
    project: Path,
    profile_input: str,
) -> tuple[dict[str, Any] | None, str | None]:
    """Refresh a missing/stale detected profile before recording evidence."""

    project_root = Path(project).resolve()
    profile = harness_profile.load_profile(project_root)
    inputs = profile.get("verificationInputs") if isinstance(profile, dict) else None
    graph = profile.get("verificationGraph") if isinstance(profile, dict) else None
    targets = graph.get("targets") if isinstance(graph, dict) else None
    if (
        isinstance(inputs, dict)
        and profile_input in inputs
        and isinstance(targets, dict)
        and isinstance(targets.get(profile_input), dict)
    ):
        return profile, None

    # An unreadable profile needs an explicit repair; silently replacing it can
    # discard user-owned overrides.
    common = harness_paths.common_root(project_root)
    for root in dict.fromkeys((common, project_root)):
        candidate = root / ".harness" / "config" / "build-profile.json"
        if not candidate.is_file():
            continue
        try:
            json.loads(candidate.read_text(encoding="utf-8-sig"))
        except (OSError, json.JSONDecodeError) as exc:
            return None, f"build-profile.json unreadable: {exc}"

    detected = harness_profile.detect(project_root)
    if not detected.get("ok"):
        return None, str(
            detected.get("message")
            or detected.get("code")
            or "build profile detection failed"
        )
    refreshed = harness_profile.load_profile(project_root)
    if not isinstance(refreshed, dict):
        return None, "build-profile.json missing after detection"
    return refreshed, None


def _state_dir(change_dir: Path) -> Path:
    contract = Path(change_dir).resolve()
    # 标准布局（.harness/changes/<id>）按路径形态把项目根作为解析 cwd：
    # 无 git 的恢复/解压目录也能解析 split state（对齐 harness_archive
    # .find_project_root 长期行为，F1/O6 收敛 archive 旧 load_ledger 后
    # 由此处统一承担该语义）；git 可用时 resolve_main_project_root 仍按
    # common-dir 裁决，worktree 等形态行为不变。
    cwd = contract
    if contract.parent.name == "changes" and contract.parent.parent.name == ".harness":
        cwd = contract.parents[2]
    return Path(harness_paths.resolve_state_dir_for_contract(contract, cwd))


def ledger_candidates(change_dir: Path) -> list[Path]:
    state = _state_dir(change_dir)
    contract = Path(change_dir)
    candidates = [state / "evidence" / "verification-ledger.json"]
    if state != contract:
        candidates.append(contract / "evidence" / "verification-ledger.json")
    candidates.append(state / "verification-ledger.json")
    if state != contract:
        candidates.append(contract / "verification-ledger.json")
    return candidates


def find_ledger_path(change_dir: Path) -> Path | None:
    for path in ledger_candidates(change_dir):
        if path.is_file():
            return path
    return None


def preferred_write_path(change_dir: Path) -> Path:
    # New writes always go to evidence/ (protocol preferred path); split-v1
    # changes route to the dynamic state root.
    return _state_dir(change_dir) / "evidence" / "verification-ledger.json"


def load_ledger(change_dir: Path) -> tuple[dict[str, Any] | None, Path | None]:
    path = find_ledger_path(change_dir)
    if path is None:
        return None, None
    text = path.read_text(encoding="utf-8-sig")
    if not text.strip():
        return {}, path
    data = json.loads(text)
    if data is None:
        return {}, path
    if not isinstance(data, dict):
        raise ValueError(f"ledger must be a JSON object: {path}")
    return data, path


def load_validations(change_dir: Path) -> dict[str, Any]:
    """盘上 ledger validations 的唯一宽容读取器（F3/O6）。

    任何失败（无账本 / 坏 JSON / 非 dict / validations 非 dict）一律
    返回 {}——调用方不需要区分缺失与损坏，只需要「没有可用验证事实」。
    """
    path = find_ledger_path(change_dir)
    if path is None:
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, ValueError):
        return {}
    if not isinstance(data, dict):
        return {}
    validations = data.get("validations")
    return validations if isinstance(validations, dict) else {}


def all_validations_ok(validations: Any) -> bool:
    """终态账本完整性判定（F3/O6 唯一权威）：非空且每条目 status 全 OK。"""
    if not isinstance(validations, dict) or not validations:
        return False
    return all(
        isinstance(entry, dict) and entry.get("status") == "OK"
        for entry in validations.values()
    )


def find_reusable_evidence(
    entries: list[dict[str, Any]],
    *,
    evidence_id: str,
    target_id: str,
    product_identity: str,
) -> dict[str, Any] | None:
    """REUSE 身份匹配判定（F3/O6 唯一权威）。

    命中条件：evidenceId 精确相等 + targetId（缺省回退 verification 键）
    精确相等 + productIdentity 精确相等 + status=="OK"；按列表序取首个。
    """
    for item in entries:
        if (
            str(item.get("evidenceId") or "") == evidence_id
            and str(item.get("targetId") or item.get("verification") or "")
            == target_id
            and str(item.get("productIdentity") or "") == product_identity
            and item.get("status") == "OK"
        ):
            return item
    return None


def write_ledger(path: Path, data: dict[str, Any]) -> None:
    """Atomic ledger write: temp -> fsync -> replace (ledger v3)."""
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
    tmp = path.with_name(f".{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
    try:
        with open(tmp, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(tmp, path)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise


LEDGER_SCHEMA_VERSION = 3
LEDGER_IDENTITY_FIELDS = (
    "repositoryId",
    "changeName",
    "baseCommit",
    "currentHead",
    "diffHash",
    "ownershipHash",
)


def validate_ledger_identity(ledger: dict[str, Any]) -> list[str]:
    """Missing/invalid top-level identity fields (ledger v3)."""
    missing: list[str] = []
    if not isinstance(ledger, dict):
        return ["ledger"]
    if ledger.get("schemaVersion") != LEDGER_SCHEMA_VERSION:
        missing.append("schemaVersion")
    for field in LEDGER_IDENTITY_FIELDS:
        value = ledger.get(field)
        if not isinstance(value, str) or not value.strip():
            missing.append(field)
    return missing


def record_integration_hashes(
    ledger_path: Path,
    *,
    change_dir: Path | None = None,
    repository_id: str,
    merge_final_hash: str,
    ci_expected_head: str,
    remote_head: str,
) -> dict[str, Any]:
    """Atomically attach post-push hashes using the change contract's ledger rules."""
    path = Path(ledger_path).resolve()
    if not path.is_file():
        return {"ok": False, "code": "LEDGER_MISSING", "ledgerPath": str(path)}
    try:
        ledger = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, json.JSONDecodeError) as exc:
        return {
            "ok": False, "code": "LEDGER_INVALID", "ledgerPath": str(path),
            "message": str(exc),
        }
    legacy_contract = False
    if change_dir is not None:
        resolved_change_dir = Path(change_dir).resolve()
        if ledger.get("changeName") != resolved_change_dir.name:
            return {
                "ok": False,
                "code": "LEDGER_CHANGE_MISMATCH",
                "ledgerPath": str(path),
            }
        legacy_contract = not _contract_is_v2(resolved_change_dir)

    missing = validate_ledger_identity(ledger)
    if missing and not legacy_contract:
        return {
            "ok": False,
            "code": "LEDGER_IDENTITY_INVALID",
            "ledgerPath": str(path),
            "missing": missing,
        }
    if not missing and ledger.get("repositoryId") != repository_id:
        return {
            "ok": False,
            "code": "LEDGER_REPOSITORY_MISMATCH",
            "ledgerPath": str(path),
        }
    values = {
        "mergeFinalHash": merge_final_hash,
        "ciExpectedHead": ci_expected_head,
        "remoteHead": remote_head,
    }
    invalid = [
        field for field, value in values.items()
        if not isinstance(value, str) or re.fullmatch(r"[0-9a-f]{40}", value) is None
    ]
    if invalid:
        return {
            "ok": False,
            "code": "FINAL_HASH_INVALID",
            "ledgerPath": str(path),
            "invalid": invalid,
        }
    if len(set(values.values())) != 1:
        return {
            "ok": False,
            "code": "FINAL_HASH_MISMATCH",
            "ledgerPath": str(path),
            **values,
        }
    ledger.update(values)
    ledger["integrationFinalizedAt"] = now_iso()
    write_ledger(path, ledger)
    return {"ok": True, "code": "INTEGRATION_HASHES_RECORDED", "ledgerPath": str(path), **values}


def _contract_is_v2(change_dir: Path) -> bool:
    try:
        contract = harness_paths.load_change_contract(change_dir)
    except (OSError, ValueError):
        return False
    if harness_paths.contract_layout_kind(contract) == "split-v1":
        return True
    version = contract.get("schemaVersion")
    return isinstance(version, int) and version >= 2


def ownership_hash(contract: dict[str, Any]) -> str:
    ownership = contract.get("ownership") or {}
    canonical = json.dumps(ownership, ensure_ascii=False, sort_keys=True)
    return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class LedgerMigrationError(ValueError):
    """A legacy ledger could not be upgraded without inventing identity."""


def ledger_evidence_identity(ledger: dict[str, Any]) -> str:
    """Stable identity of evidence-bearing fields, excluding schema metadata."""
    evidence = {
        key: ledger[key]
        for key in (
            "validations",
            "verificationTargets",
            "integration",
            "artifacts",
        )
        if key in ledger
    }
    canonical = json.dumps(
        evidence,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(canonical).hexdigest()


def _immutable_change_base(
    change_dir: Path,
    project_root: Path,
) -> str:
    roots = [change_dir]
    try:
        state_root = harness_paths.resolve_state_dir_for_contract(
            change_dir,
            project_root,
        )
        if state_root.resolve() != change_dir.resolve():
            roots.insert(0, state_root)
    except (OSError, ValueError):
        pass
    for root in roots:
        for relative in (
            Path("meta") / "state-snapshot.json",
            Path("state-snapshot.json"),
        ):
            path = root / relative
            if not path.is_file():
                continue
            try:
                payload = json.loads(path.read_text(encoding="utf-8-sig"))
            except (OSError, json.JSONDecodeError, TypeError):
                continue
            if not isinstance(payload, dict):
                continue
            value = str(payload.get("changeBase") or "").strip()
            if value:
                return value
    return ""


def migrate_ledger_for_write(
    change_dir: Path,
    ledger: dict[str, Any],
    *,
    project_root: Path | None,
    base_commit: str | None,
    diff_hash: str | None,
    record_migration: bool,
) -> tuple[dict[str, Any], dict[str, Any] | None]:
    """Upgrade one supported legacy window to v3 before the caller mutates evidence."""
    raw_schema = ledger.get("schemaVersion")
    source_schema = raw_schema if isinstance(raw_schema, int) else 1
    if source_schema not in {1, 2, LEDGER_SCHEMA_VERSION}:
        raise LedgerMigrationError(
            f"unsupported ledger schemaVersion={raw_schema!r}; "
            f"supported source versions are 1, 2, and {LEDGER_SCHEMA_VERSION}"
        )
    repo_probe = project_root or change_dir
    repo_root_raw = _git_text(repo_probe, "rev-parse", "--show-toplevel")
    current_head = _git_text(repo_probe, "rev-parse", "--verify", "HEAD")
    if not repo_root_raw or not current_head:
        raise LedgerMigrationError(
            "repository root/current HEAD could not be established"
        )
    repo_root = Path(repo_root_raw).resolve()
    try:
        contract = harness_paths.load_change_contract(change_dir)
    except (OSError, ValueError) as exc:
        raise LedgerMigrationError(
            f"change ownership contract could not be loaded: {exc}"
        ) from exc

    resolved_base = base_commit
    if not _nonempty_str(resolved_base):
        resolved_base = ledger.get("baseCommit")
    if not _nonempty_str(resolved_base):
        resolved_base = _immutable_change_base(change_dir, repo_root)
    if not _nonempty_str(resolved_base):
        if record_migration:
            raise LedgerMigrationError(
                "immutable change base is missing; recapture the Plan boundary "
                "or pass an explicit --base-commit instead of using write-time HEAD"
            )
        # A genuinely new ledger establishes its identity at the first write.
        # Existing ledgers are handled above and may never take this fallback.
        resolved_base = current_head
    resolved_diff = diff_hash
    if not _nonempty_str(resolved_diff):
        try:
            resolved_diff = compute_ownership_diff(
                repo_root,
                base=str(resolved_base).strip(),
                change_dir=change_dir,
            )["diffHash"]
        except (OSError, ValueError, RuntimeError) as exc:
            raise LedgerMigrationError(
                f"ownership diff identity could not be computed: {exc}"
            ) from exc

    before_identity = ledger_evidence_identity(ledger)
    migrated = dict(ledger)
    migrated["schemaVersion"] = LEDGER_SCHEMA_VERSION
    try:
        migrated["repositoryId"] = harness_paths.repository_identity(repo_root)
        migrated["ownershipHash"] = ownership_hash(contract)
    except (OSError, TypeError, ValueError, RuntimeError) as exc:
        raise LedgerMigrationError(
            f"repository/ownership identity could not be established: {exc}"
        ) from exc
    migrated["changeName"] = change_dir.name
    migrated["baseCommit"] = str(resolved_base).strip()
    migrated["currentHead"] = str(current_head).strip()
    migrated["diffHash"] = str(resolved_diff).strip()
    missing = validate_ledger_identity(migrated)
    if missing:
        raise LedgerMigrationError(
            "migrated ledger identity is incomplete: " + ", ".join(missing)
        )

    receipt: dict[str, Any] | None = None
    if source_schema != LEDGER_SCHEMA_VERSION and record_migration:
        after_identity = ledger_evidence_identity(migrated)
        if after_identity != before_identity:
            raise LedgerMigrationError(
                "migration changed pre-existing evidence identity"
            )
        receipt = {
            "schemaVersion": 1,
            "action": "ledger-schema-migration",
            "originalSchemaVersion": source_schema,
            "targetSchemaVersion": LEDGER_SCHEMA_VERSION,
            "migratedAt": now_iso(),
            "evidenceIdentityBefore": before_identity,
            "evidenceIdentityAfter": after_identity,
            "repositoryId": migrated["repositoryId"],
            "baseCommit": migrated["baseCommit"],
            "currentHead": migrated["currentHead"],
            "diffHash": migrated["diffHash"],
        }
        receipt_identity = json.dumps(
            receipt,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
        receipt["receiptId"] = (
            "sha256:" + hashlib.sha256(receipt_identity).hexdigest()
        )
        history = migrated.get("migrationHistory")
        if history is None:
            history = []
        if not isinstance(history, list):
            raise LedgerMigrationError("migrationHistory must be an array")
        migrated["migrationHistory"] = [*history, receipt]
    return migrated, receipt


def _deterministic_rerecord_command(
    args: argparse.Namespace,
    *,
    files: list[str],
    project_root: Path | None,
) -> list[str]:
    command = [
        sys.executable,
        str(Path(__file__).resolve()),
        "--json",
        "record",
        "--change-dir",
        str(resolve_path(args.change_dir)),
        "--verification",
        str(args.verification),
        "--status",
        str(args.status),
        "--command",
        str(args.command),
        "--exit-code",
        str(args.exit_code),
        "--duration-ms",
        str(args.duration_ms),
        "--evidence",
        str(args.evidence),
        "--files",
        ",".join(sorted(str(Path(item).resolve()) for item in files)),
    ]
    if project_root is not None:
        command.extend(["--project", str(project_root)])
    for flag, attr in (
        ("--scope", "scope"),
        ("--coverage", "coverage"),
        ("--base-commit", "base_commit"),
        ("--diff-hash", "diff_hash"),
    ):
        value = getattr(args, attr, None)
        if _nonempty_str(value):
            command.extend([flag, str(value).strip()])
    return command


_METRICS_SCHEMAS: dict[str, dict[str, tuple[str, ...]]] = {
    "unitTest": {"required": ("total", "passed", "failed"), "optional": ("errors", "skipped")},
    "unitTestFull": {"required": ("total", "passed", "failed"), "optional": ("errors", "skipped")},
    "apiTest": {"required": ("total", "passed", "failed"), "optional": ("blocked",)},
    "browserTest": {"required": ("total", "passed", "failed"), "optional": ("skipped", "retries")},
    "apiContract": {"required": ("scenariosTotal", "passed", "failed"), "optional": ("blocked",)},
    "browserE2E": {"required": ("total", "passed", "failed"), "optional": ("skipped", "retries")},
}


def validate_metrics(verification: str, metrics: Any) -> list[str]:
    """Typed metrics schema check; unknown verification types pass through."""
    problems: list[str] = []
    if not isinstance(metrics, dict):
        return ["metrics must be an object"]
    if verification == "dbCompatibility":
        applicability = metrics.get("applicability")
        if applicability not in ("APPLICABLE", "NOT_APPLICABLE"):
            return ["dbCompatibility.applicability must be APPLICABLE|NOT_APPLICABLE"]
        if applicability == "NOT_APPLICABLE":
            reason = metrics.get("reason")
            if not isinstance(reason, str) or not reason.strip():
                problems.append("dbCompatibility.reason required for NOT_APPLICABLE")
            unknown = sorted(set(metrics) - {"applicability", "reason"})
            problems.extend(
                f"dbCompatibility.{field} is not valid for NOT_APPLICABLE"
                for field in unknown
            )
        else:
            allowed = {
                "applicability", "status", "total", "passed", "failed", "evidenceHash"
            }
            status = metrics.get("status")
            if status not in {"OK", "FAIL"}:
                problems.append("dbCompatibility.status must be OK|FAIL for APPLICABLE")
            for field in ("total", "passed", "failed"):
                value = metrics.get(field)
                if not isinstance(value, int) or isinstance(value, bool) or value < 0:
                    problems.append(f"dbCompatibility.{field} must be a non-negative int")
            total = metrics.get("total")
            passed = metrics.get("passed")
            failed = metrics.get("failed")
            if all(isinstance(value, int) and not isinstance(value, bool)
                   for value in (total, passed, failed)) and passed + failed != total:
                problems.append("dbCompatibility counts must satisfy passed + failed == total")
            evidence_hash = metrics.get("evidenceHash")
            if not isinstance(evidence_hash, str) or not re.fullmatch(
                r"sha256:[0-9a-f]{64}", evidence_hash
            ):
                problems.append("dbCompatibility.evidenceHash must be sha256:<64 lowercase hex>")
            problems.extend(
                f"dbCompatibility.{field} is not a supported field"
                for field in sorted(set(metrics) - allowed)
            )
        return problems
    schema = _METRICS_SCHEMAS.get(verification)
    if schema is None:
        return problems
    allowed = set(schema["required"]) | set(schema["optional"])
    for field in schema["required"]:
        value = metrics.get(field)
        if not isinstance(value, int) or isinstance(value, bool) or value < 0:
            problems.append(f"metrics.{field} must be a non-negative int")
    for field, value in metrics.items():
        if field not in allowed:
            problems.append(f"metrics.{field} is not a {verification} field")
        elif not isinstance(value, int) or isinstance(value, bool) or value < 0:
            problems.append(f"metrics.{field} must be a non-negative int")
    return problems


def build_applicability_entry(value: str, reason: str | None = None) -> dict[str, Any]:
    if value not in ("APPLICABLE", "NOT_APPLICABLE"):
        raise ValueError("applicability must be APPLICABLE|NOT_APPLICABLE")
    if value == "NOT_APPLICABLE" and not (isinstance(reason, str) and reason.strip()):
        raise ValueError("NOT_APPLICABLE requires a scope reason")
    entry: dict[str, Any] = {"applicability": value}
    if reason and reason.strip():
        entry["reason"] = reason.strip()
    return entry


def applicability_counts_as_success(entry: dict[str, Any]) -> bool:
    """Applicability never contributes to success counters (RET-24)."""
    return False


def applicability_counts_as_failure(entry: dict[str, Any]) -> bool:
    """NOT_APPLICABLE is not a failure; status decides for APPLICABLE."""
    return False


_DYNAMIC_OWN_DIRS = ("events.ndjson", "logs", "evidence", "reports", "runtime", "backups")


def _matches_ownership_path(rel: str, declared: Any) -> bool:
    scope = str(declared).replace("\\", "/").strip("/")
    if scope.endswith("/**"):
        scope = scope[:-3].rstrip("/")
    if not scope:
        return False
    normalized = rel.replace("\\", "/").strip("/")
    return normalized == scope or normalized.startswith(scope + "/")


def _classify_ownership_path(
    rel: str, change_name: str, ownership: dict[str, Any]
) -> str:
    """owned | staticEvidence | excludedRuntime | foreign."""
    normalized = rel.replace("\\", "/")
    if normalized.startswith(".harness/state/changes/"):
        owner = normalized.split("/")[3] if len(normalized.split("/")) > 3 else ""
        return "excludedRuntime" if owner == change_name else "foreign"
    if normalized.startswith(".harness/state/"):
        return "excludedRuntime"
    if normalized.startswith(".harness/changes/"):
        parts = normalized.split("/")
        owner = parts[2] if len(parts) > 2 else ""
        if owner and owner != change_name:
            return "foreign"
        remainder = "/".join(parts[3:]) if len(parts) > 3 else ""
        head = remainder.split("/")[0] if remainder else ""
        if remainder in _DYNAMIC_OWN_DIRS or head in _DYNAMIC_OWN_DIRS:
            return "excludedRuntime"
    for excluded in ownership.get("excludedPaths") or []:
        if _matches_ownership_path(normalized, excluded):
            return "excludedRuntime"
    for static_path in ownership.get("staticEvidencePaths") or []:
        if _matches_ownership_path(normalized, static_path):
            return "staticEvidence"
    for product_path in ownership.get("productPaths") or []:
        if _matches_ownership_path(normalized, product_path):
            return "owned"
    return "foreign"


def compute_ownership_diff(
    repo_root: Path, *, base: str, change_dir: Path, head: str | None = None
) -> dict[str, Any]:
    """diffHash over the change's ownership scope only (RET-18).

    Excludes .harness/state/** and dynamic evidence; reports foreign change
    paths separately instead of folding them into the hash.
    """
    repo_root = Path(repo_root).resolve()
    change_dir = Path(change_dir).resolve()
    try:
        contract = harness_paths.load_change_contract(change_dir)
    except (OSError, ValueError):
        contract = {}
    ownership = contract.get("ownership") or {}
    diff_args = ["diff", "--name-only", base]
    if head:
        diff_args.append(head)
    raw = _git_text(repo_root, *diff_args) or ""
    changed_paths = {line.strip() for line in raw.splitlines() if line.strip()}
    if head is None:
        untracked = _git_text(
            repo_root, "ls-files", "--others", "--exclude-standard"
        ) or ""
        changed_paths.update(
            line.strip() for line in untracked.splitlines() if line.strip()
        )
    owned: list[str] = []
    static_evidence: list[str] = []
    foreign: list[str] = []
    excluded_runtime = 0
    for rel in sorted(changed_paths):
        verdict = _classify_ownership_path(rel, change_dir.name, ownership)
        if verdict == "foreign":
            foreign.append(rel)
        elif verdict == "staticEvidence":
            static_evidence.append(rel)
        elif verdict == "excludedRuntime":
            excluded_runtime += 1
        else:
            owned.append(rel)
    owned.sort()
    static_evidence.sort()
    foreign.sort()

    hasher = hashlib.sha256()
    for rel in owned:
        hasher.update(rel.encode("utf-8"))
        hasher.update(b"\x00")
        if head:
            try:
                content = _git_bytes(["show", f"{head}:{rel}"], repo_root)
            except RuntimeError:
                content = None
            hasher.update(
                hashlib.sha256(content).hexdigest().encode("ascii")
                if content is not None
                else b"<deleted>"
            )
        else:
            content_path = (repo_root / rel).resolve()
            if content_path.is_file() and _inside(content_path, repo_root):
                hasher.update(hashlib.sha256(content_path.read_bytes()).hexdigest().encode("ascii"))
            else:
                hasher.update(b"<deleted>")
        hasher.update(b"\x00")
    return {
        "diffHash": "sha256:" + hasher.hexdigest(),
        "files": owned,
        "staticEvidenceFiles": static_evidence,
        "foreignPaths": foreign,
        "excludedRuntimeCount": excluded_runtime,
        "ownedFileCount": len(owned),
        "ownershipHash": ownership_hash(contract),
    }


def normalize_status(raw: str) -> str:
    if raw not in STATUS_MAP:
        raise ValueError(
            f"unsupported status: {raw}; expected one of ok|fail|not_run (case variants OK/FAIL/NOT_RUN)"
        )
    return STATUS_MAP[raw]


def evidence_summary(entry: dict[str, Any]) -> dict[str, Any]:
    return {
        "status": entry.get("status"),
        "command": entry.get("command"),
        "evidence": entry.get("evidence"),
        "scope": entry.get("scope"),
        "inputsHash": entry.get("inputsHash"),
        "inputsFiles": entry.get("inputsFiles"),
        "durationMs": entry.get("durationMs"),
        "exitCode": entry.get("exitCode"),
        "finishedAt": entry.get("finishedAt"),
    }


def _is_relative_to(path: Path, base: Path) -> bool:
    """Python 3.9 兼容的 is_relative_to。"""
    try:
        path.relative_to(base)
        return True
    except ValueError:
        return False


def _nonempty_str(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def _scenario_receipt_error(code: str, message: str) -> dict[str, Any]:
    return {"ok": False, "code": code, "error": message}


def resolve_scenario_manifest(manifest: Any) -> dict[str, Any]:
    """把 v2 artifact 包装体解成 legacy 形状；legacy 输入原样返回。

    解包规则住在 harness_plan_finalize（它定义了 legacy manifest 的 schema），
    门禁与本模块共用同一份，两边不会各推一套。

    此前这里只判包装体、一律 fail-closed；更早的时候连判都不判——cmd_record
    的版本探测初值是 0，包装体探测不出 schemaVersion 就停在 0，
    `manifest_schema >= 2` 判假，--scenario-receipt-file 的强制要求被静默跳过。

    返回 ``{"ok": True, "manifest": ...}`` 或 ``{"ok": False, "code": ...}``。
    """
    unpacked = hpf.unpack_v2_scenario_manifest(manifest)
    if unpacked is None:
        return {"ok": True, "manifest": manifest}
    return unpacked


def _resolve_receipt_path(
    raw: str, change_dir: Path
) -> tuple[Path | None, list[Path]]:
    """Resolve --scenario-receipt-file against CWD *and* the change dir.

    Skills write receipts under ``<change-dir>/runtime/``, so a bare
    ``runtime/scenario-receipt-*.json`` is the natural thing to pass. Resolving
    only against the CWD made that fail with a bare ENOENT. Return the first
    existing candidate plus every path tried, so the error can name them.
    """
    candidate = Path(raw).expanduser()
    if candidate.is_absolute():
        resolved = candidate.resolve()
        return (resolved if resolved.is_file() else None), [resolved]
    candidates = [
        (Path.cwd() / candidate).resolve(),
        (change_dir / candidate).resolve(),
    ]
    # De-duplicate while preserving order (CWD == change_dir is common).
    unique: list[Path] = []
    for item in candidates:
        if item not in unique:
            unique.append(item)
    for item in unique:
        if item.is_file():
            return item, unique
    return None, unique


def _canonical_string_list(
    receipt: dict[str, Any],
    field: str,
) -> tuple[list[str] | None, dict[str, Any] | None]:
    raw = receipt.get(field)
    if not isinstance(raw, list) or any(not _nonempty_str(item) for item in raw):
        return None, _scenario_receipt_error(
            "SCENARIO_RECEIPT_INVALID",
            f"scenario receipt field '{field}' must be an array of non-empty strings",
        )
    values = [str(item).strip() for item in raw]
    if len(values) != len(set(values)):
        return None, _scenario_receipt_error(
            "SCENARIO_RECEIPT_INVALID",
            f"scenario receipt field '{field}' contains duplicate test IDs",
        )
    return values, None


def validate_scenario_execution_receipt(
    *,
    change_dir: Path,
    scenario_ids: list[str],
    receipt: dict[str, Any],
) -> dict[str, Any]:
    """Validate and minimize runner output before it can satisfy scenario evidence."""
    manifest_path = change_dir / "meta" / "scenario-manifest.json"
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
    except FileNotFoundError:
        return _scenario_receipt_error(
            "SCENARIO_MANIFEST_MISSING",
            f"scenario manifest does not exist: {manifest_path}",
        )
    except (OSError, json.JSONDecodeError) as exc:
        return _scenario_receipt_error("SCENARIO_MANIFEST_INVALID", str(exc))
    resolved = resolve_scenario_manifest(manifest)
    if not resolved["ok"]:
        return _scenario_receipt_error(
            str(resolved["code"]), str(resolved.get("message") or "")
        )
    manifest = resolved["manifest"]
    if not isinstance(manifest, dict) or not isinstance(manifest.get("scenarios"), list):
        return _scenario_receipt_error(
            "SCENARIO_MANIFEST_INVALID",
            "scenario-manifest.json must contain a scenarios array",
        )
    if int(manifest.get("schemaVersion") or 0) < 2:
        return _scenario_receipt_error(
            "SCENARIO_RECEIPT_UNSUPPORTED",
            "structured scenario receipts require scenario-manifest schemaVersion 2",
        )
    if receipt.get("schemaVersion") != 1:
        return _scenario_receipt_error(
            "SCENARIO_RECEIPT_INVALID",
            "scenario receipt schemaVersion must be 1",
        )
    runner = receipt.get("runner")
    if not isinstance(runner, dict) or not _nonempty_str(runner.get("name")):
        return _scenario_receipt_error(
            "SCENARIO_RECEIPT_INVALID",
            "scenario receipt runner.name is required",
        )
    if "version" in runner and not _nonempty_str(runner.get("version")):
        return _scenario_receipt_error(
            "SCENARIO_RECEIPT_INVALID",
            "scenario receipt runner.version must be a non-empty string when present",
        )
    attempt = receipt.get("attempt")
    if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
        return _scenario_receipt_error(
            "SCENARIO_RECEIPT_INVALID",
            "scenario receipt attempt must be a positive integer",
        )

    declared, error = _canonical_string_list(receipt, "declared")
    if error:
        return error
    selected, error = _canonical_string_list(receipt, "selected")
    if error:
        return error
    assert declared is not None and selected is not None

    collected_raw = receipt.get("collected")
    executed_raw = receipt.get("executed")
    if not isinstance(collected_raw, list) or not isinstance(executed_raw, list):
        return _scenario_receipt_error(
            "SCENARIO_RECEIPT_INVALID",
            "scenario receipt collected and executed fields must be arrays",
        )

    collected: list[dict[str, str]] = []
    collected_keys: set[tuple[str, str, str]] = set()
    for index, item in enumerate(collected_raw):
        if not isinstance(item, dict) or any(
            not _nonempty_str(item.get(field))
            for field in ("testId", "file", "title")
        ):
            return _scenario_receipt_error(
                "SCENARIO_RECEIPT_INVALID",
                f"scenario receipt collected[{index}] has an invalid test identity",
            )
        canonical = {
            "testId": str(item["testId"]).strip(),
            "file": str(item["file"]).strip(),
            "title": str(item["title"]).strip(),
        }
        key = (canonical["testId"], canonical["file"], canonical["title"])
        if key in collected_keys:
            return _scenario_receipt_error(
                "SCENARIO_RECEIPT_INVALID",
                f"scenario receipt collected[{index}] duplicates a test identity",
            )
        collected_keys.add(key)
        collected.append(canonical)

    executed: list[dict[str, Any]] = []
    executed_keys: set[tuple[str, str, str, int]] = set()
    allowed_statuses = {"PASSED", "FAILED", "SKIPPED"}
    for index, item in enumerate(executed_raw):
        if not isinstance(item, dict) or any(
            not _nonempty_str(item.get(field))
            for field in ("testId", "file", "title", "status")
        ):
            return _scenario_receipt_error(
                "SCENARIO_RECEIPT_INVALID",
                f"scenario receipt executed[{index}] has an invalid test identity",
            )
        item_attempt = item.get("attempt")
        if (
            not isinstance(item_attempt, int)
            or isinstance(item_attempt, bool)
            or item_attempt != attempt
        ):
            return _scenario_receipt_error(
                "SCENARIO_RECEIPT_INVALID",
                f"scenario receipt executed[{index}].attempt must equal receipt attempt",
            )
        item_status = str(item["status"]).strip().upper()
        if item_status not in allowed_statuses:
            return _scenario_receipt_error(
                "SCENARIO_RECEIPT_INVALID",
                f"scenario receipt executed[{index}].status is unsupported",
            )
        canonical = {
            "testId": str(item["testId"]).strip(),
            "file": str(item["file"]).strip(),
            "title": str(item["title"]).strip(),
            "attempt": item_attempt,
            "status": item_status,
        }
        key = (
            canonical["testId"],
            canonical["file"],
            canonical["title"],
            item_attempt,
        )
        if key in executed_keys:
            return _scenario_receipt_error(
                "SCENARIO_RECEIPT_INVALID",
                f"scenario receipt executed[{index}] duplicates a test attempt",
            )
        executed_keys.add(key)
        executed.append(canonical)

    manifest_scenarios = {
        str(item.get("id") or "").strip(): item
        for item in manifest["scenarios"]
        if isinstance(item, dict) and _nonempty_str(item.get("id"))
    }
    unknown = sorted(set(scenario_ids) - set(manifest_scenarios))
    if unknown:
        return _scenario_receipt_error(
            "SCENARIO_ID_UNKNOWN",
            "scenario IDs are not declared in scenario-manifest.json: "
            + ", ".join(unknown),
        )

    coverage = {
        key: []
        for key in (
            "declared",
            "selected",
            "collected",
            "executed",
            "passed",
            "skipped",
            "failed",
            "unexecuted",
        )
    }
    for scenario_id in scenario_ids:
        scenario = manifest_scenarios[scenario_id]
        if scenario.get("requiredEvidenceKind") != "ledger":
            continue
        if any(
            not _nonempty_str(scenario.get(field))
            for field in ("executableTestId", "testFile", "testTitle")
        ):
            return _scenario_receipt_error(
                "SCENARIO_MANIFEST_INVALID",
                f"scenario {scenario_id} is missing its executable test identity",
            )
        test_id = str(scenario["executableTestId"]).strip()
        identity = (
            test_id,
            str(scenario["testFile"]).strip(),
            str(scenario["testTitle"]).strip(),
        )
        if test_id in declared:
            coverage["declared"].append(scenario_id)
        if test_id in selected:
            coverage["selected"].append(scenario_id)
        if identity in collected_keys:
            coverage["collected"].append(scenario_id)
        selection_closed = (
            test_id in declared
            and test_id in selected
            and identity in collected_keys
        )
        matches = [
            item
            for item in executed
            if (
                item["testId"],
                item["file"],
                item["title"],
            )
            == identity
        ]
        if matches:
            coverage["executed"].append(scenario_id)
            terminal = matches[-1]["status"]
            if terminal == "PASSED" and selection_closed:
                coverage["passed"].append(scenario_id)
            elif terminal == "SKIPPED":
                coverage["skipped"].append(scenario_id)
            elif terminal == "FAILED":
                coverage["failed"].append(scenario_id)
            if not selection_closed:
                coverage["unexecuted"].append(scenario_id)
        else:
            coverage["unexecuted"].append(scenario_id)

    canonical_runner = {"name": str(runner["name"]).strip()}
    if _nonempty_str(runner.get("version")):
        canonical_runner["version"] = str(runner["version"]).strip()
    return {
        "ok": True,
        "receipt": {
            "schemaVersion": 1,
            "runner": canonical_runner,
            "attempt": attempt,
            "declared": declared,
            "selected": selected,
            "collected": collected,
            "executed": executed,
        },
        "coverage": coverage,
    }


def _scope_covers(ledger_scope: Any, requested_scope: str | None) -> bool:
    """unitTest: ledger scope must cover requested scope (broad scopes cover all)."""
    if not _nonempty_str(ledger_scope) and not isinstance(ledger_scope, list):
        return False
    if requested_scope is None or not str(requested_scope).strip():
        # No requested scope → only require ledger to have some scope recorded.
        return True

    req = str(requested_scope).strip()
    if isinstance(ledger_scope, list):
        ledger_items = {str(x).strip() for x in ledger_scope if str(x).strip()}
    else:
        text = str(ledger_scope).strip()
        if text in BROAD_SCOPES:
            return True
        ledger_items = {p.strip() for p in text.split(",") if p.strip()}

    if req in BROAD_SCOPES:
        # Requesting broad scope only reusable if ledger also broad (same or broader).
        return str(ledger_scope).strip() in BROAD_SCOPES if not isinstance(ledger_scope, list) else False

    req_items = {p.strip() for p in req.split(",") if p.strip()}
    return req_items.issubset(ledger_items)


def worktree_ready(ledger: dict[str, Any], change_dir: Path) -> bool:
    root = ledger.get("worktreeRoot")
    if root is not None and _nonempty_str(root):
        return True
    # Metadata aliases are accepted only when they resolve to a real linked
    # worktree of this repository. This keeps install reuse aligned with the
    # execution root used by context and gate, including `worktreePath`.
    return infer_execution_project_root(change_dir) is not None


def decide_can_reuse(
    *,
    change_dir: Path,
    verification: str,
    files: list[str],
    requested_scope: str | None = None,
    requested_command: str | None = None,
    requested_toolchain_hash: str | None = None,
    requested_profile_hash: str | None = None,
    requested_environment_hash: str | None = None,
    requested_db_schema_hash: str | None = None,
    requested_product_tree_hash: str | None = None,
    requested_command_set_hash: str | None = None,
    requested_lock_hash: str | None = None,
    requested_target_identity: str | None = None,
    required_coverage: str | None = None,
    project_root: Path | str | None = None,
) -> dict[str, Any]:
    ledger, ledger_path = load_ledger(change_dir)
    if ledger is None:
        return {
            "ok": True,
            "reuse": False,
            "reason": "insufficient-evidence",
            "code": "LEDGER_MISSING",
            "executionNeed": "first-run",
            "verification": verification,
            "detail": "ledger missing",
        }

    validations = ledger.get("validations")
    targets_raw = ledger.get("verificationTargets")
    targets = (
        list(targets_raw.values())
        if isinstance(targets_raw, dict)
        else targets_raw
        if isinstance(targets_raw, list)
        else []
    )
    target_entries = [
        target
        for target in targets
        if isinstance(target, dict) and target.get("verification") == verification
    ]
    target_entries.sort(key=lambda item: str(item.get("finishedAt") or ""))
    # A ledger can contain both a feature-tip and a later merge target. Select
    # the newest candidate matching the requested canonical key, rather than
    # letting an unrelated newest commit mask reusable evidence.
    requested_key_values = {
        "productTreeHash": requested_product_tree_hash,
        "commandSetHash": requested_command_set_hash,
        "toolchainHash": requested_toolchain_hash,
        "profileHash": requested_profile_hash,
        "environmentHash": requested_environment_hash,
        "lockHash": requested_lock_hash,
        "dbSchemaHash": requested_db_schema_hash,
    }
    matching_entries = [
        candidate
        for candidate in target_entries
        if all(
            not _nonempty_str(requested)
            or str(candidate.get(field) or "").strip().removeprefix("sha256:")
            == str(requested).strip().removeprefix("sha256:")
            for field, requested in requested_key_values.items()
        )
        and (
            not _nonempty_str(requested_command)
            or canonical_command(str(candidate.get("command") or ""))
            == canonical_command(str(requested_command))
        )
    ]
    entry = (
        matching_entries[-1]
        if matching_entries
        else target_entries[-1]
        if target_entries
        else None
    )
    if not isinstance(validations, dict) and entry is None:
        return {
            "ok": True,
            "reuse": False,
            "reason": "insufficient-evidence",
            "code": "VALIDATIONS_MISSING",
            "executionNeed": "first-run",
            "verification": verification,
            "detail": "validations missing",
            "ledger_path": str(ledger_path) if ledger_path else None,
        }

    if entry is None and isinstance(validations, dict):
        entry = validations.get(verification)
    if not isinstance(entry, dict):
        return {
            "ok": True,
            "reuse": False,
            "reason": "insufficient-evidence",
            "code": "VALIDATION_MISSING",
            "executionNeed": "first-run",
            "verification": verification,
            "detail": f"validation '{verification}' missing",
            "ledger_path": str(ledger_path) if ledger_path else None,
        }

    stored_hash = entry.get("inputsHash")
    stored_files = entry.get("inputsFiles")
    status = entry.get("status")
    evidence = entry.get("evidence")
    command = entry.get("command")
    scope = entry.get("scope")
    algorithm_version = entry.get("algorithmVersion")
    coverage = entry.get("coverage")
    stored_target_identity = entry.get("targetIdentity")

    if entry.get("reusable") is False or isinstance(entry.get("invalidation"), dict):
        return {
            "ok": True,
            "reuse": False,
            "reason": "rerun",
            "code": "EVIDENCE_INVALIDATED",
            "executionNeed": "rerun",
            "verification": verification,
            "detail": "recorded target was invalidated by workflow fixback",
            "invalidation": entry.get("invalidation"),
        }

    # v2 fields: a v1 entry (no algorithmVersion/coverage) is conservatively
    # invalidated once and must be re-recorded with v2 fields (COM-002). No
    # silent upgrade of stale evidence.
    v2_missing: list[str] = []
    if not _nonempty_str(algorithm_version):
        v2_missing.append("algorithmVersion")
    if not (_nonempty_str(coverage) and str(coverage).strip() in COVERAGE_RANK):
        v2_missing.append("coverage")

    missing: list[str] = []
    if not _nonempty_str(stored_hash):
        missing.append("inputsHash")
    if not isinstance(stored_files, list):
        missing.append("inputsFiles")
    elif verification == "unitTestFull" and not stored_files:
        # 全量门禁的依赖闭包文件集必须非空，禁止空/staged-only 闭包冒充全量。
        missing.append("inputsFiles")
    if status != "OK":
        missing.append("status=OK")
    if not _nonempty_str(evidence):
        missing.append("evidence")
    if not _nonempty_str(command):
        missing.append("command")
    if verification == "unitTest" and not (
        _nonempty_str(scope) or isinstance(scope, list)
    ):
        missing.append("scope")
    if verification == "unitTestFull":
        # 独立 full-scope 检查：增量范围（如 FooTest）不能冒充全量门禁。
        # 不并入 _scope_covers()，避免依赖增量复用的隐含行为。
        if not isinstance(scope, str) or scope.strip() not in BROAD_SCOPES:
            missing.append("scope=module|full")
    if verification == "install" and not worktree_ready(ledger, change_dir):
        missing.append("worktree")

    all_missing = v2_missing + missing
    if all_missing:
        return {
            "ok": True,
            "reuse": False,
            "reason": "insufficient-evidence",
            "code": "MISSING_V2_FIELDS" if v2_missing else "MISSING_FIELDS",
            "executionNeed": "evidence-incomplete",
            "verification": verification,
            "detail": "missing or invalid: " + ", ".join(all_missing),
            "ledger_path": str(ledger_path) if ledger_path else None,
        }

    # Coverage lattice: recorded coverage rank must meet the verification's
    # required rank. Stops incremental evidence satisfying a module/full gate
    # (UT-015 / API-005).
    required = (
        COVERAGE_RANK.get(required_coverage, 1)
        if required_coverage is not None
        else REQUIRED_COVERAGE.get(verification, 1)
    )
    if COVERAGE_RANK.get(str(coverage).strip(), -1) < required:
        return {
            "ok": True,
            "reuse": False,
            "reason": "insufficient-evidence",
            "code": "COVERAGE_INSUFFICIENT",
            "verification": verification,
            "detail": f"coverage '{coverage}' below required rank {required} for {verification}",
            "ledger_path": str(ledger_path) if ledger_path else None,
            "stored_coverage": coverage,
        }

    if _nonempty_str(requested_target_identity):
        if not _nonempty_str(stored_target_identity):
            return {
                "ok": True,
                "reuse": False,
                "reason": "insufficient-evidence",
                "code": "MISSING_TARGET_IDENTITY",
                "verification": verification,
                "detail": "recorded evidence predates the current verification target identity",
                "ledger_path": str(ledger_path) if ledger_path else None,
                "requested_target_identity": requested_target_identity,
            }
        if str(stored_target_identity).strip() != str(requested_target_identity).strip():
            return {
                "ok": True,
                "reuse": False,
                "reason": "rerun",
                "code": "TARGET_IDENTITY_CHANGED",
                "verification": verification,
                "detail": "verification target definition changed",
                "ledger_path": str(ledger_path) if ledger_path else None,
                "stored_target_identity": stored_target_identity,
                "requested_target_identity": requested_target_identity,
            }

    if requested_command is not None and str(requested_command).strip():
        if canonical_command(str(command)) != canonical_command(str(requested_command)):
            return {
                "ok": True,
                "reuse": False,
                "reason": "rerun",
                "code": "COMMAND_CHANGED",
                "verification": verification,
                "detail": "command changed",
                "ledger_path": str(ledger_path) if ledger_path else None,
                "stored_command": command,
                "requested_command": requested_command,
            }

    # The reusable identity is content and execution context, never commit SHA.
    # productCommit remains provenance only so a feature tip can serve its no-ff
    # merge when the merge has exactly the same product tree.
    # Only compared when both the stored entry and the request carry the field.
    for field, requested, code_name in (
        ("productTreeHash", requested_product_tree_hash, "PRODUCT_TREE_CHANGED"),
        ("commandSetHash", requested_command_set_hash, "COMMAND_SET_CHANGED"),
        ("toolchainHash", requested_toolchain_hash, "TOOLCHAIN_CHANGED"),
        ("profileHash", requested_profile_hash, "PROFILE_CHANGED"),
        ("environmentHash", requested_environment_hash, "ENVIRONMENT_CHANGED"),
        ("lockHash", requested_lock_hash, "LOCK_CHANGED"),
        ("dbSchemaHash", requested_db_schema_hash, "DB_SCHEMA_CHANGED"),
    ):
        stored = entry.get(field)
        if requested and str(requested).strip() and not _nonempty_str(stored):
            return {
                "ok": True,
                "reuse": False,
                "reason": "insufficient-evidence",
                "code": "MISSING_TARGET_IDENTITY",
                "verification": verification,
                "detail": f"{field} missing from recorded target",
                "field": field,
            }
        if requested and str(requested).strip() and str(stored).strip() != str(requested).strip():
            return {
                "ok": True,
                "reuse": False,
                "reason": "rerun",
                "code": code_name,
                "verification": verification,
                "detail": f"{field} changed",
                "ledger_path": str(ledger_path) if ledger_path else None,
                "field": field,
                "stored": stored,
                "requested": requested,
            }

    if verification == "unitTest" and not _scope_covers(scope, requested_scope):
        return {
            "ok": True,
            "reuse": False,
            "reason": "insufficient-evidence",
            "code": "SCOPE_INSUFFICIENT",
            "verification": verification,
            "detail": "scope does not cover requested tests",
            "ledger_path": str(ledger_path) if ledger_path else None,
            "stored_scope": scope,
            "requested_scope": requested_scope,
        }

    try:
        current_hash, current_files = compute_inputs_hash(
            files, project_root=project_root
        )
    except (FileNotFoundError, ValueError) as exc:
        return {
            "ok": True,
            "reuse": False,
            "reason": "insufficient-evidence",
            "code": "INPUT_FILE_MISSING",
            "verification": verification,
            "detail": str(exc),
            "ledger_path": str(ledger_path) if ledger_path else None,
        }

    if current_hash != stored_hash:
        return {
            "ok": True,
            "reuse": False,
            "reason": "rerun",
            "code": "INPUTS_HASH_CHANGED",
            "verification": verification,
            "detail": "inputsHash changed",
            "ledger_path": str(ledger_path) if ledger_path else None,
            "stored_inputsHash": stored_hash,
            "current_inputsHash": current_hash,
            "inputsFiles": current_files,
        }

    return {
        "ok": True,
        "reuse": True,
        "reason": "reuse",
        "code": "REUSED",
        "executionNeed": "reuse",
        "invalidationCode": None,
        "verification": verification,
        "ledger_path": str(ledger_path) if ledger_path else None,
        "inputsHash": stored_hash,
        "inputsFiles": stored_files,
        "evidence_summary": evidence_summary(entry),
        "reusedFrom": {
            "targetId": entry.get("id"),
            "sourceCommit": entry.get("productCommit") or entry.get("currentHead"),
            "sourceCandidate": entry.get("candidateId"),
        },
        "reuseReason": "canonical reuse key matched; commit SHA is provenance only",
        "reuseKey": canonical_reuse_key(entry),
        "marker": "REUSED",
    }


_INVALIDATION_ALIASES = {
    "INPUTS_HASH_CHANGED": "INPUT_CHANGED",
    "ENVIRONMENT_CHANGED": "ENV_CHANGED",
    "MISSING_FIELDS": "MISSING_IDENTITY_FIELD",
    "MISSING_V2_FIELDS": "MISSING_IDENTITY_FIELD",
    "LEDGER_MISSING": "MISSING_IDENTITY_FIELD",
    "VALIDATIONS_MISSING": "MISSING_IDENTITY_FIELD",
    "VALIDATION_MISSING": "MISSING_IDENTITY_FIELD",
}


def _attach_invalidation_code(payload: dict[str, Any]) -> dict[str, Any]:
    """Wave-2 H-9: expose stable invalidationCode alongside legacy code."""
    if payload.get("reuse") is True:
        payload.setdefault("invalidationCode", None)
        return payload
    code = str(payload.get("code") or "").strip()
    if code:
        payload["invalidationCode"] = _INVALIDATION_ALIASES.get(code, code)
    return payload


def verification_graph_path(change_dir: Path) -> Path:
    return Path(change_dir) / "evidence" / "verification-graph.json"


def upsert_verification_graph_node(
    change_dir: Path,
    *,
    verification: str,
    status: str,
    inputs_hash: str,
    command: str | None = None,
    toolchain_hash: str | None = None,
    environment_hash: str | None = None,
    db_schema_hash: str | None = None,
    diff_hash: str | None = None,
    target_identity: str | None = None,
) -> dict[str, Any]:
    """Persist a verification graph node keyed by canonical identity (H-9)."""
    path = verification_graph_path(change_dir)
    path.parent.mkdir(parents=True, exist_ok=True)
    graph: dict[str, Any]
    if path.is_file():
        try:
            loaded = json.loads(path.read_text(encoding="utf-8-sig"))
            graph = loaded if isinstance(loaded, dict) else {"schemaVersion": 1, "nodes": []}
        except (OSError, json.JSONDecodeError):
            graph = {"schemaVersion": 1, "nodes": []}
    else:
        graph = {"schemaVersion": 1, "nodes": []}
    nodes = graph.get("nodes")
    if not isinstance(nodes, list):
        nodes = []
    identity = {
        "verification": verification,
        "inputsHash": inputs_hash,
        "command": command,
        "toolchainHash": toolchain_hash,
        "environmentHash": environment_hash,
        "dbSchemaHash": db_schema_hash,
        "diffHash": diff_hash,
        "targetIdentity": target_identity,
    }
    node = {
        "identity": identity,
        "status": status,
        "recordedAt": dt.datetime.now().astimezone().isoformat(timespec="milliseconds"),
    }
    replaced = False
    for index, existing in enumerate(nodes):
        if not isinstance(existing, dict):
            continue
        existing_id = existing.get("identity")
        if isinstance(existing_id, dict) and existing_id == identity:
            nodes[index] = node
            replaced = True
            break
    if not replaced:
        nodes.append(node)
    graph["schemaVersion"] = 1
    graph["nodes"] = nodes
    write_ledger(path, graph)  # atomic json write helper already used for ledger
    return {"ok": True, "path": str(path), "nodeCount": len(nodes), "replaced": replaced}


def cmd_hash(args: argparse.Namespace) -> int:
    as_json = bool(args.json)
    try:
        files, project_root = input_files_from_args(args)
        if not files:
            return emit_error("hash requires --files or --files-from", as_json=as_json)
        inputs_hash, inputs_files = compute_inputs_hash(
            files, project_root=project_root
        )
    except (OSError, FileNotFoundError) as exc:
        return emit_error(str(exc), as_json=as_json)

    payload = {
        "ok": True,
        "action": "hash",
        "inputsHash": inputs_hash,
        "inputsFiles": inputs_files,
        "fileCount": len(inputs_files),
        "resolvedProjectRoot": str(project_root) if project_root else None,
    }
    emit_json(payload, as_json=as_json)
    return 0


def cmd_can_reuse(args: argparse.Namespace) -> int:
    as_json = bool(args.json)
    verbose = bool(getattr(args, "verbose", False))
    verification = args.verification
    change_dir = resolve_path(args.change_dir)
    apply_inferred_project_root(args, change_dir)
    ownership_check = frozen_ownership_check(change_dir)
    if not ownership_check.get("ok"):
        return emit_error(
            str(ownership_check.get("message")),
            as_json=as_json,
            error_code=str(ownership_check.get("code")),
            extra={
                key: value
                for key, value in ownership_check.items()
                if key not in {"ok", "code", "message"}
            },
        )
    try:
        files, project_root = input_files_from_args(args)
    except ProjectRootRequiredError as exc:
        return emit_error(
            str(exc),
            as_json=as_json,
            error_code="PROJECT_ROOT_REQUIRED",
        )
    except (OSError, ValueError) as exc:
        return emit_error(f"can-reuse failed: {exc}", as_json=as_json)
    target = resolve_verification_target(verification, project_root)
    if target is None:
        return emit_error(
            "unsupported verification: "
            f"{verification}; declare it in build-profile.json "
            "verificationGraph.targets",
            as_json=as_json,
        )
    explicit_files = list(files)
    profile_input = getattr(args, "profile_input", None)
    project_raw = getattr(args, "project", None)
    if profile_input:
        if not project_raw:
            return emit_error("--profile-input requires --project", as_json=as_json)
        resolved_files, err = expand_profile_input_files(project_root, profile_input)
        if err:
            # profile 未正确配置：不允许缓存复用，返回 insufficient-evidence（exit 0）。
            payload = {
                "ok": True,
                "reuse": False,
                "reason": "insufficient-evidence",
                "verification": verification,
                "detail": err,
            }
            emit_compact_or_verbose(
                payload, as_json=as_json, verbose=verbose,
                compact_fn=_compact_can_reuse_payload,
            )
            return 0
        profile_files = resolve_input_files(resolved_files, project_root)
        if explicit_files:
            _explicit_hash, explicit_names = compute_inputs_hash(
                explicit_files, project_root=project_root
            )
            _profile_hash, profile_names = compute_inputs_hash(
                profile_files, project_root=project_root
            )
            if explicit_names != profile_names:
                return emit_error(
                    "PROFILE_INPUT_FILES_CONFLICT: --files does not match the "
                    "declared build-profile input set",
                    as_json=as_json,
                    error_code="PROFILE_INPUT_FILES_CONFLICT",
                )
        files = profile_files
    if not files:
        return emit_error(
            "can-reuse requires --files or a non-empty --profile-input file set",
            as_json=as_json,
        )
    try:
        payload = decide_can_reuse(
            change_dir=change_dir,
            verification=verification,
            files=files,
            requested_scope=getattr(args, "scope", None),
            requested_command=(
                canonical_command(str(getattr(args, "command", "")))
                if _nonempty_str(getattr(args, "command", None))
                else None
            ),
            requested_toolchain_hash=(
                getattr(args, "toolchain_hash", None)
                or default_toolchain_hash(project_root)
            ),
            requested_profile_hash=getattr(args, "profile_hash", None),
            requested_environment_hash=(
                getattr(args, "environment_hash", None)
                or default_environment_hash(project_root)
            ),
            requested_db_schema_hash=getattr(args, "db_schema_hash", None),
            requested_product_tree_hash=(
                getattr(args, "product_tree_hash", None) or product_tree_hash(project_root)
            ),
            requested_command_set_hash=(
                getattr(args, "command_set_hash", None)
                or (
                    command_set_hash(str(getattr(args, "command", "")))
                    if _nonempty_str(getattr(args, "command", None))
                    else None
                )
            ),
            requested_lock_hash=(
                getattr(args, "lock_hash", None) or lock_hash(project_root)
            ),
            requested_target_identity=verification_target_identity(
                verification, target
            ),
            required_coverage=str(target["requiredCoverage"]),
            project_root=project_root,
        )
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        return emit_error(f"can-reuse failed: {exc}", as_json=as_json)

    payload = _attach_invalidation_code(payload)
    payload["resolvedProjectRoot"] = (
        str(project_root) if project_root is not None else None
    )
    emit_compact_or_verbose(
        payload, as_json=as_json, verbose=verbose,
        compact_fn=_compact_can_reuse_payload,
    )
    return 0


def _profile_input_missing_warning(
    verification: str, target: Any, profile_input: str | None
) -> str | None:
    """S-4：profile 声明了 target 却没带 --profile-input 时提前预警。"""
    if profile_input or not isinstance(target, dict):
        return None
    return (
        "PROFILE_INPUT_MISSING: 未传 --profile-input，本条记录缺少 profile 推导的"
        " scope/coverage/inputsHash 身份字段，can-reuse 将无法复用。建议重跑："
        f"record --verification {verification} --profile-input <key> --project . "
        "（key 见 build-profile.json verificationInputs）"
    )


def cmd_record(args: argparse.Namespace) -> int:
    as_json = bool(args.json)
    verbose = bool(getattr(args, "verbose", False))
    change_dir = resolve_path(args.change_dir)
    apply_inferred_project_root(args, change_dir)
    ownership_check = frozen_ownership_check(change_dir)
    if not ownership_check.get("ok"):
        return emit_error(
            str(ownership_check.get("message")),
            as_json=as_json,
            error_code=str(ownership_check.get("code")),
            extra={
                key: value
                for key, value in ownership_check.items()
                if key not in {"ok", "code", "message"}
            },
        )
    verification = args.verification
    contract_v2 = _contract_is_v2(change_dir)
    migration_receipt: dict[str, Any] | None = None
    try:
        files, project_root = input_files_from_args(args)
    except (OSError, ValueError) as exc:
        return emit_error(f"record failed: {exc}", as_json=as_json)
    explicit_files = list(files)
    profile_input = getattr(args, "profile_input", None)
    project_raw = getattr(args, "project", None)
    if profile_input:
        if not project_raw:
            return emit_error("--profile-input requires --project", as_json=as_json)
        _profile, profile_error = ensure_profile_input_target(
            project_root, profile_input
        )
        if profile_error:
            return emit_error(
                f"record failed: {profile_error}",
                as_json=as_json,
                error_code="BUILD_PROFILE_REFRESH_FAILED",
            )
    target = resolve_verification_target(verification, project_root)
    if target is None:
        return emit_error(
            "unsupported verification: "
            f"{verification}; declare it in build-profile.json "
            "verificationGraph.targets",
            as_json=as_json,
        )
    # S-4：profile 声明了 target 却没带 --profile-input，条目缺身份字段，
    # can-reuse 事后才报 MISSING_FIELDS——只能重跑一遍（2026-08-31 实测）。
    profile_input_missing_warn = _profile_input_missing_warning(
        verification, target, profile_input
    )
    if profile_input:
        resolved_files, err = expand_profile_input_files(project_root, profile_input)
        if err:
            return emit_error(f"record failed: {err}", as_json=as_json)
        profile_files = resolve_input_files(resolved_files, project_root)
        if explicit_files:
            _explicit_hash, explicit_names = compute_inputs_hash(
                explicit_files, project_root=project_root
            )
            _profile_hash, profile_names = compute_inputs_hash(
                profile_files, project_root=project_root
            )
            if explicit_names != profile_names:
                return emit_error(
                    "PROFILE_INPUT_FILES_CONFLICT: --files does not match the "
                    "declared build-profile input set",
                    as_json=as_json,
                    error_code="PROFILE_INPUT_FILES_CONFLICT",
                )
        files = profile_files
    # NOT_APPLICABLE 的 validation 没有输入文件可言。门禁 close 又要求
    # requiredValidations 每项都有 entry——两条规则叠加会逼调用方拿无关文件凑数，
    # ledger 从此声称该 validation 的输入是那些文件，是假证据。
    not_applicable = str(getattr(args, "applicability", "") or "") == "NOT_APPLICABLE"
    if not files and not not_applicable:
        return emit_error(
            "record requires --files or a non-empty --profile-input file set; "
            "unless --applicability NOT_APPLICABLE (then --applicability-reason is required)",
            as_json=as_json,
        )
    if not_applicable and not _nonempty_str(getattr(args, "applicability_reason", None)):
        return emit_error(
            "NOT_APPLICABLE requires --applicability-reason",
            as_json=as_json,
        )

    try:
        status = normalize_status(args.status)
        inputs_hash, inputs_files = compute_inputs_hash(
            files, project_root=project_root
        )
        ledger, existing_path = load_ledger(change_dir)
        if ledger is None:
            ledger = {
                "changeName": change_dir.name,
                "stateDir": str(change_dir),
                "validations": {},
            }
        elif not isinstance(ledger.get("validations"), dict):
            ledger["validations"] = {}
        explicit_identity = any(
            _nonempty_str(getattr(args, field, None))
            for field in ("base_commit", "diff_hash")
        )
        existing_schema = ledger.get("schemaVersion")
        identity_write = (
            contract_v2
            or explicit_identity
            or existing_schema in {2, LEDGER_SCHEMA_VERSION}
        )
        if identity_write:
            try:
                ledger, migration_receipt = migrate_ledger_for_write(
                    change_dir,
                    ledger,
                    project_root=project_root,
                    base_commit=getattr(args, "base_commit", None),
                    diff_hash=getattr(args, "diff_hash", None),
                    record_migration=existing_path is not None,
                )
            except LedgerMigrationError as exc:
                return emit_error(
                    f"ledger migration failed: {exc}",
                    as_json=as_json,
                    error_code="LEDGER_MIGRATION_REQUIRED",
                    extra={
                        "originalSchemaVersion": ledger.get("schemaVersion", 1),
                        "targetSchemaVersion": LEDGER_SCHEMA_VERSION,
                        "rerecordCommand": _deterministic_rerecord_command(
                            args,
                            files=files,
                            project_root=project_root,
                        ),
                    },
                )

        # Preserve top-level diffHash and all other existing fields (backward compatible).
        entry = {}
        prev = ledger["validations"].get(verification)
        if isinstance(prev, dict):
            entry.update(prev)

        entry.update(
            {
                "status": status,
                "command": canonical_command(args.command),
                "runnerCommand": (
                    str(getattr(args, "runner_command", None)).strip()
                    if _nonempty_str(getattr(args, "runner_command", None))
                    else str(args.command).strip()
                ),
                "evidence": args.evidence,
                "exitCode": args.exit_code,
                "durationMs": args.duration_ms,
                "inputsHash": inputs_hash,
                "inputsFiles": inputs_files,
                "finishedAt": now_iso(),
            }
        )
        # A fresh successful execution supersedes the prior target's fixback
        # invalidation. Carrying these projection-only fields forward would
        # make every rerecord permanently non-reusable.
        entry.pop("reusable", None)
        entry.pop("invalidation", None)
        metrics_raw = getattr(args, "metrics_json", None)
        metrics_file = getattr(args, "metrics_file", None)
        if metrics_file and metrics_raw is not None and str(metrics_raw).strip():
            return emit_error(
                "use only one of --metrics-json and --metrics-file",
                as_json=as_json,
            )
        if metrics_file:
            try:
                metrics_raw = Path(str(metrics_file)).expanduser().resolve().read_text(
                    encoding="utf-8-sig"
                )
            except OSError as exc:
                return emit_error(f"invalid --metrics-file: {exc}", as_json=as_json)
        if metrics_raw is not None and str(metrics_raw).strip() != "":
            try:
                metrics_obj = json.loads(str(metrics_raw))
            except json.JSONDecodeError:
                return emit_error("invalid --metrics-json", as_json=as_json)
            if not isinstance(metrics_obj, dict):
                return emit_error("invalid --metrics-json", as_json=as_json)
            if contract_v2:
                # Typed metrics schemas are ledger v3 (contract-gated); legacy
                # changes keep the loose run/failures shape (zero regression).
                problems = validate_metrics(verification, metrics_obj)
                if problems:
                    return emit_error(
                        "invalid metrics: " + "; ".join(problems), as_json=as_json
                    )
            entry["metrics"] = metrics_obj
        elif "metrics" in entry:
            # Fresh record without metrics must not keep stale metrics from prev.
            entry.pop("metrics", None)
        applicability_raw = getattr(args, "applicability", None)
        if applicability_raw:
            try:
                entry["applicability"] = build_applicability_entry(
                    str(applicability_raw).strip(),
                    reason=getattr(args, "applicability_reason", None),
                )
            except ValueError as exc:
                return emit_error(str(exc), as_json=as_json)
        effective_scope = (
            str(args.scope).strip()
            if args.scope is not None and str(args.scope).strip()
            else (
                str(
                    target.get("coverageLevel")
                    or target.get("requiredCoverage")
                    or ""
                ).strip()
                if profile_input
                else ""
            )
        )
        if effective_scope:
            entry["scope"] = effective_scope
        # No default scope: recording an incremental run as broad "module" scope
        # would let can-reuse wrongly approve untested classes (D13 guardrail).
        # Missing scope → can-reuse treats unitTest as insufficient-evidence.

        # v2 fields (cluster 2): algorithmVersion + coverage lattice + optional
        # toolchain/profile/environment hashes. v1 entries missing these are
        # conservatively invalidated by can-reuse (COM-002).
        entry["algorithmVersion"] = LEDGER_VERSION
        entry["targetIdentity"] = verification_target_identity(
            verification, target
        )
        coverage = getattr(args, "coverage", None)
        if not (_nonempty_str(coverage) and str(coverage).strip() in COVERAGE_RANK):
            coverage = derive_coverage(verification, effective_scope or None)
        entry["coverage"] = coverage
        for field, attr in (
            ("toolchainHash", "toolchain_hash"),
            ("profileHash", "profile_hash"),
            ("environmentHash", "environment_hash"),
            ("dbSchemaHash", "db_schema_hash"),
        ):
            val = getattr(args, attr, None)
            if field == "toolchainHash" and not _nonempty_str(val):
                val = default_toolchain_hash(project_root)
            elif field == "environmentHash" and not _nonempty_str(val):
                val = default_environment_hash(project_root)
            if _nonempty_str(val):
                entry[field] = str(val).strip()
        entry["commandSetHash"] = command_set_hash(args.command)
        tree_hash = product_tree_hash(project_root)
        if tree_hash:
            entry["productTreeHash"] = tree_hash
        dependency_lock_hash = lock_hash(project_root)
        if dependency_lock_hash:
            entry["lockHash"] = dependency_lock_hash
        # package verification: record build artifact + test-reuse provenance.
        if verification == "package":
            if _nonempty_str(getattr(args, "deploy_artifact", None)):
                entry["deployArtifact"] = str(args.deploy_artifact).strip()
            if _nonempty_str(getattr(args, "artifact_hash", None)):
                entry["sha256"] = str(args.artifact_hash).strip()
            entry["testsExecuted"] = bool(getattr(args, "tests_executed", False))
            if _nonempty_str(getattr(args, "tests_reused_from", None)):
                entry["testsReusedFrom"] = str(args.tests_reused_from).strip()

        # C9: bind scenario IDs from --scenario-ids to this ledger entry.
        scenario_ids_raw = getattr(args, "scenario_ids", None)
        scenario_receipt_file = getattr(args, "scenario_receipt_file", None)
        if _nonempty_str(scenario_receipt_file) and not _nonempty_str(
            scenario_ids_raw
        ):
            return emit_error(
                "--scenario-receipt-file requires --scenario-ids",
                as_json=as_json,
                error_code="SCENARIO_IDS_REQUIRED",
            )
        if _nonempty_str(scenario_ids_raw):
            ids = [s.strip() for s in str(scenario_ids_raw).split(",") if s.strip()]
            if len(ids) != len(set(ids)):
                return emit_error(
                    "scenario IDs must be unique",
                    as_json=as_json,
                    error_code="SCENARIO_ID_DUPLICATE",
                )
            if ids:
                entry["scenarioIds"] = ids
                receipt_file = scenario_receipt_file
                manifest_path = change_dir / "meta" / "scenario-manifest.json"
                manifest_schema = 0
                if manifest_path.is_file():
                    try:
                        manifest_probe = json.loads(
                            manifest_path.read_text(encoding="utf-8-sig")
                        )
                    except (OSError, json.JSONDecodeError) as exc:
                        return emit_error(
                            f"invalid scenario manifest: {exc}",
                            as_json=as_json,
                            error_code="SCENARIO_MANIFEST_INVALID",
                        )
                    resolved_probe = resolve_scenario_manifest(manifest_probe)
                    if not resolved_probe["ok"]:
                        return emit_error(
                            str(resolved_probe.get("message") or ""),
                            as_json=as_json,
                            error_code=str(resolved_probe["code"]),
                            extra={"artifactType": "scenario_manifest"},
                        )
                    manifest_probe = resolved_probe["manifest"]
                    raw_schema = (
                        manifest_probe.get("schemaVersion")
                        if isinstance(manifest_probe, dict)
                        else None
                    )
                    if (
                        isinstance(raw_schema, int)
                        and not isinstance(raw_schema, bool)
                    ):
                        manifest_schema = raw_schema
                if manifest_schema >= 2 and not _nonempty_str(receipt_file):
                    return emit_error(
                        "scenario-manifest schemaVersion 2 requires "
                        "--scenario-receipt-file; generate a skeleton with: "
                        "harness_ledger.py scenario-receipt-template "
                        f"--change-dir {change_dir} --scenario-ids "
                        f"{','.join(ids)} --runner <name> --out "
                        "runtime/scenario-receipt-<verification>.json",
                        as_json=as_json,
                        error_code="SCENARIO_RECEIPT_REQUIRED",
                        extra={"template": "scenario-receipt-template"},
                    )
                if _nonempty_str(receipt_file):
                    resolved_receipt, receipt_candidates = _resolve_receipt_path(
                        str(receipt_file), change_dir
                    )
                    if resolved_receipt is None:
                        return emit_error(
                            "scenario receipt not found; tried: "
                            + ", ".join(str(c) for c in receipt_candidates),
                            as_json=as_json,
                            error_code="SCENARIO_RECEIPT_NOT_FOUND",
                            extra={
                                "triedPaths": [str(c) for c in receipt_candidates],
                                "hint": (
                                    "--scenario-receipt-file accepts an absolute "
                                    "path, a CWD-relative path, or a path relative "
                                    "to --change-dir"
                                ),
                            },
                        )
                    try:
                        receipt_payload = json.loads(
                            resolved_receipt.read_text(encoding="utf-8-sig")
                        )
                    except (OSError, json.JSONDecodeError) as exc:
                        return emit_error(
                            f"invalid scenario receipt: {exc}",
                            as_json=as_json,
                            error_code="SCENARIO_RECEIPT_INVALID",
                        )
                    if not isinstance(receipt_payload, dict):
                        return emit_error(
                            "scenario receipt top level must be an object",
                            as_json=as_json,
                            error_code="SCENARIO_RECEIPT_INVALID",
                        )
                    scenario_result = validate_scenario_execution_receipt(
                        change_dir=change_dir,
                        scenario_ids=ids,
                        receipt=receipt_payload,
                    )
                    if not scenario_result.get("ok"):
                        return emit_error(
                            str(scenario_result.get("error") or "invalid scenario receipt"),
                            as_json=as_json,
                            error_code=str(
                                scenario_result.get("code")
                                or "SCENARIO_RECEIPT_INVALID"
                            ),
                        )
                    entry["scenarioReceipt"] = scenario_result["receipt"]
                    entry["scenarioCoverage"] = scenario_result["coverage"]
                    if status == "OK":
                        passed = set(scenario_result["coverage"]["passed"])
                        required = set(ids)
                        missing = sorted(required - passed)
                        if missing:
                            return emit_error(
                                "required scenarios were not executed successfully: "
                                + ", ".join(missing),
                                as_json=as_json,
                                error_code="REQUIRED_SCENARIO_NOT_EXECUTED",
                                extra={
                                    "missing": missing,
                                    "scenarioCoverage": scenario_result["coverage"],
                                },
                            )
                else:
                    entry.pop("scenarioReceipt", None)
                    entry.pop("scenarioCoverage", None)
        elif "scenarioIds" in entry:
            # Fresh record without scenario-ids must not keep stale ids from prev.
            entry.pop("scenarioIds", None)
            entry.pop("scenarioReceipt", None)
            entry.pop("scenarioCoverage", None)

        target_identity = {
            "verification": verification,
            "command": entry.get("command"),
            "scope": entry.get("scope"),
            "coverage": entry.get("coverage"),
            "inputsHash": entry.get("inputsHash"),
            "toolchainHash": entry.get("toolchainHash"),
            "profileHash": entry.get("profileHash"),
            "environmentHash": entry.get("environmentHash"),
            "productTreeHash": entry.get("productTreeHash"),
            "commandSetHash": entry.get("commandSetHash"),
            "lockHash": entry.get("lockHash"),
            "dbSchemaHash": entry.get("dbSchemaHash"),
        }
        target_id = (
            verification
            + "-"
            + hashlib.sha256(
                json.dumps(
                    target_identity,
                    ensure_ascii=False,
                    sort_keys=True,
                    separators=(",", ":"),
                ).encode("utf-8")
            ).hexdigest()[:16]
        )
        target = {"id": target_id, "verification": verification, **entry}
        targets = ledger.setdefault("verificationTargets", {})
        if not isinstance(targets, dict):
            targets = {}
            ledger["verificationTargets"] = targets
        targets[target_id] = target
        ledger["validations"][verification] = {
            key: value
            for key, value in target.items()
            if key not in {"id", "verification"}
        }
        if "changeName" not in ledger:
            ledger["changeName"] = change_dir.name
        if "stateDir" not in ledger:
            ledger["stateDir"] = str(change_dir)

        out_path = preferred_write_path(change_dir)
        write_ledger(out_path, ledger)
        try:
            upsert_verification_graph_node(
                change_dir,
                verification=verification,
                status=str(status),
                inputs_hash=str(inputs_hash),
                command=str(entry.get("command") or "") or None,
                toolchain_hash=(
                    str(entry.get("toolchainHash")).strip()
                    if _nonempty_str(entry.get("toolchainHash"))
                    else None
                ),
                environment_hash=(
                    str(entry.get("environmentHash")).strip()
                    if _nonempty_str(entry.get("environmentHash"))
                    else None
                ),
                db_schema_hash=(
                    str(entry.get("dbSchemaHash")).strip()
                    if _nonempty_str(entry.get("dbSchemaHash"))
                    else None
                ),
                diff_hash=(
                    str(ledger.get("diffHash")).strip()
                    if _nonempty_str(ledger.get("diffHash"))
                    else None
                ),
                target_identity=(
                    str(entry.get("targetIdentity")).strip()
                    if _nonempty_str(entry.get("targetIdentity"))
                    else None
                ),
            )
        except (OSError, ValueError, TypeError):
            # Graph is advisory persistence; ledger write already succeeded.
            pass
    except (OSError, ValueError, FileNotFoundError, json.JSONDecodeError) as exc:
        return emit_error(f"record failed: {exc}", as_json=as_json)

    payload = {
        "ok": True,
        "action": "record",
        "verification": verification,
        "status": status,
        "inputsHash": inputs_hash,
        "inputsFiles": inputs_files,
        "ledger_path": str(out_path),
        "diffHash": ledger.get("diffHash"),
        "resolvedProjectRoot": str(project_root) if project_root else None,
        "scenarioCoverage": entry.get("scenarioCoverage"),
        "migrationReceipt": migration_receipt,
    }
    zero_warn = _zero_tests_with_selector_warning(
        str(args.command), args.evidence, project_root
    )
    record_warnings = [
        warning for warning in (zero_warn, profile_input_missing_warn)
        if warning is not None
    ]
    if record_warnings:
        payload["warnings"] = record_warnings
        for warning in record_warnings:
            print(f"[harness-ledger] WARNING {warning}", file=sys.stderr)
    emit_compact_or_verbose(
        payload, as_json=as_json, verbose=verbose,
        compact_fn=_compact_record_payload,
    )
    return 0


# --- 批次 2 WI-1b：record-from-receipt（证据直接采集） ---

RECEIPT_REQUIRED_FIELDS = (
    "argv",
    "exitCode",
    "timedOut",
    "durationMs",
    "outputTail",
)


def _load_exec_result_receipt(
    receipt_path: Path,
) -> tuple[dict[str, Any] | None, str | None, str | None]:
    """读取并校验 exec 结果收据。

    返回 (receipt, field_path, problem)；receipt 为 None 时 field_path
    指向第一个问题字段。错误信封带 field_path 是 F3 的教训（无
    field_path 时排障靠读源码）。
    """

    try:
        raw = receipt_path.read_text(encoding="utf-8-sig")
    except OSError as exc:
        return None, "receipt", f"unreadable: {exc}"
    try:
        receipt = json.loads(raw)
    except json.JSONDecodeError as exc:
        return None, "receipt", f"invalid JSON: {exc}"
    if not isinstance(receipt, dict):
        return None, "receipt", "top level must be an object"
    if receipt.get("schemaVersion") != 1:
        return (
            None,
            "schemaVersion",
            f"expected 1, got {receipt.get('schemaVersion')!r}",
        )
    if receipt.get("action") != "exec-result":
        return (
            None,
            "action",
            f"expected 'exec-result', got {receipt.get('action')!r}",
        )
    for field in RECEIPT_REQUIRED_FIELDS:
        if field not in receipt:
            return None, field, "missing required field"
    if not isinstance(receipt["argv"], list) or not receipt["argv"]:
        return None, "argv", "must be a non-empty list"
    if any(not isinstance(item, str) for item in receipt["argv"]):
        return None, "argv", "all items must be strings"
    if not isinstance(receipt["exitCode"], int) or isinstance(
        receipt["exitCode"], bool
    ):
        return None, "exitCode", "must be an integer"
    if not isinstance(receipt["timedOut"], bool):
        return None, "timedOut", "must be a boolean"
    if not isinstance(receipt["durationMs"], int) or isinstance(
        receipt["durationMs"], bool
    ):
        return None, "durationMs", "must be an integer"
    if not isinstance(receipt["outputTail"], str):
        return None, "outputTail", "must be a string"
    return receipt, None, None


def _receipt_evidence(output_tail: str) -> str:
    """从 outputTail 提取 ledger evidence 文本。

    优先保留含测试计数的尾部行（如 "Tests run: N, Failures: M"），
    最多 4 行；无匹配行时回退到尾部非空行。空输出返回占位说明
    （evidence 是 record 必填字段，不能为空串）。
    """

    lines = [line.strip() for line in output_tail.splitlines()]
    lines = [line for line in lines if line]
    if not lines:
        return "(no captured output; see runner logs)"
    count_lines = [
        line
        for line in lines
        if re.search(r"tests? run|passed|failed|failures?|error", line, re.IGNORECASE)
    ]
    selected = count_lines[-4:] if count_lines else lines[-4:]
    return "\n".join(selected)


def cmd_record_from_receipt(args: argparse.Namespace) -> int:
    as_json = bool(args.json)
    receipt_path = resolve_path(args.receipt)
    receipt, field_path, problem = _load_exec_result_receipt(receipt_path)
    if receipt is None:
        return emit_error(
            f"invalid exec result receipt at {receipt_path}: {problem}",
            as_json=as_json,
            error_code="RECEIPT_INVALID",
            extra={
                "fieldPath": field_path,
                "recoveryAction": (
                    "rerun the verification with harness_test_runner.py exec "
                    "--result-receipt to produce a fresh receipt, or fall back "
                    "to the manual record subcommand"
                ),
            },
        )
    verification = args.verification
    status = "OK" if (receipt["exitCode"] == 0 and not receipt["timedOut"]) else "FAIL"
    command = " ".join(receipt["argv"])
    evidence = _receipt_evidence(receipt["outputTail"])
    # 沿 harness_task.py _record_ledger_entry 的 Namespace 构造模式：
    # 复用 cmd_record 全部语义（ownership 检查、profile 展开、ledger v3
    # 迁移、场景绑定），不复制其逻辑。cmd_record 的 stdout 输出被捕获
    # 吞掉——record-from-receipt 只输出自己的信封，避免两个 JSON 拼接
    # 破坏机器可读性；其 stderr（emit_error + WARNING）原样透传。
    record_args = argparse.Namespace(
        change_dir=args.change_dir,
        verification=verification,
        status=status,
        command=command,
        runner_command=None,
        exit_code=receipt["exitCode"],
        duration_ms=receipt["durationMs"],
        files=args.files,
        files_from=None,
        evidence=evidence,
        project=args.project,
        profile_input=args.profile_input,
        scope=None,
        coverage=None,
        toolchain_hash=None,
        profile_hash=None,
        environment_hash=None,
        db_schema_hash=None,
        deploy_artifact=None,
        artifact_hash=None,
        tests_executed=False,
        tests_reused_from=None,
        metrics_json=None,
        metrics_file=None,
        base_commit=None,
        diff_hash=None,
        applicability=None,
        applicability_reason=None,
        # B2-1：场景绑定透传（收据路径 + 场景绑定一次完成）。
        scenario_ids=getattr(args, "scenario_ids", None),
        scenario_receipt_file=getattr(args, "scenario_receipt_file", None),
        verbose=bool(getattr(args, "verbose", False)),
        json=as_json,
    )

    class _Capture:
        def __init__(self) -> None:
            self.chunks: list[str] = []

        def write(self, text: str) -> int:
            self.chunks.append(text)
            return len(text)

        def flush(self) -> None:
            pass

    captured_stdout = _Capture()
    original_stdout = sys.stdout
    sys.stdout = captured_stdout  # type: ignore[assignment]
    try:
        rc = cmd_record(record_args)
    finally:
        sys.stdout = original_stdout  # type: ignore[assignment]
    if rc != 0:
        # cmd_record 已输出结构化错误信封（stderr）；此处只补收据上下文。
        return rc
    payload = {
        "ok": True,
        "action": "record-from-receipt",
        "verification": verification,
        "status": status,
        "receiptPath": str(receipt_path),
        "derivedFrom": {
            "command": command,
            "exitCode": receipt["exitCode"],
            "timedOut": receipt["timedOut"],
            "durationMs": receipt["durationMs"],
        },
    }
    emit_json(payload, as_json=as_json)
    return 0


# --- 批次 3 WI-3.2：CI 证据导入（校验式落账，fail-closed） ---

CI_IMPORTABLE_VERIFICATIONS = frozenset({"compile", "unitTestFull"})
CI_RECEIPT_SCHEMA_VERSION = 1


def _ci_receipt_hash(receipt: dict[str, Any]) -> str:
    body = {k: v for k, v in receipt.items() if k != "receiptHash"}
    canonical = json.dumps(
        body, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(canonical).hexdigest()


def _load_ci_evidence_receipt(
    receipt_path: Path,
) -> tuple[dict[str, Any] | None, str | None, str | None, str]:
    """读取并校验 CI 证据收据（schema + 完整性哈希）。

    返回 (receipt, field_path, problem, error_code)；失败 fail-closed。
    """
    try:
        raw = receipt_path.read_text(encoding="utf-8-sig")
    except OSError as exc:
        return None, "receipt", f"unreadable: {exc}", "RECEIPT_INVALID"
    try:
        receipt = json.loads(raw)
    except json.JSONDecodeError as exc:
        return None, "receipt", f"invalid JSON: {exc}", "RECEIPT_INVALID"
    if not isinstance(receipt, dict):
        return None, "receipt", "must be a JSON object", "RECEIPT_INVALID"
    for field in (
        "schemaVersion", "repository", "runId", "headSha", "headTree",
        "toolchain", "conclusion", "jobs", "concludedAt", "receiptHash",
    ):
        if field not in receipt:
            return None, field, "missing required field", "RECEIPT_INVALID"
    if receipt["schemaVersion"] != CI_RECEIPT_SCHEMA_VERSION:
        return None, "schemaVersion", (
            f"expected {CI_RECEIPT_SCHEMA_VERSION}, got {receipt['schemaVersion']!r}"
        ), "RECEIPT_INVALID"
    if not isinstance(receipt["jobs"], list) or not receipt["jobs"]:
        return None, "jobs", "must be a non-empty list", "RECEIPT_INVALID"
    for index, job in enumerate(receipt["jobs"]):
        if not isinstance(job, dict) or not job.get("name"):
            return None, f"jobs[{index}]", "must be an object with a name", "RECEIPT_INVALID"
        if "conclusion" not in job:
            return None, f"jobs[{index}].conclusion", "missing required field", "RECEIPT_INVALID"
    if not str(receipt["headTree"]).startswith("sha256:"):
        return None, "headTree", "must be sha256: prefixed", "RECEIPT_INVALID"
    expected = _ci_receipt_hash(receipt)
    if receipt["receiptHash"] != expected:
        return None, "receiptHash", (
            f"hash mismatch: expected {expected}, got {receipt['receiptHash']!r}"
        ), "RECEIPT_HASH_MISMATCH"
    return receipt, None, None, "RECEIPT_INVALID"


def _validate_ci_evidence_receipt(
    receipt: dict[str, Any],
    *,
    project_root: Path,
    verification: str,
    job_name: str,
    step_name: str,
    shard: str | None = None,
) -> tuple[dict[str, Any] | None, str, str, dict[str, Any] | None]:
    """语义校验链：结果/代码版本/来源/verification 集合。

    返回 (step, field_path, problem, detail)；step 非 None 表示全部通过。
    """
    if receipt.get("conclusion") != "success":
        return None, "conclusion", "RECEIPT_CONCLUSION_NOT_SUCCESS", {
            "actual": receipt.get("conclusion"),
        }
    job = next(
        (j for j in receipt["jobs"] if j.get("name") == job_name), None
    )
    if job is None:
        return None, "job", "RECEIPT_JOB_NOT_FOUND", {
            "requested": job_name,
            "available": [j.get("name") for j in receipt["jobs"]],
        }
    if job.get("conclusion") != "success":
        return None, f"jobs[{job_name}].conclusion", (
            "RECEIPT_CONCLUSION_NOT_SUCCESS"
        ), {"actual": job.get("conclusion")}
    step = next(
        (
            s for s in job.get("steps", [])
            if isinstance(s, dict) and s.get("name") == step_name
        ),
        None,
    )
    if step is None:
        return None, "step", "RECEIPT_STEP_NOT_FOUND", {
            "requested": step_name,
            "available": [
                s.get("name") for s in job.get("steps", []) if isinstance(s, dict)
            ],
        }
    if step.get("conclusion") != "success":
        return None, f"steps[{step_name}].conclusion", (
            "RECEIPT_CONCLUSION_NOT_SUCCESS"
        ), {"actual": step.get("conclusion")}
    local_tree = product_tree_hash(project_root)
    if not local_tree or receipt.get("headTree") != local_tree:
        return None, "headTree", "RECEIPT_TREE_MISMATCH", {
            "receipt": receipt.get("headTree"),
            "local": local_tree,
            "hint": (
                "CI 证据绑定整个产品 tree；工作区与 CI 运行时不一致即拒绝。"
                "请在当前代码上重新触发 CI 并导入新收据。"
            ),
        }
    # R4：productTree 必须绑定候选实际内容（含未提交漂移、未跟踪产品文件，
    # 排除 .harness/）——只比对 HEAD^{tree} 时工作区漂移不可见。旧收据无
    # productTree 字段时回退 headTree（二者在清洁工作区同值）。
    bound_tree = receipt.get("productTree") or receipt.get("headTree")
    worktree_tree = worktree_product_tree_hash(project_root)
    if not worktree_tree or bound_tree != worktree_tree:
        return None, "productTree", "RECEIPT_TREE_MISMATCH", {
            "receipt": bound_tree,
            "local": worktree_tree,
            "hint": (
                "收据绑定的是 CI 运行时的产品内容；当前工作区存在未提交的"
                "产品漂移（修改或未跟踪文件）。请先提交/还原漂移，或在当前"
                "内容上重新触发 CI 并导入新收据。"
            ),
        }
    # R5：声明了分片身份的收据必须与本次导入的分片一致（错分片/重复导入
    # 同一分片以外的串扰直接拒绝）。
    receipt_shard = receipt.get("shard")
    if isinstance(receipt_shard, dict):
        expected_shard = (
            f"{receipt_shard.get('index')}/{receipt_shard.get('total')}"
        )
        if shard != expected_shard:
            return None, "shard", "RECEIPT_SHARD_MISMATCH", {
                "receipt": expected_shard,
                "actual": shard,
                "hint": (
                    "收据声明的分片与 --shard 不一致；请按收据实际分片导入，"
                    "或重新生成分片收据。"
                ),
            }
    local_repo = harness_paths._primary_remote_url(project_root)
    if local_repo is None:
        return None, "repository", "RECEIPT_REPOSITORY_MISMATCH", {
            "receipt": receipt.get("repository"),
            "local": None,
            "hint": "本地仓库没有配置 remote，无法核验收据来源",
        }
    local_repo_normalized = harness_paths._normalize_remote(local_repo)
    if receipt.get("repository") != local_repo_normalized:
        return None, "repository", "RECEIPT_REPOSITORY_MISMATCH", {
            "receipt": receipt.get("repository"),
            "local": local_repo_normalized,
        }
    if verification not in CI_IMPORTABLE_VERIFICATIONS:
        return None, "verification", "RECEIPT_VERIFICATION_NOT_IMPORTABLE", {
            "requested": verification,
            "importable": sorted(CI_IMPORTABLE_VERIFICATIONS),
        }
    return step, None, "", None


def cmd_import_ci_evidence(args: argparse.Namespace) -> int:
    """WI-3.2：导入宿主 CI 验证证据（提案 §4.6）。

    校验链（任一失败即拒绝，fail-closed）：收据完整性 → conclusion →
    headTree == 本地 product tree → repository == 本地 origin →
    verification ∈ {compile, unitTestFull}。
    """
    as_json = bool(args.json)
    change_dir = resolve_path(args.change_dir)
    apply_inferred_project_root(args, change_dir)
    receipt_path = resolve_path(args.receipt)
    receipt, field_path, problem, error_code = _load_ci_evidence_receipt(
        receipt_path
    )
    if receipt is None:
        return emit_error(
            f"invalid CI evidence receipt at {receipt_path}: {problem}",
            as_json=as_json,
            error_code=error_code,
            extra={
                "fieldPath": field_path,
                "recoveryAction": (
                    "使用 CI workflow 末尾生成的原始收据文件；不要手改任何字段"
                    "（receiptHash 会失配）。"
                ),
            },
        )
    job_name = args.job
    step_name = args.step
    shard = str(args.shard).strip() if getattr(args, "shard", None) else None
    if shard:
        match = re.fullmatch(r"([1-9]\d*)/([1-9]\d*)", shard)
        if match is None or int(match.group(1)) > int(match.group(2)):
            return emit_error(
                f"invalid --shard {shard!r}: 期望 I/N（1<=I<=N）",
                as_json=as_json,
                error_code="SHARD_INVALID",
                extra={
                    "fieldPath": "shard",
                    "recoveryAction": "按收据实际分片传 --shard I/N（如 1/3）",
                },
            )
    step, field_path, code, detail = _validate_ci_evidence_receipt(
        receipt,
        project_root=Path(str(getattr(args, "project", None) or change_dir)),
        verification=args.verification,
        job_name=job_name,
        step_name=step_name,
        shard=shard,
    )
    if field_path is not None:
        return emit_error(
            f"CI evidence receipt rejected at {field_path}: {code}",
            as_json=as_json,
            error_code=code,
            extra={"fieldPath": field_path, **(detail or {})},
        )

    project_root = Path(str(getattr(args, "project", None) or change_dir))
    head_tree = receipt["headTree"]
    ci_command = str(step.get("command") or f"{job_name}/{step_name}")
    status = "OK"
    # R5：分片聚合——只有与本次收据同 runId 且同 productTree 的分片可以
    # 聚合；内容分叉或旧 run 的碎片记入 supersededShards（不再贡献 full
    # 覆盖判定）。无 --shard 时维持整单收据语义（coverage=full）。
    coverage = "full"
    shard_records: dict[str, dict[str, Any]] | None = None
    superseded_shards: list[dict[str, Any]] = []
    if shard is not None:
        bound_tree = receipt.get("productTree") or head_tree
        current_run = str(receipt.get("runId") or "")
        shard_total = int(shard.split("/", 1)[1])
        prev_entry: dict[str, Any] | None = None
        try:
            prev_ledger, _ = load_ledger(change_dir)
            if isinstance(prev_ledger, dict):
                prev_validations = prev_ledger.get("validations")
                if isinstance(prev_validations, dict):
                    candidate = prev_validations.get(args.verification)
                    if isinstance(candidate, dict):
                        prev_entry = candidate
        except (OSError, ValueError, json.JSONDecodeError):
            prev_entry = None
        active: dict[str, dict[str, Any]] = {}
        if prev_entry is not None:
            prev_superseded = prev_entry.get("supersededShards")
            if isinstance(prev_superseded, list):
                superseded_shards.extend(
                    item for item in prev_superseded if isinstance(item, dict)
                )
            prev_shards = prev_entry.get("shards")
            if isinstance(prev_shards, dict):
                for key, record in prev_shards.items():
                    if not isinstance(record, dict):
                        continue
                    same_run = str(record.get("runId") or "") == current_run
                    same_tree = (
                        str(record.get("productTree") or "") == bound_tree
                    )
                    same_total = str(key).endswith(f"/{shard_total}")
                    if same_run and same_tree and same_total:
                        active[str(key)] = record
                    else:
                        superseded_shards.append(record)
        active[shard] = {
            "runId": receipt.get("runId"),
            "runAttempt": receipt.get("runAttempt"),
            "productTree": bound_tree,
            "job": job_name,
            "step": step_name,
            "importedAt": now_iso(),
        }
        have = {
            int(key.split("/", 1)[0]) for key in active if "/" in key
        }
        coverage = "full" if len(have) >= shard_total else "incremental"
        shard_records = active
    entry = {
        "status": status,
        "command": ci_command,
        "runnerCommand": f"ci:{receipt.get('runUrl') or receipt.get('runId')}",
        "evidence": str(receipt_path),
        "exitCode": 0,
        "durationMs": 0,
        "inputsHash": head_tree,
        "inputsFiles": [],
        "coverage": coverage,
        "algorithmVersion": LEDGER_VERSION,
        "finishedAt": now_iso(),
        "imported": True,
        "importedFrom": {
            "importVersion": "ci-import-v1",
            "repository": receipt["repository"],
            "runId": receipt["runId"],
            "runAttempt": receipt.get("runAttempt"),
            "runUrl": receipt.get("runUrl"),
            "headSha": receipt["headSha"],
            "headTree": head_tree,
            "job": job_name,
            "step": step_name,
            "concludedAt": receipt["concludedAt"],
        },
    }
    if shard_records is not None:
        entry["shards"] = shard_records
        if superseded_shards:
            entry["supersededShards"] = superseded_shards
    try:
        ledger, existing_path = load_ledger(change_dir)
        if ledger is None:
            ledger = {
                "changeName": change_dir.name,
                "stateDir": str(change_dir),
                "validations": {},
            }
        elif not isinstance(ledger.get("validations"), dict):
            ledger["validations"] = {}
        ledger, _migration = migrate_ledger_for_write(
            change_dir,
            ledger,
            project_root=project_root,
            base_commit=None,
            diff_hash=None,
            record_migration=existing_path is not None,
        )
        prev = ledger["validations"].get(args.verification)
        superseded = None
        if isinstance(prev, dict) and isinstance(prev.get("importedFrom"), dict):
            superseded = prev["importedFrom"].get("runId")
        if superseded is not None:
            entry["importedFrom"]["supersededRunId"] = superseded
        target_identity = {
            "verification": args.verification,
            "command": entry["command"],
            "coverage": entry["coverage"],
            "inputsHash": entry["inputsHash"],
            "imported": True,
        }
        target_id = (
            args.verification
            + "-"
            + hashlib.sha256(
                json.dumps(
                    target_identity,
                    ensure_ascii=False,
                    sort_keys=True,
                    separators=(",", ":"),
                ).encode("utf-8")
            ).hexdigest()[:16]
        )
        target = {"id": target_id, "verification": args.verification, **entry}
        targets = ledger.setdefault("verificationTargets", {})
        if not isinstance(targets, dict):
            targets = {}
            ledger["verificationTargets"] = targets
        targets[target_id] = target
        ledger["validations"][args.verification] = {
            key: value
            for key, value in target.items()
            if key not in {"id", "verification"}
        }
        ledger["changeName"] = change_dir.name
        ledger["stateDir"] = str(change_dir)
        out_path = preferred_write_path(change_dir)
        write_ledger(out_path, ledger)
    except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
        return emit_error(f"import failed: {exc}", as_json=as_json)
    payload = {
        "ok": True,
        "action": "import-ci-evidence",
        "verification": args.verification,
        "status": status,
        "coverage": coverage,
        "inputsHash": head_tree,
        "receiptPath": str(receipt_path),
        "importedFrom": entry["importedFrom"],
    }
    if shard_records is not None:
        payload["aggregatedShards"] = sorted(shard_records)
        payload["supersededShards"] = len(superseded_shards)
    emit_json(payload, as_json=as_json)
    return 0


def _zero_tests_with_selector_warning(
    command: str, evidence: str | None, project_root: Path | None
) -> str | None:
    """选择器存在却 0 命中 → WARN（E-2）。

    surefire 插件级 ``<excludedGroups>`` 会压过命令行空值覆盖，
    `-Dgroups=X` 选择器 0 命中仍 exit 0——只看退出码就是假阳性
    （2026-08-30 demo-datasource 集成测试静默 0 执行实测）。
    证据文本出现 Tests run: 0 且命令含选择器时，record 给出醒目警告。
    """
    if not evidence:
        return None
    selectors = ("-Dgroups=", "-Dtest=", "--groups", "--filter", "-DfailIfNoSpecifiedTests")
    if not any(selector in str(command) for selector in selectors):
        return None
    candidates = [Path(evidence)]
    if project_root is not None and not Path(evidence).is_absolute():
        candidates.insert(0, project_root / evidence)
    text: str | None = None
    for candidate in candidates:
        try:
            if candidate.is_file() and candidate.stat().st_size <= 4 * 1024 * 1024:
                text = candidate.read_text(encoding="utf-8", errors="replace")
                break
        except OSError:
            continue
    if text is None:
        return None
    zero = re.search(
        r"Tests run:\s*0,\s*Failures:\s*0,\s*Errors:\s*0", text
    )
    ran_something = re.search(r"Tests run:\s*[1-9]", text)
    if zero is not None and ran_something is None:
        return (
            "ZERO_TESTS_WITH_SELECTOR: 命令含测试选择器但 Tests run=0（exit 0 不代表通过）；"
            "常见于 surefire 插件级 excludedGroups 压过命令行覆盖——检查 pom 是否写死 "
            "<excludedGroups>（应改为 ${excludedGroups} 属性占位）"
        )
    return None


def cmd_scenario_receipt_template(args: argparse.Namespace) -> int:
    """Emit a schema-v2 receipt skeleton built from the scenario manifest.

    Without this, agents had to read harness_ledger.py source to learn the
    receipt shape. The skeleton is already valid for `record`; the caller only
    has to correct `status` for any test that did not pass.
    """
    as_json = bool(args.json)
    change_dir = resolve_path(args.change_dir)
    ids = [s.strip() for s in str(args.scenario_ids).split(",") if s.strip()]
    if not ids:
        return emit_error(
            "--scenario-ids must list at least one scenario ID",
            as_json=as_json,
            error_code="SCENARIO_IDS_REQUIRED",
        )
    if len(ids) != len(set(ids)):
        return emit_error(
            "scenario IDs must be unique",
            as_json=as_json,
            error_code="SCENARIO_ID_DUPLICATE",
        )
    manifest_path = change_dir / "meta" / "scenario-manifest.json"
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
    except FileNotFoundError:
        return emit_error(
            f"scenario manifest does not exist: {manifest_path}",
            as_json=as_json,
            error_code="SCENARIO_MANIFEST_MISSING",
        )
    except (OSError, json.JSONDecodeError) as exc:
        return emit_error(
            f"scenario manifest unreadable: {exc}",
            as_json=as_json,
            error_code="SCENARIO_MANIFEST_INVALID",
        )
    resolved = resolve_scenario_manifest(manifest)
    if not resolved["ok"]:
        return emit_error(
            str(resolved.get("message") or ""),
            as_json=as_json,
            error_code=str(resolved["code"]),
            extra={"artifactType": "scenario_manifest"},
        )
    manifest = resolved["manifest"]
    scenarios = manifest.get("scenarios") if isinstance(manifest, dict) else None
    if not isinstance(scenarios, list):
        return emit_error(
            "scenario-manifest.json must contain a scenarios array",
            as_json=as_json,
            error_code="SCENARIO_MANIFEST_INVALID",
        )
    by_id = {
        str(item.get("id") or "").strip(): item
        for item in scenarios
        if isinstance(item, dict) and _nonempty_str(item.get("id"))
    }
    unknown = sorted(set(ids) - set(by_id))
    if unknown:
        return emit_error(
            "scenario IDs are not declared in scenario-manifest.json: "
            + ", ".join(unknown),
            as_json=as_json,
            error_code="SCENARIO_ID_UNKNOWN",
        )
    status = str(args.status).strip().upper()
    if status not in {"PASSED", "FAILED", "SKIPPED"}:
        return emit_error(
            "--status must be one of PASSED|FAILED|SKIPPED",
            as_json=as_json,
            error_code="SCENARIO_RECEIPT_INVALID",
        )
    attempt = int(args.attempt)
    if attempt < 1:
        return emit_error(
            "--attempt must be a positive integer",
            as_json=as_json,
            error_code="SCENARIO_RECEIPT_INVALID",
        )

    identities: list[tuple[str, str, str]] = []
    incomplete: list[str] = []
    for scenario_id in ids:
        scenario = by_id[scenario_id]
        if any(
            not _nonempty_str(scenario.get(field))
            for field in ("executableTestId", "testFile", "testTitle")
        ):
            incomplete.append(scenario_id)
            continue
        identities.append(
            (
                str(scenario["executableTestId"]).strip(),
                str(scenario["testFile"]).strip(),
                str(scenario["testTitle"]).strip(),
            )
        )
    if incomplete:
        return emit_error(
            "scenarios are missing executableTestId/testFile/testTitle in the "
            "manifest: " + ", ".join(incomplete),
            as_json=as_json,
            error_code="SCENARIO_MANIFEST_INVALID",
            extra={"incomplete": incomplete},
        )

    runner: dict[str, Any] = {"name": str(args.runner).strip()}
    if _nonempty_str(getattr(args, "runner_version", None)):
        runner["version"] = str(args.runner_version).strip()
    receipt = {
        "schemaVersion": 1,
        "runner": runner,
        "attempt": attempt,
        "declared": [test_id for test_id, _, _ in identities],
        "selected": [test_id for test_id, _, _ in identities],
        "collected": [
            {"testId": test_id, "file": file, "title": title}
            for test_id, file, title in identities
        ],
        "executed": [
            {
                "testId": test_id,
                "file": file,
                "title": title,
                "attempt": attempt,
                "status": status,
            }
            for test_id, file, title in identities
        ],
    }
    body = json.dumps(receipt, ensure_ascii=False, indent=2) + "\n"

    out_raw = getattr(args, "out", None)
    if not _nonempty_str(out_raw):
        sys.stdout.write(body)
        return 0
    out_path = Path(str(out_raw)).expanduser()
    if not out_path.is_absolute():
        # 防双重拼接（E-1）：调用方按 CLI 惯例传 cwd 相对路径时（如
        # .harness/changes/<cn>/evidence/x.json），直接拼 change_dir 会产生嵌套
        # 幽灵目录。相对路径若以 change-dir 自身前缀开头，按 cwd 解析；否则按
        # 文档规则相对 change-dir 解析。
        cwd = Path.cwd().resolve()
        change_resolved = change_dir.resolve()
        if _is_relative_to(change_resolved, cwd):
            change_rel = change_resolved.relative_to(cwd)
            if out_path.parts[: len(change_rel.parts)] == change_rel.parts:
                out_path = cwd / out_path
            else:
                out_path = change_dir / out_path
        else:
            out_path = change_dir / out_path
    out_path = out_path.resolve()
    change_resolved = change_dir.resolve()
    cwd = Path.cwd().resolve()
    if not (
        _is_relative_to(out_path, change_resolved)
        or _is_relative_to(out_path, cwd)
    ):
        return emit_error(
            "--out 解析后越出项目目录: " + str(out_path),
            as_json=as_json,
            error_code="SCENARIO_RECEIPT_OUT_OF_PROJECT",
        )
    try:
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(body, encoding="utf-8", newline="\n")
    except OSError as exc:
        return emit_error(
            f"cannot write scenario receipt: {exc}",
            as_json=as_json,
            error_code="SCENARIO_RECEIPT_WRITE_FAILED",
        )
    payload = {
        "ok": True,
        "action": "scenario-receipt-template",
        "path": str(out_path),
        "scenarioIds": ids,
    }
    if as_json:
        sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\n")
    else:
        sys.stdout.write(f"{out_path}\n")
    return 0


# --- 批次 2 WI-4a：render-report（测试报告派生，消不对称 E） ---

REPORT_RENDER_VERSION = "harness-render-report-1"
# 五态状态沿用归档 summary-data 语义（设计 §8），不新造枚举：
# OK/FAIL/NOT_RUN 来自 ledger 条目 status；REUSED 表示复用标记；
# RETESTED 表示同 target 多 attempt（rerunCount>0）。
_FIVE_STATE_REUSED_MARKERS = ("REUSED",)


def _render_report_changed_files(
    project: Path | None,
    base: str | None,
    head: str | None,
) -> list[dict[str, Any]]:
    """变更文件表：git diff --numstat base..head（与归档同口径）。"""
    if project is None or not base or not head or base == head:
        return []
    proc = subprocess.run(
        ["git", "diff", "--numstat", f"{base}..{head}"],
        cwd=str(project),
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        check=False,
    )
    if proc.returncode != 0 or not proc.stdout:
        return []
    changed: list[dict[str, Any]] = []
    for line in proc.stdout.splitlines():
        parts = line.split("\t")
        if len(parts) < 3:
            continue
        ins = int(parts[0]) if parts[0].isdigit() else 0
        dele = int(parts[1]) if parts[1].isdigit() else 0
        changed.append(
            {"path": parts[2], "insertions": ins, "deletions": dele}
        )
    return changed


def _render_report_five_state(entry: dict[str, Any]) -> str:
    """单条 ledger 条目 → 五态状态（OK/FAIL/NOT_RUN/REUSED/RETESTED）。

    语义沿用归档 summary-data：REUSED 看 reused 标记或 status 含
    REUSED；RETESTED 看 attempts 历史（>1 次终态记录即重测过）；
    其余按 status 原样（OK/FAIL/NOT_RUN）。无法判定时保留原始
    status 字符串，不猜（设计 §8：视图字段缺失时报 UNKNOWN 不猜）。
    """
    status = str(entry.get("status") or "").strip().upper()
    reused_flag = str(entry.get("reused") or "").strip().lower()
    if reused_flag in {"true", "1"} or any(
        marker in status for marker in _FIVE_STATE_REUSED_MARKERS
    ):
        return "REUSED"
    attempts = entry.get("attempts")
    if isinstance(attempts, list):
        terminal = [
            item
            for item in attempts
            if isinstance(item, dict)
            and str(item.get("status") or "").strip().upper()
            in {"OK", "FAIL", "NOT_RUN"}
        ]
        if len(terminal) > 1:
            return "RETESTED"
    history = entry.get("history")
    if isinstance(history, list) and len(history) > 1:
        return "RETESTED"
    if status in {"OK", "FAIL", "NOT_RUN"}:
        return status
    return status or "UNKNOWN"


def _render_report_metrics_line(entry: dict[str, Any]) -> str:
    """metrics dict → 单行摘要（total/passed/failed 或 run/failures）。"""
    metrics = entry.get("metrics")
    if not isinstance(metrics, dict):
        return ""
    parts: list[str] = []
    for key in (
        "total",
        "run",
        "testsRun",
        "passed",
        "failed",
        "failures",
        "errors",
        "skipped",
        "blocked",
        "deselected",
    ):
        if key in metrics:
            parts.append(f"{key}={metrics[key]}")
    return " ".join(parts)


def _render_report_evidence_excerpt(entry: dict[str, Any]) -> str:
    """evidence 字段 → 最多 4 行摘录（与 record-from-receipt 同上限）。"""
    evidence = entry.get("evidence")
    if not isinstance(evidence, str) or not evidence.strip():
        return ""
    lines = [line.strip() for line in evidence.splitlines() if line.strip()]
    return "\n".join(lines[-4:])


def _render_report_scenario_summary(change_dir: Path) -> dict[str, Any] | None:
    """场景覆盖摘要：复用 gate 的 _validate_scenario_coverage（只读）。

    harness_gate 顶部 import harness_ledger，此处必须延迟导入避免环。
    """
    try:
        import harness_gate as hgate
    except ImportError:
        return None
    try:
        return hgate._validate_scenario_coverage(change_dir)
    except (OSError, ValueError, json.JSONDecodeError):
        return None


def _render_report_markdown(
    *,
    change_dir: Path,
    change_name: str,
    rendered_at: str,
    ledger: dict[str, Any] | None,
    events_summary: dict[str, Any] | None,
    changed_files: list[dict[str, Any]],
    base_commit: str | None,
    head_commit: str | None,
    scenario_summary: dict[str, Any] | None,
) -> str:
    lines: list[str] = []
    # frontmatter：生成标记（设计 §5：渲染报告与手写报告不并存于同一
    # change；归档按 generated 标记区分事实来源）。
    lines.extend(
        [
            "---",
            "generated: true",
            f"generator: {REPORT_RENDER_VERSION}",
            f"change-name: {change_name}",
            f"rendered-at: {rendered_at}",
            "source: ledger+events",
            "---",
            "",
            f"# 测试报告 — {change_name}",
            "",
            "> 本报告由 harness_ledger.py render-report 从 ledger 与 events 派生，",
            "> 可随时重建，不作为第二份可写状态（提案 §4.9）。",
            "> 模型只在文末「解读」段落追加残余风险与下一步。",
            "",
        ]
    )

    # --- 变更文件表 ---
    lines.extend(["## 变更文件", ""])
    if base_commit and head_commit:
        lines.append(f"范围: `{base_commit[:12]}..{head_commit[:12]}`")
        lines.append("")
    if changed_files:
        lines.extend(["| 路径 | + | - |", "|---|---:|---:|"])
        for item in changed_files:
            lines.append(
                f"| {item.get('path') or ''} | "
                f"{item.get('insertions', 0)} | {item.get('deletions', 0)} |"
            )
        total_ins = sum(int(i.get("insertions") or 0) for i in changed_files)
        total_del = sum(int(i.get("deletions") or 0) for i in changed_files)
        lines.extend(
            [
                "",
                f"共 {len(changed_files)} 个文件，+{total_ins}/-{total_del}。",
            ]
        )
    else:
        lines.append("（无提交间差异或 base/head 不可得）")
    lines.append("")

    # --- 验证证据 ---
    lines.extend(["## 验证证据", ""])
    validations: dict[str, Any] = {}
    if isinstance(ledger, dict):
        raw = ledger.get("validations")
        if isinstance(raw, dict):
            validations = raw
    if validations:
        lines.extend(
            [
                "| 验证 | 五态 | exit | 耗时 | scope/coverage | metrics |",
                "|---|:---:|---:|---:|---|---|",
            ]
        )
        for kind in sorted(validations):
            entry = validations.get(kind)
            if not isinstance(entry, dict):
                continue
            five = _render_report_five_state(entry)
            exit_code = entry.get("exitCode")
            exit_text = str(exit_code) if exit_code is not None else "—"
            duration = entry.get("durationMs")
            duration_text = (
                f"{int(duration) / 1000:.1f}s"
                if isinstance(duration, int) and not isinstance(duration, bool)
                else "—"
            )
            scope = str(entry.get("scope") or "—")
            coverage = str(entry.get("coverage") or "—")
            metrics = _render_report_metrics_line(entry) or "—"
            lines.append(
                f"| {kind} | {five} | {exit_text} | {duration_text} "
                f"| {scope}/{coverage} | {metrics} |"
            )
        lines.append("")
        # 每条验证的命令与证据摘录（表格放不下的事实）
        for kind in sorted(validations):
            entry = validations.get(kind)
            if not isinstance(entry, dict):
                continue
            command = str(entry.get("command") or "").strip()
            finished = str(entry.get("finishedAt") or "").strip()
            excerpt = _render_report_evidence_excerpt(entry)
            if not command and not excerpt:
                continue
            lines.append(f"### {kind}")
            if command:
                lines.append(f"- 命令: `{command}`")
            if finished:
                lines.append(f"- 完成时间: {finished}")
            if excerpt:
                lines.extend(["- 证据摘录:", ""])
                for evidence_line in excerpt.splitlines():
                    lines.append(f"  > {evidence_line}")
                lines.append("")
            else:
                lines.append("")
    else:
        lines.append("（ledger 无验证记录）")
        lines.append("")

    # --- 场景覆盖摘要 ---
    lines.extend(["## 场景覆盖摘要", ""])
    if scenario_summary is not None and not scenario_summary.get("skipped"):
        code = str(scenario_summary.get("code") or "")
        lines.append(f"检查结果: `{code}`")
        lines.append("")
        for field, label in (
            ("covered", "已覆盖（passed）"),
            ("missing", "缺失（未绑定 ledger 条目）"),
            ("unexecuted", "未执行（绑定但无 passed 收据）"),
            ("deferred", "移交后续阶段（ownerPhase 靠后）"),
        ):
            values = scenario_summary.get(field)
            if isinstance(values, list) and values:
                lines.append(f"- {label}: {', '.join(str(v) for v in values)}")
        detail = scenario_summary.get("executed")
        if isinstance(detail, list) and detail:
            lines.append(
                f"- 已执行收据: {len(detail)} 个场景"
            )
        attempts = scenario_summary.get("attempts")
        if isinstance(attempts, dict) and attempts:
            multi = {
                sid: vals
                for sid, vals in attempts.items()
                if isinstance(vals, list) and len(vals) > 1
            }
            if multi:
                lines.append(
                    "- 多次尝试: "
                    + ", ".join(
                        f"{sid}(attempt {','.join(str(v) for v in vals)})"
                        for sid, vals in sorted(multi.items())
                    )
                )
        lines.append("")
    else:
        lines.append("（无 scenario-manifest 或检查不适用）")
        lines.append("")

    # --- 五态状态总览 ---
    lines.extend(["## 五态状态总览", ""])
    if validations:
        lines.extend(["| 验证 | 五态 |", "|---|:---:|"])
        for kind in sorted(validations):
            entry = validations.get(kind)
            if isinstance(entry, dict):
                lines.append(f"| {kind} | {_render_report_five_state(entry)} |")
        lines.append("")
    else:
        lines.append("（无）")
        lines.append("")

    # --- 事件侧阶段摘要（补充 ledger 覆盖不到的执行事实） ---
    if isinstance(events_summary, dict):
        phases = events_summary.get("phases")
        if isinstance(phases, dict) and phases:
            lines.extend(["## 阶段执行摘要（events）", ""])
            lines.extend(["| 阶段 | 状态 |", "|---|---|"])
            for phase in sorted(phases):
                info = phases.get(phase)
                if isinstance(info, dict):
                    lines.append(
                        f"| {phase} | {info.get('status') or 'UNKNOWN'} |"
                    )
                else:
                    lines.append(f"| {phase} | {info} |")
            lines.append("")

    # --- 模型解读占位（模型只追加，不改写派生内容） ---
    lines.extend(
        [
            "## 解读（模型追加）",
            "",
            "<!-- 以下两段由模型在 render-report 输出后追加；派生部分勿改。 -->",
            "",
            "### 残余风险",
            "",
            "（待模型补充）",
            "",
            "### 下一步",
            "",
            "（待模型补充）",
            "",
        ]
    )
    return "\n".join(lines)


def _render_report_project_root(change_dir: Path) -> Path | None:
    """渲染用项目根：.harness 祖先；声明执行 worktree 时优先 worktree。

    与归档 find_project_root 同法（.harness/changes/<name> → 项目根），
    但产品文件可能在链接 worktree——infer_execution_project_root 只认
    可信 worktree 元数据，无声明时回退主检出。
    """
    resolved = change_dir.resolve()
    project_root = next(
        (ancestor.parent for ancestor in resolved.parents if ancestor.name == ".harness"),
        None,
    )
    if project_root is None:
        return None
    execution_root = infer_execution_project_root(change_dir)
    if execution_root is not None:
        return execution_root
    return project_root


def cmd_render_report(args: argparse.Namespace) -> int:
    """从 ledger+events 派生测试报告（只读渲染，可重建）。"""
    as_json = bool(args.json)
    change_dir = resolve_path(args.change_dir)
    if not change_dir.is_dir():
        return emit_error(
            f"change directory does not exist: {change_dir}",
            as_json=as_json,
            error_code="CHANGE_DIR_NOT_FOUND",
        )

    # 渲染报告与手写报告不并存（设计 §5）：已存在手写报告（无 generated
    # 标记）时拒绝渲染，避免同一 change 出现两份事实来源。
    existing_reports = _find_existing_test_reports(change_dir)
    handwritten = [
        path
        for path in existing_reports
        if not _report_has_generated_marker(path)
    ]
    if handwritten:
        return emit_error(
            "hand-written test report already exists; rendered and hand-written "
            "reports must not coexist in one change: "
            + ", ".join(str(p) for p in handwritten),
            as_json=as_json,
            error_code="HANDWRITTEN_REPORT_EXISTS",
            extra={
                "existingReports": [str(p) for p in handwritten],
                "recoveryAction": (
                    "旧 change 沿用手写报告完成；新 change 不要手写报告，"
                    "直接使用 render-report"
                ),
            },
        )

    import harness_events as he

    try:
        ledger, ledger_path = load_ledger(change_dir)
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        return emit_error(
            f"ledger unreadable: {exc}",
            as_json=as_json,
            error_code="LEDGER_UNREADABLE",
        )

    events_summary: dict[str, Any] | None = None
    events_file = he.events_path(change_dir)
    if events_file.is_file() and events_file.stat().st_size > 0:
        try:
            events = he.load_events_cached(events_file)
            events_summary = he.build_summary(change_dir, events)
        except (OSError, ValueError):
            events_summary = None

    # base/head：ledger v3 优先，其次 state-snapshot（与 diff-hash 同序）。
    # 项目根：.harness 祖先（与归档 find_project_root 同法）；变更声明
    # 执行 worktree 时优先 worktree 根（产品文件在那里）。
    project = _render_report_project_root(change_dir)
    base_commit = None
    head_commit = None
    if isinstance(ledger, dict):
        base_commit = str(ledger.get("baseCommit") or "").strip() or None
    if project is not None:
        head_commit = _git_text(project, "rev-parse", "--verify", "HEAD")
    if not base_commit and project is not None:
        base_commit = _immutable_change_base(change_dir, project) or None

    changed_files = _render_report_changed_files(
        project, base_commit, head_commit
    )
    scenario_summary = _render_report_scenario_summary(change_dir)

    rendered_at = dt.datetime.now().strftime("%Y-%m-%d %H:%M")
    body = _render_report_markdown(
        change_dir=change_dir,
        change_name=change_dir.name,
        rendered_at=rendered_at,
        ledger=ledger,
        events_summary=events_summary,
        changed_files=changed_files,
        base_commit=base_commit,
        head_commit=head_commit,
        scenario_summary=scenario_summary,
    )

    out_raw = getattr(args, "out", None)
    if _nonempty_str(out_raw):
        out_path = Path(str(out_raw)).expanduser()
        if not out_path.is_absolute():
            out_path = change_dir / out_path
    else:
        stamp = dt.datetime.now().strftime("%Y%m%d-%H%M")
        out_path = change_dir / "reports" / "test" / f"test-report-{stamp}.md"
    out_path = out_path.resolve()
    # 越界防护（与 scenario-receipt-template --out 同规则）。
    change_resolved = change_dir.resolve()
    cwd = Path.cwd().resolve()
    if not (
        _is_relative_to(out_path, change_resolved)
        or _is_relative_to(out_path, cwd)
    ):
        return emit_error(
            "--out 解析后越出项目目录: " + str(out_path),
            as_json=as_json,
            error_code="REPORT_OUT_OF_PROJECT",
        )
    try:
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(body, encoding="utf-8", newline="\n")
    except OSError as exc:
        return emit_error(
            f"cannot write report: {exc}",
            as_json=as_json,
            error_code="REPORT_WRITE_FAILED",
        )

    payload = {
        "ok": True,
        "action": "render-report",
        "path": str(out_path),
        "changeName": change_dir.name,
        "generator": REPORT_RENDER_VERSION,
        "changedFileCount": len(changed_files),
        "verificationCount": (
            len(
                [
                    k
                    for k, v in (ledger or {}).get("validations", {}).items()
                    if isinstance(v, dict)
                ]
            )
            if isinstance(ledger, dict)
            else 0
        ),
        "scenarioCoverageCode": (
            str(scenario_summary.get("code"))
            if isinstance(scenario_summary, dict)
            else None
        ),
        "ledgerPath": str(ledger_path) if ledger_path else None,
    }
    if as_json:
        sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\n")
    else:
        sys.stdout.write(f"{out_path}\n")
    return 0


def _find_existing_test_reports(change_dir: Path) -> list[Path]:
    """与 harness_archive.find_test_reports 同一 glob 集（本地实现避免环）。"""
    patterns = [
        "tests/test-report-*.md",
        "reports/test/test-report-*.md",
        "reports/test/*.md",
    ]
    found: list[Path] = []
    seen: set[Path] = set()
    for pattern in patterns:
        for path in sorted(change_dir.glob(pattern)):
            if path in seen:
                continue
            seen.add(path)
            found.append(path)
    return found


def _report_has_generated_marker(path: Path) -> bool:
    """frontmatter 含 generated: true 即渲染产物（只读头部探测）。"""
    try:
        with path.open("r", encoding="utf-8-sig") as handle:
            head = handle.read(2048)
    except OSError:
        return False
    if not head.startswith("---"):
        return False
    end = head.find("\n---", 3)
    if end == -1:
        return False
    frontmatter = head[:end]
    return re.search(r"^generated:\s*true\s*$", frontmatter, re.MULTILINE) is not None


def cmd_diff_hash(args: argparse.Namespace) -> int:
    as_json = bool(args.json)
    repo_raw = getattr(args, "repo", None)
    base = getattr(args, "base", None)
    change_dir_raw = getattr(args, "change_dir", None)
    try:
        change_dir = resolve_path(change_dir_raw) if change_dir_raw else None
        repo = Path(repo_raw).expanduser().resolve() if repo_raw else Path.cwd().resolve()
        # Historical skill text passed ``--repo .`` even when the active change
        # ran in a linked worktree.  Treat that shorthand like the default and
        # recover the authoritative execution root captured during Plan.
        repo_hint = str(repo_raw or "").strip().replace("\\", "/")
        if change_dir is not None and repo_hint in {"", ".", "./"}:
            inferred_repo = infer_execution_project_root(change_dir)
            if inferred_repo is not None:
                repo = inferred_repo
            elif declares_execution_worktree(change_dir):
                return emit_error(
                    "change metadata declares an execution worktree that is missing or belongs "
                    "to another repository",
                    as_json=as_json,
                    error_code="EXECUTION_WORKTREE_INVALID",
                )
        if change_dir is not None and _contract_is_v2(change_dir):
            resolved_base = str(base or "").strip()
            if not resolved_base:
                existing, _ = load_ledger(change_dir)
                if isinstance(existing, dict):
                    resolved_base = str(existing.get("baseCommit") or "").strip()
            if not resolved_base:
                resolved_base = _immutable_change_base(change_dir, repo)
            if not resolved_base:
                raise ValueError(
                    "immutable change base is missing; capture the Plan boundary "
                    "or pass --base"
                )
            detail = compute_ownership_diff(
                repo,
                base=resolved_base,
                change_dir=change_dir,
            )
            payload = {
                "ok": True,
                "action": "diff-hash",
                "identityScope": "change-ownership",
                "algorithmVersion": "ownership-content-changeset-1",
                "base": resolved_base,
                "head": _git_text(repo, "rev-parse", "--verify", "HEAD"),
                **detail,
            }
            emit_json(payload, as_json=as_json)
            return 0
        diff_hash, meta = compute_diff_hash(
            repo,
            base=base,
            change_dir=str(change_dir) if change_dir is not None else None,
        )
    except (OSError, RuntimeError, ValueError) as exc:
        return emit_error(str(exc), as_json=as_json)
    payload = {"ok": True, "action": "diff-hash", "diffHash": diff_hash}
    payload.update(meta)
    emit_json(payload, as_json=as_json)
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="harness_ledger.py",
        description="Compute inputsHash and manage verification-ledger reuse",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="emit machine-readable JSON on stdout",
    )
    sub = parser.add_subparsers(dest="command_name", required=True)

    # --json 也注册到每个子命令（default=SUPPRESS），使 --json 可放在子命令之后
    # （skill / Gate 命令均把 --json 放最后），且不会用子命令默认值覆盖
    # 在子命令之前传入的顶层 --json=True。
    shared_json = argparse.ArgumentParser(add_help=False)
    shared_json.add_argument(
        "--json",
        action="store_true",
        default=argparse.SUPPRESS,
    )

    p_hash = sub.add_parser("hash", parents=[shared_json], help="compute inputsHash for a file set")
    hash_files = p_hash.add_mutually_exclusive_group()
    hash_files.add_argument("--files", default=None, help="comma-separated source file paths")
    hash_files.add_argument("--files-from", default=None, help="UTF-8 newline-delimited source paths")
    p_hash.add_argument("--project", default=None, help="root for relative input paths")
    p_hash.set_defaults(func=cmd_hash)

    p_reuse = sub.add_parser("can-reuse", parents=[shared_json], help="decide whether a verification can be reused")
    p_reuse.add_argument("--change-dir", "--change", dest="change_dir", required=True)
    p_reuse.add_argument(
        "--verification",
        required=True,
        help="built-in or build-profile v3 verificationGraph target",
    )
    p_reuse.add_argument(
        "--files",
        default=None,
        help="comma-separated source file paths for current inputsHash",
    )
    p_reuse.add_argument(
        "--files-from",
        default=None,
        help="UTF-8 newline-delimited source paths",
    )
    p_reuse.add_argument(
        "--project",
        default=None,
        help="project root containing .harness/config/build-profile.json (for --profile-input)",
    )
    p_reuse.add_argument(
        "--profile-input",
        default=None,
        help="expand verificationInputs.<key> globs from build-profile as the file set; "
        "unitTestFull 最终门禁用此展开依赖闭包，禁止用仅含 staged 文件的 --files 冒充",
    )
    p_reuse.add_argument(
        "--scope",
        default=None,
        help="optional requested scope (unitTest coverage check)",
    )
    p_reuse.add_argument(
        "--command",
        default=None,
        help="optional command to compare against ledger entry",
    )
    p_reuse.add_argument(
        "--toolchain-hash",
        default=None,
        help="optional toolchain hash to compare against ledger entry (UT-017)",
    )
    p_reuse.add_argument(
        "--profile-hash",
        default=None,
        help="optional profile hash to compare against ledger entry (UT-017)",
    )
    p_reuse.add_argument(
        "--environment-hash",
        default=None,
        help="optional environment hash to compare against ledger entry (UT-017)",
    )
    p_reuse.add_argument(
        "--db-schema-hash",
        default=None,
        help="optional DB schema hash to compare against ledger entry (H-9)",
    )
    p_reuse.add_argument("--product-tree-hash", default=None)
    p_reuse.add_argument("--command-set-hash", default=None)
    p_reuse.add_argument("--lock-hash", default=None)
    p_reuse.add_argument(
        "--verbose",
        action="store_true",
        help="emit full payload (default: compact ok/reuse/code)",
    )
    p_reuse.set_defaults(func=cmd_can_reuse)

    p_record = sub.add_parser("record", parents=[shared_json], help="write validation result into ledger")
    p_record.add_argument("--change-dir", "--change", dest="change_dir", required=True)
    p_record.add_argument("--verification", required=True)
    p_record.add_argument("--status", required=True)
    p_record.add_argument("--command", required=True)
    p_record.add_argument(
        "--runner-command",
        default=None,
        help="actual launcher command; stored as execution metadata, not target identity",
    )
    p_record.add_argument("--exit-code", type=int, required=True)
    p_record.add_argument("--duration-ms", type=int, required=True)
    p_record_files = p_record.add_mutually_exclusive_group()
    p_record_files.add_argument("--files", default=None)
    p_record_files.add_argument("--files-from", default=None)
    p_record.add_argument("--evidence", required=True)
    p_record.add_argument(
        "--project",
        default=None,
        help="project root containing .harness/config/build-profile.json (for --profile-input)",
    )
    p_record.add_argument(
        "--profile-input",
        default=None,
        help="expand verificationInputs.<key> globs from build-profile as the file set",
    )
    p_record.add_argument(
        "--scope",
        default=None,
        help="optional scope (default module when absent on new entries)",
    )
    p_record.add_argument(
        "--coverage",
        default=None,
        help="optional coverage lattice value (incremental|module|module-am|full); derived when absent",
    )
    p_record.add_argument("--toolchain-hash", default=None)
    p_record.add_argument("--profile-hash", default=None)
    p_record.add_argument("--environment-hash", default=None)
    p_record.add_argument("--db-schema-hash", default=None)
    p_record.add_argument("--deploy-artifact", default=None, help="package: built artifact path")
    p_record.add_argument("--artifact-hash", default=None, help="package: artifact sha256")
    p_record.add_argument(
        "--tests-executed",
        type=lambda v: str(v).lower() in ("1", "true", "yes", "y"),
        default=False,
        help="package: whether tests ran in this package lifecycle",
    )
    p_record.add_argument(
        "--tests-reused-from",
        default=None,
        help="package: prior verifications reused when testsExecuted=false",
    )
    p_record.add_argument(
        "--metrics-json",
        default=None,
        help='structured counts, e.g. \'{"run":155,"failures":0}\' or \'{"total":3,"passed":3}\'',
    )
    p_record.add_argument(
        "--metrics-file",
        default=None,
        help="UTF-8 JSON file containing typed verification metrics",
    )
    p_record.add_argument(
        "--base-commit",
        default=None,
        help="ledger v3: base commit for identity (default: existing ledger value, else HEAD)",
    )
    p_record.add_argument(
        "--diff-hash",
        default=None,
        help="ledger v3: precomputed ownership diff hash (default: existing, else computed)",
    )
    p_record.add_argument(
        "--applicability",
        choices=("APPLICABLE", "NOT_APPLICABLE"),
        default=None,
        help="ledger v3: applicability of this verification to the change",
    )
    p_record.add_argument(
        "--applicability-reason",
        default=None,
        help="ledger v3: scope reason (required when --applicability NOT_APPLICABLE)",
    )
    p_record.add_argument(
        "--scenario-ids",
        default=None,
        help="comma-separated scenario IDs from scenario-manifest.json to bind to this entry",
    )
    p_record.add_argument(
        "--scenario-receipt-file",
        default=None,
        help="UTF-8 JSON runner receipt proving declared/selected/collected/executed tests",
    )
    p_record.add_argument(
        "--verbose",
        action="store_true",
        help="emit full payload (default: compact ok/action/verification/status)",
    )
    p_record.set_defaults(func=cmd_record)

    p_rfr = sub.add_parser(
        "record-from-receipt",
        parents=[shared_json],
        help="write validation result from an exec result receipt (batch 2 WI-1b)",
    )
    p_rfr.add_argument("--change-dir", "--change", dest="change_dir", required=True)
    p_rfr.add_argument("--receipt", required=True, help="exec result receipt path")
    p_rfr.add_argument(
        "--verification",
        required=True,
        help="verification kind recorded into the ledger (e.g. unitTest)",
    )
    p_rfr.add_argument(
        "--project",
        default=None,
        help="project root containing .harness/config/build-profile.json (for --profile-input)",
    )
    p_rfr.add_argument(
        "--profile-input",
        default=None,
        help="expand verificationInputs.<key> globs from build-profile as the file set",
    )
    p_rfr.add_argument(
        "--files",
        default=None,
        help="comma-separated explicit file paths (targeted runs; bypasses --profile-input)",
    )
    p_rfr.add_argument(
        "--scenario-ids",
        default=None,
        help="comma-separated scenario IDs from scenario-manifest.json to bind to this entry (B2-1)",
    )
    p_rfr.add_argument(
        "--scenario-receipt-file",
        default=None,
        help="UTF-8 JSON runner receipt proving declared/selected/collected/executed tests (B2-1)",
    )
    p_rfr.add_argument(
        "--verbose",
        action="store_true",
        help="emit full payload (default: compact ok/action/verification/status)",
    )
    p_rfr.set_defaults(func=cmd_record_from_receipt)

    p_imp = sub.add_parser(
        "import-ci-evidence",
        parents=[shared_json],
        help="import host CI verification evidence from a signed receipt (WI-3.2)",
    )
    p_imp.add_argument("--change-dir", "--change", dest="change_dir", required=True)
    p_imp.add_argument(
        "--receipt", required=True, help="ci-evidence-receipt.json path"
    )
    p_imp.add_argument(
        "--verification",
        required=True,
        help="ledger verification to satisfy (only compile|unitTestFull importable)",
    )
    p_imp.add_argument(
        "--job", required=True, help="CI job name inside the receipt"
    )
    p_imp.add_argument(
        "--step", required=True, help="CI step name inside the job"
    )
    p_imp.add_argument(
        "--shard",
        default=None,
        help=(
            "shard identity I/N (e.g. 1/3); shards aggregate to full coverage "
            "only when all share runId+productTree, else incremental (R5)"
        ),
    )
    p_imp.add_argument(
        "--project",
        default=None,
        help="project root used for tree hash and origin remote checks",
    )
    p_imp.add_argument(
        "--verbose",
        action="store_true",
        help="emit full payload (default: compact)",
    )
    p_imp.set_defaults(func=cmd_import_ci_evidence)

    p_diff = sub.add_parser(
        "diff-hash",
        parents=[shared_json],
        help="compute commit-invariant byte-level diff hash for a repo",
    )
    p_diff.add_argument(
        "--repo",
        default=None,
        help="repo root (default: change worktree when recorded, otherwise cwd)",
    )
    p_diff.add_argument("--base", default=None, help="base commit (default: root commit)")
    p_diff.add_argument(
        "--change-dir",
        default=None,
        help="change directory whose evidence/test-tracking.json contributes ignored tests",
    )
    p_diff.set_defaults(func=cmd_diff_hash)

    p_receipt = sub.add_parser(
        "scenario-receipt-template",
        parents=[shared_json],
        help=(
            "emit a schema-v2 scenario execution receipt skeleton "
            "pre-filled from meta/scenario-manifest.json"
        ),
    )
    p_receipt.add_argument("--change-dir", "--change", dest="change_dir", required=True)
    p_receipt.add_argument(
        "--scenario-ids",
        required=True,
        help="comma-separated scenario IDs to include (must exist in the manifest)",
    )
    p_receipt.add_argument(
        "--runner",
        required=True,
        help="test runner name recorded in the receipt (e.g. vitest / maven-surefire)",
    )
    p_receipt.add_argument("--runner-version", default=None)
    p_receipt.add_argument("--attempt", type=int, default=1)
    p_receipt.add_argument(
        "--status",
        default="PASSED",
        help="per-test status written into executed[] (default: PASSED)",
    )
    p_receipt.add_argument(
        "--out",
        default=None,
        help=(
            "write the skeleton to this path instead of stdout; relative paths "
            "resolve against --change-dir (same rule as --scenario-receipt-file)"
        ),
    )
    p_receipt.set_defaults(func=cmd_scenario_receipt_template)

    p_render = sub.add_parser(
        "render-report",
        parents=[shared_json],
        help=(
            "derive the test report from ledger+events "
            "(batch 2 WI-4a; read-only render, rebuildable)"
        ),
    )
    p_render.add_argument("--change-dir", "--change", dest="change_dir", required=True)
    p_render.add_argument(
        "--out",
        default=None,
        help=(
            "write the report to this path instead of "
            "reports/test/test-report-YYYYMMDD-HHmm.md; relative paths "
            "resolve against --change-dir"
        ),
    )
    p_render.set_defaults(func=cmd_render_report)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    return int(args.func(args))


if __name__ == "__main__":
    raise SystemExit(main())