#!/usr/bin/env python3
"""Local Work evaluation: prepare references, freeze a signed build, run serially, grade.

Public artifacts are opt-in. Account/form/local-fixture lanes remain pending until
an appropriate scoped runner and oracle are provided. Process completion is never
semantic correctness. All output files are private and baseline files immutable.
"""
import argparse
import collections
import contextlib
import datetime
import fcntl
import hashlib
import html.parser
import json
import os
from pathlib import Path
import re
import shutil
import signal
import statistics
import stat
import subprocess
import sys
import tempfile
import time
import urllib.parse
import urllib.request
import uuid

ROOT = Path(__file__).resolve().parent.parent
DEFAULT_MANIFEST = ROOT / "eval/work-heldout-v1.json"
MODEL = "gpt-6-luna"
SCHEMA = 1
EXPORT_SCHEMA = 1
INSTRUMENTATION = {
    "crates/zephium-terra-macos-probe/src/acceptance.rs",
    "crates/zephium-terra-macos-probe/src/work_durable.rs",
    "crates/zephium-terra-macos-probe/src/loopback_support.rs",
    "scripts/work-eval", "scripts/test_work_eval.py",
}


FORM_INSTRUMENTATION = {
    "crates/zephium-engine/src/host/work_resource.rs",
    "crates/zephium-engine/src/host/work_resource_form_oracle.rs",
    "crates/zephium-engine/src/host/work_form_oracle.js",
    "crates/zephium-engine/src/host/work_resource_action.rs",
    "crates/zephium-engine/src/platform/mod.rs",
    "crates/zephium-engine/src/platform/form_qualification.rs",
    "crates/zephium-engine/src/platform/form_submit_observer.js",
    "crates/zephium-engine/src/platform/work_document_navigation.rs",
    "crates/zephium-engine/src/platform/macos/agent_context.rs",
    "crates/zephium-engine/src/platform/macos/semantic_runtime.rs",
    "crates/zephium-engine/src/platform/macos/agentic_semantic_program_probe.rs",
    "crates/zephium-engine/src/platform/macos/agentic_form_diagnostic_probe.rs",
    "crates/zephium-engine/src/platform/macos/agentic_form_diagnostic_probe.js",
}
LOCAL_INSTRUMENTATION = {
    "crates/zephium-terra-macos-probe/src/main.rs",
    "crates/zephium-terra-macos-probe/src/work_local_fixture.rs",
    "crates/zephium-terra-macos-probe/src/work_heldout_blocks.rs",
    "crates/zephium-terra-macos-probe/Cargo.toml", "Cargo.lock", "eval/work-heldout-v1.json",
}
LOCAL_ELIGIBILITY = {"pending_confined_fixture", "pending_blocks_state_oracle"}

FORM_CHECK_COUNTS = {
    "form-httpbin-draft": 11, "form-selenium-controls": 3, "form-selenium-date": 2,
    "form-expand-inputs": 3, "form-heroku-numeric": 2, "form-heroku-dropdown": 1,
    "form-heroku-checkboxes": 2, "form-heroku-editor": 1, "form-qa-select": 1, "form-qa-date": 1,
}


def digest_bytes(data):
    return hashlib.sha256(data).hexdigest()


def digest(path):
    with open(path, "rb") as source:
        return hashlib.file_digest(source, "sha256").hexdigest()


def now():
    return datetime.datetime.now(datetime.timezone.utc).isoformat()


def write_private(path, value):
    path = Path(path)
    data = value if isinstance(value, bytes) else (json.dumps(value, indent=2, ensure_ascii=False) + "\n").encode()
    with open(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600), "wb") as output:
        output.write(data)


def read_json(path, maximum=32 * 1024 * 1024):
    path = Path(path)
    if path.is_symlink() or path.stat().st_size > maximum:
        raise ValueError(f"unsafe or oversized evidence: {path}")
    return json.loads(path.read_text())


def host(url):
    parsed = urllib.parse.urlsplit(url)
    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
        raise ValueError("references must be public HTTPS URLs without credentials")
    return parsed.hostname.lower().removeprefix("www.")


def excluded(url, manifest):
    value = host(url)
    return any(value == domain or value.endswith("." + domain) for domain in manifest["excluded_domains"])


def load_manifest(path):
    manifest = read_json(path)
    expected = {"work-heldout-v1": 100, "work-heldout-v2": 150, "work-external-webvoyager-v1": 50}.get(manifest.get("suite"))
    if manifest.get("schema_version") != SCHEMA or expected is None or len(manifest["tasks"]) != expected:
        raise ValueError("expected an unsealed versioned Work manifest")
    ids = [task["id"] for task in manifest["tasks"]]
    if len(set(ids)) != len(ids) or any(not re.fullmatch(r"[a-z0-9-]{1,64}", name) for name in ids):
        raise ValueError("invalid or duplicate task IDs")
    if len(manifest["tuned_cases"]) != 23 or "unsplash" not in {case["id"] for case in manifest["tuned_cases"]}:
        raise ValueError("the exact tuned 23-case exclusion inventory is required")
    for task in manifest["tasks"]:
        if not task["predicates"] or len(task["prompt"].encode()) > 2400 or "\n" in task["prompt"]:
            raise ValueError(f"invalid task contract: {task['id']}")
        followups = task.get("follow_ups", [])
        if any(not isinstance(follow.get("prompt"), str) or not follow["prompt"] or "\n" in follow["prompt"] or not follow.get("criteria") for follow in followups):
            raise ValueError(f"invalid follow-up criteria: {task['id']}")
        if len(task_request(task).encode()) > 2400:
            raise ValueError(f"combined request exceeds probe limit: {task['id']}")
        for url in task.get("urls", []):
            if excluded(url, manifest):
                exceptions = manifest.get("domain_overlap_exceptions", {}).get(task.get("holdout_scope"), [])
                allowed = task["eligibility"] == "pending_user_session" and any(host(url) == domain or host(url).endswith("." + domain) for domain in exceptions)
                if not allowed:
                    raise ValueError(f"tuned/recipe site in holdout: {task['id']}")
        for source in task.get("oracle", {}).get("sources", {}).values():
            if excluded(source["url"], manifest):
                raise ValueError(f"tuned/recipe oracle in holdout: {task['id']}")
    return manifest


def task_request(task):
    return "\n".join([task["prompt"], *[follow["prompt"] for follow in task.get("follow_ups", [])]])


class Metadata(html.parser.HTMLParser):
    def __init__(self):
        super().__init__()
        self.values = collections.defaultdict(list)

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        if tag == "meta" and attrs.get("name") and "content" in attrs:
            self.values[attrs["name"]].append(attrs["content"])


def reference_fields(source, body):
    if source["format"] == "json":
        data = json.loads(body)
    elif source["format"] == "html_meta":
        parser = Metadata()
        parser.feed(body.decode("utf-8"))
        data = dict(parser.values)
    else:
        raise ValueError("unknown reference format")
    result = {}
    for name, rule in source["fields"].items():
        value = data
        for key in rule["path"]:
            value = value[key]
        if rule.get("transform") == "length":
            value = len(value)
        if not isinstance(value, (str, int, float, bool, type(None))):
            raise ValueError("reference predicate must yield a scalar")
        result[name] = value
    return result


def prepare_references(args):
    manifest = load_manifest(args.manifest)
    directory = Path(args.out).resolve()
    directory.mkdir(mode=0o700, parents=True, exist_ok=False)
    references = {"schema_version": SCHEMA, "manifest_sha256": digest(args.manifest), "captured_at": now(), "tasks": {}}
    cache = {}
    for task in manifest["tasks"]:
        if not task.get("oracle"):
            continue
        expected, sources = {}, []
        try:
            for name, source in task["oracle"]["sources"].items():
                url = source["url"]
                if url not in cache:
                    request = urllib.request.Request(url, headers={"User-Agent": "Zephium-local-evaluation/1.0 (public reference capture)"})
                    with urllib.request.urlopen(request, timeout=25) as response:
                        final = response.geturl()
                        if host(final) != host(url):
                            raise ValueError("reference redirected to a different host")
                        body = response.read(2 * 1024 * 1024 + 1)
                        if len(body) > 2 * 1024 * 1024:
                            raise ValueError("reference too large")
                        cache[url] = body
                body = cache[url]
                values = reference_fields(source, body)
                if set(values) & set(expected):
                    raise ValueError("duplicate oracle fields")
                expected.update(values)
                sources.append({"name": name, "url": url, "sha256": digest_bytes(body)})
                saved = directory / (digest_bytes(body) + ".source")
                if not saved.exists():
                    write_private(saved, body)
            references["tasks"][task["id"]] = {"state": "ready", "expected": expected, "sources": sources}
        except Exception as error:
            references["tasks"][task["id"]] = {"state": "blocked_reference", "reason": type(error).__name__}
        print(task["id"], references["tasks"][task["id"]]["state"], flush=True)
    write_private(directory / "references.json", references)
    print(directory / "references.json")


def signature(probe, identifier=None):
    subprocess.run(["/usr/bin/codesign", "--verify", "--strict", str(probe)], check=True, capture_output=True)
    shown = subprocess.run(["/usr/bin/codesign", "-dv", "--verbose=4", str(probe)], check=True, capture_output=True, text=True).stderr.splitlines()
    required = "Identifier=" + identifier if identifier else None
    if (required is not None and required not in shown) or not any(line.startswith("Identifier=") for line in shown):
        raise ValueError("probe signing identifier mismatch")
    return [line for line in shown if line.startswith(("Identifier=", "CDHash=", "TeamIdentifier=", "Authority="))]


def git(root, *args):
    return subprocess.run(["git", "-C", str(root), *args], check=True, capture_output=True).stdout


def freeze(args):
    manifest = load_manifest(args.manifest)
    references = read_json(args.references)
    if references["manifest_sha256"] != digest(args.manifest):
        raise ValueError("reference/manifest mismatch")
    source = Path(args.source_root).resolve()
    patch = git(source, "diff", "--binary", args.base_ref)
    changed = set(git(source, "diff", "--name-only", args.base_ref).decode().splitlines())
    forms = getattr(args, "form_oracle", False)
    locals_enabled = getattr(args, "local_fixtures", False)
    allowed = INSTRUMENTATION | (FORM_INSTRUMENTATION if forms else set()) | (LOCAL_INSTRUMENTATION if locals_enabled else set())
    tuned = getattr(args, "tuned_matrix", False)
    if tuned:
        if not args.matrix_source or args.timeout != 480:
            raise ValueError("tuned matrix requires original matrix source and historical 480-second cap")
        if matrix_pairs(args.matrix_source) != manifest["tuned_cases"]:
            raise ValueError("matrix source/manifest URL-goal pairs differ")
    if tuned and (args.label != "candidate" or forms or locals_enabled):
        raise ValueError("tuned matrix requires its own candidate freeze without form/local modes")
    if args.label == "baseline" and not changed <= allowed:
        raise ValueError("baseline source includes production changes after base revision")
    if forms:
        # New instrumentation must actually be represented in the recorded patch.
        for name in sorted(FORM_INSTRUMENTATION):
            git(source, "ls-files", "--error-unmatch", name)
    probe = Path(args.probe).resolve()
    signed = signature(probe, args.identifier)
    original_hash = digest(probe)
    out = Path(args.out).resolve()
    out.mkdir(mode=0o700, parents=True, exist_ok=False)
    binary = out / "probe"
    shutil.copyfile(probe, binary)
    if digest(binary) != original_hash or digest(probe) != original_hash:
        raise ValueError("probe changed during freeze")
    os.chmod(binary, 0o500)
    if signature(binary) != signed:
        raise ValueError("copied signature mismatch")
    for target, raw in [("manifest.json", Path(args.manifest).read_bytes()), ("references.json", Path(args.references).read_bytes()),
                        ("source.patch", patch), ("work-eval.py", Path(__file__).read_bytes())]:
        write_private(out / target, raw)
    record = {"schema_version": SCHEMA, "label": args.label, "frozen_at": now(),
              "base_revision": git(source, "rev-parse", args.base_ref).decode().strip(),
              "source_head": git(source, "rev-parse", "HEAD").decode().strip(), "source_root": str(source),
              "source_patch_sha256": digest_bytes(patch), "changed_files": sorted(changed),
              "probe_sha256": original_hash, "signature": signed,
              "manifest_sha256": digest(out / "manifest.json"), "references_sha256": digest(out / "references.json"),
              "runner_sha256": digest(out / "work-eval.py"), "model": MODEL, "evaluation_lane": manifest["suite"],
              "read_decisions": getattr(args, "read_decisions", "recommended"),
              "model_setup": "Luna lead/page/light; read-decision mode is frozen separately; actual primary/emulation counters are reported separately",
              "platform": list(os.uname()), "python": sys.version, "timeout_seconds": args.timeout,
              "oracle_scope": "deterministic predicates and separately identified human or agent source audits; form task effects use bounded native action/navigation and DOM submit receipts, not universal HTTP observation"}
    if tuned:
        write_private(out / "matrix.sh", Path(args.matrix_source).read_bytes())
        write_private(out / "matrix-pairs.json", matrix_pairs(args.matrix_source))
        record["matrix_source_sha256"] = digest(out / "matrix.sh")
        record["matrix_pairs_sha256"] = digest(out / "matrix-pairs.json")
        record["evaluation_lane"] = "tuned_matrix_v1"
        record["tuned_matrix_qualification"] = "Exact original 23 URL/goal pairs, including Unsplash. Modern public-only lead path declines all consequential confirmations. Historical live-site terminal-status scores are not matched semantic grades. No automatic pass without a separately recorded source audit."
    if forms:
        directory = out / "form-observer"
        directory.mkdir(mode=0o700)
        record["form_oracle"] = {"version": 2, "resource_selection": "single fresh page, or unique action-bearing native document with complete matching retirement ledgers and p2 evidence for every opened page; otherwise unknown", "files": {}, "predicate_files": {}}
        for name in sorted(FORM_INSTRUMENTATION):
            copied = "form-observer/" + Path(name).name
            write_private(out / copied, (source / name).read_bytes())
            record["form_oracle"]["files"][copied] = digest(out / copied)
            if Path(name).name in ("work_resource_form_oracle.rs", "work_form_oracle.js", "form_qualification.rs", "form_submit_observer.js", "agentic_form_diagnostic_probe.rs", "agentic_form_diagnostic_probe.js"):
                record["form_oracle"]["predicate_files"][copied] = digest(out / copied)
    if locals_enabled:
        directory = out / "local-fixtures"
        directory.mkdir(mode=0o700)
        record["local_fixtures"] = {"version": 1, "files": {}, "predicate_files": {},
            "authority": "Only exact predeclared synthetic scratch changes or new synthetic Work artifacts; concrete frozen prompts override generic read-only boilerplate. Existing approval and confinement gates remain mandatory.",
            "calendar": "No typed timestamp/timezone in v1 product plan; full predicate remains unsupported/unknown."}
        for name in sorted(LOCAL_INSTRUMENTATION):
            git(source, "ls-files", "--error-unmatch", name)
            copied = "local-fixtures/" + Path(name).name
            write_private(out / copied, (source / name).read_bytes())
            record["local_fixtures"]["files"][copied] = digest(out / copied)
            if Path(name).name in ("work_local_fixture.rs", "work_heldout_blocks.rs", "work-heldout-v1.json"):
                record["local_fixtures"]["predicate_files"][copied] = digest(out / copied)
    write_private(out / "freeze.json", record)
    print(out / "freeze.json")



def matrix_pairs(path):
    source = Path(path).read_text()
    pattern = re.compile(r'^url\[([a-z0-9-]+)\]="([^"\n]+)";[ \t]*goal\[\1\]="([^"\n]+)"$', re.MULTILINE)
    pairs = [{"id": name, "url": url, "goal": goal} for name, url, goal in pattern.findall(source)]
    if len(pairs) != 23 or len({row["id"] for row in pairs}) != 23 or "unsplash" not in {row["id"] for row in pairs}:
        raise ValueError("expected exactly 23 literal URL/goal pairs including Unsplash")
    return pairs


def evaluation_manifest(manifest, frozen):
    if frozen.get("evaluation_lane") != "tuned_matrix_v1":
        return manifest
    # This is a separately labelled tuned set, never held-out contamination relief.
    result = dict(manifest, excluded_domains=[])
    result["tasks"] = [{"id": case["id"], "urls": [case["url"]],
                        "prompt": "Open " + case["url"] + ". " + case["goal"],
                        "original_goal": case["goal"], "category": "tuned_matrix",
                        "eligibility": "eligible_public", "grading": "source_audit",
                        "predicates": [{"id": "goal_correctness", "description": case["goal"]},
                                       {"id": "source_support", "description": "Returned claims and record associations are supported by observed primary sources; unavailable or time-sensitive proof remains unknown."}]}
                       for case in manifest["tuned_cases"]]
    return result


def verify_freeze(path):
    root = Path(path).resolve().parent
    record = read_json(path)
    for name, key in [("probe", "probe_sha256"), ("manifest.json", "manifest_sha256"), ("references.json", "references_sha256"),
                      ("work-eval.py", "runner_sha256"), ("source.patch", "source_patch_sha256")]:
        if digest(root / name) != record[key]:
            raise ValueError(f"frozen {name} changed")
    for name, expected in record.get("form_oracle", {}).get("files", {}).items():
        if not name.startswith("form-observer/") or Path(name).name != name.split("/", 1)[1] or digest(root / name) != expected:
            raise ValueError("frozen form observer changed")
    for name, expected in record.get("local_fixtures", {}).get("files", {}).items():
        if not name.startswith("local-fixtures/") or Path(name).name != name.split("/", 1)[1] or digest(root / name) != expected:
            raise ValueError("frozen local fixture changed")
    if record.get("evaluation_lane") == "tuned_matrix_v1":
        if digest(root / "matrix.sh") != record["matrix_source_sha256"] or digest(root / "matrix-pairs.json") != record["matrix_pairs_sha256"]:
            raise ValueError("frozen tuned matrix provenance changed")
        if matrix_pairs(root / "matrix.sh") != read_json(root / "matrix-pairs.json") or read_json(root / "matrix-pairs.json") != read_json(root / "manifest.json")["tuned_cases"]:
            raise ValueError("frozen tuned matrix pairs differ")
    if digest(__file__) != record["runner_sha256"]:
        raise ValueError("invoke the frozen work-eval.py to preserve grading behavior")
    return root, record


def normal_url(url):
    parsed = urllib.parse.urlsplit(url)
    return (parsed.scheme, parsed.hostname.removeprefix("www.") if parsed.hostname else "", parsed.port, parsed.path.rstrip("/"), parsed.query)


def evidence_key(link):
    return (link.get("extraction_id"), link.get("source_id"))


def current_publications(artifacts):
    by_id = {artifact["id"]: artifact for artifact in artifacts}
    revised = {artifact.get("revises") for artifact in artifacts if artifact.get("revises")}
    current = []
    for artifact in artifacts:
        if artifact["id"] in revised:
            continue
        at, seen, published = artifact, set(), False
        while at is not None:
            if at["id"] in seen:
                raise ValueError("cyclic artifact revision history")
            seen.add(at["id"])
            published |= bool(at.get("published"))
            prior = at.get("revises")
            if prior and prior not in by_id:
                raise ValueError("missing artifact revision predecessor")
            at = by_id.get(prior)
        if published:
            current.append(dict(artifact, published=True))
    # A cycle with no current head would otherwise silently disappear.
    for artifact in artifacts:
        at, seen = artifact, set()
        while at is not None:
            if at["id"] in seen:
                raise ValueError("cyclic artifact revision history")
            seen.add(at["id"])
            at = by_id.get(at.get("revises"))
    return current


def answer_object(artifacts):
    candidates = [artifact for artifact in artifacts if artifact.get("published")
                  and artifact["data"].get("kind") in ("code", "reply")]
    for artifact in reversed(candidates):
        data = artifact["data"]
        if data.get("kind") == "code" and data.get("language", "").lower() != "json":
            continue
        text = data.get("text", "").strip()
        if text.startswith("```json") and text.endswith("```"):
            text = text[7:-3].strip()
        try:
            value = json.loads(text)
        except (ValueError, TypeError):
            continue
        if isinstance(value, dict):
            return value, artifact
    return None, None


def interval_union(intervals):
    merged = []
    for start, end in sorted(intervals):
        if end < start:
            raise ValueError("reversed timing interval")
        if merged and start <= merged[-1][1]:
            merged[-1][1] = max(end, merged[-1][1])
        else:
            merged.append([start, end])
    return merged


def timing(evidence, log):
    timeline = evidence.get("timeline") or {}
    start, end = timeline.get("started_ms"), timeline.get("finished_ms")
    result = {"model_busy_ms": None, "page_without_model_ms": None, "other_ms": None, "post_browser_finalization_gaps_ms": [], "finalization_windows": []}
    # A finalization gap is per execution; follow-ups must not share its last part.
    last = None
    first_reply = None
    previous_end = -1
    for line in log.splitlines():
        match = re.match(r"\[\s*(\d+)ms\] lead-run: event=(.*)", line)
        if not match:
            continue
        at, event = int(match[1]), match[2]
        if event.startswith("PartEnded { helper: Browser"):
            last = at
        elif event.startswith("ReplyPublished {") and first_reply is None:
            first_reply = at
        elif event.startswith("Ended {"):
            finished_pages = [page["finished_ms"] for page in evidence.get("page_intervals", []) if isinstance(page.get("finished_ms"), int) and previous_end < page["finished_ms"] <= at]
            candidates = finished_pages + ([last] if last is not None else [])
            last = max(candidates) if candidates else None
            if last is not None:
                result["post_browser_finalization_gaps_ms"].append(max(0, at-last))
                result["finalization_windows"].append({
                    "last_browser_ms": last, "ended_ms": at, "first_reply_ms": first_reply,
                    "first_reply_relative_to_last_browser_ms": first_reply-last if first_reply is not None else None,
                    "model_busy_ms": None, "non_model_ms": None,
                })
            last = None
            first_reply = None
            previous_end = at
    rows = evidence.get("model_calls", []) + evidence.get("page_intervals", [])
    valid = isinstance(start, int) and isinstance(end, int) and end >= start and not evidence.get("model_calls_truncated") and all(isinstance(row.get("started_ms"), int) and isinstance(row.get("finished_ms"), int) and row["finished_ms"] >= row["started_ms"] for row in rows)
    if not valid:
        return result
    if isinstance(evidence.get("wall_ms"), int) and abs(end-start-evidence["wall_ms"]) > 1:
        return result
    def spans(rows):
        return [[max(start,row["started_ms"]),min(end,row["finished_ms"])] for row in rows if row["finished_ms"] >= start and row["started_ms"] <= end]
    models = interval_union(spans(evidence.get("model_calls", [])))
    pages = interval_union(spans(evidence.get("page_intervals", [])))
    duration = lambda intervals: sum(end-start for start,end in intervals)
    together = duration(interval_union(models+pages))
    for window in result["finalization_windows"]:
        left, right = window["last_browser_ms"], window["ended_ms"]
        if not start <= left <= right <= end:
            continue
        busy = sum(max(0, min(right, finish)-max(left, begin)) for begin, finish in models)
        window.update(model_busy_ms=busy, non_model_ms=right-left-busy)
    result.update(model_busy_ms=duration(models), page_without_model_ms=together-duration(models), other_ms=end-start-together,
                  qualification="Disjoint wall intervals, model first then page. Other includes orchestration, storage, and uninstrumented search-provider I/O; it is not a pure CPU-overhead measurement.")
    return result


def metrics(evidence, log):
    executions = evidence.get("executions", [])
    attempts = [attempt for execution in executions for attempt in execution.get("attempts", [])]
    usage = [attempt.get("usage") for attempt in attempts]
    steps = [step for execution in executions for step in execution.get("steps", [])]
    calls = evidence.get("model_calls")
    call_rows = calls if isinstance(calls, list) else []
    failures = collections.Counter()
    handoffs = collections.Counter()
    for step in steps:
        measured = step.get("measurements") or {}
        outcome = measured.get("outcome") or {}
        if outcome.get("kind") not in (None, "completed"):
            failures[outcome["kind"] + (":" + outcome["stage"] if "stage" in outcome else "")] += 1
        if measured.get("resolved_handoff"):
            handoffs[measured["resolved_handoff"]] += 1
    internal = collections.Counter(re.findall(r"lead-run: event=(ObjectRefused|CommitRefused|ModelRefused|BudgetSpent|Closed)\b", log))
    reasons = collections.Counter(re.findall(r"ObjectRefused \{ reason: Some\((\w+)\)", log))
    page_calls = sum((step.get("measurements") or {}).get("planner_calls", 0) for step in steps)
    page_reasks = sum(map(int, re.findall(r"ModelAskedAgain\((\d+)\)", log)))
    page_turns = page_calls + page_reasks if evidence.get("page_diagnostics_version") == 1 else None
    malformed = len(re.findall(r"ModelActionRejected\(", log)) if page_turns is not None else None
    selected_cost = sum(row["cost_micro_usd"] for row in call_rows) if isinstance(calls, list) and all(row.get("state") == "settled" and isinstance(row.get("cost_micro_usd"), int) for row in call_rows) else None
    fully_observed_cost = selected_cost is not None and not evidence.get("model_calls_truncated") and not any(step.get("kind") == "search" or (step.get("measurements") or {}).get("decision_calls", 0) for step in steps)
    return {"wall_ms": evidence.get("wall_ms"), "timing": timing(evidence, log),
            "selected_client_reported_cost_micro_usd": selected_cost,
            "cost_for_target_micro_usd": selected_cost if fully_observed_cost else None,
            "cost_for_target_qualification": "Available only when all selected-client calls settled with reported cost and neither public search nor primary decision calls can add unobserved charges.",
            "malformed_action_events": malformed, "page_agent_turns": page_turns,
            "malformed_actions_per_100_page_turns": 100 * malformed / page_turns if page_turns else None,
            "selected_model_call_attempts": len(calls) if isinstance(calls, list) else None,
            "model_call_scope": evidence.get("model_call_scope", "unavailable"),
            "unsettled_model_calls": sum(call.get("state") == "unsettled" for call in call_rows) if isinstance(calls, list) else None,
            "reported_cost_micro_usd": sum(item["cost_micro_usd"] for item in usage) if usage and all(item is not None for item in usage) else None,
            "reported_tokens": sum(item["model_tokens"] for item in usage) if usage and all(item is not None for item in usage) else None,
            "accounting_bases": [item["accounting"] if item else "unknown" for item in usage],
            "primary_decision_calls": sum((step.get("measurements") or {}).get("decision_calls", 0) for step in steps) if "executions" in evidence else None,
            "emulation_calls_overlapping_selected_client": sum((step.get("measurements") or {}).get("emulation_calls", 0) for step in steps) if "executions" in evidence else None,
            "cost_scope": "Selected-client or durable-attempt accounting basis as exported; primary Jev/direct search may be excluded. Never add overlapping page and attempt costs.",
            "page_model_calls": sum((step.get("measurements") or {}).get("planner_calls", 0) for step in steps),
            "search_steps": sum(step.get("kind") == "search" for step in steps),
            "asks": sum(step.get("kind") == "ask" for step in steps), "confirmations": sum(step.get("kind") == "confirm" for step in steps),
            "failure_taxonomy": dict(failures), "resolved_handoffs": dict(handoffs),
            "internal_failures": dict(internal), "object_refusal_reasons": dict(reasons)}


def grade(task, evidence, reference, manifest):
    result = {"task_outcome": "unknown", "predicates": [], "review": None}
    if evidence.get("schema_version") != EXPORT_SCHEMA or evidence.get("case_id") != task["id"]:
        return dict(result, reason="missing_or_mismatched_evidence")
    if evidence.get("public_artifacts") != "exported":
        return dict(result, reason="private_evidence_withheld")
    for role in ("lead", "page", "light"):
        model = evidence.get("models", {}).get(role, {}).get("model", {})
        if model.get("provider") != "open_ai" or model.get("model") != MODEL:
            return dict(result, reason="model_policy_mismatch")
    executions = evidence.get("executions", [])
    steps = [step for execution in executions for step in execution.get("steps", [])]
    for step in steps:
        if step.get("url") and excluded(step["url"], manifest):
            return dict(result, reason="tuned_site_contamination")
    if task["grading"] == "source_audit":
        return dict(result, reason="source_audit_required", predicates=[{"id": p["id"], "state": "unreviewed"} for p in task["predicates"]])
    if not reference or reference.get("state") != "ready":
        return dict(result, reason="reference_unavailable")
    artifacts = [artifact for execution in executions for artifact in execution.get("artifacts", [])]
    try:
        current = current_publications(artifacts)
    except ValueError:
        return dict(result, reason="invalid_artifact_revision_history")
    candidates = [artifact for artifact in current if answer_object([artifact])[0] is not None]
    if len(candidates) > 1 and any("revises" not in artifact for artifact in artifacts):
        return dict(result, reason="legacy_artifact_revisions_unknown")
    answer, published = answer_object(current)
    if answer is None:
        return dict(result, task_outcome="failed", reason="required_answer_object_missing_or_invalid")
    checks = [{"id": "exact_answer_keys", "state": "passed" if set(answer) == set(reference["expected"]) else "failed"}]
    for key, expected in reference["expected"].items():
        actual = answer.get(key)
        if isinstance(expected, str):
            passed = isinstance(actual, str) and " ".join(actual.split()).casefold() == " ".join(expected.split()).casefold()
        else:
            passed = type(actual) is type(expected) and actual == expected
        checks.append({"id": key, "state": "passed" if passed else "failed"})
    answer_evidence = {evidence_key(link) for link in published.get("evidence", [])}
    by_id = {artifact["id"]: artifact for artifact in artifacts}
    for index, url in enumerate(task["urls"]):
        allowed = {normal_url(url)}
        oracle_sources = list(task.get("oracle", {}).get("sources", {}).values())
        # A single-source task's predeclared ground-truth endpoint is also
        # valid native provenance. This never admits arbitrary same-host URLs.
        if manifest.get("suite") == "work-heldout-v2" and len(task["urls"]) == 1 and len(oracle_sources) == 1:
            allowed.add(normal_url(oracle_sources[0]["url"]))
        reads = [step for step in steps if step.get("kind") == "read" and step.get("status") == "succeeded"
                 and step.get("url") and normal_url(step["url"]) in allowed]
        proven = any(any(answer_evidence & {evidence_key(link) for link in by_id.get(identity, {}).get("evidence", [])}
                         for identity in step.get("artifacts", [])) for step in reads)
        checks.append({"id": f"native_source_{index + 1}", "state": "passed" if proven else "failed"})
    outcome = "passed" if checks and all(check["state"] == "passed" for check in checks) else "failed"
    # A source-backed answer with some correct requested fields is partial,
    # never a pass. Preserve v1's historical binary grades unchanged.
    if outcome == "failed" and manifest.get("suite") == "work-heldout-v2":
        fields = [check for check in checks if check["id"] in reference["expected"]]
        structure = [check for check in checks if check["id"] not in reference["expected"]]
        if fields and any(check["state"] == "passed" for check in fields) and all(check["state"] == "passed" for check in structure):
            outcome = "partial"
    return dict(result, task_outcome=outcome, predicates=checks)


@contextlib.contextmanager
def live_lease():
    descriptor = os.open("/private/tmp/zephium-native-probe.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
    try:
        fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
        yield
    finally:
        os.close(descriptor)


def stop_child(child):
    try:
        os.killpg(child.pid, signal.SIGTERM)
    except ProcessLookupError:
        child.wait()
        return
    try:
        child.wait(timeout=3)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(child.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        child.wait()


def form_effects(value):
    if not isinstance(value, dict) or value.get("version") != 1:
        return None
    for key in ("native_actions_started", "native_actions_settled", "native_commits_applied"):
        if type(value.get(key)) is not int or not 0 <= value[key] <= 256:
            return None
    for key in ("no_task_submit_or_navigation", "fill_then_clear", "fill_then_increment_once"):
        if value.get(key) is not None and type(value[key]) is not bool:
            return None
    nav = value.get("navigation")
    if not isinstance(nav, dict) or type(value.get("incomplete")) is not bool:
        return None
    for key in ("callbacks", "accepted_main_forms", "accepted_subframe_forms", "post_initial_main", "consent_navigation"):
        if type(nav.get(key)) is not int or not 0 <= nav[key] <= 256:
            return None
    if any(type(nav.get(key)) is not bool for key in ("initial_finished", "uncertain")):
        return None
    # Validate the positive receipt invariants independently of its asserted bit.
    if value.get("no_task_submit_or_navigation") is True and (
        value["incomplete"] or value["native_actions_started"] != value["native_actions_settled"] or
        value["native_commits_applied"] or nav["uncertain"] or not nav["initial_finished"] or
        not nav["callbacks"] or any(nav[key] for key in ("accepted_main_forms", "accepted_subframe_forms", "post_initial_main", "consent_navigation"))):
        return None
    return value


def grade_form(task, evidence, log):
    unknown = {"task_outcome": "unknown", "reason": "form_observation_unavailable", "predicates": [
        {"id": "p1", "state": "unknown"}, {"id": "p2", "state": "unknown"}]}
    if evidence.get("case_id") != task["id"] or evidence.get("public_artifacts") != "exported":
        return unknown
    observations, settlements = [], []
    for line in log.splitlines():
        prefix = next((prefix for prefix in ("form-oracle: ", "form-effects: ") if line.startswith(prefix)), None)
        if prefix is None:
            continue
        try:
            row = json.loads(line.removeprefix(prefix))
        except ValueError:
            return unknown
        if not isinstance(row, dict):
            return unknown
        if row.get("case") == task["id"]:
            (observations if prefix == "form-oracle: " else settlements).append(row)
    if not observations:
        return unknown
    for row in observations:
        checks = row.get("checks")
        if row.get("version") != 1 or row.get("reason") != "observed" or row.get("content_exported") is not False or not isinstance(checks, list) or len(checks) != FORM_CHECK_COUNTS.get(task["id"]) or any(value is not None and type(value) is not bool for value in checks):
            return unknown
    effects = [form_effects(row.get("effects")) for row in observations]
    settled = [form_effects(row.get("effects")) for row in settlements if row.get("version") == 1 and row.get("content_exported") is False]
    complete = len(settled) == len(settlements) == len(observations) and all(item is not None for item in effects + settled)
    if complete:
        # Each owned resource reports its fresh oracle then exactly one retirement
        # ledger. Matching the entire multiset detects skipped/extra/unsettled pages;
        # equivalent ledgers need not be assigned an invented identity.
        complete = collections.Counter(json.dumps(item, sort_keys=True) for item in effects) == collections.Counter(json.dumps(item, sort_keys=True) for item in settled)
    selected = 0
    rule = "single_fresh_page"
    if len(observations) > 1:
        stamps = [row.get("document") for row in observations]
        profiles = [row.get("profile") for row in observations]
        exact_stamps = all(isinstance(stamp, str) and re.fullmatch(r"WorkDocumentStamp \{ native_id: NavigationId\(\d+\), epoch: \d+ \}", stamp) for stamp in stamps)
        if not complete or any(item["incomplete"] for item in effects) or not exact_stamps or len(set(stamps)) != len(stamps) or any(not isinstance(profile, str) for profile in profiles) or len(set(profiles)) != 1:
            return unknown
        action_pages = [index for index, item in enumerate(effects) if item["native_actions_started"] > 0]
        if len(action_pages) != 1:
            return dict(unknown, reason="ambiguous_form_action_resource")
        selected = action_pages[0]
        rule = "unique_action_page_with_every_opened_page_accounted"
    row = observations[selected]
    chosen = effects[selected]
    checks = list(row["checks"])
    if complete and not chosen["incomplete"]:
        if task["id"] == "form-expand-inputs":
            checks[2] = chosen.get("fill_then_clear")
        elif task["id"] == "form-heroku-numeric":
            checks[1] = chosen.get("fill_then_increment_once")
    state = "failed" if False in checks else "unknown" if None in checks else "passed"
    events = [row.get("no_task_submit_event") for row in observations]
    native = [item.get("no_task_submit_or_navigation") for item in effects] if complete else [None]
    p2 = "failed" if any(value is False for value in events + native) else "passed" if all(value is True for value in events + native) else "unknown"
    outcome = "failed" if "failed" in (state, p2) else "passed" if state == p2 == "passed" else "unknown"
    return {"task_outcome": outcome, "reason": "native_form_predicates_and_effect_receipts",
            "predicates": [{"id": "p1", "state": state, "checks": checks},
                           {"id": "p2", "state": p2, "reason": "Observed main-document task-submit plus native commit/navigation receipts for every accounted page; closed shadow roots and unrelated background HTTP activity are not measured or claimed."}],
            "form_observer": row, "form_resource_selection": rule, "observed_pages": len(observations),
            "grading_evidence": "native_fixed_form_predicates_and_bounded_effect_ledger"}


def grade_local(task, evidence, setup, manifest_sha):
    unknown = {"task_outcome": "unknown", "reason": "local_fixture_evidence_unavailable", "predicates": []}
    if any(not isinstance(row, dict) or row.get("schema_version") != 1 or row.get("case_id") != task["id"] or row.get("manifest_sha256") != manifest_sha or row.get("frozen_case") != task for row in (evidence, setup)):
        return unknown
    for role in ("lead", "page", "light"):
        model = evidence.get("models", {}).get(role, {}).get("model", {})
        if model.get("provider") != "open_ai" or model.get("model") != MODEL:
            return dict(unknown, reason="local_model_setup_mismatch")
    checks = evidence.get("predicates")
    if not isinstance(checks, dict) or set(checks) != {check["id"] for check in task["predicates"]} or any(value is not None and type(value) is not bool for value in checks.values()):
        return unknown
    if task["category"] == "local-fixture":
        if not setup.get("input_hashes") or evidence.get("input_hashes") != setup.get("input_hashes") or evidence.get("expected_changes") != setup.get("expected_changes"):
            return dict(unknown, reason="local_fixture_input_or_authority_mismatch")
    else:
        sentinel = evidence.get("sentinel", {})
        if setup.get("target_seeded") is not False or not setup.get("fixture_appendix_sha256") or evidence.get("fixture_appendix_sha256") != setup.get("fixture_appendix_sha256") or sentinel.get("id") != setup.get("sentinel_id") or sentinel.get("before_sha256") != setup.get("sentinel_sha256"):
            return dict(unknown, reason="blocks_fixture_input_or_identity_mismatch")
        if task["id"] == "blocks-calendar" and checks.get("p2") is not None:
            return dict(unknown, reason="unsupported_typed_timezone_cannot_pass")
    predicates = [{"id": key, "state": "passed" if value is True else "failed" if value is False else "unknown"} for key, value in checks.items()]
    state = "failed" if any(value is False for value in checks.values()) else "passed" if all(value is True for value in checks.values()) else "unknown"
    if state == "passed" and evidence.get("status") not in ("completed", "needs_review"):
        state = "unknown"
    return {"task_outcome": state, "reason": "typed_synthetic_fixture_predicates", "predicates": predicates,
            "grading_evidence": "independent typed state or fixed parser plus actual command/approval receipts, never model-reported success",
            "fixture_limitation": evidence.get("unsupported") or evidence.get("failure"), "runtime_status": evidence.get("status")}


def local_metrics(evidence, log):
    diagnostics = evidence.get("diagnostics") or {}
    usage = diagnostics.get("attempt_usage")
    converted = {"wall_ms": evidence.get("wall_ms"), "model_call_scope": diagnostics.get("scope", "unavailable")}
    calls = diagnostics.get("selected_client_calls")
    if isinstance(calls, list) and diagnostics.get("truncated") is False:
        converted["model_calls"] = calls
    if isinstance(usage, list):
        converted["executions"] = [{"attempts": [{"usage": item} for item in usage], "steps": []}]
    result = metrics(converted, log)
    result.update(observed_selected_client_calls=len(calls) if isinstance(calls, list) else None,
                  selected_client_calls_truncated=diagnostics.get("truncated"),
                  primary_decision_calls=diagnostics.get("primary_decision_calls"),
                  emulation_calls_overlapping_selected_client=diagnostics.get("emulation_calls"),
                  call_scope_qualification="No browser/search dispatch allowed in fixed local mode; selected-client observations only. Missing/truncated totals and unexported primary counters stay unknown.")
    if evidence.get("failure"):
        result["failure_taxonomy"] = {evidence["failure"]: 1}
    if evidence.get("unsupported"):
        result["failure_taxonomy"] = {"unsupported_typed_state": 1}
    return result


def cleanup_local_fixture(task, setup, manifest_sha, *, reaped, home=None):
    """Remove only the recorded owned child, using pinned nofollow directory FDs.

    No discovery by prefix is permitted. Unexpected content or identity leaves
    cleanup unresolved; the original evidence is never rewritten or removed.
    This is post-reap cleanup of a probe-owned0700tree, not an atomic unlink
    guarantee against arbitrary concurrent same-UID writers.
    """
    if task.get("category") != "local-fixture":
        return "not_applicable"
    if not reaped:
        return "refused_not_reaped"
    if not isinstance(setup, dict) or setup.get("schema_version") != 1 or setup.get("case_id") != task.get("id") or setup.get("manifest_sha256") != manifest_sha or setup.get("frozen_case") != task:
        return "refused_setup"
    locator = setup.get("cleanup")
    if not isinstance(locator, dict) or locator.get("kind") != "owned_home_child_v1" or locator.get("mode") != 0o700:
        return "refused_locator"
    name = locator.get("home_basename")
    if not isinstance(name, str) or not re.fullmatch(r"zephium-work-fixture-[A-Za-z0-9_-]{6,64}", name):
        return "refused_locator"
    def matches(info, identity):
        return isinstance(identity, dict) and all(type(identity.get(key)) is int for key in ("device", "inode", "uid")) and (info.st_dev, info.st_ino, info.st_uid) == (identity["device"], identity["inode"], identity["uid"]) and info.st_uid == os.getuid() == os.geteuid()
    def same(a, b):
        return (a.st_dev, a.st_ino, a.st_uid, stat.S_IFMT(a.st_mode)) == (b.st_dev, b.st_ino, b.st_uid, stat.S_IFMT(b.st_mode))
    def unchanged_file(a, b):
        return same(a,b) and (a.st_size,a.st_mtime_ns,a.st_ctime_ns,a.st_nlink,stat.S_IMODE(a.st_mode)) == (b.st_size,b.st_mtime_ns,b.st_ctime_ns,b.st_nlink,stat.S_IMODE(b.st_mode))
    def known_file(fd, info, hashes):
        if not unchanged_file(info,os.fstat(fd)):
            return False
        os.lseek(fd,0,os.SEEK_SET)
        body=bytearray()
        while len(body)<=128*1024:
            block=os.read(fd,16*1024)
            if not block:
                break
            body.extend(block)
        return len(body)<=128*1024 and digest_bytes(body) in hashes and unchanged_file(info,os.fstat(fd))
    inputs, changes = setup.get("input_hashes"), setup.get("expected_changes")
    if not isinstance(inputs, dict) or not isinstance(changes, list) or any(not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value) for value in inputs.values()):
        return "refused_locator"
    if any(not isinstance(locator.get(key), dict) or set(locator[key]) != {"device", "inode", "uid"} or any(type(value) is not int or value < 0 for value in locator[key].values()) for key in ("root_identity", "parent_identity")):
        return "refused_locator"
    allowed = {}
    for path, expected in inputs.items():
        allowed.setdefault(path, set()).add(expected)
    for change in changes:
        if not isinstance(change, dict) or not isinstance(change.get("path"), str) or not isinstance(change.get("after_sha256"), str) or not re.fullmatch(r"[0-9a-f]{64}", change["after_sha256"]):
            return "refused_locator"
        allowed.setdefault(change.get("path"), set()).add(change.get("after_sha256"))
    if not allowed or len(allowed) > 32 or any(not isinstance(path, str) or path.startswith("/") or any(part in ("", ".", "..") for part in path.split("/")) or any(not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value) for value in hashes) for path, hashes in allowed.items()):
        return "refused_locator"
    directories = {"/".join(path.split("/")[:index]) for path in allowed for index in range(1, len(path.split("/")))}
    if len(directories) > 16 or any(len(path.split("/")) > 4 for path in allowed):
        return "refused_locator"
    if not all(hasattr(os, name) for name in ("O_NOFOLLOW", "O_DIRECTORY", "getuid", "geteuid")):
        return "refused_platform"
    fds, files, dirs = [], [], []
    try:
        home_path = Path(home if home is not None else os.environ["HOME"]).resolve(strict=True)
        flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
        parent = os.open(home_path, flags)
        fds.append(parent)
        if not matches(os.fstat(parent), locator.get("parent_identity")):
            return "refused_parent_identity"
        try:
            root = os.open(name, flags, dir_fd=parent)
        except FileNotFoundError:
            return "already_missing"
        fds.append(root)
        root_info = os.fstat(root)
        if not matches(root_info, locator.get("root_identity")) or stat.S_IMODE(root_info.st_mode) != 0o700:
            return "refused_root_identity"
        pending = [(root, "")]
        seen = 0
        # Preflight the entire bounded synthetic tree before deleting any entry.
        while pending:
            directory, prefix = pending.pop()
            names = os.listdir(directory)
            seen += len(names)
            if seen > 64:
                return "refused_tree"
            for child in names:
                relative = f"{prefix}/{child}" if prefix else child
                info = os.stat(child, dir_fd=directory, follow_symlinks=False)
                if stat.S_ISDIR(info.st_mode) and relative in directories:
                    nested = os.open(child, flags, dir_fd=directory)
                    fds.append(nested)
                    if not same(info, os.fstat(nested)) or info.st_uid != root_info.st_uid or info.st_dev != root_info.st_dev:
                        return "refused_tree"
                    dirs.append((directory, child, info, relative, nested))
                    pending.append((nested, relative))
                elif stat.S_ISREG(info.st_mode) and relative in allowed and info.st_nlink == 1 and info.st_size <= 128 * 1024:
                    fd = os.open(child, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory)
                    fds.append(fd)
                    if not same(info, os.fstat(fd)) or info.st_uid != root_info.st_uid or info.st_dev != root_info.st_dev or not known_file(fd,info,allowed[relative]):
                        return "refused_tree"
                    files.append((directory, child, info, fd, relative))
                else:
                    return "refused_tree"
        def root_bound():
            return same(os.fstat(parent),os.stat(home_path,follow_symlinks=False)) and matches(os.fstat(parent),locator["parent_identity"]) and same(root_info,os.stat(name,dir_fd=parent,follow_symlinks=False)) and stat.S_IMODE(os.fstat(root).st_mode)==0o700
        def tree_current():
            if not root_bound():
                return False
            active_dirs={root}|{entry[4] for entry in dirs}
            for directory in active_dirs:
                expected={entry[1] for entry in files if entry[0]==directory}|{entry[1] for entry in dirs if entry[0]==directory}
                if set(os.listdir(directory))!=expected:
                    return False
            for directory,child,info,_,nested in dirs:
                observed=os.stat(child,dir_fd=directory,follow_symlinks=False)
                if not same(info,observed) or not same(info,os.fstat(nested)) or stat.S_IMODE(info.st_mode)!=stat.S_IMODE(observed.st_mode):
                    return False
            for directory,child,info,fd,relative in files:
                if not unchanged_file(info,os.stat(child,dir_fd=directory,follow_symlinks=False)) or not known_file(fd,info,allowed[relative]):
                    return False
            return True
        # Revalidate parent/root placement, every ancestor and retained file's
        # content before the destructive phase and before each bounded mutation.
        while files:
            if not root_bound():
                return "refused_root_changed"
            if not tree_current():
                return "refused_tree_changed"
            directory,child,_,_,_=files.pop(0)
            os.unlink(child,dir_fd=directory)
        for entry in sorted(list(dirs),key=lambda item:item[3].count("/"),reverse=True):
            if not root_bound():
                return "refused_root_changed"
            if not tree_current():
                return "refused_tree_changed"
            directory,child,_,_,_=entry
            os.rmdir(child,dir_fd=directory)
            dirs.remove(entry)
        if not root_bound():
            return "refused_root_changed"
        if not tree_current():
            return "refused_tree_changed"
        os.rmdir(name,dir_fd=parent)
        return "removed"
    except (OSError, KeyError, TypeError, ValueError):
        return "refused_or_failed"
    finally:
        for fd in reversed(fds):
            os.close(fd)


def run_local_case(root, frozen, task, out):
    directory = out / task["id"]
    directory.mkdir(mode=0o700)
    export = directory / "evidence.json"
    setup_path = export.with_suffix(".freeze.json")
    env = {key: value for key, value in os.environ.items() if not key.startswith("ZEPHIUM_") and key not in ("RUST_LOG", "RUST_BACKTRACE")}
    if digest(root / "probe") != frozen["probe_sha256"]:
        raise ValueError("frozen probe changed before launch")
    started = time.monotonic()
    timed_out = False
    child = None
    cleanup = "refused_setup"
    log_path = directory / "run.log"
    def optional(path):
        try:
            return read_json(path) if path.exists() else {}
        except (OSError,ValueError):
            return {}
    try:
        with open(os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as log:
            child = subprocess.Popen([str(root / "probe"), "--live-local-fixture", task["id"], str(export)], cwd=directory, env=env, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
            write_private(directory / "child.json", {"pid":child.pid,"process_group":child.pid,"probe_sha256":frozen["probe_sha256"]})
            try:
                code = child.wait(timeout=frozen["timeout_seconds"])
            except subprocess.TimeoutExpired:
                timed_out = True
                stop_child(child)
                code = child.returncode
        wall = int((time.monotonic() - started) * 1000)
        if digest(root / "probe") != frozen["probe_sha256"]:
            raise ValueError("frozen probe changed during launch")
        evidence, setup = optional(export), optional(setup_path)
        log = log_path.read_text(errors="replace")
    finally:
        if child is not None:
            original_error = sys.exc_info()[0] is not None
            try:
                if child.poll() is None:
                    stop_child(child)
                child.wait()
                cleanup = cleanup_local_fixture(task,optional(setup_path),frozen["manifest_sha256"],reaped=child.poll() is not None)
                write_private(directory / "cleanup.json", {"schema_version":1,"child_reaped":child.poll() is not None,"result":cleanup,
                    "scope":"Stopped probe-owned0700fixture; bindings/content revalidated, no atomic guarantee against arbitrary concurrent same-UID writers."})
            except BaseException:
                if not original_error:
                    raise
    row = {"id":task["id"], "eligibility":"eligible_synthetic_fixture", "exit":code, "timed_out":timed_out,
           "process_ms":wall, "process_completed":code==0, "evidence_sha256":digest(export) if export.exists() else None,
           "fixture_freeze_sha256":digest(setup_path) if setup_path.exists() else None,
           "log_sha256":digest(log_path), "metrics":local_metrics(evidence,log), "fixture_cleanup":cleanup,
           **grade_local(task,evidence,setup,frozen["manifest_sha256"])}
    write_private(directory / "result.json", row)
    return row


def run_case(root, frozen, task, reference, manifest, out):
    directory = out / task["id"]
    directory.mkdir(mode=0o700)
    export = directory / "evidence.json"
    env = dict(os.environ)
    for key in list(env):
        if key.startswith("ZEPHIUM_") or key in ("RUST_LOG", "RUST_BACKTRACE"):
            env.pop(key)
    env.update(ZEPHIUM_EVAL_PUBLIC_EXPORT=str(export), ZEPHIUM_EVAL_CASE_ID=task["id"], ZEPHIUM_EVAL_READ_DECISIONS=frozen.get("read_decisions", "recommended"))
    form = task.get("eligibility") == "pending_form_state_oracle"
    if form:
        if frozen.get("form_oracle", {}).get("version") != 2 or task["id"] not in FORM_CHECK_COUNTS:
            raise ValueError("form task requires an explicit frozen observer")
        env["ZEPHIUM_EVAL_FORM_ORACLE"] = task["id"]
    if digest(root / "probe") != frozen["probe_sha256"]:
        raise ValueError("frozen probe changed before launch")
    command = [str(root / "probe"), "--live-lead-request", task_request(task),
               "Use only the specified public sources. Do not sign in, submit forms, send messages, or access local files."]
    started = time.monotonic()
    timed_out = False
    log_path = directory / "run.log"
    with open(os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as log:
        child = subprocess.Popen(command, cwd=directory, env=env, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
        try:
            code = child.wait(timeout=frozen["timeout_seconds"])
        except subprocess.TimeoutExpired:
            timed_out = True
            stop_child(child)
            code = child.returncode
        except BaseException:
            stop_child(child)
            raise
    wall = int((time.monotonic() - started) * 1000)
    if digest(root / "probe") != frozen["probe_sha256"]:
        raise ValueError("frozen probe changed during launch")
    evidence = read_json(export) if export.exists() else {}
    log = log_path.read_text(errors="replace")
    row = {"id": task["id"], "eligibility": "eligible_public", "exit": code, "timed_out": timed_out,
           "process_ms": wall, "process_completed": code == 0, "evidence_sha256": digest(export) if export.exists() else None,
           "log_sha256": digest(log_path), "metrics": metrics(evidence, log), **grade(task, evidence, reference, manifest)}
    if form:
        row["eligibility"] = "eligible_form_observer"
        if row.get("reason") == "source_audit_required":
            row.update(grade_form(task, evidence, log))
        row["instrumentation_timeout"] = any('"reason":"timed_out"' in line.replace(" ", "") for line in log.splitlines() if line.startswith("form-oracle: "))
    write_private(directory / "result.json", row)
    return row


def distribution(values, excluded_runs=0):
    known = sorted(value for value in values if isinstance(value, (int, float)) and not isinstance(value, bool))
    return {"known_runs": len(known), "unknown_runs": len(values) - len(known),
            "excluded_unattempted_runs": excluded_runs,
            "median": statistics.median(known) if known else None,
            "p90": known[(9 * len(known) + 9) // 10 - 1] if known else None,
            "quantile_method": "nearest rank: sorted[ceil(0.90*n)-1]; median averages the two middle values"}


def summarize(rows):
    attempted = [row for row in rows if "exit" in row]
    outcomes = collections.Counter(row.get("task_outcome", "pending") for row in rows)
    eligible = sum(row["eligibility"] == "eligible_public" for row in rows)
    costs = [row["metrics"]["reported_cost_micro_usd"] for row in attempted]
    times = [row.get("process_ms") for row in attempted]
    excluded_runs = len(rows) - len(attempted)
    distributions = {
        "reported_cost_micro_usd": distribution(costs, excluded_runs),
        "selected_client_calls": distribution([row["metrics"].get("selected_model_call_attempts") for row in attempted], excluded_runs),
        "process_latency_ms": distribution(times, excluded_runs),
        "work_wall_latency_ms": distribution([row["metrics"].get("wall_ms") for row in attempted], excluded_runs),
        "fully_observed_cost_micro_usd": distribution([row["metrics"].get("cost_for_target_micro_usd") for row in attempted], excluded_runs),
        "post_browser_finalization_gap_ms": distribution([gap for row in attempted for gap in (row["metrics"].get("timing", {}).get("post_browser_finalization_gaps_ms", []) or [None])], excluded_runs),
    }
    cost_groups = collections.defaultdict(list)
    for row in attempted:
        basis = "+".join(sorted(set(row["metrics"].get("accounting_bases") or ["unknown"])))
        cost_groups[basis].append(row["metrics"].get("reported_cost_micro_usd"))
    internal, taxonomy, handoffs = collections.Counter(), collections.Counter(), collections.Counter()
    for row in attempted:
        internal.update(row["metrics"]["internal_failures"])
        taxonomy.update(row["metrics"]["failure_taxonomy"])
        handoffs.update(row["metrics"]["resolved_handoffs"])
    return {"tasks": len(rows), "eligible_public": eligible, "attempted": len(attempted), "task_outcomes": dict(outcomes),
            "verified_passes_of_attempted": f"{outcomes['passed']}/{len(attempted)}",
            "process_completions": sum(row["process_completed"] for row in attempted),
            "source_audit_root_causes": dict(collections.Counter(row["root_cause"] for row in attempted if row.get("review") and row.get("root_cause") not in (None, "none"))),
            "reported_cost_micro_usd_known": sum(cost for cost in costs if cost is not None), "unknown_cost_runs": sum(cost is None for cost in costs),
            "latency_median_ms": distributions["process_latency_ms"]["median"],
            "distributions": distributions,
            "reported_cost_by_accounting_basis": {basis: distribution(values) for basis, values in sorted(cost_groups.items())},
            "latency_qualification": {"timed_out_runs": sum(bool(row.get("timed_out")) for row in attempted),
                                      "timeout_policy": "Include observed time to timeout; censored observations are not completed-task latency."},
            "selected_model_call_attempts": sum(row["metrics"]["selected_model_call_attempts"] or 0 for row in attempted),
            "unknown_selected_model_call_runs": sum(row["metrics"]["selected_model_call_attempts"] is None for row in attempted),
            "primary_decision_calls_known": sum(row["metrics"].get("primary_decision_calls") or 0 for row in attempted),
            "unknown_primary_decision_call_runs": sum(row["metrics"].get("primary_decision_calls") is None for row in attempted),
            "unknown_emulation_call_runs": sum(row["metrics"].get("emulation_calls_overlapping_selected_client") is None for row in attempted),
            "emulation_calls_overlapping_selected_client_known": sum(row["metrics"].get("emulation_calls_overlapping_selected_client") or 0 for row in attempted),
            "internal_failures": dict(internal), "failure_taxonomy": dict(taxonomy), "resolved_handoffs": dict(handoffs),
            "qualification": "Constrained deterministic predicates or recorded source audits only; operational completion and historical tuned-site passes are not task correctness."}


def run_once(args):
    root, frozen = verify_freeze(args.freeze)
    manifest = evaluation_manifest(load_manifest(root / "manifest.json"), frozen)
    references = read_json(root / "references.json")["tasks"]
    requested = set(args.tasks or [])
    if requested - {task["id"] for task in manifest["tasks"]}:
        raise ValueError("unknown requested task")
    out = Path(args.out).resolve()
    out.mkdir(mode=0o700, parents=True, exist_ok=False)
    run_id = uuid.uuid4().hex
    write_private(out / "attempt.json", {"run_id": run_id, "freeze_sha256": digest(args.freeze), "started_at": now()})
    rows = []
    with live_lease():
        signature(root / "probe")
        for task in manifest["tasks"]:
            selected = not requested or task["id"] in requested
            reason = task["eligibility"]
            if not selected:
                row = {"id": task["id"], "eligibility": reason, "task_outcome": "pending", "reason": "not_selected"}
            elif reason != "eligible_public" and not (reason == "pending_form_state_oracle" and frozen.get("form_oracle", {}).get("version") == 2) and not (reason in LOCAL_ELIGIBILITY and frozen.get("local_fixtures", {}).get("version") == 1):
                row = {"id": task["id"], "eligibility": reason, "task_outcome": "pending", "reason": reason}
            elif task.get("oracle") and references.get(task["id"], {}).get("state") != "ready":
                row = {"id": task["id"], "eligibility": "blocked_reference", "task_outcome": "pending", "reason": "reference_unavailable"}
            else:
                print(f"{task['id']}: running", flush=True)
                row = run_local_case(root, frozen, task, out) if reason in LOCAL_ELIGIBILITY else run_case(root, frozen, task, references.get(task["id"]), manifest, out)
                print(f"{task['id']}: task={row['task_outcome']} process={row['exit']} {row['process_ms']}ms", flush=True)
            row.update(category=task["category"], grading=task["grading"])
            rows.append(row)
            # Append-only progress survives interruptions; a resumed run gets a new directory.
            with open(os.open(out / "progress.jsonl", os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600), "a") as stream:
                stream.write(json.dumps(row) + "\n")
    record = {"schema_version": SCHEMA, "run_id": run_id, "freeze": str(Path(args.freeze).resolve()), "freeze_sha256": digest(args.freeze),
              "started_from": frozen["label"], "finished_at": now(), "rows": rows, "summary": summarize(rows)}
    write_private(out / "run.json", record)
    print(json.dumps(record["summary"], indent=2))
    print(out / "run.json")


def reliability(runs):
    if len(runs) != 3 or len({run["freeze_sha256"] for run in runs}) != 1:
        raise ValueError("three runs of exactly the same frozen setup are required")
    identities = [run.get("run_id") for run in runs]
    if any(not isinstance(identity, str) or not re.fullmatch(r"[0-9a-f]{32}", identity) for identity in identities) or len(set(identities)) != 3:
        raise ValueError("three independently recorded attempt identities are required")
    indexes = [{row["id"]: row for row in run["rows"]} for run in runs]
    if any(len(index) != len(run["rows"]) for index, run in zip(indexes, runs)) or any(set(index) != set(indexes[0]) for index in indexes[1:]):
        raise ValueError("each run must contain the same task IDs exactly once")
    tasks = [key for key, row in indexes[0].items() if row["eligibility"] == "eligible_public"]
    if not tasks:
        raise ValueError("no public tasks")
    if any(index[key].get("eligibility") != "eligible_public" for index in indexes for key in tasks):
        raise ValueError("eligibility changed between attempts")
    outcomes = [[index[key].get("task_outcome", "pending") for index in indexes] for key in tasks]
    known = {"passed", "partial", "failed"}
    return {
        "tasks": len(tasks), "attempts": len(tasks) * 3,
        "pass_at_1": sum(row[0] == "passed" for row in outcomes) / len(tasks),
        "pass_power_3": sum(all(value == "passed" for value in row) for row in outcomes) / len(tasks),
        "ungraded_attempts": sum(value not in known for row in outcomes for value in row),
        "fully_graded": all(value in known for row in outcomes for value in row),
        "qualification": "Until fully_graded, rates are lower bounds; partial, unknown, missing and completed-without-a-grade are never passes. pass_at_1 uses the first attempt; pass_power_3 requires all three.",
        "attempt_summary": summarize([index[key] for index in indexes for key in tasks]),
        "outcomes": dict(zip(tasks, outcomes)),
    }


def run(args):
    repetitions = getattr(args, "repetitions", 1)
    if repetitions == 1:
        return run_once(args)
    out = Path(args.out).resolve()
    out.mkdir(mode=0o700, parents=True, exist_ok=False)
    for attempt in range(1, repetitions + 1):
        one = argparse.Namespace(**vars(args))
        one.out = str(out / f"attempt-{attempt}")
        run_once(one)
    if repetitions == 3:
        runs = [read_json(out / f"attempt-{attempt}" / "run.json") for attempt in (1, 2, 3)]
        write_private(out / "round.json", reliability(runs))


def round_report(args):
    if len(set(str(Path(path).resolve()) for path in args.runs)) != 3:
        raise ValueError("the same run cannot count as multiple attempts")
    runs = [read_json(path) for path in args.runs]
    report = reliability(runs)
    report["inputs"] = [{"path": str(Path(path).resolve()), "sha256": digest(path)} for path in args.runs]
    write_private(args.out, report)
    print(json.dumps({key: value for key, value in report.items() if key not in ("outcomes", "attempt_summary")}, indent=2))


def predicate_outcome(checks):
    states = [check["state"] for check in checks]
    if not states or any(state not in ("passed", "partial", "failed", "unknown") for state in states):
        raise ValueError("invalid predicate states")
    if "failed" in states:
        return "failed"
    if "unknown" in states:
        return "unknown"
    if "partial" in states:
        return "partial"
    return "passed"


def validate_source_audit_binding(entry, row):
    # Bind judgments to the captured evidence, not merely a task name reused
    # in another repetition. Missing exports cannot substantiate an audit.
    for key in ("evidence_sha256", "log_sha256"):
        expected = row.get(key)
        if not isinstance(expected, str) or not re.fullmatch(r"[0-9a-f]{64}", expected) or entry.get(key) != expected:
            raise ValueError("source audit does not match this repetition's evidence and log")
    cause = entry.get("root_cause")
    if not isinstance(cause, str) or not re.fullmatch(r"[a-z][a-z0-9_]{0,79}", cause):
        raise ValueError("source audit requires a bounded root-cause category")
    notes = entry.get("quality_notes", [])
    if not isinstance(notes, list) or len(notes) > 16 or any(not isinstance(note, str) or len(note) > 2048 for note in notes):
        raise ValueError("source audit quality notes exceed their limits")


def review(args):
    run = read_json(args.run)
    audit = read_json(args.audit)
    if audit.get("run_sha256") != digest(args.run) or audit.get("reviewer_type") not in ("human", "agent_source_audit"):
        raise ValueError("audit must identify its exact run and source-audit reviewer type")
    if not audit.get("reviewer"):
        raise ValueError("audit reviewer is required")
    manifest = evaluation_manifest(load_manifest(Path(run["freeze"]).parent / "manifest.json"), read_json(run["freeze"]))
    tasks = {task["id"]: task for task in manifest["tasks"]}
    rows = {row["id"]: row for row in run["rows"]}
    for entry in audit["tasks"]:
        task, row = tasks[entry["id"]], rows[entry["id"]]
        if task["grading"] != "source_audit" or "exit" not in row or row.get("reason") != "source_audit_required":
            raise ValueError("only attempted, uncontaminated source-audit tasks can be reviewed")
        if manifest["suite"] != "work-heldout-v1":
            validate_source_audit_binding(entry, row)
        checks = entry["predicates"]
        if {check["id"] for check in checks} != {check["id"] for check in task["predicates"]} or len(checks) != len(task["predicates"]):
            raise ValueError("audit must address every declared predicate exactly once")
        allowed_states = ("passed", "failed", "unknown") if manifest["suite"] == "work-heldout-v1" else ("passed", "partial", "failed", "unknown")
        if any(check["state"] not in allowed_states or not check.get("evidence") or not check.get("reason") for check in checks):
            raise ValueError("each audit result needs concrete evidence references and reasoning")
        if "root_cause" in entry:
            row["root_cause"] = entry["root_cause"]
        if "quality_notes" in entry:
            row["quality_notes"] = entry["quality_notes"]
        row["task_outcome"] = predicate_outcome(checks)
        row["predicates"] = checks
        row["review"] = {"reviewer": audit["reviewer"], "type": audit["reviewer_type"], "audit_sha256": digest(args.audit), "processor_sha256": digest(__file__)}
        row["reason"] = "recorded_source_audit"
    run["parent_run_sha256"] = digest(args.run)
    run["summary"] = summarize(run["rows"])
    write_private(args.out, run)
    print(json.dumps(run["summary"], indent=2))


def compare(args):
    a, b = read_json(args.before), read_json(args.after)
    freezes = [read_json(run["freeze"]) for run in (a, b)]
    matching = all(freezes[0].get(key) == freezes[1].get(key) for key in
                   ("manifest_sha256", "references_sha256", "runner_sha256", "model", "platform", "python", "timeout_seconds", "evaluation_lane", "read_decisions"))
    form_versions = [record.get("form_oracle", {}).get("version") for record in freezes]
    if any(form_versions):
        matching = matching and form_versions == [2, 2] and bool(freezes[0]["form_oracle"].get("predicate_files")) and freezes[0]["form_oracle"].get("predicate_files") == freezes[1]["form_oracle"].get("predicate_files")
    local_versions = [record.get("local_fixtures", {}).get("version") for record in freezes]
    if any(local_versions):
        matching = matching and local_versions == [1, 1] and bool(freezes[0]["local_fixtures"].get("predicate_files")) and freezes[0]["local_fixtures"].get("predicate_files") == freezes[1]["local_fixtures"].get("predicate_files")
    if not matching:
        raise ValueError("unmatched manifest/reference/grader/model/platform/setup; no comparative success claim")
    old = {row["id"]: row for row in a["rows"]}
    changes = [{"id": row["id"], "before": old.get(row["id"], {}).get("task_outcome", "missing"),
                "after": row["task_outcome"], "before_metrics": old.get(row["id"], {}).get("metrics"), "after_metrics": row.get("metrics")}
               for row in b["rows"]]
    print(json.dumps({"comparison": "matched recorded setup; live-site drift and stochastic variation remain", "rows": changes}, indent=2))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest="command", required=True)
    listing = commands.add_parser("list")
    listing.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
    preparation = commands.add_parser("prepare")
    preparation.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
    preparation.add_argument("--out", required=True)
    freezing = commands.add_parser("freeze")
    freezing.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
    freezing.add_argument("--references", required=True)
    freezing.add_argument("--probe", required=True)
    freezing.add_argument("--identifier", default="app.zephium.work-integration.probe")
    freezing.add_argument("--source-root", required=True)
    freezing.add_argument("--base-ref", default="3fcd07c5")
    freezing.add_argument("--label", choices=["baseline", "candidate"], required=True)
    freezing.add_argument("--form-oracle", action="store_true", help="freeze exact probe-only final-form, action, navigation and submit-event observers")
    freezing.add_argument("--local-fixtures", action="store_true", help="freeze exact synthetic files/Blocks fixtures and independent predicates")
    freezing.add_argument("--tuned-matrix", action="store_true", help="separate candidate lane for exact 23 tuned goals; source audits required")
    freezing.add_argument("--matrix-source", type=Path, help="original matrix.sh literal-pair provenance; never executed")
    freezing.add_argument("--timeout", type=int, default=300)
    freezing.add_argument("--read-decisions", choices=["recommended", "emulation", "disabled"], default="recommended")
    freezing.add_argument("--out", required=True)
    running = commands.add_parser("run")
    running.add_argument("--freeze", required=True)
    running.add_argument("--out", required=True)
    running.add_argument("--tasks", nargs="*")
    running.add_argument("--repetitions", type=int, choices=[1, 3], default=1)
    reporting = commands.add_parser("round-report")
    reporting.add_argument("--runs", nargs=3, required=True)
    reporting.add_argument("--out", required=True)
    reviewing = commands.add_parser("review")
    reviewing.add_argument("--run", required=True)
    reviewing.add_argument("--audit", required=True)
    reviewing.add_argument("--out", required=True)
    comparison = commands.add_parser("compare")
    comparison.add_argument("before")
    comparison.add_argument("after")
    args = parser.parse_args()
    if args.command == "list":
        manifest = load_manifest(args.manifest)
        print(json.dumps({"tasks": len(manifest["tasks"]), "eligibility": dict(collections.Counter(task["eligibility"] for task in manifest["tasks"])),
                          "grading": dict(collections.Counter(task["grading"] for task in manifest["tasks"])),
                          "tuned_cases_excluded": len(manifest["tuned_cases"]), "ids": [task["id"] for task in manifest["tasks"]]}, indent=2))
    else:
        {"prepare": prepare_references, "freeze": freeze, "run": run, "review": review, "compare": compare, "round-report": round_report}[args.command](args)


if __name__ == "__main__":
    try:
        main()
    except (ValueError, OSError, subprocess.SubprocessError) as error:
        sys.exit(f"work-eval: {error}")
