#!/usr/bin/env python3
"""Drive a real `codex app-server` and check whether stalling the client's
stdout reader makes code-mode host negotiation time out.

usage: driver.py <outdir> <stall_seconds> [model]
"""
import fcntl, json, os, subprocess, sys, threading, time

outdir, stall = sys.argv[1], float(sys.argv[2])
model = sys.argv[3] if len(sys.argv) > 3 else None
os.makedirs(outdir, exist_ok=True)
log = open(os.path.join(outdir, "events.jsonl"), "w")
t0 = time.time()

def rec(kind, obj):
    log.write(json.dumps({"t": round(time.time() - t0, 3), "kind": kind, "msg": obj}) + "\n")
    log.flush()

proc = subprocess.Popen(["codex", "app-server"], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                        stderr=open(os.path.join(outdir, "stderr.log"), "w"), cwd=outdir)
F_SETPIPE_SZ = 1031
try:
    fcntl.fcntl(proc.stdout.fileno(), F_SETPIPE_SZ, 4096)
    rec("note", f"stdout pipe size set to {fcntl.fcntl(proc.stdout.fileno(), 1032)}")
except OSError as e:
    rec("note", f"pipe resize failed {e}")

reading = threading.Event(); reading.set()
msgs = []; cond = threading.Condition()

def reader():
    for line in proc.stdout:
        reading.wait()
        try:
            m = json.loads(line)
        except Exception:
            continue
        rec("in", m)
        with cond:
            msgs.append(m); cond.notify_all()
threading.Thread(target=reader, daemon=True).start()

nid = [0]
def send(method, params, notify=False):
    m = {"method": method, "params": params}
    if not notify:
        nid[0] += 1; m["id"] = nid[0]
    rec("out", m)
    proc.stdin.write((json.dumps(m) + "\n").encode()); proc.stdin.flush()
    return m.get("id")

def wait_for(pred, timeout):
    end = time.time() + timeout
    with cond:
        while True:
            for m in msgs:
                if pred(m):
                    return m
            left = end - time.time()
            if left <= 0:
                return None
            cond.wait(left)

def resp(i, timeout=120):
    return wait_for(lambda m: m.get("id") == i and "method" not in m, timeout)

r = resp(send("initialize", {"clientInfo": {"name": "repro-24322", "version": "0"}}))
send("initialized", {}, notify=True)
tp = {"cwd": outdir, "approvalPolicy": "never", "sandbox": "danger-full-access", "ephemeral": True}
if model:
    tp["model"] = model
r = resp(send("thread/start", tp))
tid = r["result"]["thread"]["id"]

PROMPT = ("Use your shell/exec tool to run each of these as a separate tool call, one at a time: "
          "`echo one`, `echo two`, `echo three`, `seq 1 3000`. Then reply DONE.")

KILL_HOST = os.environ.get("KILL_HOST") == "1"

def descendants(pid):
    out = []
    for task in os.listdir(f"/proc/{pid}/task"):
        try:
            kids = open(f"/proc/{pid}/task/{task}/children").read().split()
        except OSError:
            kids = []
        for k in kids:
            out.append(int(k)); out.extend(descendants(int(k)))
    return out

def turn(label, stall_s):
    if stall_s:
        reading.clear(); rec("note", f"{label}: reader stalled for {stall_s}s")
        if KILL_HOST:
            for d in descendants(proc.pid):
                try:
                    exe = os.readlink(f"/proc/{d}/exe")
                except OSError:
                    continue
                if exe.endswith("/codex-code-mode-host"):
                    os.kill(d, 9); rec("note", f"killed code-mode host pid {d}")
    send("turn/start", {"threadId": tid, "input": [{"type": "text", "text": PROMPT}]})
    if stall_s:
        time.sleep(stall_s)
        rec("note", f"{label}: reader resumed"); reading.set()
    done = wait_for(lambda m: m.get("method") == "turn/completed" and m["params"].get("turn", {}).get("id") not in seen, 600)
    seen.add(done["params"]["turn"]["id"] if done else None)
    rec("note", f"{label}: turn completed={done is not None}")

seen = set()
turn("control", 0)
turn("stalled", stall)
proc.terminate(); proc.wait(10)
rec("note", "exit")