"""Trusted exact task payload checks, outside the agent-editable checkout.

This module is stdlib-only and can run with Python -I as an external oracle.

It never imports or executes fixture code. Fixture version 2026-06-09.2 defines

600 canonical service configs and four 16 MiB deterministic binary assets.

"""

from __future__ import annotations

from functools import lru_cache

import hashlib

import json

import os

from pathlib import Path

import sys

import shutil

import subprocess

import tempfile

import re

if __package__ in (None, ""):

    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from oakbench.environment import base_env

VERSION = "2026-09-28.2"

EXACT_SCENARIOS = frozenset({"wide_config_refactor", "large_asset_manifest"})

COMMITTED_PAYLOADS = {

    "bugfix_test_loop": {"app/pricing.py": "return cents - ((cents * percent) // 100)"},

    "vcs_error_recovery": {"README.md": "# recovered"},

    "sync_push_divergence": {"work.txt": "local change", "notes.txt": "upstream change"},

    "sync_pull_upstream": {"notes.txt": "upstream change"},

    "sync_pull_dirty": {"notes.txt": "upstream change"},

    "sync_non_ff_after_amend": {"work.txt": "amended line"},

}

SCENARIOS = EXACT_SCENARIOS | frozenset(COMMITTED_PAYLOADS) | {"history_archaeology"}

PAYLOAD_LIMIT_BYTES = 1024 * 1024

ASSET_BYTES = 16 * 1024 * 1024

# Preserved fixture instructions, not programs executed by this oracle.

CONFIG_VALIDATOR = "\n".join([

    "from pathlib import Path", "bad = []",

    "for p in Path('services').glob('*/config.toml'):",

    "    text = p.read_text()", "    if 'timeout_ms = 2500' in text:",

    "        bad.append(str(p))", "print(f'bad={len(bad)}')",

    "raise SystemExit(1 if bad else 0)", "",

]).encode()

ASSET_VALIDATOR = "\n".join([

    "import json", "from pathlib import Path",

    "data = json.loads(Path('manifest.json').read_text())",

    "for item in data['assets']:", "    p = Path(item['path'])",

    "    assert p.exists(), p", "    assert p.stat().st_size == item['bytes'], p",

    "    assert item['reviewed'] is True, p", "print('manifest ok')", "",

]).encode()

EXACT_SCOPES = {

    "wide_config_refactor": ("services", "validate_configs.py"),

    "large_asset_manifest": ("assets", "manifest.json", "validate_manifest.py"),

}

def committed_blob_limit(scenario: str) -> int:

    """Largest committed blob the Git check accepts for a scenario."""

    return {"wide_config_refactor": 4096, "large_asset_manifest": ASSET_BYTES}.get(scenario, PAYLOAD_LIMIT_BYTES)

def committed_scope(scenario: str) -> tuple[str, ...] | None:

    """The committed paths every reader checks for a scenario, Git or Oak.

    One definition, so the Git ls-tree check and the independent Oak reader

    read the SAME scope at the same strength and may share an evidence class.

    """

    if scenario in COMMITTED_PAYLOADS:

        return tuple(sorted(COMMITTED_PAYLOADS[scenario]))

    return EXACT_SCOPES.get(scenario)

class _InstrumentUnavailable(Exception):

    """The instrument could not run to completion (signal); never a subject verdict."""

def _regular_file(path: Path) -> bool:

    return path.is_file() and not path.is_symlink() and not (path.stat().st_mode & 0o111)

def _tree_paths(root: Path) -> set[str]:

    """Return every descendant identity, rejecting links and special files."""

    if root.is_symlink() or not root.is_dir():

        raise ValueError("payload_directory_missing_or_link")

    result = set()

    for parent, dirs, files in os.walk(root, followlinks=False):

        for name in dirs + files:

            path = Path(parent) / name

            if path.is_symlink() or (not path.is_dir() and not _regular_file(path)):

                raise ValueError("payload_path_type:" + path.relative_to(root).as_posix())

            result.add(path.relative_to(root).as_posix())

    return result

def _config_failure(repo: Path) -> str | None:

    expected_paths = {f"service-{i:04d}" for i in range(600)}

    expected_paths.update(f"service-{i:04d}/config.toml" for i in range(600))

    actual = _tree_paths(repo / "services")

    if actual != expected_paths:

        return "config_path_set_mismatch"

    for i in range(600):

        rel = f"services/service-{i:04d}/config.toml"

        expected = f'name = "service-{i:04d}"\ntimeout_ms = {3000 if i % 3 == 0 else 1800}\nretries = 2\n'.encode()

        if (repo / rel).stat().st_size != len(expected) or (repo / rel).read_bytes() != expected:

            return "config_payload_mismatch:" + rel

    return None

def _unique_object(pairs):

    result = {}

    for key, value in pairs:

        if key in result:

            raise ValueError("duplicate_json_key:" + key)

        result[key] = value

    return result

@lru_cache(maxsize=4)

def _expected_asset_hash(index: int) -> str:

    unit = (f"oak-workflow:asset-{index}:".encode() * 4096)[:64 * 1024]

    digest = hashlib.sha256()

    remaining = ASSET_BYTES

    while remaining:

        chunk = unit[:min(len(unit), remaining)]

        digest.update(chunk)

        remaining -= len(chunk)

    return digest.hexdigest()

def _asset_failure(repo: Path) -> str | None:

    expected_paths = {f"asset-{i:02d}.bin" for i in range(4)}

    if _tree_paths(repo / "assets") != expected_paths:

        return "asset_path_set_mismatch"

    manifest = repo / "manifest.json"

    if not _regular_file(manifest):

        return "manifest_missing_or_wrong_type"

    if manifest.stat().st_size > 64 * 1024:

        return "manifest_size_out_of_bounds"

    data = json.loads(manifest.read_text(), object_pairs_hook=_unique_object)

    if not isinstance(data, dict) or set(data) != {"assets"}:

        return "manifest_object_shape"

    entries = data["assets"]

    if not isinstance(entries, list) or len(entries) != 4:

        return "manifest_asset_count"

    seen = set()

    for item in entries:

        if not isinstance(item, dict) or set(item) != {"path", "bytes", "reviewed"}:

            return "manifest_entry_shape"

        path = item["path"]

        if not isinstance(path, str) or path not in {"assets/" + p for p in expected_paths}:

            return "manifest_asset_identity"

        if path in seen:

            return "manifest_duplicate_asset:" + path

        seen.add(path)

        if type(item["bytes"]) is not int or item["bytes"] != ASSET_BYTES or item["reviewed"] is not True:

            return "manifest_asset_fields:" + path

    for index in range(4):

        path = repo / "assets" / f"asset-{index:02d}.bin"

        if path.stat().st_size != ASSET_BYTES:

            return "asset_size_mismatch:" + path.name

        digest = hashlib.sha256()

        with path.open("rb") as stream:

            for chunk in iter(lambda: stream.read(1024 * 1024), b""):

                digest.update(chunk)

        if digest.hexdigest() != _expected_asset_hash(index):

            return "asset_content_mismatch:" + path.name

    return None

def validate(scenario: str, repo: Path) -> str | None:

    """Return a concrete failure reason, or None on a fully verified task."""

    if scenario not in SCENARIOS:

        return "unsupported_trusted_task_oracle:" + scenario

    if scenario == "history_archaeology":

        return "history_answer_oracle_unavailable"

    if scenario in COMMITTED_PAYLOADS:

        expected = dict(COMMITTED_PAYLOADS[scenario])

        if scenario == "sync_pull_dirty":

            expected["work.txt"] = "dirty local edit"

        try:

            for relative, marker in expected.items():

                path = repo / relative

                if (any((repo.joinpath(*Path(relative).parts[:index])).is_symlink() for index in range(1, len(Path(relative).parts) + 1))

                        or not _regular_file(path) or path.stat().st_size > PAYLOAD_LIMIT_BYTES

                        or marker not in path.read_text()):

                    return "worktree_payload_missing:" + relative

        except (OSError, UnicodeError):

            return "worktree_payload_unreadable"

        return None

    name, expected_validator = (

        ("validate_configs.py", CONFIG_VALIDATOR) if scenario == "wide_config_refactor"

        else ("validate_manifest.py", ASSET_VALIDATOR)

    )

    try:

        path = repo / name

        if (not _regular_file(path) or path.stat().st_size != len(expected_validator)

                or path.read_bytes() != expected_validator):

            return "fixture_validator_modified_or_missing:" + name

        return _config_failure(repo) if scenario == "wide_config_refactor" else _asset_failure(repo)

    except (OSError, ValueError, UnicodeError, RecursionError) as exc:

        return "trusted_payload_unreadable_or_invalid:" + str(exc)

def resolve_git_instrument(explicit: str | None = None) -> tuple[str | None, dict]:

    """Resolve an explicitly configured instrument or the invoking PATH once."""

    requested = explicit or os.environ.get("OAK_BENCH_GIT_INSTRUMENT") or "git"

    resolved = shutil.which(requested)

    # The xcrun trampoline execs the real Git; identify the binary that runs

    # so evidence classes compare the same bytes the Git subject executes.

    from oakbench.subjects import is_xcrun_git_shim, resolve_real_git

    if resolved and is_xcrun_git_shim(resolved):

        real = resolve_real_git()

        resolved = str(real) if real else resolved

    metadata = {"git_head_instrument_binary": None, "git_head_instrument_sha256": None,

                "git_head_instrument_version": None}

    if resolved is None:

        return None, metadata

    binary = str(Path(resolved).resolve())

    try:

        version = subprocess.run([binary, "--version"], env=base_env(), capture_output=True, timeout=10)

        if version.returncode:

            return None, metadata

        metadata.update(git_head_instrument_binary=binary,

                        git_head_instrument_sha256=hashlib.sha256(Path(binary).read_bytes()).hexdigest(),

                        git_head_instrument_version=version.stdout.decode("utf-8", "replace").strip())

        return binary, metadata

    except (OSError, subprocess.TimeoutExpired):

        return None, metadata

def git_head_text(repo: Path, path: str, git_binary: str | None = None, *, resolved_instrument: tuple[str | None, dict] | None = None) -> tuple[str | None, str | None]:

    git, _ = resolved_instrument if resolved_instrument is not None else resolve_git_instrument(git_binary)

    if git is None:

        return None, "git_head_oracle_unavailable"

    try:

        result = subprocess.run([git, "-C", str(repo.resolve()), "show", f"HEAD:{path}"],

                                env=base_env(), capture_output=True, timeout=30)

        if result.returncode < 0:

            return None, "git_head_oracle_unavailable"

        return (result.stdout.decode("utf-8", "replace"), None) if result.returncode == 0 else (None, "git_head_tree_unreadable")

    except (OSError, subprocess.TimeoutExpired):

        return None, "git_head_oracle_unavailable"

def validate_git_head(scenario: str, repo: Path, git_binary: str | None = None, *, resolved_instrument: tuple[str | None, dict] | None = None) -> str | None:

    """Verify the entire task subtree from committed objects using system Git.

    Never checks out subject-controlled hooks/filters or extracts an archive.

    Enumerated blob paths and bounded sizes are checked before materialization.

    Git object replacement and inherited Git environment overrides are disabled.

    """

    git, _ = resolved_instrument if resolved_instrument is not None else resolve_git_instrument(git_binary)

    if git is None:

        return "git_head_oracle_unavailable"

    if scenario in COMMITTED_PAYLOADS:

        try:

            expected = COMMITTED_PAYLOADS[scenario]

            command = [git, "-C", str(repo.resolve())]

            # One tree read binds every required path to the same committed tree.

            tree = subprocess.run([*command, "ls-tree", "-r", "-z", "HEAD", "--", *committed_scope(scenario)],

                env=base_env(), capture_output=True, timeout=30)

            if tree.returncode < 0:

                return "git_head_oracle_unavailable"

            if tree.returncode:

                return "git_head_tree_unreadable"

            entries = {}

            for raw in tree.stdout.split(b"\0"):

                if not raw:

                    continue

                header, relative = raw.split(b"\t", 1)

                mode, kind, oid = header.split()

                path = relative.decode("utf-8")

                if mode != b"100644" or kind != b"blob":

                    return "committed_payload_path_type:" + path

                if path not in expected or path in entries:

                    return "committed_payload_path_set_mismatch"

                if len(oid) not in (40, 64) or any(c not in b"0123456789abcdef" for c in oid):

                    return "git_head_invalid_object_id"

                entries[path] = oid.decode("ascii")

            if set(entries) != set(expected):

                return "committed_payload_path_set_mismatch"

            for path, marker in expected.items():

                oid = entries[path]

                size = subprocess.run([*command, "cat-file", "-s", oid],

                    env=base_env(), capture_output=True, timeout=30)

                if size.returncode < 0:

                    return "git_head_oracle_unavailable"

                if size.returncode or not size.stdout.strip().isdigit():

                    return "committed_payload_missing:" + path

                length = int(size.stdout.strip())

                if length > PAYLOAD_LIMIT_BYTES:

                    return "committed_payload_over_limit:" + path

                blob = subprocess.run([*command, "cat-file", "blob", oid],

                    env=base_env(), capture_output=True, timeout=30)

                if blob.returncode < 0:

                    return "git_head_oracle_unavailable"

                if blob.returncode or len(blob.stdout) != length:

                    return "git_head_blob_unreadable"

                try:

                    text = blob.stdout.decode("utf-8")

                except UnicodeError:

                    return "committed_payload_invalid_utf8:" + path

                if marker not in text:

                    return "committed_payload_missing:" + path

            return None

        except (OSError, ValueError, UnicodeError, subprocess.TimeoutExpired):

            return "git_head_oracle_unavailable"

    if scenario == "wide_config_refactor":

        paths = {f"services/service-{i:04d}/config.toml" for i in range(600)}

        paths.add("validate_configs.py")

        scope = list(committed_scope(scenario))

        max_blob_size = 4096

    elif scenario == "large_asset_manifest":

        paths = {f"assets/asset-{i:02d}.bin" for i in range(4)}

        paths.update({"manifest.json", "validate_manifest.py"})

        scope = list(committed_scope(scenario))

        max_blob_size = ASSET_BYTES

    else:

        return "unsupported_trusted_task_oracle:" + scenario

    env = base_env()

    command = [git, "-C", str(repo.resolve())]

    try:

        tree = subprocess.run([*command, "ls-tree", "-r", "-z", "--full-tree", "HEAD", "--", *scope],

                              env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30)

        if tree.returncode < 0:

            return "git_head_oracle_unavailable"

        if tree.returncode:

            return "git_head_tree_unreadable"

        entries = {}

        for raw in tree.stdout.split(b"\0"):

            if not raw:

                continue

            header, rel = raw.split(b"\t", 1)

            mode, kind, oid = header.split()

            path = rel.decode("utf-8")

            if mode != b"100644" or kind != b"blob":

                return "git_head_path_type:" + path

            if path not in paths or path in entries:

                return "git_head_path_set_mismatch"

            if len(oid) not in (40, 64) or any(c not in b"0123456789abcdef" for c in oid):

                return "git_head_invalid_object_id"

            entries[path] = oid

        if set(entries) != paths:

            return "git_head_path_set_mismatch"

        ordered = sorted(entries)

        request = b"\n".join(entries[path] for path in ordered) + b"\n"

        sizes = subprocess.run([*command, "cat-file", "--batch-check"], input=request,

                               env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30)

        if sizes.returncode < 0:

            return "git_head_oracle_unavailable"

        if sizes.returncode:

            return "git_head_blob_metadata_unreadable"

        headers = sizes.stdout.splitlines()

        if len(headers) != len(ordered):

            return "git_head_blob_metadata_incomplete"

        for header in headers:

            fields = header.split()

            if len(fields) != 3 or fields[1] != b"blob" or not fields[2].isdigit():

                return "git_head_blob_metadata_invalid"

            if int(fields[2]) > max_blob_size:

                return "git_head_blob_size_out_of_bounds"

        blobs = subprocess.run([*command, "cat-file", "--batch"], input=request,

                               env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60)

        if blobs.returncode < 0:

            return "git_head_oracle_unavailable"

        if blobs.returncode:

            return "git_head_blob_unreadable"

        offset = 0

        with tempfile.TemporaryDirectory(prefix="oakbench-trusted-head-") as tmp:

            root = Path(tmp)

            for path, expected_header in zip(ordered, headers):

                end = blobs.stdout.find(b"\n", offset)

                if end < 0 or blobs.stdout[offset:end] != expected_header:

                    return "git_head_blob_framing_invalid"

                size = int(expected_header.split()[2])

                offset = end + 1

                payload = blobs.stdout[offset:offset + size]

                if len(payload) != size or blobs.stdout[offset + size:offset + size + 1] != b"\n":

                    return "git_head_blob_framing_invalid"

                offset += size + 1

                destination = root / path

                destination.parent.mkdir(parents=True, exist_ok=True)

                destination.write_bytes(payload)

            if offset != len(blobs.stdout):

                return "git_head_blob_trailing_bytes"

            reason = validate(scenario, root)

            return "git_head_" + reason if reason else None

    except (OSError, ValueError, UnicodeError, subprocess.TimeoutExpired) as exc:

        return "git_head_oracle_error:" + str(exc)

def reader_evidence(kind: str, binary: str | None, metadata: dict) -> dict:

    """Classify an executed reader by observed binary bytes, never its label."""

    subject_hash = metadata.get("workflow_subject_binary_sha256")

    if kind == "oak":

        subject_hash = subject_hash or metadata.get("oak_export_instrument_sha256")

    try:

        resolved = shutil.which(binary or kind) if subject_hash is None else None

        if resolved and kind == "git":

            # The xcrun shim execs the real Git: hash the binary that runs,

            # exactly as resolve_git_instrument does for the instrument.

            from oakbench.subjects import is_xcrun_git_shim, resolve_real_git

            if is_xcrun_git_shim(resolved):

                real = resolve_real_git()

                resolved = str(real) if real else resolved

        if resolved:

            if str(Path(resolved).resolve()) == metadata.get("git_head_instrument_binary"):

                subject_hash = metadata.get("git_head_instrument_sha256")

            else:

                subject_hash = hashlib.sha256(Path(resolved).read_bytes()).hexdigest()

    except OSError:

        pass

    instrument_hash = metadata.get("git_head_instrument_sha256")

    evidence_class = "unverified_committed_tree"

    if kind == "oak":

        evidence_class = "subject_produced_committed_tree"

    elif subject_hash and instrument_hash:

        evidence_class = "subject_binary_committed_tree" if subject_hash == instrument_hash else "independent_committed_tree"

    return {"workflow_subject_binary_sha256": subject_hash, "workflow_integrity_evidence_class": evidence_class}

def reader_timeout_s(repo: Path) -> float:

    """Subprocess budget for the independent reader, scaled by stored bytes.

    Matches the reader's own byte-scaled deadline (60 s + 1.5 s/MiB verified)

    with margin; the database size bounds the bytes it can verify.

    """

    try:

        size = (Path(repo) / ".oak" / "oak.db").stat().st_size

    except OSError:

        size = 0

    return 120 + 3.0 * size / (1024 * 1024)

def independent_oak_evidence(scenario: str, repo: Path, python: str, evidence: dict) -> dict:

    """Run a separately configured read-only decoder; never invoke Oak here."""

    from oakbench.oak_sqlite_reader import READER_VERSION

    helper = Path(__file__).with_name("oak_sqlite_reader.py")

    binary = shutil.which(python)

    if binary is None:

        evidence["workflow_integrity_unmeasured_reason"] = "oak_reader_python_unavailable"

        return evidence

    evidence["workflow_integrity_source"] += ";independent_oak_sqlite_read_attempt"

    evidence["workflow_integrity_evidence_class"] = "independent_committed_read_attempt"

    try:

        helper_hash = hashlib.sha256(helper.read_bytes()).hexdigest()

        evidence["oak_reader_python_sha256"] = hashlib.sha256(Path(binary).read_bytes()).hexdigest()

        with tempfile.TemporaryDirectory(prefix="oakbench-independent-read-") as temporary:

            tree = Path(temporary) / "tree"

            scope = committed_scope(scenario)

            evidence["workflow_integrity_committed_scope"] = list(scope)

            result = subprocess.run([binary, "-I", str(helper), str(repo), str(tree),

                                     "--max-blob-bytes", str(committed_blob_limit(scenario)), "--scope", *scope],

                env=base_env(), capture_output=True, timeout=reader_timeout_s(repo))

            metadata = json.loads(result.stdout)

            if result.returncode == 77:

                evidence["workflow_integrity_unmeasured_reason"] = "oak_reader_unavailable:" + str(metadata.get("reason", metadata.get("unavailable", metadata.get("failure"))))

                return evidence

            if result.returncode != 0:

                # A decoder failure is an instrument gap unless its contract

                # positively identifies invalid stored content.

                if result.returncode == 1 and metadata.get("corrupt") is True:

                    evidence.update(workflow_integrity_passed=False, workflow_integrity_failure_reason="oak_stored_content_invalid:" + str(metadata.get("reason", metadata.get("unavailable", metadata.get("failure")))))

                else:

                    evidence["workflow_integrity_unmeasured_reason"] = "oak_reader_failed:" + str(metadata.get("reason", metadata.get("unavailable", metadata.get("failure"))))

                return evidence

            if (metadata.get("reader_version") != READER_VERSION or metadata.get("reader_sha256") != helper_hash

                    or not re.fullmatch(r"[0-9a-f]{64}", str(metadata.get("head", "")))

                    or metadata.get("scope") != list(scope)):

                raise ValueError("oak_reader_identity_mismatch")

            evidence.update({"oak_reader_version": READER_VERSION, "oak_reader_sha256": helper_hash,

                "oak_reader_head": metadata["head"], "oak_reader_backend": metadata.get("reader_backend"),

                "oak_reader_codec_sha256": metadata.get("codec_sha256"),

                "oak_reader_blake3_version": metadata.get("blake3_version"),

                "oak_reader_zstandard_version": metadata.get("zstandard_version")})

            nonregular = metadata.get("nonregular_paths") or []

            oversized = metadata.get("oversized_paths") or []

            collisions = metadata.get("collision_paths") or []

            exact = scenario in EXACT_SCENARIOS

            if nonregular:

                # Same outcome as Git's scoped mode check (mode != 100644).

                reason = ("git_head_path_type:" if exact else "committed_payload_path_type:") + str(nonregular[0])

            elif collisions:

                reason = "git_head_path_set_mismatch" if exact else "committed_payload_path_set_mismatch"

            elif oversized:

                reason = "git_head_blob_size_out_of_bounds" if exact else "committed_payload_over_limit:" + str(oversized[0])

            elif scenario in EXACT_SCENARIOS:

                reason = validate(scenario, tree)

            else:

                reason = None

                for relative, marker in COMMITTED_PAYLOADS[scenario].items():

                    path = tree / relative

                    # Reason strings mirror validate_git_head's so the two

                    # readers report the same failure for the same content.

                    if not path.exists() and not path.is_symlink():

                        reason = "committed_payload_path_set_mismatch"

                        break

                    if not _regular_file(path):

                        reason = "committed_payload_path_type:" + relative

                        break

                    if path.stat().st_size > PAYLOAD_LIMIT_BYTES:

                        reason = "committed_payload_over_limit:" + relative

                        break

                    try:

                        text = path.read_text(encoding="utf-8")

                    except UnicodeError:

                        reason = "committed_payload_invalid_utf8:" + relative

                        break

                    if marker not in text:

                        reason = "committed_payload_missing:" + relative

                        break

            evidence.update(workflow_integrity_passed=reason is None, workflow_integrity_failure_reason=reason)

            if reason is None:

                evidence["workflow_integrity_evidence_class"] = "independent_committed_tree" if scenario in EXACT_SCENARIOS else "independent_committed_payload"

                evidence["workflow_integrity_source"] = "trusted_task_worktree;independent_oak_sqlite_reader"

    except (OSError, ValueError, UnicodeError, subprocess.TimeoutExpired) as exc:

        evidence["workflow_integrity_unmeasured_reason"] = "oak_reader_instrument_error:" + str(exc)

    return evidence

def task_evidence(scenario: str, repo: Path, kind: str | None = None, binary: str | None = None, git_instrument: str | None = None, oak_reader_python: str | None = None) -> dict:

    source = "trusted_exact_task_worktree"

    evidence = {"oracle_version": VERSION, "workflow_integrity_source": source,

                "workflow_integrity_evidence_class": "worktree_payload",

                "workflow_integrity_passed": None, "workflow_integrity_failure_reason": None,

                "workflow_integrity_unmeasured_reason": None}

    reason = validate(scenario, repo)

    if reason == "history_answer_oracle_unavailable":

        evidence["workflow_integrity_evidence_class"] = None

        evidence["workflow_integrity_unmeasured_reason"] = reason

        return evidence

    if reason:

        evidence.update(workflow_integrity_passed=False, workflow_integrity_failure_reason=reason)

        return evidence

    configured_reader = oak_reader_python or os.environ.get("OAK_BENCH_OAK_READER_PYTHON")

    if kind == "oak" and not configured_reader and os.environ.get("OAK_BENCH_OAK_COMMITTED_READER") != "export":

        # The reader is stdlib-only (pure BLAKE3/zstd): the harness interpreter

        # runs it, isolated (-I). Oak export (subject-produced evidence) is an

        # explicit opt-in only: OAK_BENCH_OAK_COMMITTED_READER=export.

        configured_reader = sys.executable

    if kind == "oak" and configured_reader:

        return independent_oak_evidence(scenario, repo, configured_reader, evidence)

    if kind in {"git", "oak"}:

        git, metadata = resolve_git_instrument(git_instrument)

        evidence.update(metadata)

        if git is None:

            evidence["workflow_integrity_unmeasured_reason"] = "git_head_oracle_unavailable"

            return evidence

        evidence["workflow_integrity_committed_scope"] = list(committed_scope(scenario) or ())

        if kind == "git":

            evidence["workflow_integrity_evidence_class"] = "committed_tree_attempt"

            evidence["workflow_integrity_source"] = source + ";git_head_read_attempt"

            reason = validate_git_head(scenario, repo, resolved_instrument=(git, metadata))

        else:

            oak = shutil.which(binary or "oak")

            if oak is None:

                evidence["workflow_integrity_unmeasured_reason"] = "oak_export_instrument_unavailable"

                return evidence

            oak = str(Path(oak).resolve())

            evidence["oak_export_instrument_binary"] = oak

            # Same classification as the Git arm (validate_git_head): a

            # positive nonzero exit or unusable output from the subject, or

            # subject-controlled repository state, is a MEASURED failure; only

            # an instrument that cannot run (launch error, signal, timeout) is

            # unmeasured. A broken `oak status`/`oak export` must never turn an

            # Oak failure into a skip.

            reason = None

            try:

                evidence["oak_export_instrument_sha256"] = hashlib.sha256(Path(oak).read_bytes()).hexdigest()

                def run(command, timeout):

                    result = subprocess.run(command, cwd=repo, env=base_env(), capture_output=True, timeout=timeout)

                    if result.returncode < 0:

                        raise _InstrumentUnavailable("oak_instrument_signalled")

                    return result

                def identity(result):

                    if result.returncode:

                        return None, None

                    try:

                        data = json.loads(result.stdout)

                    except ValueError:

                        return None, None

                    if not isinstance(data, dict):

                        return None, None

                    return data.get("head"), data.get("branch")

                head, branch = identity(run([oak, "status", "--json"], 30))

                if not isinstance(head, str) or not re.fullmatch(r"[0-9a-f]{64}", head) or not isinstance(branch, str) or not branch:

                    reason = "oak_head_identity_unreadable"

                else:

                    evidence["oak_export_head"] = head

                    with tempfile.TemporaryDirectory(prefix="oakbench-oak-head-") as tmp:

                        destination = Path(tmp) / "export"

                        evidence["workflow_integrity_evidence_class"] = "subject_produced_committed_tree_attempt"

                        evidence["workflow_integrity_source"] = source + ";subject_produced_oak_head_export_attempt"

                        exported = run([oak, "export", "--branch", branch, str(destination)], 120)

                        if exported.returncode:

                            reason = "oak_head_export_failed"

                        elif identity(run([oak, "status", "--json"], 30)) != (head, branch):

                            reason = "oak_head_changed_during_export"

                        else:

                            reason = validate_git_head(scenario, destination, resolved_instrument=(git, metadata))

            except (OSError, subprocess.TimeoutExpired, _InstrumentUnavailable) as exc:

                evidence["workflow_integrity_unmeasured_reason"] = "oak_export_instrument_error:" + str(exc)

                return evidence

    if reason == "git_head_oracle_unavailable" or (reason and reason.startswith("git_head_oracle_error:")):

        evidence["workflow_integrity_unmeasured_reason"] = reason

        return evidence

    if reason is None and kind in {"git", "oak"}:

        evidence.update(reader_evidence(kind, binary, evidence))

        if scenario not in EXACT_SCENARIOS:

            evidence["workflow_integrity_evidence_class"] = evidence["workflow_integrity_evidence_class"].replace("_tree", "_payload")

        suffix = "subject_produced_oak_head_export" if kind == "oak" else evidence["workflow_integrity_evidence_class"]

        evidence["workflow_integrity_source"] = source + ";" + suffix

    evidence.update(workflow_integrity_passed=reason is None, workflow_integrity_failure_reason=reason)

    return evidence

def main() -> int:

    import argparse

    parser = argparse.ArgumentParser()

    parser.add_argument("scenario"); parser.add_argument("repo", type=Path)

    parser.add_argument("--git-head", action="store_true")

    parser.add_argument("--oak-head", action="store_true")

    parser.add_argument("--binary")

    parser.add_argument("--evidence-json", action="store_true")

    parser.add_argument("--git-instrument")

    parser.add_argument("--oak-reader-python")

    args = parser.parse_args()

    evidence = task_evidence(args.scenario, args.repo, "git" if args.git_head else "oak" if args.oak_head else None, args.binary, args.git_instrument, args.oak_reader_python)

    reason = evidence["workflow_integrity_failure_reason"] or evidence["workflow_integrity_unmeasured_reason"]

    print(json.dumps(evidence) if args.evidence_json else reason or "trusted_task_payload_verified:" + args.scenario)

    passed = evidence["workflow_integrity_passed"]

    return 77 if passed is None else 0 if passed else 1

if __name__ == "__main__":

    raise SystemExit(main())