#!/usr/bin/env python3
"""Persistent, independently verified Blender Web migration supervisor."""
from __future__ import annotations

import argparse
from contextlib import contextmanager
import fcntl
import hashlib
import json
import os
import re
import shutil
import signal
import subprocess
import sys
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path

from wbv2_timing import WorkerTrace, command_phase, display_times, format_report, minute_label, timing_report

ROOT = Path(__file__).resolve().parents[1]
CONTROL = "tests/golden/WBV2/autopilot"
POLICY = "my-tools/wbv2_autopilot_policy.json"
REQUIRED = {"source_build", "typecheck", "lint", "chromium", "positive", "negative", "undo_redo", "save_reopen", "determinism"}
ACTIVE = {"implementing", "validating", "finalizing", "committing", "committed", "pushing", "pushed"}


class ProviderUnavailable(ValueError):
    """No agent action started before a bounded provider failure."""


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


def read_json(path, default=None):
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        if default is not None:
            return default
        raise


def write_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + f".tmp-{os.getpid()}")
    with temporary.open("w", encoding="utf-8") as stream:
        json.dump(value, stream, ensure_ascii=True, indent=2)
        stream.write("\n")
        stream.flush()
        os.fsync(stream.fileno())
    temporary.replace(path)


def sha(path):
    if not path.is_file():
        return None
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()


def contained(root, relative):
    if not isinstance(relative, str) or not relative or Path(relative).is_absolute() or ".." in Path(relative).parts:
        raise ValueError(f"invalid relative path: {relative!r}")
    result = (root / relative).resolve()
    if not result.is_relative_to(root.resolve()):
        raise ValueError(f"path escapes repository: {relative}")
    return result


def alive(pid):
    if not isinstance(pid, int) or pid <= 0:
        return False
    try:
        os.kill(pid, 0)
        return True
    except ProcessLookupError:
        return False
    except PermissionError:
        return True


def identity(pid):
    try:
        return Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[19]
    except (OSError, IndexError):
        return None


class Lease:
    def __init__(self, path, inherited=None):
        self.path = path
        self.inherited = inherited
        self.transferred = False

    def __enter__(self):
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.stream = os.fdopen(self.inherited, "a+") if self.inherited is not None else self.path.open("a+")
        if os.fstat(self.stream.fileno()).st_ino != self.path.stat().st_ino:
            self.stream.close()
            raise RuntimeError("inherited lease does not match the owner lock")
        try:
            fcntl.flock(self.stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            self.stream.close()
            raise RuntimeError("another Python autopilot owns the project") from None
        self.stream.seek(0)
        self.stream.truncate()
        json.dump({"pid": os.getpid(), "identity": identity(os.getpid()), "startedAt": now()}, self.stream)
        self.stream.flush()
        return self

    def __exit__(self, *_):
        # Never unlink a flock inode while another process can hold it open.
        if not self.transferred:
            fcntl.flock(self.stream, fcntl.LOCK_UN)
        self.stream.close()


class ProjectFence:
    """Share the existing scheduler's exclusive marker with Python runs."""
    def __init__(self, root):
        self.directory = root / "tests/golden/WBV2"
        self.path = self.directory / "continuous-scheduler.lock"

    def __enter__(self):
        self.directory.mkdir(parents=True, exist_ok=True)
        for name in ("continuous-scheduler.lock", "continuous-guardian.lock", "continuous-supervisor.lock", "web-product-migration.lock"):
            owner = read_json(self.directory / name, {})
            if alive(owner.get("pid")):
                raise RuntimeError(f"another project runner is active: {name}")
        try:
            with self.path.open("x", encoding="utf-8") as stream:
                json.dump({"pid": os.getpid(), "identity": identity(os.getpid()), "runtime": "python-autopilot", "startedAt": now()}, stream)
        except FileExistsError:
            raise RuntimeError("stale scheduler lock requires reconciliation before Python can claim it") from None
        return self

    def __exit__(self, *_):
        owner = read_json(self.path, {})
        if owner.get("pid") == os.getpid() and owner.get("runtime") == "python-autopilot":
            self.path.unlink()


def queue_context(root):
    queue = (root / "docs/EXECUTION_QUEUE.md").read_text(encoding="utf-8")
    task = re.search(r"\| 当前任务 \| `([^`]+)` \|", queue)
    parent = re.search(r"\| parent manifest \| `([^`]+)` \|", queue)
    if not task or not parent or not re.fullmatch(r"WBV2-P\d+-\d+", task[1]):
        raise ValueError("authoritative queue pointer is malformed")
    manifest = read_json(contained(root, parent[1]))
    if manifest.get("status") != "done" or manifest.get("nextTask") != task[1]:
        raise ValueError("parent manifest does not authorize the current task")
    card = f"docs/tasks/{task[1]}.md"
    if not (root / card).is_file():
        raise ValueError("current task card is missing")
    return {"task": task[1], "parentManifest": parent[1], "card": card, "parentSha256": sha(root / parent[1])}


def manifest_proof(root, task):
    manifest = read_json(contained(root, f"tests/golden/{task}/manifest.json"), {})
    reasons = []
    if manifest.get("task") != task or manifest.get("status") != "done":
        reasons.append("task_not_done")
    if manifest.get("verificationDeferred") or manifest.get("acceptanceComplete") is not True:
        reasons.append("runtime_acceptance_not_complete")
    artifacts = manifest.get("artifacts", {})
    if not isinstance(artifacts, dict) or not artifacts:
        reasons.append("artifacts_missing")
    else:
        for name, item in artifacts.items():
            try:
                if not item.get("sha256") or sha(contained(root, item["path"])) != item["sha256"]:
                    reasons.append(f"artifact_drift:{name}")
            except (KeyError, ValueError, AttributeError):
                reasons.append(f"artifact_invalid:{name}")
    return {"ok": not reasons, "reasons": reasons, "nextTask": manifest.get("nextTask")}


def inventory_proof(root):
    inventory = read_json(root / "tests/golden/WBV2/dynamic-inventory.json")
    units = inventory.get("units", [])
    unverified = sum(unit.get("coverageStatus") != "VERIFIED" for unit in units)
    invalid, checked = 0, {}
    for unit in units:
        if unit.get("coverageStatus") != "VERIFIED":
            continue
        matches = True
        for key in ("evidenceManifest", "productionEvidence"):
            item = unit.get("coverage", {}).get(key, {})
            try:
                signature = (item["path"], item["sha256"])
                if signature not in checked:
                    checked[signature] = bool(item["sha256"]) and sha(contained(root, item["path"])) == item["sha256"]
                matches = matches and checked[signature]
            except (KeyError, ValueError):
                matches = False
        invalid += not matches
    return {"ok": bool(units) and unverified == 0 and invalid == 0, "unitCount": len(units), "recordedVerified": len(units) - unverified, "unverifiedUnits": unverified, "unitsWithInvalidReferences": invalid}


def result_schema():
    strings = {"type": "array", "items": {"type": "string"}}
    def obj(properties):
        return {"type": "object", "additionalProperties": False, "required": list(properties), "properties": properties}
    check = obj({"id": {"type": "string"}, "kinds": strings, "argv": strings, "cwd": {"type": "string"}})
    case = obj({"id": {"type": "string"}, "kind": {"type": "string"}, "checkId": {"type": "string"}, "receipt": {"type": "string"}, "unitIds": strings})
    return obj({"status": {"type": "string", "enum": ["ready", "blocked"]}, "summary": {"type": "string"}, "workflow": {"type": "string"}, "routeIds": strings, "sourceReferences": strings, "changedFiles": strings, "sourceUnits": strings, "dependencies": strings, "artifacts": strings, "checks": {"type": "array", "items": check}, "cases": {"type": "array", "items": case}})


def validate_command(root, argv, cwd):
    if not argv or not all(isinstance(v, str) and v and "\0" not in v for v in argv):
        raise ValueError("command must be an argv array")
    directory = root if cwd == "." else contained(root, cwd)
    if not directory.is_dir():
        raise ValueError("command cwd does not exist")
    if argv[0] not in {"node", "python3", "npm", "cmake", "ninja", "ctest", "git"}:
        raise ValueError("unsupported validation executable")
    if any(arg in {"-e", "--eval", "-c", "--command", "--input-type=module"} for arg in argv[1:]):
        raise ValueError("inline code is not a validation entrypoint")
    if any(re.search(r"firefox|webkit", arg, re.I) for arg in argv):
        raise ValueError("only Chromium is permitted")
    if argv[0] == "git" and argv != ["git", "diff", "--check"]:
        raise ValueError("validation cannot change Git state")
    if argv[0] in {"node", "python3"}:
        scripts = [arg for arg in argv[1:] if arg.endswith((".mjs", ".js", ".py"))]
        if not scripts or any(not (directory / p).resolve().is_relative_to(root) or not (directory / p).is_file() for p in scripts):
            raise ValueError("a concrete repository validation script is required")
    if argv[0] == "npm" and "run" not in argv:
        raise ValueError("only repository npm run scripts are permitted")
    return argv, directory


class Autopilot:
    def __init__(self, root=ROOT, policy=None):
        self.root = root.resolve()
        self.directory = self.root / CONTROL
        self.policy = policy or read_json(self.root / POLICY)
        if self.policy.get("sourceFirst") is not True:
            raise ValueError("source-first business workflow policy is mandatory")
        if not REQUIRED.issubset(self.policy.get("requiredChecks", [])):
            raise ValueError("policy cannot omit mandatory behavioral gates")
        for key in ("workerTimeoutSeconds", "checkTimeoutSeconds", "heartbeatSeconds", "maxAttempts", "maxTokensPerRun", "maxLogBytes", "maxBuildJobs", "monitorPollSeconds", "modelStartTimeoutSeconds", "modelIdleTimeoutSeconds", "providerBackoffSeconds", "providerMaxBackoffSeconds", "maxProviderFailures"):
            if not isinstance(self.policy.get(key), (float, int)) or self.policy[key] <= 0:
                raise ValueError(f"invalid positive policy limit: {key}")
        self.state_path = self.directory / "state.json"
        self.state = read_json(self.state_path, {"schemaVersion": 1, "state": "idle", "jobs": {}, "tokensUsed": 0})
        self.stop_requested = False
        self.child = None
        self.run_tokens = 0

    def checkpoint(self, state, **values):
        self.state.update(state=state, updatedAt=now(), pid=os.getpid(), identity=identity(os.getpid()), **values)
        write_json(self.state_path, self.state)
        write_json(self.root / "tests/golden/WBV2/web-product-migration-status.json", {"schemaVersion": 2, "pid": os.getpid(), "status": state, "updatedAt": now(), "activeTask": self.state.get("activeJob"), "checkpoint": str(self.state_path.relative_to(self.root)), "verifiedJobs": sum(j.get("state") == "verified" for j in self.state["jobs"].values()), "reason": self.state.get("reason")})

    def event(self, kind, **values):
        self.directory.mkdir(parents=True, exist_ok=True)
        record = {"time": now(), "event": kind, **values}
        line = json.dumps(record, ensure_ascii=True)
        with (self.directory / "events.jsonl").open("a", encoding="utf-8") as stream:
            stream.write(line + "\n")
        print(json.dumps(display_times({**record, "time": minute_label(record["time"])}), ensure_ascii=True), flush=True)

    @contextmanager
    def timed(self, phase):
        started, wall = time.monotonic(), now()
        try:
            yield
        finally:
            self.event("timing.span", phase=phase, startedAt=wall, finishedAt=now(), durationSeconds=round(time.monotonic() - started, 6))

    def request_stop(self, *_):
        self.stop_requested = True
        if self.child and self.child.poll() is None:
            try:
                os.killpg(self.child.pid, signal.SIGTERM)
            except ProcessLookupError:
                pass

    def run_process(self, argv, directory, log, timeout, stdin=None, env=None, monitor_worker=False, phase=None):
        log.parent.mkdir(parents=True, exist_ok=True)
        started = time.monotonic()
        phase = phase or command_phase(argv)
        result = {"argv": argv, "cwd": str(directory.relative_to(self.root)), "phase": phase, "startedAt": now(), "log": str(log.relative_to(self.root))}
        trace = WorkerTrace() if monitor_worker else None
        provider_abort = None
        next_heartbeat = started + self.policy["heartbeatSeconds"]
        last_provider_error = None
        interrupted = timed_out = log_limit = False
        self.event("process.started", **result)
        with log.open("wb") as output:
            self.child = subprocess.Popen(argv, cwd=directory, env={**os.environ, **(env or {})}, stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL, stdout=output, stderr=subprocess.STDOUT, start_new_session=True)
            self.checkpoint(self.state["state"], childPid=self.child.pid, childIdentity=identity(self.child.pid), currentProcess={**result, "status": "running", "elapsedSeconds": 0, "phase": "provider_wait" if trace else phase})
            if stdin is not None:
                try:
                    self.child.stdin.write(stdin.encode())
                    self.child.stdin.close()
                except BrokenPipeError:
                    pass
            while self.child.poll() is None:
                elapsed = time.monotonic() - started
                if trace:
                    trace.read(log, elapsed)
                    if trace.provider_error and not trace.actions_seen:
                        provider_abort = "provider_unavailable"
                    elif not trace.actions_seen and elapsed >= self.policy.get("modelStartTimeoutSeconds", 60):
                        provider_abort = "model_start_timeout"
                    elif trace.actions_seen and not trace.active_items and elapsed - trace.last_progress >= self.policy.get("modelIdleTimeoutSeconds", 180):
                        provider_abort = "model_idle_timeout"
                interrupted = self.stop_requested or (self.directory / "pause.request").exists()
                timed_out = elapsed > timeout
                log_limit = log.stat().st_size > self.policy["maxLogBytes"]
                if interrupted or timed_out or log_limit or provider_abort:
                    try:
                        os.killpg(self.child.pid, signal.SIGTERM)
                        self.child.wait(timeout=5)
                    except ProcessLookupError:
                        pass
                    except subprocess.TimeoutExpired:
                        os.killpg(self.child.pid, signal.SIGKILL)
                        self.child.wait()
                    break
                changed_error = trace and trace.provider_error != last_provider_error
                if time.monotonic() >= next_heartbeat or changed_error:
                    current = {**result, "status": "running", "elapsedSeconds": round(elapsed, 3)}
                    if trace:
                        current.update(phase="provider_wait" if not trace.actions_seen else "agent_work", providerError=trace.provider_error, firstActionSeconds=trace.first_action, idleSeconds=round(elapsed - trace.last_progress, 3), activeItems=list(trace.active_items.values()))
                        last_provider_error = trace.provider_error
                    self.checkpoint(self.state["state"], heartbeatAt=now(), currentProcess=current)
                    next_heartbeat = time.monotonic() + self.policy["heartbeatSeconds"]
                wait_for = max(0.001, min(self.policy.get("monitorPollSeconds", 0.25), next_heartbeat - time.monotonic(), timeout - elapsed))
                try:
                    self.child.wait(timeout=wait_for)
                except subprocess.TimeoutExpired:
                    pass
            code = self.child.wait()
            interrupted = interrupted or self.stop_requested or (self.directory / "pause.request").exists()
            if trace:
                trace.read(log, time.monotonic() - started)
            try:
                os.killpg(self.child.pid, signal.SIGTERM)
                time.sleep(0.1)
                os.killpg(self.child.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
            self.child = None
        elapsed = time.monotonic() - started
        result.update(exitCode=code, durationSeconds=round(elapsed, 6), interrupted=interrupted, timedOut=timed_out, logLimit=log_limit, providerAbort=provider_abort, logSha256=sha(log), finishedAt=now())
        if trace:
            result["workerTrace"] = trace.snapshot(elapsed)
        self.state.pop("childPid", None)
        self.state.pop("childIdentity", None)
        write_json(log.with_suffix(log.suffix + ".process.json"), result)
        self.event("process.finished", **result)
        self.checkpoint(self.state["state"], currentProcess={"phase": phase, "startedAt": result["startedAt"], "finishedAt": result["finishedAt"], "status": "finished", "durationSeconds": result["durationSeconds"], "providerAbort": provider_abort})
        return result

    def protected_hashes(self, context):
        files = ["docs/EXECUTION_QUEUE.md", "nextTask.md", context["parentManifest"], "tests/golden/WBV2/web-product-backlog.json", "tests/golden/WBV2/dynamic-inventory.json", "tests/golden/WBV2/task-index.json", "tests/golden/WBV2-P0-002/parity-map.json", "my-tools/wbv2_autopilot.py", POLICY, f"tests/golden/{context['task']}/manifest.json"]
        inventory = read_json(self.root / "tests/golden/WBV2/dynamic-inventory.json", {})
        for unit in inventory.get("units", []):
            for key in ("evidenceManifest", "productionEvidence"):
                artifact = unit.get("coverage", {}).get(key)
                if artifact:
                    files.append(artifact["path"])
        files.append("my-tools/wbv2_timing.py")
        with self.timed("evidence_hashing"):
            return {p: sha(self.root / p) for p in sorted(set(files))}

    def jobs(self, context):
        backlog = read_json(self.root / "tests/golden/WBV2/web-product-backlog.json", {})
        jobs = [dict(j) for j in backlog.get("tasks", [])] if backlog.get("queueTask") == context["task"] else [{"id": context["task"], "title": "Implement the current source-backed module", "ownedPaths": [], "implementation": "Follow the current module card and its source/case contract."}]
        priority = self.policy.get("priority", [])
        jobs.sort(key=lambda j: priority.index(j["id"]) if j["id"] in priority else len(priority))
        for job in jobs:
            if not re.fullmatch(r"[A-Z0-9_-]+", job["id"]):
                raise ValueError("unsafe job identifier")
            job["queueTask"] = context["task"]
        return jobs

    def receipt_valid(self, record):
        if record.get("state") != "verified" or not record.get("receipt"):
            return False
        try:
            receipt = read_json(contained(self.root, record["receipt"]))
            if sha(self.root / record["receipt"]) != record.get("receiptSha256"):
                return False
            return all(sha(contained(self.root, p)) == digest for p, digest in receipt["inputs"].items()) and all(sha(contained(self.root, a["path"])) == a["sha256"] for a in receipt["artifacts"])
        except (OSError, KeyError, ValueError):
            return False

    def choose_job(self, jobs):
        # Complete untouched workflows before revisiting shared-file changes.
        for job in jobs:
            if self.state["jobs"].get(job["id"], {}).get("state") != "verified":
                return job
        return next((job for job in jobs if not self.receipt_valid(self.state["jobs"][job["id"]])), None)

    def worker_prompt(self, context, job, run_dir, previous):
        receipt_dir = str((run_dir / "cases").relative_to(self.root))
        return f"""Implement one bounded Blender Web workflow in {self.root}.
Queue: {json.dumps(context, ensure_ascii=True)}
Workflow inside this task: {json.dumps(job, ensure_ascii=True)}
Previous attempt: {json.dumps(previous, ensure_ascii=True)[:4000]}
Known findings: {json.dumps(self.policy.get('knownFindings', []), ensure_ascii=True)}

Follow AGENTS.md: read the queue, print-task-context, check-task-context, then only named owned paths and bounded source searches.
Drive this task by its business module, user workflow and concrete function routes, then the exact source/operator case matrix.
FIRST search the supplied Blender 5.2 sources and task-scoped related references to resolve behavior before editing the Web implementation.
Use node tools/web/search-wbv2-batch-source.mjs --unit <source-unit-id> --pattern <symbol> to read bounded native source locations.
Blender 5.2 source/Main is authoritative. Never simulate Blender in JS or count route classification, timing, builds or bookkeeping as completed behavior.
Preserve dirty files. Reconcile previous edits before changing code. Never replay an unknown generator or external write.
Implement production code and executable focused tests, adding only directly related helpers/tests.
Do not modify queue, parent manifests, nextTask.md, backlog, historical evidence, source inventory or autopilot control files.
Do not commit, push, SSH, deploy, install services, start agents or change the live website.
Do not run expensive validation yourself: the independent supervisor executes your checks after your process exits successfully.
Return ready only when implementation and runnable tests are prepared; ready is NOT completion.
Checks use argv arrays and repo-relative cwd, with no shell or inline code and only Chromium.
Required check kinds: {', '.join(sorted(REQUIRED))}. Reuse existing build/test entrypoints; one command may cover multiple kinds.
Tests must prove positive Main mutation or exact Main reader, negative no-mutation, undo/redo, save/reopen and determinism.
Declare cases for positive, negative, undo_redo, save_reopen and determinism, bound to checkId and exact source inventory unitIds.
Tests must write fresh case receipts under {receipt_dir}/ at validation time using WBV2_RECEIPT_DIR. Do NOT write these receipts now.
Case JSON: status=PASS, runtime=CHROMIUM_WEBENGINE or BLENDER_NATIVE, caseId, kind, unitIds, observations, artifacts=[{{path,sha256}}].
observations use SHA-256 of actual canonical native Main snapshots: positive before/after differ, or reader expected/actual equal with reader=true; negative before/after equal; undo_redo before/after/undo/redo prove reversal; save_reopen saved/reopened equal; determinism first/second equal. At least one case must run in Chromium.
Use actual native/Chromium receipts, screenshots, .blend or WASM artifacts. Test assertions must verify behavior, not static markers.
List source/protocol/test/fixture/build dependencies for hash invalidation and compiled WASM/native paths in artifacts.
Return workflow (the concrete user workflow), routeIds (exact current route catalog IDs) and sourceReferences (native Blender files actually used).
sourceUnits must be existing dynamic inventory IDs. Do not claim verified coverage; the supervisor checks and hashes results.
Return blocked with a concrete reason if prerequisites cannot be established.
"""

    def worker(self, prompt, run_dir):
        schema_path, output_path = run_dir / "result.schema.json", run_dir / "worker-result.json"
        write_json(schema_path, result_schema())
        command = ["codex", "exec", "--sandbox", self.policy["sandbox"], "-c", 'approval_policy="never"', "--ephemeral", "--cd", str(self.root), "--json", "--output-schema", str(schema_path), "--output-last-message", str(output_path), "-"]
        proc = self.run_process(command, self.root, run_dir / "worker.jsonl", self.policy["workerTimeoutSeconds"], stdin=prompt, monitor_worker=True, phase="handoff_model" if self.state["state"] == "finalizing" else "model_request")
        completed = False
        for line in (run_dir / "worker.jsonl").read_text(encoding="utf-8", errors="replace").splitlines():
            try:
                event = json.loads(line)
            except json.JSONDecodeError:
                continue
            if event.get("type") == "turn.completed":
                completed = True
                usage = event.get("usage", {})
                tokens = int(usage.get("input_tokens", 0)) + int(usage.get("output_tokens", 0))
                self.run_tokens += tokens
                self.state["tokensUsed"] = self.state.get("tokensUsed", 0) + tokens
        if proc["interrupted"] or proc["timedOut"] or proc["logLimit"]:
            self.checkpoint("needs_reconcile", reason="worker interrupted; inspect edits and process receipt before resuming")
            raise InterruptedError(self.state["reason"])
        trace = proc.get("workerTrace", {})
        if proc.get("providerAbort") or trace.get("providerError"):
            if trace.get("safeToRetry"):
                raise ProviderUnavailable(trace.get("providerError") or proc["providerAbort"])
            if proc.get("providerAbort"):
                raise InterruptedError("model stalled after actions or unrecognized output; inspect the bounded trace before retry")
        if proc["exitCode"] != 0 or not completed:
            raise ValueError(f"worker failed: exit={proc['exitCode']}, turnCompleted={completed}")
        return read_json(output_path)

    def validate_result(self, result, run_dir):
        if result.get("status") != "ready":
            raise ValueError(result.get("summary", "worker reported blocked"))
        known = {u["id"] for u in read_json(self.root / "tests/golden/WBV2/dynamic-inventory.json")["units"]}
        units = set(result.get("sourceUnits", []))
        if not units or not units.issubset(known):
            raise ValueError("missing or unknown source unit bindings")
        catalog = read_json(self.root / "web/app/src/capabilities/blender-feature-route-catalog.json")
        route_ids = set(result.get("routeIds", []))
        if not result.get("workflow", "").strip() or not route_ids or not route_ids.issubset({entry["id"] for entry in catalog["entries"]}) or units != route_ids:
            raise ValueError("a concrete business workflow and source-bound function routes are required")
        references = result.get("sourceReferences", [])
        if not references or any(not p.startswith("blender-5.2.0/source/blender/") or not contained(self.root, p).is_file() for p in references):
            raise ValueError("native Blender source references are required before Web implementation")
        kinds, ids = set(), {}
        for check in result.get("checks", []):
            if not re.fullmatch(r"[A-Za-z0-9_-]+", check["id"]) or check["id"] in ids:
                raise ValueError("check IDs must be unique and path-safe")
            ids[check["id"]] = check
            kinds.update(check["kinds"])
            validate_command(self.root, check["argv"], check["cwd"])
        if not REQUIRED.issubset(kinds):
            raise ValueError(f"missing validation kinds: {sorted(REQUIRED - kinds)}")
        case_kinds, case_ids, paths, covered = set(), set(), set(), set()
        positive_units = set()
        for case in result.get("cases", []):
            destination = contained(self.root, case["receipt"])
            if not destination.is_relative_to(run_dir / "cases") or destination.exists():
                raise ValueError("behavioral receipts must be fresh validation output in this attempt")
            if case["checkId"] not in ids or case["kind"] not in ids[case["checkId"]]["kinds"] or case["id"] in case_ids or str(destination) in paths:
                raise ValueError("invalid or duplicate case/check binding")
            case_ids.add(case["id"])
            paths.add(str(destination))
            case_kinds.add(case["kind"])
            if not case["unitIds"] or not set(case["unitIds"]).issubset(units):
                raise ValueError("invalid source-unit case binding")
            covered.update(case["unitIds"])
            if case["kind"] == "positive":
                positive_units.update(case["unitIds"])
        if not {"positive", "negative", "undo_redo", "save_reopen", "determinism"}.issubset(case_kinds) or covered != units or positive_units != units:
            raise ValueError("incomplete behavioral case matrix")
        for p in result.get("changedFiles", []) + result.get("dependencies", []) + result.get("artifacts", []):
            contained(self.root, p)
        if not any(p.endswith((".wasm", "/blender", "/web_engine")) for p in result.get("artifacts", [])):
            raise ValueError("a compiled Blender/WASM artifact is required")

    def verify_cases(self, result):
        artifacts = []
        chromium = False
        for case in result["cases"]:
            p = contained(self.root, case["receipt"])
            receipt = read_json(p)
            if receipt.get("status") != "PASS" or receipt.get("runtime") not in {"CHROMIUM_WEBENGINE", "BLENDER_NATIVE"}:
                raise ValueError(f"runtime case failed: {case['id']}")
            if receipt.get("caseId") != case["id"] or receipt.get("kind") != case["kind"] or set(receipt.get("unitIds", [])) != set(case["unitIds"]):
                raise ValueError("runtime receipt does not match the requested case")
            chromium = chromium or receipt["runtime"] == "CHROMIUM_WEBENGINE"
            observations = receipt.get("observations", {})
            def hashes(*keys):
                values = [observations.get(key, "") for key in keys]
                if not all(isinstance(v, str) and re.fullmatch(r"[a-f0-9]{64}", v) for v in values):
                    raise ValueError(f"missing native state hashes: {case['id']}")
                return values
            kind = case["kind"]
            if kind == "positive" and observations.get("reader"):
                left, right = hashes("expected", "actual")
                valid = left == right
            elif kind in {"positive", "negative"}:
                left, right = hashes("before", "after")
                valid = (left != right) if kind == "positive" else (left == right)
            elif kind == "undo_redo":
                before, after, undo, redo = hashes("before", "after", "undo", "redo")
                valid = before != after and before == undo and after == redo
            elif kind == "save_reopen":
                saved, reopened = hashes("saved", "reopened")
                valid = saved == reopened
            elif kind == "determinism":
                first, second = hashes("first", "second")
                valid = first == second
            else:
                valid = False
            if not valid:
                raise ValueError(f"native behavior assertion failed: {case['id']}")
            if not receipt.get("artifacts"):
                raise ValueError("runtime case has no production artifacts")
            for artifact in receipt["artifacts"]:
                if not artifact.get("sha256") or sha(contained(self.root, artifact["path"])) != artifact["sha256"]:
                    raise ValueError(f"runtime artifact hash drift: {case['id']}")
                artifacts.append(artifact)
            artifacts.append({"path": case["receipt"], "sha256": sha(p)})
        if not chromium:
            raise ValueError("no Chromium runtime case receipt")
        return artifacts

    def attempt(self, context, job):
        previous = self.state["jobs"].get(job["id"], {})
        count = int(previous.get("attempts", 0)) + 1
        run_dir = self.directory / "runs" / f"{job['id']}-{uuid.uuid4().hex[:12]}"
        run_dir.mkdir(parents=True)
        record = {"state": "implementing", "attempts": count, "providerFailures": previous.get("providerFailures", 0), "runDirectory": str(run_dir.relative_to(self.root))}
        self.state["jobs"][job["id"]] = record
        protected = self.protected_hashes(context)
        with self.timed("source_fingerprint"):
            owned_before = {p: sha(contained(self.root, p)) for p in job.get("ownedPaths", [])}
        write_json(run_dir / "before-worker.json", {"protected": protected, "owned": owned_before})
        self.checkpoint("implementing", activeJob=job["id"], queue=context, reason=None)
        try:
            result = self.worker(self.worker_prompt(context, job, run_dir, previous), run_dir)
            if self.protected_hashes(context) != protected:
                raise InterruptedError("worker changed protected queue/evidence files")
            self.validate_result(result, run_dir)
            inputs = sorted(set(job.get("ownedPaths", []) + result["changedFiles"] + result["dependencies"] + result["sourceReferences"] + [context["card"], context["parentManifest"], POLICY]))
            before = {p: sha(contained(self.root, p)) for p in inputs}
            if any(value is None for value in before.values()):
                raise ValueError("declared dependency does not exist")
            record["state"] = "validating"
            self.checkpoint("validating")
            checks, cache = [], {}
            env = {"WBV2_RECEIPT_DIR": str(run_dir / "cases"), "WBV2_DEPLOYED_ORIGIN": self.policy["origin"], "WEB_BUILD_JOBS": str(min(self.policy["maxBuildJobs"], 8)), "CMAKE_BUILD_PARALLEL_LEVEL": str(min(self.policy["maxBuildJobs"], 8))}
            for check in result["checks"]:
                argv, cwd = validate_command(self.root, check["argv"], check["cwd"])
                key = json.dumps([argv, str(cwd)])
                if key not in cache:
                    cache[key] = self.run_process(argv, cwd, run_dir / f"check-{check['id']}.log", self.policy["checkTimeoutSeconds"], env=env, phase="validation:" + "+".join(sorted(check["kinds"])))
                else:
                    self.event("check.reused", checkId=check["id"], originalLog=cache[key]["log"], avoidedSeconds=cache[key]["durationSeconds"])
                receipt = {**cache[key], "id": check["id"], "kinds": check["kinds"]}
                checks.append(receipt)
                if receipt["exitCode"] != 0 or receipt["interrupted"] or receipt["timedOut"] or receipt["logLimit"]:
                    raise ValueError(f"check {check['id']} failed: exit={receipt['exitCode']}, timeout={receipt['timedOut']}")
            if self.protected_hashes(context) != protected or any(sha(self.root / p) != digest for p, digest in before.items()):
                raise InterruptedError("source/protected evidence changed during validation")
            artifacts = self.verify_cases(result)
            for p in result["artifacts"]:
                digest = sha(contained(self.root, p))
                if digest is None:
                    raise ValueError(f"compiled artifact missing: {p}")
                artifacts.append({"path": p, "sha256": digest})
            for check in checks:
                artifacts.append({"path": check["log"], "sha256": check["logSha256"]})
            receipt_path = run_dir / "verified-receipt.json"
            write_json(receipt_path, {"schemaVersion": 1, "queueTask": context["task"], "job": job["id"], "workflow": result["workflow"], "routeIds": result["routeIds"], "sourceReferences": result["sourceReferences"], "status": "verified", "verifiedAt": now(), "sourceUnits": result["sourceUnits"], "inputs": before, "checks": checks, "artifacts": artifacts, "coverageIncrementClaimed": 0, "summary": result["summary"]})
            record.update(state="verified", receipt=str(receipt_path.relative_to(self.root)), receiptSha256=sha(receipt_path), changedFiles=result["changedFiles"], attempts=0, providerFailures=0)
            self.checkpoint("ready", activeJob=None)
            self.event("job.verified", job=job["id"], receipt=record["receipt"])
            return True
        except ProviderUnavailable as error:
            if self.protected_hashes(context) != protected or any(sha(self.root / p) != value for p, value in owned_before.items()):
                record.update(state="needs_reconcile", reason="provider failure coincided with source/evidence changes")
                self.checkpoint("needs_reconcile", reason=record["reason"])
                return False
            failures = int(previous.get("providerFailures", 0)) + 1
            delay = min(self.policy["providerMaxBackoffSeconds"], self.policy["providerBackoffSeconds"] * 2 ** min(failures - 1, 20))
            retry_at = datetime.fromtimestamp(time.time() + delay, timezone.utc).isoformat()
            state = "provider_blocked" if failures >= self.policy["maxProviderFailures"] else "waiting_provider"
            record.update(state=state, attempts=int(previous.get("attempts", 0)), providerFailures=failures, reason=str(error), nextRetryAt=retry_at)
            self.checkpoint(state, reason=str(error), nextRetryAt=retry_at, providerFailures=failures)
            self.event("provider.unavailable", job=job["id"], failures=failures, nextRetryAt=retry_at, reason=str(error), noAgentActions=True)
            return False
        except InterruptedError as error:
            record.update(state="needs_reconcile", reason=str(error))
            self.checkpoint("needs_reconcile", reason=str(error))
            return False
        except (ValueError, OSError, KeyError, TypeError) as error:
            record.update(state="blocked" if count >= self.policy["maxAttempts"] else "retry_pending", reason=str(error))
            self.checkpoint("paused" if self.stop_requested else record["state"], reason=str(error))
            self.event("job.failed", job=job["id"], attempts=count, reason=str(error))
            return False

    def git(self, args):
        log = self.directory / "git" / f"{uuid.uuid4().hex}.log"
        result = self.run_process(["git", *args], self.root, log, 120)
        if result["interrupted"] or result["timedOut"] or result["logLimit"]:
            raise InterruptedError("Git command interrupted; reconcile its outcome before continuing")
        output = log.read_text(encoding="utf-8", errors="surrogateescape")
        if result["exitCode"] != 0:
            raise subprocess.CalledProcessError(result["exitCode"], ["git", *args], output=output)
        return output

    def remember_baseline(self, refresh=False):
        location = self.directory / "baseline.json"
        if location.exists() and not refresh:
            return
        dirty = set(self.git(["diff", "--name-only", "-z", "HEAD"]).split("\0"))
        dirty.update(self.git(["ls-files", "--others", "--exclude-standard", "-z"]).split("\0"))
        write_json(location, {"head": self.git(["rev-parse", "HEAD"]).strip(), "dirtyPaths": sorted(dirty - {""}), "recordedAt": now()})

    def publish(self, context, changed_files):
        if not self.policy.get("autoCommit"):
            self.checkpoint("ready_to_publish", reason="automatic commit disabled")
            return False
        baseline = read_json(self.directory / "baseline.json")
        paths = sorted(set(changed_files))
        if not paths:
            raise ValueError("handoff has no declared files to commit")
        for p in paths:
            contained(self.root, p)
            if p == ".git" or p.startswith(".git/"):
                raise ValueError("cannot publish Git internals")
        dirty_overlap = set(paths) & set(baseline["dirtyPaths"])
        if dirty_overlap:
            raise ValueError(f"publication would include pre-existing edits: {sorted(dirty_overlap)[:12]}")
        if self.git(["branch", "--show-current"]).strip() != "main":
            raise ValueError("automatic publication requires branch main")
        if self.git(["rev-parse", "HEAD"]).strip() != baseline["head"]:
            raise ValueError("HEAD changed since the supervisor baseline")
        self.checkpoint("committing", publishPaths=paths)
        self.git(["add", "-f", "--", *paths])
        # --only excludes unrelated staged changes from this commit.
        self.git(["commit", "--only", "-m", f"feat(web): verified workflow {context['task']}", "--", *paths])
        head = self.git(["rev-parse", "HEAD"]).strip()
        self.checkpoint("committed", commit=head)
        if self.policy.get("autoPush"):
            self.checkpoint("pushing")
            pushed = False
            for attempt in range(self.policy.get("pushRetries", 6) + 1):
                result = self.run_process(["git", "push", "origin", "HEAD:refs/heads/main"], self.root, self.directory / f"push-{attempt}.log", 120)
                try:
                    remote = self.git(["ls-remote", "--refs", "origin", "refs/heads/main"]).split()[0]
                except (subprocess.CalledProcessError, IndexError):
                    remote = None
                if remote == head:
                    pushed = True
                    self.checkpoint("pushed", remoteHead=remote)
                    break
                if result["interrupted"] or self.stop_requested:
                    break
                deadline = time.monotonic() + min(30, 2 ** attempt)
                while time.monotonic() < deadline and not self.stop_requested:
                    time.sleep(0.2)
            if not pushed:
                raise InterruptedError("push outcome not confirmed; reconcile remote HEAD before continuing")
        baseline["head"] = head
        write_json(self.directory / "baseline.json", baseline)
        return True

    def finalize(self, context):
        run_dir = self.directory / "runs" / f"handoff-{uuid.uuid4().hex[:12]}"
        run_dir.mkdir(parents=True)
        records = self.state["jobs"]
        if not records or not all(self.receipt_valid(r) for r in records.values()):
            raise ValueError("cannot finalize stale workflow receipts")
        references = [{"path": r["receipt"], "sha256": r["receiptSha256"]} for r in records.values()]
        self.checkpoint("finalizing")
        prompt = f"""Finalize only the already verified Blender Web task {context['task']}.
Read the minimal queue/context/card first. Current pointer: {json.dumps(context)}.
Independent runtime receipts: {json.dumps(references)}.
Use repository generators to write aggregate report, status, manifest, task-context and publish the successor pointer.
Bind these verified receipts and their source/unit/build/test hashes. Set acceptanceComplete=true only if the module contract is fully met.
No source changes, tests, historical inventory/provenance rewrites, blind hash refresh, commit, push, SSH or production deployment.
Do not rewrite the parent manifest. Preserve dirty files. Never use deferred verification or route counts as completion.
If source inventory remains unverified and the current module is complete, use the repository generator to select the next source-backed module; do not invent task IDs or declare full parity.
Return structured status=ready and exact changedFiles; other array fields may be empty for this handoff-only turn.
If a required generator/evidence is unavailable, return blocked with the exact cause. Do not advance the queue on failure.
"""
        result = self.worker(prompt, run_dir)
        if result.get("status") != "ready":
            raise ValueError(result.get("summary", "handoff worker blocked"))
        if sha(self.root / context["parentManifest"]) != context["parentSha256"]:
            raise InterruptedError("handoff changed the parent manifest")
        for record in records.values():
            receipt = read_json(self.root / record["receipt"])
            for p, digest in receipt["inputs"].items():
                if p == context["card"]:
                    continue
                if sha(self.root / p) != digest:
                    raise InterruptedError("handoff changed a verified implementation dependency")
        for index, argv in enumerate(self.policy["handoffChecks"]):
            process = self.run_process(argv, self.root, run_dir / f"gate-{index}.log", self.policy["checkTimeoutSeconds"])
            if process["exitCode"] != 0 or process["interrupted"] or process["timedOut"] or process["logLimit"]:
                raise ValueError(f"handoff gate failed: {argv}")
        proof = manifest_proof(self.root, context["task"])
        if not proof["ok"]:
            raise ValueError(f"aggregate acceptance missing: {proof['reasons']}")
        after = queue_context(self.root)
        if proof["nextTask"] is not None and after["task"] != proof["nextTask"]:
            raise ValueError("generator did not publish the authorized successor")
        if proof["nextTask"] is None and after["task"] != context["task"]:
            raise ValueError("terminal handoff changed the queue unexpectedly")
        changed = list(result["changedFiles"])
        for record in records.values():
            changed.extend(record.get("changedFiles", []))
            receipt = read_json(self.root / record["receipt"])
            changed.append(record["receipt"])
            changed.extend(a["path"] for a in receipt["artifacts"])
        if not self.publish(context, changed):
            return False
        self.checkpoint("handed_off", before=context, after=after)
        return True

    def doctor(self):
        context = queue_context(self.root)
        jobs = self.jobs(context)
        return {"schemaVersion": 1, "queue": context, "taskCompletion": manifest_proof(self.root, context["task"]), "inventory": inventory_proof(self.root), "jobCount": len(jobs), "nextJob": (self.choose_job(jobs) or {}).get("id"), "tools": {n: bool(shutil.which(n)) for n in ("python3", "node", "git", "codex", "cmake")}, "policy": POLICY, "deployment": "implementation workers do not alter the live server", "state": self.state.get("state")}

    def wait_for_provider(self, record):
        retry_at = record.get("nextRetryAt")
        if not retry_at:
            return
        remaining = max(0, datetime.fromisoformat(retry_at).timestamp() - time.time())
        if remaining == 0:
            return
        self.checkpoint("waiting_provider", nextRetryAt=retry_at, reason=record.get("reason"))
        deadline, heartbeat = time.monotonic() + remaining, time.monotonic()
        with self.timed("provider_backoff"):
            while time.monotonic() < deadline and not self.stop_requested and not (self.directory / "pause.request").exists():
                if time.monotonic() >= heartbeat:
                    self.checkpoint("waiting_provider", heartbeatAt=now())
                    heartbeat = time.monotonic() + self.policy["heartbeatSeconds"]
                time.sleep(min(0.25, max(0, deadline - time.monotonic())))

    def run(self, max_cycles=0):
        inherited = os.environ.pop("WBV2_AUTOPILOT_LEASE_FD", None)
        with Lease(self.directory / "owner.lock", int(inherited) if inherited else None), ProjectFence(self.root):
            if self.state.get("state") in ACTIVE | {"needs_reconcile"}:
                self.checkpoint("needs_reconcile", reason="previous writer has no completed handoff; run read-only reconcile")
                return 3
            if self.state.get("state") in {"blocked", "provider_blocked"}:
                return 2
            if (self.directory / "pause.request").exists():
                self.checkpoint("paused")
                return 0
            signal.signal(signal.SIGINT, self.request_stop)
            signal.signal(signal.SIGTERM, self.request_stop)
            self.remember_baseline()
            cycles = 0
            last_success = True
            self.checkpoint("preflight", reason=None)
            try:
                while not self.stop_requested and (max_cycles == 0 or cycles < max_cycles):
                    if self.run_tokens >= self.policy["maxTokensPerRun"]:
                        self.checkpoint("budget_exhausted", reason="per-run token budget reached")
                        return 4
                    if (self.directory / "pause.request").exists():
                        self.checkpoint("paused")
                        return 0
                    context = queue_context(self.root)
                    prior = self.state.get("queue")
                    if prior and prior["task"] != context["task"]:
                        proof = manifest_proof(self.root, prior["task"])
                        if not proof["ok"] or proof["nextTask"] != context["task"]:
                            raise ValueError("queue advanced without an accepted parent handoff")
                        self.state["jobs"] = {}
                    self.checkpoint("preflight", queue=context)
                    for index, argv in enumerate(self.policy["preflight"]):
                        proc = self.run_process(argv, self.root, self.directory / f"preflight-{index}.log", 120)
                        if proc["exitCode"] != 0 or proc["interrupted"] or proc["timedOut"] or proc["logLimit"]:
                            raise ValueError(f"context preflight failed: {argv}")
                    job = self.choose_job(self.jobs(context))
                    if job is None:
                        if not self.finalize(context):
                            return 2
                        proof = manifest_proof(self.root, context["task"])
                        if proof["nextTask"] is None:
                            inventory = inventory_proof(self.root)
                            if not inventory["ok"]:
                                self.checkpoint("planning_required", reason="no successor but source inventory remains unverified; parity not complete")
                                return 2
                            self.checkpoint("complete", reason="accepted terminal manifest and no unverified source units")
                            return 0
                        continue
                    record = self.state["jobs"].get(job["id"], {})
                    if record.get("state") in {"blocked", "provider_blocked"}:
                        self.checkpoint(record["state"], activeJob=job["id"], reason=record.get("reason"))
                        return 2
                    if record.get("state") == "waiting_provider":
                        self.wait_for_provider(record)
                        if self.stop_requested or (self.directory / "pause.request").exists():
                            self.checkpoint("paused")
                            return 2
                    success = self.attempt(context, job)
                    last_success = success
                    cycles += 1
                    if self.state["state"] in {"needs_reconcile", "blocked", "provider_blocked", "paused"}:
                        return 3 if self.state["state"] == "needs_reconcile" else 2
                    if not success and self.state["state"] != "waiting_provider":
                        with self.timed("retry_backoff"):
                            deadline = time.monotonic() + self.policy.get("retryDelaySeconds", 10)
                            while time.monotonic() < deadline and not self.stop_requested:
                                time.sleep(0.2)
                final_state = "paused" if self.stop_requested else "idle" if last_success else self.state["state"]
                self.checkpoint(final_state, reason="stop requested" if self.stop_requested else self.state.get("reason") or "cycle limit reached")
                return 0 if last_success and not self.stop_requested else 2
            except InterruptedError as error:
                self.checkpoint("needs_reconcile", reason=str(error))
                return 3
            except (ValueError, OSError, KeyError, RuntimeError, subprocess.CalledProcessError) as error:
                self.checkpoint("blocked", reason=str(error))
                self.event("supervisor.blocked", reason=str(error))
                return 2


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("action", choices=["run", "start", "status", "timings", "doctor", "pause", "stop", "resume", "reconcile", "service"], nargs="?", default="run")
    parser.add_argument("--once", action="store_true")
    parser.add_argument("--max-cycles", type=int, default=0)
    parser.add_argument("--retry-blocked", action="store_true")
    parser.add_argument("--refresh-baseline", action="store_true", help="adopt a reviewed maintenance commit before any workflow edits/receipts")
    parser.add_argument("--json", action="store_true", help="JSON timing report")
    parser.add_argument("--write", action="store_true", help="write timing report snapshots")
    args = parser.parse_args(argv)
    if args.max_cycles < 0:
        parser.error("--max-cycles must be non-negative")
    runner = Autopilot()
    directory = runner.directory
    if args.action == "doctor":
        print(json.dumps(runner.doctor(), ensure_ascii=False, indent=2))
        return 0
    if args.action == "status":
        state = dict(runner.state)
        state["alive"] = alive(state.get("pid")) and identity(state["pid"]) == state.get("identity")
        state["timezone"] = "Asia/Shanghai"
        print(json.dumps(display_times(state), ensure_ascii=False, indent=2))
        return 0
    if args.action == "timings":
        report = timing_report(directory, runner.state)
        if args.write:
            write_json(directory / "timing-report.json", report)
            (directory / "timing-report.md").write_text(format_report(report) + "\n", encoding="utf-8")
        print(json.dumps(report, ensure_ascii=False, indent=2) if args.json else format_report(report))
        return 0
    if args.action == "service":
        def quote(value):
            if "\n" in value or "\r" in value:
                raise ValueError("newline in systemd value")
            return '"' + value.replace("\\", "\\\\").replace('"', '\\"').replace("%", "%%") + '"'
        unit = Path.home() / ".config/systemd/user/blender-web-autopilot.service"
        unit.parent.mkdir(parents=True, exist_ok=True)
        executable = Path(__file__).with_name("run_web_product_migration.py")
        unit.write_text("[Unit]\nDescription=Blender Web source migration supervisor\nAfter=network-online.target\n\n[Service]\nType=simple\n" + f"WorkingDirectory={quote(str(ROOT))}\nExecStart={quote(sys.executable)} {quote(str(executable))} run\nEnvironment={quote('PATH=' + os.environ.get('PATH', '/usr/bin:/bin'))}\n" + "Restart=on-failure\nRestartPreventExitStatus=2 3 4\nRestartSec=10\nKillMode=control-group\nTimeoutStopSec=15\n\n[Install]\nWantedBy=default.target\n", encoding="utf-8")
        print(f"Created {unit}; enable with systemctl --user daemon-reload and systemctl --user enable --now blender-web-autopilot")
        return 0
    if args.action in {"pause", "stop"}:
        directory.mkdir(parents=True, exist_ok=True)
        write_json(directory / "pause.request", {"requestedAt": now()})
        owner = read_json(directory / "owner.lock", {})
        if alive(owner.get("pid")) and identity(owner["pid"]) == owner.get("identity"):
            os.kill(owner["pid"], signal.SIGTERM)
            deadline = time.monotonic() + 10
            while alive(owner["pid"]) and identity(owner["pid"]) == owner.get("identity") and time.monotonic() < deadline:
                time.sleep(0.1)
        print("Pause requested; interrupted work requires reconciliation before retry.")
        return 0
    if args.action == "reconcile":
        print(json.dumps({"queue": queue_context(ROOT), "state": runner.state, "childAlive": alive(runner.state.get("childPid")), "receipts": {key: runner.receipt_valid(record) for key, record in runner.state["jobs"].items()}, "decision": "read-only; inspect interrupted edits before resume --retry-blocked"}, ensure_ascii=False, indent=2))
        return 0
    if args.action == "resume":
        with Lease(directory / "owner.lock"):
            if alive(runner.state.get("childPid")) and identity(runner.state["childPid"]) == runner.state.get("childIdentity"):
                parser.error("previous child is alive; stop and reconcile first")
            if runner.state["state"] in ACTIVE | {"needs_reconcile", "blocked", "provider_blocked"} and not args.retry_blocked:
                parser.error("run reconcile; --retry-blocked acknowledges the inspected repository state")
            for record in runner.state["jobs"].values():
                if record.get("state") in ACTIVE | {"needs_reconcile", "blocked", "provider_blocked"} or (args.retry_blocked and record.get("state") != "verified"):
                    record.update(state="retry_pending", attempts=0, providerFailures=0)
                    record.pop("nextRetryAt", None)
            if args.refresh_baseline and any(record.get("receipt") or record.get("changedFiles") for record in runner.state["jobs"].values()):
                parser.error("cannot refresh a baseline containing workflow edits or verified receipts")
            (directory / "pause.request").unlink(missing_ok=True)
            if args.refresh_baseline:
                runner.remember_baseline(refresh=True)
            runner.checkpoint("idle", reason="explicit resume")
    if args.action in {"start", "resume"}:
        with Lease(directory / "owner.lock") as lease:
            if (directory / "pause.request").exists():
                parser.error("paused; use resume")
            if runner.state["state"] in ACTIVE | {"needs_reconcile", "blocked", "provider_blocked"}:
                parser.error("run reconcile before restarting interrupted or blocked work")
            command = [sys.executable, str(Path(__file__).with_name("run_web_product_migration.py")), "run", "--max-cycles", str(1 if args.once else args.max_cycles)]
            fd = lease.stream.fileno()
            with (directory / "daemon.log").open("ab") as log:
                child = subprocess.Popen(command, cwd=ROOT, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, start_new_session=True, pass_fds=(fd,), env={**os.environ, "WBV2_AUTOPILOT_LEASE_FD": str(fd)})
            lease.transferred = True
        print(json.dumps({"launchedPid": child.pid, "statusCommand": "python3 my-tools/run_web_product_migration.py status", "log": str(directory / "daemon.log")}))
        return 0
    return runner.run(1 if args.once else args.max_cycles)


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