#!/usr/bin/env python3
"""Authorized #308 Linux validation after the successor receipt has been delivered."""
import datetime
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import signal
import subprocess
import sys
import time
import tempfile

SHA = "7890ead39bb7f14ed44aaae44b0951f098ffe9ac"
ROOT = Path("/home/reavus/projects/spt-core/spt-core")
TREE = ROOT / ".worktrees/308-registry-process-lock"
PROOF = Path(__file__).resolve().parent
PROOF.mkdir(exist_ok=True)
assert not (PROOF / "receipt.json").exists(), "consumer already has a receipt"
TARGET = TREE / "target"
FLOOR = 32 * 1024**3
GOLDEN_ENV = {
    "SPT_TEST_EPHEMERAL_ADVISORY_PORTS": "1",
    "SPT_ATTACH_IPC_DEADLINE_MS": "30000",
    "SPT_ATTACH_GATE_WATCHDOG_MS": "120000",
}
INHERITED_ENV = dict(os.environ)
ENV = dict(INHERITED_ENV)
for key in INHERITED_ENV:
    if key.startswith(("OWL_", "SPT_")):
        ENV.pop(key, None)
        os.environ.pop(key, None)
ENVIRONMENT = {"allowed_owl_spt_names": sorted(GOLDEN_ENV), "snapshots": []}
OWNED_TEMP_ROOTS = []
ENV.update(PATH="/home/reavus/.cargo/bin:" + ENV["PATH"], CARGO_BUILD_JOBS="2",
           CARGO_TARGET_DIR=str(TARGET), CARGO_INCREMENTAL="0",
           RUSTFLAGS="-C link-arg=-fuse-ld=mold", NEXTEST_PROFILE="default",
           CARGO_TERM_COLOR="never", CI="true", **GOLDEN_ENV)
for name in ("CARGO_ENCODED_RUSTFLAGS", "NEXTEST_TEST_THREADS", "NEXTEST_RETRIES"):
    ENV.pop(name, None)
START = datetime.datetime.now(datetime.timezone.utc).isoformat()
SOURCE_MANIFEST = json.loads((PROOF / "draft-source-manifest.json").read_text())
MANIFEST_HASH = hashlib.sha256((PROOF / "draft-source-manifest.json").read_bytes()).hexdigest()
EXPECTED_FILES = dict(SOURCE_MANIFEST["files"])
NEGATIVE_CONTROL = False


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


def save(name, value):
    (PROOF / name).write_text(json.dumps(value, indent=2) + "\n")


def capture_environment(label, mapping=None, require_scrubbed=True):
    mapping = dict(ENV if mapping is None else mapping)
    forbidden = sorted(key for key in mapping if key.startswith(("OWL_", "SPT_")) and key not in GOLDEN_ENV)
    ENVIRONMENT["snapshots"].append({"label": label, "utc": utc(),
        "environment_names": sorted(mapping),
        "owl_spt_keys": sorted(key for key in mapping if key.startswith(("OWL_", "SPT_"))),
        "forbidden_present": forbidden})
    if require_scrubbed:
        ENVIRONMENT["snapshots"][-1]["golden_values"] = {key: mapping.get(key) for key in GOLDEN_ENV}
    save("environment.json", ENVIRONMENT)
    if require_scrubbed and forbidden:
        raise RuntimeError("IDENTITY_ENV_REFUSED " + repr(forbidden))


def source_guard():
    head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=TREE, text=True).strip()
    changed = subprocess.check_output(["git", "diff", "--name-only"], cwd=TREE, text=True).splitlines()
    new = subprocess.check_output(["git", "ls-files", "--others", "--exclude-standard"], cwd=TREE, text=True).splitlines()
    assert head == SHA == SOURCE_MANIFEST["base"], head
    assert set(changed + new) == set(EXPECTED_FILES), (changed, new)
    observed = {name: hashlib.sha256((TREE / name).read_bytes()).hexdigest() for name in EXPECTED_FILES}
    assert observed == EXPECTED_FILES, "draft source changed outside the declared control"
    return {"base_sha": head, "draft_manifest_sha256": MANIFEST_HASH,
            "negative_control": NEGATIVE_CONTROL, "file_hashes": observed}


def census():
    active = []
    owned = []
    for directory in Path("/proc").iterdir():
        if not directory.name.isdigit():
            continue
        try:
            name = (directory / "comm").read_text().strip()
            exe = os.readlink(directory / "exe")
            row = {"pid": int(directory.name), "name": name, "exe": exe,
                   "cmd": (directory / "cmdline").read_bytes().replace(b"\0", b" ").decode(errors="replace"),
                   "stat": (directory / "stat").read_text()}
            if any(word in name for word in ("cargo", "rustc", "nextest", "Runner.Worker")) or "/deps/" in exe:
                active.append(row)
            if exe.startswith(str(TARGET) + "/"):
                owned.append(row)
        except (OSError, PermissionError):
            pass
    return {"utc": utc(), "free_bytes": shutil.disk_usage(TREE).free,
            "load": os.getloadavg(), "active": active, "owned_survivors": owned}


def admit(label):
    state = {**source_guard(), **census()}
    save(label + "-admission.json", state)
    if state["free_bytes"] < FLOOR or state["active"] or state["owned_survivors"]:
        raise RuntimeError("RESOURCE_REFUSED " + json.dumps(state))
    return state


def run(label, argv):
    capture_environment("before-" + label)
    started = utc()
    clock = time.monotonic()
    path = PROOF / (label + ".log")
    print(f"BEGIN {label} base={SHA} draft={MANIFEST_HASH} control={NEGATIVE_CONTROL} utc={started} argv={json.dumps(argv)}", flush=True)
    with path.open("x") as stream:
        result = subprocess.Popen(argv, cwd=TREE, env=ENV, stdout=stream, stderr=subprocess.STDOUT, start_new_session=True)
        free_samples = []
        floor_crossed = False
        while result.poll() is None:
            free = shutil.disk_usage(TREE).free
            free_samples.append({"utc": utc(), "free_bytes": free})
            if free < FLOOR and label != "pool-release":
                floor_crossed = True
                os.killpg(result.pid, signal.SIGKILL)
                result.wait()
                break
            time.sleep(5)
        save(label + "-disk-samples.json", {"floor_bytes": FLOOR, "crossed": floor_crossed, "samples": free_samples})
    capture_environment("after-" + label)
    text = path.read_text(errors="replace")
    summaries = [line.strip() for line in text.splitlines() if re.search(r"\bSummary\b", line)]
    receipt = {"base_sha": SHA, "draft_manifest_sha256": MANIFEST_HASH, "negative_control": NEGATIVE_CONTROL, "command": argv, "start_utc": started, "end_utc": utc(),
               "elapsed_s": round(time.monotonic() - clock, 3), "exit": result.returncode,
               "summary_lines": summaries, "log": str(path),
               "log_sha256": hashlib.sha256(path.read_bytes()).hexdigest()}
    save(label + "-receipt.json", receipt)
    print("END " + label + " " + json.dumps(receipt), flush=True)
    if floor_crossed:
        raise RuntimeError("DISK_FLOOR_CROSSED " + label)
    return receipt


def outside_git(path):
    result = subprocess.run(["git", "-C", str(path), "rev-parse", "--show-toplevel"], env=ENV, capture_output=True, text=True)
    with (PROOF / "temp-git-probes.jsonl").open("a") as log:
        log.write(json.dumps({"path": str(path), "exit": result.returncode, "stdout": result.stdout, "stderr": result.stderr}) + "\n")
    if result.returncode == 0 or "not a git repository" not in result.stderr:
        raise RuntimeError("TEMP_PREMISE_REFUSED " + str(path) + " " + result.stdout + result.stderr)


def reap_owned(label):
    rows = census()["owned_survivors"]
    actions = []
    def born(stat):
        return stat.split(") ", 1)[1].split()[19]
    for sig in (signal.SIGTERM, signal.SIGKILL):
        for row in rows:
            fd = None
            try:
                fd = os.pidfd_open(row["pid"])
                proc = Path("/proc") / str(row["pid"])
                exe = os.readlink(proc / "exe")
                stat = (proc / "stat").read_text()
                assert exe == row["exe"] and exe.startswith(str(TARGET) + "/") and born(stat) == born(row["stat"])
                home = next((value.split(b"=", 1)[1].decode() for value in (proc / "environ").read_bytes().split(b"\0") if value.startswith(b"SPT_HOME=")), None)
                assert Path(exe).name in ("spt", "mock-session"), exe
                command = (proc / "cmdline").read_bytes().replace(b"\0", b" ").decode(errors="replace")
                assert command == row["cmd"], (command, row["cmd"])
                assert home is not None and any(Path(home).resolve().is_relative_to(root) for root in OWNED_TEMP_ROOTS), home
                signal.pidfd_send_signal(fd, sig)
                actions.append({"pid": row["pid"], "exe": exe, "birth": born(stat), "SPT_HOME": home, "signal": sig.name, "utc": utc()})
            except (FileNotFoundError, ProcessLookupError):
                pass
            finally:
                if fd is not None:
                    os.close(fd)
        if rows:
            time.sleep(3 if sig == signal.SIGTERM else 1)
    remaining = census()["owned_survivors"]
    save(label + "-owned-reap.json", {"scope": str(TARGET), "actions": actions, "remaining": remaining})
    assert not remaining, remaining
    print("REAP " + label + " " + json.dumps({"signalled": len(actions), "remaining": len(remaining)}), flush=True)


def inventory_for(label, args, expected_count=None):
    admit(label)
    receipt = run(label, ["cargo", "nextest", "list", *args, "--message-format", "json"])
    receipt["source_after"] = source_guard()
    save(label + "-receipt.json", receipt)
    assert receipt["exit"] == 0, receipt
    documents = []
    for line in Path(receipt["log"]).read_text().splitlines():
        if line.startswith("{"):
            try:
                document = json.loads(line)
            except json.JSONDecodeError:
                continue
            if isinstance(document, dict) and "rust-suites" in document:
                documents.append(document)
    assert len(documents) == 1
    names = sorted([suite["binary-id"], name]
                   for suite in documents[0]["rust-suites"].values()
                   for name, info in suite.get("testcases", {}).items()
                   if info["filter-match"]["status"] == "matches" and not info["ignored"])
    assert names and (expected_count is None or len(names) == expected_count), names
    save(label + "-names.json", {"count": len(names), "names": names, "source": source_guard()})
    return names

phases = []
claimed = False
success = False
try:
    prerequisite = json.loads((PROOF / "successor-receipt-delivery.json").read_text())
    assert prerequisite["sent"] and prerequisite["acknowledged"] and prerequisite["pool_released"] and prerequisite["own_survivors"] == 0
    capture_environment("inherited-before-scrub", INHERITED_ENV, require_scrubbed=False)
    capture_environment("scrubbed-driver-process", os.environ)
    source_guard()
    admit("start")
    runner_temp = Path(tempfile.mkdtemp(prefix="spt-registry308-linux-", dir="/tmp"))
    outside_git(runner_temp)
    github_env = PROOF / "temp.env"
    assert not github_env.exists()
    ENV.update(RUNNER_TEMP=str(runner_temp), GITHUB_ENV=str(github_env),
               GITHUB_RUN_ID="registry308-linux", GITHUB_RUN_ATTEMPT="1")
    setup = run("temp-setup", ["bash", ".github/ci/test-temp-sandbox.sh", "setup"])
    assert setup["exit"] == 0, setup
    for line in github_env.read_text().splitlines():
        key, value = line.split("=", 1)
        ENV[key] = value
    for key in ("TEMP", "TMP", "TMPDIR", "SPT_CI_TEST_TMP"):
        outside_git(Path(ENV[key]))
    OWNED_TEMP_ROOTS.append(Path(ENV["SPT_CI_TEST_TMP"]).resolve())
    save("temp-boundary.json", {key: ENV[key] for key in ("TEMP", "TMP", "TMPDIR", "SPT_CI_TEST_TMP")})
    ENV.pop("SPT_CI_TEST_TMP")
    claim = run("pool-claim", [str(PROOF / "xtask-b8482445"),
                "pool-claim", "--pool", str(TARGET), "--label", "registry308-linux"])
    assert claim["exit"] == 0, claim
    claimed = True

    # A separately labelled negative control proves the new regression rejects
    # a public guard that opens its sentinel but never takes the OS lock.
    source_name = "crates/spt-store/src/serving.rs"
    source_path = TREE / source_name
    original = source_path.read_bytes()
    token = b"lock.lock_exclusive()?"
    assert original.count(token) == 1
    (PROOF / "serving-before-negative-control.rs").write_bytes(original)
    mutant = original.replace(token, b"if false { lock.lock_exclusive()? }", 1)
    negative_args = ["-p", "spt-store", "--test", "serving_registry_two_process_int", "-E",
                     "test(=later_reap_preserves_a_registration_published_while_it_was_waiting)"]
    try:
        source_path.write_bytes(mutant)
        EXPECTED_FILES[source_name] = hashlib.sha256(mutant).hexdigest()
        NEGATIVE_CONTROL = True
        inventory_for("negative-control-inventory", negative_args, 1)
        admit("negative-control")
        negative = run("negative-control", ["cargo", "nextest", "run", *negative_args, "--no-fail-fast"])
        negative["source_after"] = source_guard()
        save("negative-control-receipt.json", negative)
        reap_owned("negative-control")
        failure_text = Path(negative["log"]).read_text(errors="replace")
        assert negative["exit"] == 100 and "another process reported HELD but its sentinel was unlocked" in failure_text, negative
    finally:
        source_path.write_bytes(original)
        EXPECTED_FILES[source_name] = SOURCE_MANIFEST["files"][source_name]
        NEGATIVE_CONTROL = False
        save("negative-control-restoration.json", source_guard())

    groups = [
        ("process-regressions", ["-p", "spt-store", "--test", "serving_registry_two_process_int"], 3),
        ("store-serving-units", ["-p", "spt-store", "--lib", "-E", "test(/^serving::tests::/)"], None),
        ("daemon-servehost-units", ["-p", "spt-daemon", "--lib", "-E", "test(/^servehost::tests::/)"], None),
        ("affected-e2es", ["--workspace", "-E", "package(spt) & (binary(=webserve_attachment_e2e) | binary(=webserve_cross_node_e2e))"], None),
    ]
    for name, args, count in groups:
        if name == "affected-e2es":
            admit("capture-player-build")
            fixture_build = run("capture-player-build", ["cargo", "build", "-p", "mock-adapter", "--bin", "capture-player"])
            assert fixture_build["exit"] == 0, fixture_build
            fixture = TARGET / "debug/capture-player"
            assert fixture.is_file(), "capture-player fixture missing before workspace inventory"
            save("capture-player-fixture.json", {"path": str(fixture), "sha256": hashlib.sha256(fixture.read_bytes()).hexdigest()})
        names = inventory_for(name + "-inventory", args, count)
        if name == "daemon-servehost-units":
            assert any(cell == "servehost::tests::unavailable_registry_lock_cannot_publish_an_add" for _, cell in names)
            assert any(cell == "servehost::tests::concurrent_requests_publish_distinct_entries_without_lost_updates" for _, cell in names)
        admit(name)
        result = run(name, ["cargo", "nextest", "run", *args, "--no-fail-fast"])
        result["source_after"] = source_guard()
        phases.append(result)
        save("phases.json", phases)
        reap_owned(name)
        assert result["exit"] == 0, result

    trace = shutil.which("traceable-reqs", path=ENV["PATH"] + ":/home/reavus/.local/bin")
    assert trace is not None, "traceable-reqs binary unavailable"
    admit("trace-version")
    version = run("trace-version", [trace, "--version"])
    assert version["exit"] == 0, version
    admit("trace-check")
    trace_result = run("trace-check", [trace, "check", "--json"])
    trace_result["source_after"] = source_guard()
    save("trace-check-receipt.json", trace_result)
    assert trace_result["exit"] == 0, trace_result
    success = True
except BaseException as error:
    save("driver-error.json", {"base_sha": SHA, "draft_manifest_sha256": MANIFEST_HASH,
         "start_utc": START, "end_utc": utc(), "error": repr(error), "phases": phases, "census": census()})
    raise
finally:
    capture_environment("final")
    release = None
    if claimed:
        release = run("pool-release", [str(PROOF / "xtask-b8482445"),
                      "pool-release", "--pool", str(TARGET)])
        success = success and release["exit"] == 0
    final_census = census()
    success = success and not final_census["active"] and not final_census["owned_survivors"]
    save("final-census.json", final_census)
    save("receipt.json", {"base_sha": SHA, "draft_manifest_sha256": MANIFEST_HASH,
         "start_utc": START, "end_utc": utc(), "success": success, "phases": phases,
         "pool_release": release, "final": final_census})
    print("REGISTRY308_DONE " + json.dumps({"success": success, "exits": [row["exit"] for row in phases]}), flush=True)
sys.exit(0 if success else 1)