"""Real-server fixtures. All mutation uses MCP; SQLite connections are read-only.

Usage: capture.py <before|after> <F03|F04|F06|F07|probe> [cursor|codex]
Every invocation creates a timestamped run directory and a new, unseeded server home.
Models are discovered at runtime and omitted from public state dumps.
"""
import datetime
import importlib.util
import json
import os
import pathlib
import re
import signal
import shutil
import sqlite3
import subprocess
import sys
import time
import traceback

HERE = pathlib.Path(__file__).resolve().parent
SCRATCH = HERE.parent
spec = importlib.util.spec_from_file_location("mcp_client", HERE / "mcp-client.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
Mcp = module.Mcp
TERMINAL = {"completed", "failed", "cancelled", "interrupted"}


def utc():
    return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="milliseconds")


class Capture:
    def __init__(self, side, fixture, provider):
        self.side, self.fixture, self.provider = side, fixture, provider
        self.root = pathlib.Path("/home/ubuntu/wt/pr-33469-before" if side == "before" else "/home/ubuntu/wt/pr-33469")
        self.sha = "ad989057ba" if side == "before" else "827879efad"
        self.run = SCRATCH / (datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + "-" + side + "-" + fixture)
        self.run.mkdir()
        self.home = self.run / "home"
        (self.home / "userdata").mkdir(parents=True)
        self.dbpath = self.home / "userdata/statev2.sqlite"
        self.steps, self.blocks, self.private_models = [], [], set()
        self.tid = None
        self.server = None
        self.last_snapshot = None
        self.settings = {"providerInstances": {"cursor": {"driver": "cursor", "enabled": True, "config": {"enabled": True}}}, "enableProviderUpdateChecks": False}
        (self.home / "userdata/settings.json").write_text(json.dumps(self.settings))

    def step(self, text):
        line = utc() + " " + text
        self.steps.append(line)
        print(line, flush=True)

    def clean(self, value):
        if isinstance(value, dict):
            return {k: ("<model>" if k in {"model", "modelId", "modelName", "requestedModel", "resolvedModel"} and v else self.clean(v)) for k, v in value.items()}
        if isinstance(value, list):
            return [self.clean(v) for v in value]
        if isinstance(value, str):
            for name in sorted(self.private_models, key=len, reverse=True):
                value = value.replace(name, "<model>")
        return value

    def start(self):
        env = dict(os.environ, T3CODE_DEV_SEED="0", T3CODE_NO_BROWSER="1", T3CODE_AUTO_BOOTSTRAP_PROJECT_FROM_CWD="0")
        env.pop("VITE_HTTP_URL", None)
        env.pop("VITE_WS_URL", None)
        self.log = open(self.run / "server-private.log", "w")
        self.server = subprocess.Popen(["vp", "run", "dev", "--home-dir", str(self.home)], cwd=self.root, env=env, stdout=self.log, stderr=subprocess.STDOUT)
        (self.run / "server-pid.txt").write_text(str(self.server.pid))
        self.step(f"Started T3CODE_DEV_SEED=0 vp run dev --home-dir <fresh home>; captured PID={self.server.pid}; commit={self.sha}.")
        deadline = time.monotonic() + 180
        port = None
        while time.monotonic() < deadline:
            raw = (self.run / "server-private.log").read_text()
            match = re.search(r"\[dev-runner\].*serverPort=(\d+) webPort=(\d+)", raw)
            if match:
                port = int(match[1])
            if port and self.dbpath.exists() and "pairingUrl:" in raw:
                break
            if self.server.poll() is not None:
                raise RuntimeError("dev server exited; inspect private server log")
            time.sleep(0.25)
        else:
            raise RuntimeError("dev server did not become ready")
        self.step(f"Read [dev-runner] ports: server={port}, web={match[2]}; fresh database created by this checkout, no seed.")
        helper = self.run / "auth-helper"
        helper.mkdir()
        (helper / "node_modules").symlink_to(self.root / "apps/server/node_modules", target_is_directory=True)
        shutil.copyfile(HERE / "mint-auth.ts", helper / "mint-auth.ts")
        auth = subprocess.run(["node", str(helper / "mint-auth.ts"), str(self.root), str(self.home)], cwd=self.root / "apps/server", capture_output=True, text=True, timeout=60)
        (self.run / "auth-private.stderr").write_text(auth.stderr)
        if auth.returncode:
            (self.run / "auth-private.stdout").write_text(auth.stdout)
            raise RuntimeError("MCP auth mint failed; inspect private stderr")
        token = auth.stdout.strip()
        (self.run / "token-private.txt").write_text(token)
        os.chmod(self.run / "token-private.txt", 0o600)
        self.mcp = Mcp(f"http://127.0.0.1:{port}", token)

    def call(self, name, **args):
        result = self.mcp.call(name, args)
        with open(self.run / "rpc-private.jsonl", "a") as out:
            out.write(json.dumps({"at": utc(), "tool": name, "arguments": args, "result": result}) + "\n")
        return result

    def capabilities(self):
        result = self.call("orchestrator_capabilities")
        (self.run / "capabilities-private.json").write_text(json.dumps(result, indent=2))
        return result

    def read(self):
        db = sqlite3.connect(f"file:{self.dbpath}?mode=ro", uri=True)
        db.row_factory = sqlite3.Row
        db.execute("BEGIN")
        try:
            tables = [r[0] for r in db.execute("select name from sqlite_master where type='table'")]
            result = {}
            for key in ["runs", "run_attempts", "subagents", "messages", "nodes"]:
                table = "orchestration_v2_projection_" + key
                if table in tables:
                    result[key] = [json.loads(r[0]) for r in db.execute(f"select payload_json from {table} where thread_id=?", (self.tid,))]
                else:
                    result[key] = []
            result["runs"].sort(key=lambda r: r["ordinal"])
            result["run_attempts"].sort(key=lambda r: (r["runId"], r["attemptOrdinal"]))
            eventtable = "orchestration_events" if "orchestration_events" in tables else None
            if eventtable:
                columns = [r[1] for r in db.execute(f"pragma table_info({eventtable})")]
                result["eventColumns"] = columns
                seq = next((x for x in ["sequence", "seq", "event_sequence"] if x in columns), None)
                result["eventSequence"] = db.execute(f"select max({seq}) from {eventtable} where stream_id=?", (self.tid,)).fetchone()[0] if seq else None
                result["events"] = [dict(r) for r in db.execute(f"select sequence, command_id, event_type, occurred_at, payload_json from {eventtable} where stream_id=? order by sequence", (self.tid,))]
                for event in result["events"]:
                    event["payload"] = json.loads(event.pop("payload_json"))
            else:
                result["eventSequence"] = None
            return result
        finally:
            db.close()

    def snapshot(self, label, state=None):
        s = state or self.read()
        stamp = utc()
        self.last_snapshot = s
        rawfile = self.run / (f"state-{len(self.blocks):02d}-private.json")
        rawfile.write_text(json.dumps(s, indent=2))
        runs = s["runs"]
        wake_ids = {m.get("runId") for m in s["messages"] if m.get("delegatedCompletion")}
        wake_runs = [r for r in runs if r["id"] in wake_ids]
        data = {
            "eventSequence": s.get("eventSequence"),
            "runs": [{k: r.get(k) for k in ["id", "ordinal", "status", "userMessageId", "delegatedCompletion", "execution"]} for r in runs],
            "attempts": [{k: a.get(k) for k in ["id", "runId", "attemptOrdinal", "reason", "status", "userMessageId", "providerTurnId"]} for a in s["run_attempts"]],
            "tasks": [{k: t.get(k) for k in ["id", "title", "task", "runId", "childThreadId", "status", "completionDelivery", "completionWake"]} for t in s["subagents"]],
            "messages": [{k: m.get(k) for k in ["id", "role", "runId", "text", "delegatedCompletion", "createdAt"]} for m in s["messages"]],
            "nodes": [{k: n.get(k) for k in ["id", "runId", "kind", "status", "attemptId"]} for n in s["nodes"]],
            "commitEventsSincePriorMilestone": [e for e in s.get("events", []) if e["sequence"] > getattr(self, "last_sequence", 0) and any(k in e["event_type"] for k in ["run.", "run-attempt.", "attempt.", "subagent.", "message.", "node.created", "node.updated"]) and "delta" not in e["event_type"]],
            "wakeRunCounts": {"total": len(wake_runs), "queued": sum(r["status"] == "queued" for r in wake_runs), "started": sum(r["status"] not in {"queued", "cancelled"} for r in wake_runs), "terminal": sum(r["status"] in TERMINAL for r in wake_runs)},
        }
        self.blocks.append(stamp + " MILESTONE " + label + "\n" + json.dumps(self.clean(data), indent=2))
        self.last_sequence = s.get("eventSequence") or 0
        self.step("Captured " + label)
        self.write()
        return s

    def wait(self, label, predicate, timeout=300):
        self.step("Await persisted milestone: " + label)
        until = time.monotonic() + timeout
        while time.monotonic() < until:
            s = self.read()
            if predicate(s):
                return s
            time.sleep(0.25)
        self.snapshot("TIMEOUT " + label)
        raise RuntimeError("Timed out: " + label)

    def gate(self, name):
        return self.run / (name + ".release")

    def release(self, name):
        self.gate(name).touch()
        self.step("Released driver-controlled gate " + name)

    def gate_cmd(self, name):
        return f"python3 {HERE / 'gate.py'} {self.gate(name)}"

    def write(self, outcome="CAPTURE IN PROGRESS"):
        header = [f"{self.fixture} {self.side} {self.sha} — {outcome}", "Environment: dev server; fresh DB; T3CODE_DEV_SEED=0; VM fs-37 Linux; server only.", f"Provider: {self.provider}; private model identifiers replaced by <model>.", "Narrated steps (UTC):", *self.steps, "", "Timestamped persisted-state snapshots:", *self.blocks]
        (self.run / f"{self.fixture}-{self.side}.txt").write_text(self.clean("\n".join(header)) + "\n")

    def execute(self, caps):
        providers = {p["providerInstanceId"]: p for p in caps["providers"]}
        for p in caps["providers"]:
            for m in p["models"]:
                self.private_models.add(m["id"])
                if m.get("label"):
                    self.private_models.add(m["label"])
        selection_file = SCRATCH / "model-selections-private.json"
        if not selection_file.exists():
            selection_file.write_text(json.dumps({name: providers[name]["models"][0]["id"] for name in ["codex", "cursor"]}))
            os.chmod(selection_file, 0o600)
        selections = json.loads(selection_file.read_text())
        self.private_models.update(selections.values())
        parent_model = selections[self.provider]
        child_model = selections["codex"]
        project_root = self.run / "synthetic-project"
        project_root.mkdir()
        subprocess.run(["git", "init", "-q", str(project_root)], check=True)
        (project_root / "README.md").write_text("Synthetic server-only delegated completion fixture.\n")
        subprocess.run(["git", "-C", str(project_root), "add", "README.md"], check=True)
        subprocess.run(["git", "-C", str(project_root), "-c", "user.name=Synthetic Fixture", "-c", "user.email=fixture@example.invalid", "commit", "-qm", "synthetic fixture"], check=True)
        project = self.call("t3_project_create", title="33469 synthetic fixture", workspaceRoot=str(project_root))
        self.step("Registered new synthetic project through MCP; no checkout or live database used as seed.")
        if self.fixture == "F07":
            prompt = "This is an ordinary running-turn restart control. Do not delegate anything. Execute exactly this command and wait for its completion, then reply ORDINARY_DONE: " + self.gate_cmd("ordinary")
        else:
            alpha = "Execute exactly this command and wait until it exits, then reply only ALPHA. Do not delegate or modify code: " + self.gate_cmd("alpha")
            beta = "Execute exactly this command and wait until it exits, then reply only BETA. Do not delegate or modify code: " + self.gate_cmd("beta")
            target = json.dumps({"providerInstanceId": "codex", "model": child_model})
            prompt = f"""This is a synthetic T3 Code runtime fixture. Follow the exact tool sequence. Do not edit code or create files other than those the given gate command creates.
Call orchestrator_capabilities once. Then call delegate_task twice, default async mode, isolation shared, target {target}.
First title ALPHA, task: {json.dumps(alpha)}
Second title BETA, task: {json.dumps(beta)}
As soon as both delegate calls return, end this turn with exactly SPAWNED. Do not wait on, poll, or inspect child tasks. Do not call task_status, task_cancel, or t3_thread_send.
Later, on EVERY delegated completion wake: run exactly this shell command and wait until it exits: {self.gate_cmd('wake')}
Then reply with the names of completed tasks in that wake. Do not create more tasks. The gate is driver-controlled; do not bypass, release, or time out the gate yourself.
"""
        launched = self.call("t3_thread_launch", title=self.fixture + " " + self.side + " synthetic capture", projectId=project["id"], message=prompt, runtimeMode="full-access", modelSelection={"instanceId": self.provider, "model": parent_model})
        self.tid = launched["threadId"]
        self.step("Launched synthetic parent through t3_thread_launch; thread=" + self.tid)
        self.call("t3_thread_read", threadId=self.tid)
        if self.fixture == "F07":
            self.wait("ordinary initial turn running at gate", lambda s: len(s["runs"]) == 1 and s["runs"][0]["status"] == "running" and self.gate("ordinary").with_suffix(".entered").exists())
            self.snapshot("before ordinary restart; no delegated cohort")
            result = self.call("t3_thread_send", threadId=self.tid, mode="restart", message="Restart control: abandon the old gate. Execute exactly this command, wait until it exits, then reply RESTART_DONE: " + self.gate_cmd("restart"))
            self.step("t3_thread_send(mode=restart) returned " + json.dumps(self.clean(result)))
            self.snapshot("after ordinary restart commit")
            self.release("ordinary")
            self.wait("ordinary restart attempt running at gate", lambda s: len(s["run_attempts"]) >= 2 and self.gate("restart").with_suffix(".entered").exists())
            self.snapshot("restarted ordinary turn; cohort absence retained")
            self.release("restart")
            self.wait("ordinary restarted run completed", lambda s: s["runs"][0]["status"] == "completed")
            self.snapshot("ordinary run finished; no completion messages, tasks or cohort")
            return
        self.wait("P completed and A/B tasks running behind gates", lambda s: len(s["runs"]) >= 1 and s["runs"][0]["status"] == "completed" and len(s["subagents"]) == 2 and self.gate("alpha").with_suffix(".entered").exists() and self.gate("beta").with_suffix(".entered").exists(), timeout=420)
        self.snapshot("P settled with gated A and B")
        self.release("alpha")
        self.wait("W running owns D1 and wake gate entered", lambda s: len(s["runs"]) >= 2 and s["runs"][1]["status"] == "running" and self.gate("wake").with_suffix(".entered").exists())
        self.snapshot("W running with A claimed; B still gated")
        self.release("beta")
        self.wait("B terminal pending behind running W", lambda s: sum(t["status"] == "completed" and t.get("completionDelivery", {}).get("state") == "pending" for t in s["subagents"]) == 1)
        self.snapshot("before steer; B terminal pending; W retains D1")
        wake = self.read()["runs"][1]
        self.wake_id = wake["id"]
        restart_text = "Driver-controlled steering fixture. Abandon the previous gate if this is a restart. Execute exactly this command and wait until it exits, then reply ACK: " + self.gate_cmd("restart")
        if self.fixture == "F06":
            result = self.call("t3_thread_send", threadId=self.tid, mode="steer", message="Native active steering control: retain the current wake gate command and finish it normally when released. Do not start a different gate. Reply ACK after it completes.")
            self.step("t3_thread_send(mode=steer) returned " + json.dumps(self.clean(result)))
            self.snapshot("native steer accepted; original W input and attempt retained before terminal")
            self.release("wake")
        else:
            if self.fixture == "F04":
                queued = self.call("t3_thread_send", threadId=self.tid, mode="queue", message=restart_text)
                self.queue_id = queued["runId"]
                self.step("Queued Q using t3_thread_send(mode=queue): " + json.dumps(self.clean(queued)))
                self.snapshot("Q queued before promote-to-steer")
                result = self.call("t3_queue", action="promote_to_steer", threadId=self.tid, queuedRunId=self.queue_id, targetRunId=self.wake_id)
                self.step("t3_queue(action=promote_to_steer) returned " + json.dumps(self.clean(result)))
            else:
                mode = "steer" if self.provider == "cursor" else "restart"
                result = self.call("t3_thread_send", threadId=self.tid, mode=mode, message=restart_text)
                self.step("t3_thread_send(mode=" + mode + ") returned " + json.dumps(self.clean(result)))
            self.snapshot("after restart commit; journal includes exact commit transitions")
            self.wait("W steering_restart attempt entered driver gate", lambda s: any(a["runId"] == self.wake_id and a["reason"] == "steering_restart" for a in s["run_attempts"]) and self.gate("restart").with_suffix(".entered").exists())
            self.snapshot("successor queued or mailbox route while restarted W held")
            self.release("restart")
            self.release("wake")
        self.wait("W terminal", lambda s: next(r for r in s["runs"] if r["id"] == self.wake_id)["status"] == "completed")
        if self.side == "before" and self.fixture in {"F03", "F04"}:
            self.snapshot("W finished; generation 1 stranded, B pending; no successor expected on main")
        else:
            self.wait("A and B delivered and all wake runs terminal", lambda s: all(t.get("completionDelivery", {}).get("state") == "delivered" for t in s["subagents"]) and all(r["status"] in TERMINAL for r in s["runs"]))
            self.snapshot("successor finished; A/B delivered; P delivery null")

    def stop(self):
        for gate in self.run.glob("*.entered"):
            gate.with_suffix(".release").touch()
        if self.server and self.server.poll() is None:
            # Capture descendants from the PID we spawned, never from command names.
            parent_pids = {self.server.pid}
            descendants = []
            while True:
                added = []
                for proc in pathlib.Path("/proc").iterdir():
                    if not proc.name.isdigit() or int(proc.name) in parent_pids:
                        continue
                    try:
                        stat = (proc / "stat").read_text().rsplit(")", 1)[1].split()
                        if int(stat[1]) in parent_pids:
                            added.append(int(proc.name))
                    except (FileNotFoundError, ProcessLookupError, PermissionError):
                        pass
                if not added:
                    break
                descendants.extend(added)
                parent_pids.update(added)
            (self.run / "captured-descendant-pids.json").write_text(json.dumps(descendants))
            self.step("Captured spawned descendants for cleanup: " + json.dumps(descendants))
            for pid in reversed(descendants):
                try:
                    os.kill(pid, signal.SIGINT)
                except ProcessLookupError:
                    pass
            if self.server.poll() is None:
                self.server.send_signal(signal.SIGINT)
            try:
                self.server.wait(timeout=30)
                self.step(f"Stopped captured dev-runner PID {self.server.pid}; exit={self.server.returncode}.")
            except subprocess.TimeoutExpired:
                self.step(f"FRICTION: captured dev-runner PID {self.server.pid} did not stop on SIGINT.")
        if hasattr(self, "log"):
            self.log.close()


def main():
    side, fixture = sys.argv[1:3]
    provider = sys.argv[3] if len(sys.argv) > 3 else "codex"
    c = Capture(side, fixture, provider)
    outcome = "UNVERIFIED: driver unfinished"
    print("RUN_DIR " + str(c.run), flush=True)
    try:
        c.start()
        caps = c.capabilities()
        print("CAPABILITIES_FILE " + str(c.run / "capabilities-private.json"), flush=True)
        if fixture == "probe":
            c.step("Probe initialized; provider capabilities saved privately for inspection.")
            outcome = "PROBE ONLY"
            return
        c.execute(caps)
        outcome = "CAPTURED"
    except Exception as error:
        c.step("ERROR " + str(error))
        (c.run / "error-private.txt").write_text(traceback.format_exc())
        outcome = "UNVERIFIED: " + str(error)
        raise
    finally:
        c.stop()
        c.write(outcome)


if __name__ == "__main__":
    main()