"""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-10.2"

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

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()

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

    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)

    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) -> tuple[str | None, str | None]:

    git, _ = 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) -> 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, _ = resolve_git_instrument(git_binary)

    if git is None:

        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 = ["services", "validate_configs.py"]

        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 = ["assets", "manifest.json", "validate_manifest.py"]

        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 task_evidence(scenario: str, repo: Path, kind: str | None = None, binary: str | None = None, git_instrument: str | None = None) -> dict:

    source = "trusted_exact_task_worktree"

    if kind == "git":

        source += ";known_good_git_head_tree"

    elif kind == "oak":

        source += ";subject_produced_oak_head_export"

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

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

                "workflow_integrity_unmeasured_reason": None}

    reason = validate(scenario, repo)

    if reason:

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

        return evidence

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

        git, metadata = resolve_git_instrument(git_instrument or (binary if kind == "git" else None))

        evidence.update(metadata)

        if git is None:

            evidence["workflow_integrity_unmeasured_reason"] = "git_head_oracle_unavailable"

            return evidence

        if kind == "git":

            reason = validate_git_head(scenario, repo, git)

        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

            try:

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

                status = subprocess.run([oak, "status", "--json"], cwd=repo, env=base_env(), capture_output=True, timeout=30)

                before = json.loads(status.stdout) if status.returncode == 0 else {}

                head, branch = before.get("head"), before.get("branch")

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

                    raise ValueError("oak_head_identity_unreadable")

                evidence["oak_export_head"] = head

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

                    destination = Path(tmp) / "export"

                    exported = subprocess.run([oak, "export", "--branch", branch, str(destination)], cwd=repo,

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

                    if exported.returncode:

                        evidence["workflow_integrity_unmeasured_reason"] = "oak_head_export_unavailable"

                        return evidence

                    after = subprocess.run([oak, "status", "--json"], cwd=repo, env=base_env(), capture_output=True, timeout=30)

                    state = json.loads(after.stdout) if after.returncode == 0 else {}

                    if state.get("head") != head or state.get("branch") != branch:

                        raise ValueError("oak_head_changed_during_export")

                    reason = validate_git_head(scenario, destination, git)

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

                evidence["workflow_integrity_unmeasured_reason"] = 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

    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")

    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)

    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())