#!/usr/bin/env python3
"""W1 local gate driver (WEBSERVE #249). Run only after the gater releases the quiet-box hold.

Same shape as ws272-w0.py: the IR-76 three-arm pre-flight, the lane lock, the env
scrub, the cold-pool bootstrap. W1 adds spt-net to the nextest population (the
`web` record family lives there) and runs the new `twohost_web` bin ALONE as its
own leg, never beside the unit leg (it is env-gated and a silent no-op without
SPT_TWO_HOST=1, but the gate reads its run count as a separate line).

--preflight-only makes no build claim and runs no cargo. --self-test exercises
all three IR-76 decisions without network access. --run requires the explicit
hold-release acknowledgement and repeats the live meter before each cargo leg.
Every leg retains its raw output and actual subprocess exit code beside it.

SAME-LANE POOL SERIALIZATION IS THE INVARIANT: one driver at a time per
worktree, whatever its output dir. Every --run holds the lane lock
<worktree>/.spt/driver.lock (beside the target/ pool it guards) for its whole
life; a second driver in the same worktree is REFUSED with POOL_HELD naming the
holder pid. An output-dir lock would not do: two drivers with different output
dirs still share the worktree, target/ and the default docs port, which is the
shape that bit. Mechanism this guards (W0, 2026-09-06): stopping a LOCAL ssh
does not stop its remote `bash -lc` (no tty, no SIGHUP) — an orphan chain
re-ran this driver beside the intended one, two suites interleaved one
nextest.raw (two Summary lines), and the gater voided the window.

The battery runs under a SCRUBBED environment: the perch identity variables an
agent's shell carries (SPT_ENDPOINT_ID and its companions) are dropped from
every leg's child and named once as ENV_SCRUBBED. Mechanism (W0, 2026-09-06):
a driver launched from a live perch inherited SPT_ENDPOINT_ID, and every test
that stops a daemon was refused by the product's own endpoint guard
(DAEMON_STOP_REFUSED) — two reds that were the launcher's, not the tree's.
"""

import argparse
import json
import os
from pathlib import Path
import re
import subprocess
import sys

REPO = "BigscreenVR/spt-bs-core"
WORKFLOW = f"repos/{REPO}/actions/workflows/golden.yml/runs"
ACTIVE_STATES = ("in_progress", "queued", "waiting", "pending", "requested")
# The same floor xtask disk-floor enforces (32 GiB); the bootstrap reading
# below must never be more lenient than the meter it stands in for.
FLOOR_BYTES = 32 << 30
# The perch identity a live agent's shell carries. A battery must never look
# like an endpoint to the binary it tests.
# The identity trio the daemon-stop guard reads (OWL_SESSION_ID / SPT_AGENT_ID /
# SPT_ENDPOINT_ID) plus the perch companions a hosted session exports.
PERCH_ENV = ("OWL_SESSION_ID", "SPT_AGENT_ID", "SPT_ENDPOINT_ID", "SPT_SESSION_NAME",
             "SPT_ADAPTER", "SPT_HOST_PID", "SPT_INJECT_VERIFY_ECHO")


def battery_env(source=None):
    """A copy of the environment with the perch identity removed.

    Returns (env, dropped): `dropped` lists the names that were present so the
    caller can say so once — a silent scrub would hide the launcher's mistake.
    """
    source = dict(os.environ if source is None else source)
    dropped = [name for name in PERCH_ENV if name in source]
    for name in dropped:
        del source[name]
    return source, dropped


def run_rows(payload):
    if not isinstance(payload, dict) or not isinstance(payload.get("workflow_runs"), list):
        raise ValueError("missing workflow_runs array")
    rows = payload["workflow_runs"]
    if any(not isinstance(row, dict) or not isinstance(row.get("id"), int)
           or not isinstance(row.get("status"), str) for row in rows):
        raise ValueError("invalid workflow run row")
    return rows


def decision(active, control):
    """A zero is CLEAR only when the unfiltered positive control sees runs."""
    try:
        rows = run_rows(active)
        if rows:
            return "HOLD", ", ".join(f'{row["id"]}:{row["status"]}' for row in rows)
        if not run_rows(control):
            return "REFUSE", "positive control returned no runs; meter unproven"
        return "CLEAR", "no active golden runs; positive control returned runs"
    except (ValueError, TypeError) as error:
        return "REFUSE", str(error)


def query(suffix):
    result = subprocess.run(["gh", "api", WORKFLOW + suffix], capture_output=True, text=True)
    if result.returncode:
        raise RuntimeError(f"gh api exited {result.returncode}: {result.stderr.strip()}")
    return json.loads(result.stdout)


def preflight():
    try:
        active = []
        for state in ACTIVE_STATES:
            rows = run_rows(query(f"?status={state}&per_page=100"))
            if any(row["status"] != state for row in rows):
                raise ValueError(f"filtered query returned a different status than {state}")
            active.extend(rows)
            if active:
                verdict = decision({"workflow_runs": active}, None)
                break
        else:
            verdict = decision({"workflow_runs": []}, query("?per_page=1"))
    except (OSError, RuntimeError, ValueError) as error:
        verdict = "REFUSE", f"meter failed: {error}"
    print(f"GOLDEN_PREFLIGHT:{verdict[0]}: {verdict[1]}", flush=True)
    return verdict[0] == "CLEAR"


def self_test():
    empty = {"workflow_runs": []}
    running = {"workflow_runs": [{"id": 7, "status": "in_progress"}]}
    completed = {"workflow_runs": [{"id": 6, "status": "completed"}]}
    cases = [
        ("running golden", running, None, "HOLD"),
        ("earned empty", empty, completed, "CLEAR"),
        ("empty control", empty, empty, "REFUSE"),
        ("failed control", empty, None, "REFUSE"),
        ("failed active meter", None, completed, "REFUSE"),
        ("malformed row", {"workflow_runs": [{}]}, completed, "REFUSE"),
    ]
    for name, active, control, expected in cases:
        actual, detail = decision(active, control)
        if actual != expected:
            raise AssertionError(f"{name}: expected {expected}, got {actual}: {detail}")
        print(f"PASS {name}: {actual}")
    import tempfile
    with tempfile.TemporaryDirectory() as tmp:
        first = claim_pool(Path(tmp))
        if first is None:
            raise AssertionError("lane lock: first claim refused")
        probe = [sys.executable, str(Path(__file__).resolve()), "--probe-lock", tmp]
        try:
            held = subprocess.run(probe, capture_output=True, text=True)
        finally:
            first.close()
        if held.returncode != 4 or f"pid {os.getpid()}" not in held.stdout:
            raise AssertionError(f"lane lock: second driver not refused naming the holder: rc={held.returncode} "
                                 f"out={held.stdout!r} err={held.stderr!r}")
        freed = subprocess.run(probe, capture_output=True, text=True)
        if freed.returncode != 0:
            raise AssertionError(f"lane lock: not released with its holder: rc={freed.returncode} err={freed.stderr!r}")
        print("PASS lane lock: second driver refused while the holder lives, free once it exits")
    tainted = {"OWL_SESSION_ID": "owl", "SPT_AGENT_ID": "a", "SPT_ENDPOINT_ID": "agent", "SPT_HOST_PID": "1", "PATH": "keep", "SPT_HOME": "keep"}
    env, dropped = battery_env(tainted)
    if dropped != ["OWL_SESSION_ID", "SPT_AGENT_ID", "SPT_ENDPOINT_ID", "SPT_HOST_PID"] or "SPT_ENDPOINT_ID" in env or env["PATH"] != "keep" or env["SPT_HOME"] != "keep":
        raise AssertionError(f"env scrub: dropped={dropped} env={env}")
    clean_env, none_dropped = battery_env({"PATH": "keep"})
    if none_dropped or clean_env != {"PATH": "keep"}:
        raise AssertionError(f"env scrub over a clean env: dropped={none_dropped}")
    print("PASS env scrub: perch identity dropped and named, everything else kept, clean env untouched")
    with tempfile.TemporaryDirectory() as tmp:
        raw = bootstrap_floor_reading(Path(tmp), Path(tmp) / "no-such-xtask")
        admitted, verdict = floor_verdict(raw, 0)
        if "DISK_FLOOR:BOOTSTRAP" not in raw or "no-such-xtask" not in raw or verdict.split(":")[1] not in ("PASS", "REFUSE"):
            raise AssertionError(f"cold-pool bootstrap: raw={raw!r} verdict={verdict!r}")
        print(f"PASS cold-pool bootstrap: direct reading named the missing meter and rendered {verdict.split(':')[1]}")


# [impl->REQ-DISK-FLOOR-PREFLIGHT]
def floor_verdict(raw, code):
    """Render GiB from xtask's own byte reading, never a second disk probe."""
    fields = dict(re.findall(r"\b(free_bytes|floor_bytes)=(\d+)", raw))
    if len(fields) != 2:
        return False, f"DISK_FLOOR:REFUSE: xtask exit={code}; no complete disk reading"
    free, floor = int(fields["free_bytes"]), int(fields["floor_bytes"])
    admitted = code == 0 and free >= floor
    verdict = "PASS" if admitted else "REFUSE"
    return admitted, (
        f"DISK_FLOOR:{verdict}: free_gib={free / (1 << 30):.2f} "
        f"floor_gib={floor / (1 << 30):.2f} xtask_exit={code}"
    )


def bootstrap_floor_reading(root, xtask):
    """A COLD pool has no prebuilt meter yet: read free space directly against
    the same floor and say so, so the first run of a fresh checkout is guarded
    rather than refused. The claim leg builds xtask; every later floor check
    uses the real meter."""
    import shutil
    free = shutil.disk_usage(root).free
    return (
        f"DISK_FLOOR:BOOTSTRAP: prebuilt xtask absent ({xtask}); direct free-space "
        f"read free_bytes={free} floor_bytes={FLOOR_BYTES} label=ws272-w1\n"
    )


def disk_floor(root, output, name, xtask):
    # Use a prebuilt xtask: compiling the meter before its first reading would
    # make the very first build unguarded. --xtask permits a bootstrap binary;
    # a cold pool (no meter at all) takes one direct reading, named as such.
    if not Path(xtask).exists():
        raw, code = bootstrap_floor_reading(root, xtask), 0
    else:
        try:
            result = subprocess.run(
                [str(xtask), "disk-floor", "--path", str(root), "--label", "ws272-w1"],
                cwd=root, capture_output=True, text=True,
            )
            raw, code = result.stdout + result.stderr, result.returncode
        except OSError as error:
            raw, code = f"Cannot run prebuilt xtask {xtask}: {error}\n", 127
    admitted, verdict = floor_verdict(raw, code)
    (output / f"{name}-floor.raw").write_text(raw + verdict + "\n", encoding="utf-8")
    (output / f"{name}-floor.exit").write_text(f"{code}\n", encoding="utf-8")
    print(raw, end="", flush=True)
    print(verdict, flush=True)
    return admitted


def claim_pool(root):
    """Hold the lane lock <root>/.spt/driver.lock for this process's life.

    Returns the open handle (keep it alive) or None when another live driver
    holds it; the refusal names the holder pid recorded inside the lock file.
    A dead holder releases the OS lock with its process, so no stale-lock
    sweeping is needed — the lock, not the file's existence, is the claim.
    """
    path = Path(root) / ".spt" / "driver.lock"
    path.parent.mkdir(parents=True, exist_ok=True)
    # The holder pid lives in a sibling note: Windows denies reads of a byte
    # another process holds locked, so the lock file itself cannot say who.
    holder_note = path.with_suffix(".holder")
    handle = open(path, "a+", encoding="utf-8")
    try:
        if sys.platform == "win32":
            import msvcrt
            handle.seek(0)
            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
        else:
            import fcntl
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        handle.close()
        try:
            holder = holder_note.read_text(encoding="utf-8").strip() or "unknown"
        except OSError:
            holder = "unknown"
        print(f"POOL_HELD: {root} is driven by pid {holder}; same-lane pool "
              "serialization is the invariant — refusing a second driver", flush=True)
        return None
    holder_note.write_text(str(os.getpid()) + "\n", encoding="utf-8")
    return handle


BATTERY_ENV = None  # set by --run before the first leg


def leg(root, output, name, argv, guarded=True, xtask=None):
    if guarded:
        if not preflight() or not disk_floor(root, output, name, xtask):
            (output / f"{name}.exit").write_text("75\n", encoding="utf-8")
            return False
    with (output / f"{name}.raw").open("wb") as log:
        try:
            result = subprocess.run(argv, cwd=root, stdout=log, stderr=subprocess.STDOUT, env=BATTERY_ENV)
            code = result.returncode
        except OSError as error:
            log.write(str(error).encode("utf-8"))
            code = 127
    (output / f"{name}.exit").write_text(f"{code}\n", encoding="utf-8")
    print(f"{name}: exit={code}; {output / (name + '.exit')}", flush=True)
    return code == 0


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    modes = parser.add_mutually_exclusive_group(required=True)
    modes.add_argument("--self-test", action="store_true")
    modes.add_argument("--preflight-only", action="store_true")
    modes.add_argument("--run", action="store_true")
    modes.add_argument("--probe-lock", metavar="WORKTREE",
                       help="take and release the lane lock once; exit 4 with POOL_HELD if held")
    parser.add_argument("--hold-released", action="store_true",
                        help="acknowledge the gater's explicit quiet-box release")
    parser.add_argument("--output", type=Path)
    parser.add_argument("--xtask", type=Path,
                        help="prebuilt disk-floor meter; defaults to this lane's debug xtask")
    args = parser.parse_args()
    if args.self_test:
        self_test()
        return 0
    if args.preflight_only:
        return 0 if preflight() else 2
    if args.probe_lock:
        probe = claim_pool(Path(args.probe_lock))
        if probe is None:
            return 4
        probe.close()
        return 0
    if not args.hold_released:
        parser.error("--run requires --hold-released after the gater's call")
    root = Path(__file__).resolve().parents[2]
    lane_lock = claim_pool(root)  # held until this process exits
    if lane_lock is None:
        return 4
    global BATTERY_ENV
    BATTERY_ENV, dropped = battery_env()
    if dropped:
        print(f"ENV_SCRUBBED: dropped {' '.join(dropped)} from every leg (a battery is not an endpoint)", flush=True)
    output = args.output.resolve() if args.output else root / ".spt" / "ws272-w1-gate"
    output.mkdir(parents=True, exist_ok=True)
    xtask = args.xtask.resolve() if args.xtask else (
        root / "target" / "debug" / ("xtask.exe" if sys.platform == "win32" else "xtask")
    )
    commands = [
        ("treqs", ["traceable-reqs", "check", "--json"], False),
        ("claim", ["cargo", "run", "-p", "xtask", "--", "pool-claim", "--pool",
                   str(root / "target"), "--label", "ws272-w1"], True),
        ("prebuild", ["cargo", "build", "--workspace", "--bins"], True),
        ("xtask", ["cargo", "run", "-p", "xtask", "--", "check"], True),
        ("clippy", ["cargo", "clippy", "--workspace", "--all-targets", "--", "-D", "warnings"], True),
        ("nextest", ["cargo", "nextest", "run", "--no-fail-fast", "-p", "spt-store",
                     "-p", "spt-runtime", "-p", "spt-net", "-p", "spt-daemon", "-p", "spt"], True),
        # The two-host bin ALONE (its own Summary line; env-gated, a silent
        # no-op without SPT_TWO_HOST=1 + a role).
        ("twohost_web", ["cargo", "nextest", "run", "--no-fail-fast", "-p", "spt-daemon",
                         "--test", "twohost_web"], True),
        ("mdbook", ["mdbook", "build", "docs-site"], False),
    ]
    for name, argv, guarded in commands:
        if not leg(root, output, name, argv, guarded, xtask):
            return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())