from __future__ import annotations

import argparse
import hashlib
import json
import os
import pathlib
import platform
import queue
import secrets
import signal
import subprocess
import sys
import threading
import time
from typing import Any, BinaryIO

from .archive_r10r2 import extraction_fingerprint, runtime_artifacts, source_tree_manifest
from .stage_control_r10r3r2 import (
    DeadlineHierarchy,
    cleanup_wait_seconds,
    load_deadline_calibration,
)
from .suite_isolation import phase_budget_from_environment
from .process_control import (
    _assign_windows_kill_job,
    _close_windows_job,
    _enable_linux_subreaper,
    _reap_tracked,
    _terminate_identity,
    collect_descendant_identities,
    process_snapshot,
    surviving_processes,
    zombie_processes,
)


def canonical_hash(value: object) -> str:
    raw = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def atomic_json(path: pathlib.Path, value: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    os.replace(temporary, path)


class EventJournal:
    def __init__(self, path: pathlib.Path, run_id: str, challenge: str) -> None:
        self.path = pathlib.Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.run_id = run_id
        self.challenge = challenge
        self.lock = threading.Lock()
        self.sequence = 0

    def append(self, event: str, **fields: Any) -> dict[str, Any]:
        with self.lock:
            self.sequence += 1
            row = {
                "event": event,
                "sequence": self.sequence,
                "run_id": self.run_id,
                "reviewer_challenge_sha256": self.challenge,
                "pid": os.getpid(),
                "epoch": time.time(),
                "monotonic_seconds": time.monotonic(),
                **fields,
            }
            encoded = (json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n").encode("utf-8")
            descriptor = os.open(self.path, os.O_APPEND | os.O_CREAT | os.O_WRONLY, 0o600)
            try:
                os.write(descriptor, encoded)
                os.fsync(descriptor)
            finally:
                os.close(descriptor)
            return row


def load_json_lines(path: pathlib.Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    if not path.is_file():
        return rows
    for line in path.read_text(encoding="utf-8").splitlines():
        if line.strip():
            value = json.loads(line)
            if isinstance(value, dict):
                rows.append(value)
    return rows


def _load_worker_command(path: pathlib.Path | None, default: list[str]) -> list[str]:
    if path is None:
        return default
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, list) or not value or not all(isinstance(item, str) and item for item in value):
        raise ValueError("child command JSON must be a non-empty string list")
    return value


def _read_json(path: pathlib.Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return value if isinstance(value, dict) else {}


def load_controller_startup_budget(
    *,
    historical_startup_p95_seconds: float,
    current_phase_bootstrap_budget_seconds: float,
) -> float:
    """Bind load-controller admission to the freshly measured phase budget."""
    historical = float(historical_startup_p95_seconds)
    current = float(current_phase_bootstrap_budget_seconds)
    if historical <= 0 or current <= 0:
        raise ValueError("startup calibration values must be positive")
    return max(historical * 3.0, current)


def _start_load_controller(
    root: pathlib.Path,
    evidence_dir: pathlib.Path,
    *,
    startup_timeout_seconds: float,
) -> tuple[subprocess.Popen[bytes], dict[str, Any]]:
    started = time.time()
    started_monotonic = time.monotonic()
    evidence_dir.mkdir(parents=True, exist_ok=True)
    ready_path = evidence_dir / "ready.json"
    metrics_path = evidence_dir / "metrics.json"
    stop_path = evidence_dir / "stop.json"
    if ready_path.exists() or metrics_path.exists() or stop_path.exists():
        raise FileExistsError("load controller evidence paths must be fresh")
    process = subprocess.Popen(
        [
            sys.executable,
            str(root / "scripts" / "r10r2_load_worker.py"),
            "--ready-path", str(ready_path),
            "--metrics-path", str(metrics_path),
            "--stop-path", str(stop_path),
        ],
        cwd=root,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=os.name != "nt",
        creationflags=getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) if os.name == "nt" else 0,
    )
    deadline = started_monotonic + max(float(startup_timeout_seconds), 0.01)
    ready_payload: dict[str, Any] = {}
    while time.monotonic() < deadline and process.poll() is None:
        ready_payload = _read_json(ready_path)
        if ready_payload.get("ready") is True and ready_payload.get("pid") == process.pid:
            break
        time.sleep(0.01)
    ready = ready_payload.get("ready") is True and ready_payload.get("pid") == process.pid
    return process, {
        "level": "cpu-medium",
        "load_pid": process.pid,
        "started_epoch": started,
        "startup_timeout_seconds": float(startup_timeout_seconds),
        "target_cpu_fraction": 0.50,
        "observed_running_after_start": process.poll() is None,
        "ready_handshake": {
            "ready": ready,
            "elapsed_ms": round((time.monotonic() - started_monotonic) * 1000, 3),
            "payload": ready_payload or None,
        },
        "ready_path": str(ready_path),
        "metrics_path": str(metrics_path),
        "stop_path": str(stop_path),
    }


def _stop_process(process: subprocess.Popen[Any] | None, hard_deadline: float) -> dict[str, Any]:
    if process is None:
        return {
            "level": "normal", "cleanup_passed": True,
            "orphan_process_count": 0, "zombie_process_count": 0,
        }
    identity = process_snapshot().get(process.pid)
    if process.poll() is None:
        try:
            process.terminate()
            process.wait(timeout=cleanup_wait_seconds(hard_deadline))
        except (OSError, subprocess.TimeoutExpired):
            try:
                process.kill()
                process.wait(timeout=cleanup_wait_seconds(hard_deadline))
            except (OSError, subprocess.TimeoutExpired):
                pass
    survivors = surviving_processes([identity] if identity else [])
    zombies = zombie_processes([identity] if identity else [])
    return {
        "ended_epoch": time.time(),
        "cleanup_passed": process.poll() is not None and not survivors and not zombies,
        "orphan_process_count": len(survivors),
        "zombie_process_count": len(zombies),
        "exit_code": process.returncode,
    }


def load_controller_cleanup_budget(
    start_record: dict[str, Any],
    *,
    minimum_seconds: float,
) -> float:
    metrics_path = pathlib.Path(str(start_record.get("metrics_path", "")))
    metrics = _read_json(metrics_path) if str(metrics_path) else {}
    cycle_count = int(metrics.get("cycle_count", 0))
    elapsed_seconds = float(metrics.get("elapsed_seconds", 0.0))
    observed_two_cycle_budget = (
        (elapsed_seconds / cycle_count) * 2.0
        if cycle_count > 0 and elapsed_seconds > 0
        else 0.0
    )
    return max(float(minimum_seconds), observed_two_cycle_budget)


def stop_load_controller(
    process: subprocess.Popen[Any],
    start_record: dict[str, Any],
    hard_deadline: float,
) -> dict[str, Any]:
    stop_path = pathlib.Path(str(start_record.get("stop_path", "")))
    if str(stop_path) and process.poll() is None:
        atomic_json(stop_path, {"stop": True, "requested_epoch": time.time(), "pid": process.pid})
        try:
            process.wait(timeout=max(
                time.get_clock_info("monotonic").resolution,
                hard_deadline - time.monotonic(),
            ))
        except subprocess.TimeoutExpired:
            pass
    cleanup = _stop_process(process, hard_deadline)
    metrics_path = pathlib.Path(str(start_record.get("metrics_path", "")))
    metrics = _read_json(metrics_path) if str(metrics_path) else {}
    observed_fraction = float(metrics.get("observed_busy_fraction", 0.0))
    cycle_count = int(metrics.get("cycle_count", 0))
    finalized = metrics.get("finalized") is True
    load_observed = cycle_count > 0 and 0.20 < observed_fraction < 0.80
    passed = (
        (start_record.get("ready_handshake") or {}).get("ready") is True
        and load_observed
        and finalized
        and cleanup.get("cleanup_passed") is True
        and cleanup.get("orphan_process_count") == 0
        and cleanup.get("zombie_process_count") == 0
    )
    return {
        **cleanup,
        "observed_cycle_count": cycle_count,
        "observed_busy_seconds": float(metrics.get("busy_seconds", 0.0)),
        "observed_elapsed_seconds": float(metrics.get("elapsed_seconds", 0.0)),
        "observed_busy_fraction": observed_fraction,
        "metrics_finalized": finalized,
        "graceful_stop_requested": stop_path.is_file(),
        "metrics_sha256": hashlib.sha256(metrics_path.read_bytes()).hexdigest() if metrics_path.is_file() else None,
        "load_observed": load_observed,
        "passed": passed,
    }


def run_supervisor(
    root: pathlib.Path,
    *,
    order_name: str,
    random_seed: int | None,
    load_profile: str,
    total_deadline_seconds: float,
    heartbeat_stall_seconds: float,
    child_command_path: pathlib.Path | None = None,
) -> int:
    root = pathlib.Path(root).resolve()
    run_id = os.environ.get("R10_R2_RUN_ID", f"local-{os.getpid()}-{time.time_ns()}")
    challenge = os.environ.get("R10_R2_REVIEWER_CHALLENGE_SHA256", hashlib.sha256(run_id.encode()).hexdigest())
    result_zip_hash = os.environ.get("R10_R2_RESULT_ZIP_SHA256", "LOCAL_NEGATIVE_CONTROL")
    event_path = pathlib.Path(os.environ.get("R10_R2_EVENT_JOURNAL_PATH", str(root / "artifacts" / f"{run_id}.events.jsonl")))
    supervisor_path = pathlib.Path(os.environ.get("R10_R2_SUPERVISOR_RECORD_PATH", str(root / "artifacts" / f"{run_id}.supervisor.json")))
    extraction_root = root.parent
    journal = EventJournal(event_path, run_id, challenge)
    source_manifest = source_tree_manifest(root)
    artifacts_before = runtime_artifacts(root)
    process_nonce = secrets.token_hex(24)
    supervisor_pid = os.getpid()
    started_epoch = time.time()
    started_monotonic = time.monotonic()
    deadline_calibration = load_deadline_calibration()
    reserve_seconds = deadline_calibration.reserve_seconds
    reviewer_seconds = float(os.environ.get(
        "R10_R2_REVIEWER_DEADLINE_SECONDS",
        str(total_deadline_seconds + reserve_seconds),
    ))
    controller_seconds = total_deadline_seconds - reserve_seconds
    worker_seconds = controller_seconds - reserve_seconds
    deadline_hierarchy = DeadlineHierarchy(
        reviewer_seconds=reviewer_seconds,
        supervisor_seconds=total_deadline_seconds,
        controller_seconds=controller_seconds,
        worker_seconds=worker_seconds,
        reserve_seconds=reserve_seconds,
    )
    if not deadline_hierarchy.passed:
        raise ValueError("reviewer > supervisor > controller > worker deadline hierarchy is invalid")
    deadline = started_monotonic + controller_seconds
    supervisor_deadline = started_monotonic + total_deadline_seconds
    phase_times: dict[str, float] = {}
    journal.append(
        "SUITE_START", source_tree_sha256=source_manifest["tree_sha256"],
        result_zip_sha256=result_zip_hash, supervisor_pid=supervisor_pid,
        process_identity_nonce=process_nonce, deadline_seconds=total_deadline_seconds,
        deadline_hierarchy=deadline_hierarchy.to_dict(),
    )

    default_command = [sys.executable, str(root / "run_all.py"), "--tests-only", "--order", order_name, "--load-profile", load_profile]
    if random_seed is not None:
        default_command.extend(["--seed", str(random_seed)])
    command = _load_worker_command(child_command_path, default_command)
    environment = {
        **os.environ,
        "PYTHONDONTWRITEBYTECODE": "1",
        "R10_R3_DEADLINE_CALIBRATION_JSON": json.dumps(deadline_calibration.to_dict(), sort_keys=True),
        "WRLAB_SUITE_TOTAL_SECONDS": str(controller_seconds),
    }
    popen_kwargs: dict[str, Any] = {
        "cwd": root, "env": environment, "stdin": subprocess.DEVNULL,
        "stdout": subprocess.PIPE, "stderr": subprocess.PIPE,
    }
    if os.name == "nt":
        popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
    else:
        popen_kwargs["start_new_session"] = True
    _enable_linux_subreaper()
    load_process: subprocess.Popen[bytes] | None = None
    load_record: dict[str, Any] = {
        "level": "normal", "passed": True, "cleanup_passed": True,
        "orphan_process_count": 0, "zombie_process_count": 0,
    }
    if load_profile == "cpu-medium":
        phase_budget = phase_budget_from_environment(root)
        load_process, load_record = _start_load_controller(
            root,
            supervisor_path.parent / (supervisor_path.stem + ".load"),
            startup_timeout_seconds=load_controller_startup_budget(
                historical_startup_p95_seconds=deadline_calibration.startup_p95_seconds,
                current_phase_bootstrap_budget_seconds=phase_budget.bootstrap_budget_seconds,
            ),
        )
    process = subprocess.Popen(command, **popen_kwargs)
    controller_pid = process.pid
    job_handle = _assign_windows_kill_job(process)
    output_queue: queue.Queue[tuple[str, bytes | None]] = queue.Queue()
    stdout_chunks: list[bytes] = []
    stderr_chunks: list[bytes] = []
    output_limit = 32 * 1024 * 1024
    output_bytes = {"stdout": 0, "stderr": 0}

    def drain(name: str, stream: BinaryIO) -> None:
        target = sys.stdout.buffer if name == "stdout" else sys.stderr.buffer
        try:
            while True:
                line = stream.readline()
                if not line:
                    break
                output_queue.put((name, line))
                target.write(line)
                target.flush()
        finally:
            output_queue.put((name, None))
            try:
                stream.close()
            except OSError:
                pass

    if process.stdout is None or process.stderr is None:
        raise RuntimeError("child output pipes unavailable")
    stdout_thread = threading.Thread(target=drain, args=("stdout", process.stdout), name=f"r10r2-stdout-{process.pid}", daemon=True)
    stderr_thread = threading.Thread(target=drain, args=("stderr", process.stderr), name=f"r10r2-stderr-{process.pid}", daemon=True)
    stdout_thread.start(); stderr_thread.start()
    tracked: dict[tuple[int, str], dict[str, Any]] = {}
    worker_pids: list[int] = []
    last_started_test_id: str | None = None
    last_completed_test_id: str | None = None
    last_heartbeat = time.monotonic()
    active_test = False
    timed_out = False
    stall_detected = False
    signal_received: int | None = None
    signal_journaled = False
    watchdog: dict[str, Any] = {}
    policy_audit: dict[str, Any] = {}
    drain_finished: set[str] = set()

    def signal_handler(signum: int, _frame: object) -> None:
        nonlocal signal_received
        # Python signal handlers run on the main thread and may interrupt an
        # EventJournal.append() while its non-reentrant lock is held.  Keep the
        # handler async-minimal; journal from the normal supervisor loop after
        # the interrupted critical section has returned.
        signal_received = int(signum)

    handled = [signal.SIGINT, signal.SIGTERM]
    previous_handlers = {item: signal.getsignal(item) for item in handled}
    for item in handled:
        signal.signal(item, signal_handler)
    self_signal_thread: threading.Thread | None = None
    self_signal_after = os.environ.get("R10_R2_SELF_SIGNAL_AFTER_SECONDS")
    if self_signal_after:
        def issue_signal() -> None:
            time.sleep(float(self_signal_after))
            signal.raise_signal(signal.SIGTERM)
        self_signal_thread = threading.Thread(target=issue_signal, name=f"r10r2-signal-{os.getpid()}", daemon=True)
        self_signal_thread.start()

    def parse_line(raw: bytes) -> None:
        nonlocal last_started_test_id, last_completed_test_id, last_heartbeat, active_test, watchdog, policy_audit
        text = raw.decode("utf-8", errors="replace").strip()
        prefixes = {
            "TEST_HEARTBEAT ": "heartbeat",
            "GROUP_HEARTBEAT ": "group",
            "POLICY_MANIFEST_REPORT ": "manifest",
            "POLICY_AUDIT_REPORT ": "policy",
            "STARTUP_CALIBRATION_REPORT ": "calibration",
            "WATCHDOG_REPORT ": "watchdog",
        }
        kind = next((value for prefix, value in prefixes.items() if text.startswith(prefix)), None)
        if kind is None:
            return
        prefix = next(prefix for prefix, value in prefixes.items() if value == kind)
        try:
            payload = json.loads(text[len(prefix):])
        except json.JSONDecodeError:
            return
        now = time.monotonic()
        if kind == "heartbeat":
            last_heartbeat = now
            event = payload.get("event")
            test_id = payload.get("test_id")
            if event == "START":
                last_started_test_id = test_id; active_test = True
                journal.append("TEST_START", test_id=test_id, worker_pid=payload.get("pid"))
            elif event == "END":
                last_completed_test_id = test_id; active_test = False
                journal.append("TEST_END", test_id=test_id, passed=payload.get("status") == "PASS", worker_pid=payload.get("pid"))
            else:
                journal.append("TEST_HEARTBEAT", test_id=test_id, worker_pid=payload.get("pid"))
        elif kind == "group":
            journal.append("GROUP_" + str(payload.get("event", "UNKNOWN")), **{key: value for key, value in payload.items() if key != "event"})
        elif kind == "manifest":
            phase_times["discovery_policy_epoch"] = time.time()
            journal.append("DISCOVERY_COMPLETE", discovered_test_count=payload.get("discovered_count"))
        elif kind == "policy":
            policy_audit = payload
            journal.append("POLICY_AUDIT_COMPLETE", passed=payload.get("passed"), sensitive_count=len(payload.get("ast_sensitive_test_ids", [])))
        elif kind == "calibration":
            phase_times["calibration_epoch"] = time.time()
            journal.append("CALIBRATION_COMPLETE", passed=payload.get("passed"), p95_seconds=payload.get("p95_seconds"))
        elif kind == "watchdog":
            watchdog = payload
            worker_pids[:] = [int(row["worker_pid"]) for row in payload.get("groups", []) if isinstance(row, dict) and isinstance(row.get("worker_pid"), int)]

    try:
        while True:
            identities = collect_descendant_identities(process.pid, tracked.values())
            for identity in identities:
                tracked[(int(identity["pid"]), str(identity["started"]))] = identity
            try:
                name, raw = output_queue.get(timeout=0.05)
                if raw is None:
                    drain_finished.add(name)
                else:
                    output_bytes[name] += len(raw)
                    if name == "stdout":
                        stdout_chunks.append(raw); parse_line(raw)
                    else:
                        stderr_chunks.append(raw)
            except queue.Empty:
                pass
            now = time.monotonic()
            if signal_received is not None:
                if not signal_journaled:
                    journal.append("SIGNAL_RECEIVED", signal=signal_received, test_id=last_started_test_id)
                    signal_journaled = True
                break
            if now >= deadline:
                timed_out = True
                journal.append("DEADLINE_TRIGGERED", test_id=last_started_test_id)
                break
            if active_test and now - last_heartbeat > heartbeat_stall_seconds:
                stall_detected = True
                journal.append("STALL_DETECTED", test_id=last_started_test_id, stall_seconds=now - last_heartbeat)
                break
            if process.poll() is not None and len(drain_finished) == 2 and output_queue.empty():
                break
    finally:
        journal.append("CLEANUP_START", controller_pid=controller_pid)
        identities = collect_descendant_identities(process.pid, tracked.values())
        for identity in identities:
            tracked[(int(identity["pid"]), str(identity["started"]))] = identity
        if process.poll() is None:
            if os.name != "nt":
                try:
                    os.killpg(process.pid, signal.SIGTERM)
                except (ProcessLookupError, PermissionError):
                    pass
            else:
                try:
                    process.terminate()
                except OSError:
                    pass
            try:
                process.wait(timeout=cleanup_wait_seconds(supervisor_deadline))
            except subprocess.TimeoutExpired:
                if os.name != "nt":
                    try:
                        os.killpg(process.pid, signal.SIGKILL)
                    except (ProcessLookupError, PermissionError):
                        pass
                else:
                    process.kill()
                try:
                    process.wait(timeout=cleanup_wait_seconds(supervisor_deadline))
                except subprocess.TimeoutExpired:
                    pass
        _close_windows_job(job_handle)
        for identity in surviving_processes(tracked.values()):
            _terminate_identity(identity, True)
        _reap_tracked(tracked.values(), supervisor_deadline)
        drain_started = time.monotonic()
        stdout_thread.join(timeout=cleanup_wait_seconds(supervisor_deadline))
        stderr_thread.join(timeout=cleanup_wait_seconds(supervisor_deadline))
        phase_times["output_drain_seconds"] = time.monotonic() - drain_started
        while not output_queue.empty():
            name, raw = output_queue.get_nowait()
            if raw:
                output_bytes[name] += len(raw)
                if name == "stdout": stdout_chunks.append(raw); parse_line(raw)
                else: stderr_chunks.append(raw)
        if load_process is not None:
            load_record.update(stop_load_controller(load_process, load_record, supervisor_deadline))
        else:
            load_record.update(_stop_process(None, supervisor_deadline))
        if self_signal_thread is not None:
            self_signal_thread.join(timeout=cleanup_wait_seconds(supervisor_deadline))
        for item, handler in previous_handlers.items():
            signal.signal(item, handler)

    final_survivors = surviving_processes(tracked.values())
    final_zombies = zombie_processes(tracked.values())
    cleanup_passed = not final_survivors and not final_zombies and not stdout_thread.is_alive() and not stderr_thread.is_alive() and load_record.get("cleanup_passed") is True and load_record.get("passed") is True
    base_exit = process.returncode if process.returncode is not None else 1
    exit_code = 130 if signal_received is not None else 126 if stall_detected else 124 if timed_out else int(base_exit)
    failure_count = int(watchdog.get("groups") and sum(int((row.get("worker_report") or {}).get("failure_count", 0)) for row in watchdog.get("groups", [])) or 0)
    error_count = int(watchdog.get("groups") and sum(int((row.get("worker_report") or {}).get("error_count", 0)) for row in watchdog.get("groups", [])) or 0)
    test_count = int(watchdog.get("executed_test_count", 0))
    passed = exit_code == 0 and not timed_out and not stall_detected and signal_received is None and cleanup_passed and watchdog.get("passed") is True and policy_audit.get("passed") is True
    journal.append("CLEANUP_END", cleanup_passed=cleanup_passed, orphan_process_count=len(final_survivors), zombie_process_count=len(final_zombies))
    journal.append("SUITE_END", passed=passed, exit_code=exit_code, test_count=test_count)
    event_rows = load_json_lines(event_path)
    record = {
        "run_id": run_id,
        "reviewer_challenge_sha256": challenge,
        "result_zip_sha256": result_zip_hash,
        "source_tree_sha256": source_manifest["tree_sha256"],
        "source_manifest_file_count": len(source_manifest["entries"]),
        "source_origin": "final" if result_zip_hash != "LOCAL_NEGATIVE_CONTROL" else "negative-control",
        "platform": platform.system().lower(),
        "platform_detail": platform.platform(),
        "python_version": platform.python_version(),
        "python_executable": sys.executable,
        "extraction_root_fingerprint": extraction_fingerprint(
            extraction_root, os.environ.get("R10_R2_EXTRACTION_ID")
        ),
        "preexisting_runtime_artifact_count": len(artifacts_before),
        "supervisor_pid": supervisor_pid,
        "controller_pid": controller_pid,
        "worker_pids": sorted(set(worker_pids)),
        "process_identity_nonce": process_nonce,
        "started_epoch": started_epoch,
        "ended_epoch": time.time(),
        "elapsed_seconds": time.monotonic() - started_monotonic,
        "deadline_seconds": total_deadline_seconds,
        "deadline_hierarchy": deadline_hierarchy.to_dict(),
        "deadline_calibration": deadline_calibration.to_dict(),
        "cleanup_reserve_seconds": deadline_calibration.cleanup_p95_seconds,
        "timed_out": timed_out,
        "stall_detected": stall_detected,
        "signal_received": signal_received,
        "last_started_test_id": last_started_test_id,
        "last_completed_test_id": last_completed_test_id,
        "test_count": test_count,
        "failure_count": failure_count,
        "error_count": error_count,
        "duplicate_execution_count": int(watchdog.get("duplicate_execution_count", 0)),
        "missing_execution_count": int(watchdog.get("missing_execution_count", 0)),
        "unexpected_execution_count": int(watchdog.get("unexpected_execution_count", 0)),
        "orphan_process_count": len(final_survivors),
        "zombie_process_count": len(final_zombies),
        "thread_count": int(stdout_thread.is_alive()) + int(stderr_thread.is_alive()),
        "output_drain_threads_stopped": not stdout_thread.is_alive() and not stderr_thread.is_alive(),
        "output_limit_bytes": output_limit,
        "stdout_total_bytes": output_bytes["stdout"],
        "stderr_total_bytes": output_bytes["stderr"],
        "output_within_limit": output_bytes["stdout"] <= output_limit and output_bytes["stderr"] <= output_limit,
        "cleanup_passed": cleanup_passed,
        "final_record_written": True,
        "exit_code": exit_code,
        "final_status": "PASS" if passed else "TIMEOUT" if timed_out else "STALL" if stall_detected else "SIGNAL" if signal_received is not None else "FAIL",
        "passed": passed,
        "event_journal_line_count": len(event_rows),
        "event_journal_sha256": canonical_hash(event_rows),
        "policy_audit": policy_audit,
        "watchdog_report": watchdog,
        "load_controller_record": load_record,
        "phase_timings": phase_times,
        "command": command,
    }
    atomic_json(supervisor_path, record)
    print("R10_R2_SUPERVISOR_RECORD " + json.dumps(record, ensure_ascii=False, sort_keys=True), flush=True)
    return exit_code


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--order", default="default", choices=("default", "reverse", "random", "sensitive-first", "sensitive-last"))
    parser.add_argument("--seed", type=int)
    parser.add_argument("--load-profile", default="normal", choices=("normal", "cpu-medium"))
    parser.add_argument("--deadline-seconds", type=float, required=True)
    parser.add_argument("--heartbeat-stall-seconds", type=float, default=90.0)
    parser.add_argument("--deadline-calibration-json")
    parser.add_argument("--child-command-json", type=pathlib.Path)
    args = parser.parse_args()
    if args.order == "random" and args.seed is None:
        parser.error("random order requires --seed")
    if args.deadline_calibration_json is not None:
        payload = json.loads(args.deadline_calibration_json)
        if not isinstance(payload, dict):
            parser.error("deadline calibration payload must be an object")
        os.environ["R10_R3_DEADLINE_CALIBRATION_JSON"] = json.dumps(payload, sort_keys=True)
        load_deadline_calibration()
    return run_supervisor(
        pathlib.Path.cwd(), order_name=args.order, random_seed=args.seed,
        load_profile=args.load_profile, total_deadline_seconds=args.deadline_seconds,
        heartbeat_stall_seconds=args.heartbeat_stall_seconds,
        child_command_path=args.child_command_json,
    )


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