Severe WS stalls in 0.153.4: live Ping/Pong still trips the idle watchdog (Windows loopback-only reproduction)

"""Exercise an installed Codex CLI against a loopback-only synthetic peer.

No OpenAI credentials, real model requests, project tools, or image data are used.
The caller retains the scratch directory for separate, verified cleanup.
"""
import argparse
import base64
import hashlib
import http.server
import json
import os
from pathlib import Path
import select
import socket
import struct
import subprocess
import tempfile
import threading
import time


class Case:
    def __init__(self, name, heartbeat, idle_ms, delay_s=3.2):
        self.name = name
        self.heartbeat = heartbeat
        self.idle_ms = idle_ms
        self.delay_s = delay_s
        self.events = []
        self.lock = threading.Lock()
        self.stop = threading.Event()
        self.start = time.monotonic()

    def log(self, event, **fields):
        with self.lock:
            self.events.append({"t_s": round(time.monotonic() - self.start, 4),
                                "event": event, **fields})


def recv_exact(sock, count):
    chunks = bytearray()
    while len(chunks) < count:
        part = sock.recv(count - len(chunks))
        if not part:
            raise EOFError("peer closed")
        chunks.extend(part)
    return bytes(chunks)


def recv_frame(sock):
    first, second = recv_exact(sock, 2)
    opcode = first & 15
    size = second & 127
    if size == 126:
        size = struct.unpack("!H", recv_exact(sock, 2))[0]
    elif size == 127:
        size = struct.unpack("!Q", recv_exact(sock, 8))[0]
    if size > 2_000_000:
        raise ValueError("synthetic-test frame limit exceeded")
    mask = recv_exact(sock, 4) if second & 128 else None
    payload = recv_exact(sock, size)
    if mask:
        payload = bytes(value ^ mask[i % 4] for i, value in enumerate(payload))
    return opcode, payload


def send_frame(sock, opcode, payload):
    size = len(payload)
    if size < 126:
        header = bytes([128 | opcode, size])
    elif size < 65536:
        header = bytes([128 | opcode, 126]) + struct.pack("!H", size)
    else:
        header = bytes([128 | opcode, 127]) + struct.pack("!Q", size)
    sock.sendall(header + payload)


def response_events(response_id, warmup=False):
    message = {"id": "msg_synthetic", "type": "message", "role": "assistant",
               "status": "completed", "content": [{"type": "output_text", "text": "OK", "annotations": []}]}
    response = {"id": response_id, "object": "response", "status": "completed",
                "model": "gpt-6-astra", "output": [] if warmup else [message],
                "usage": {"input_tokens": 1, "output_tokens": 0 if warmup else 1,
                          "total_tokens": 1 if warmup else 2,
                          "input_tokens_details": {"cached_tokens": 0},
                          "output_tokens_details": {"reasoning_tokens": 0}}}
    if not warmup:
        yield {"type": "response.output_item.added", "output_index": 0,
               "item": {**message, "status": "in_progress", "content": []}}
        yield {"type": "response.content_part.added", "item_id": "msg_synthetic",
               "output_index": 0, "content_index": 0,
               "part": {"type": "output_text", "text": "", "annotations": []}}
        yield {"type": "response.output_text.delta", "item_id": "msg_synthetic",
               "output_index": 0, "content_index": 0, "delta": "OK"}
        yield {"type": "response.output_text.done", "item_id": "msg_synthetic",
               "output_index": 0, "content_index": 0, "text": "OK"}
        yield {"type": "response.content_part.done", "item_id": "msg_synthetic",
               "output_index": 0, "content_index": 0, "part": message["content"][0]}
        yield {"type": "response.output_item.done", "output_index": 0, "item": message}
    yield {"type": "response.completed", "response": response}


class Peer(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, *args):
        pass

    def do_GET(self):
        case = self.server.case
        if self.headers.get("Upgrade", "").lower() != "websocket":
            case.log("non_websocket_get", path=self.path)
            body = b'{"models":[]}'
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        key = self.headers.get("Sec-WebSocket-Key", "")
        accept = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()).decode()
        self.send_response(101)
        self.send_header("Upgrade", "websocket")
        self.send_header("Connection", "Upgrade")
        self.send_header("Sec-WebSocket-Accept", accept)
        self.end_headers()
        self.wfile.flush()
        self.close_connection = True
        self.connection.settimeout(2)
        case.log("websocket_upgraded")
        try:
            while not case.stop.is_set():
                opcode, payload = recv_frame(self.connection)
                if opcode == 8:
                    case.log("client_close")
                    return
                if opcode == 10:
                    case.log("pong_received")
                    continue
                if opcode != 1:
                    case.log("unexpected_client_opcode", opcode=opcode)
                    continue
                request = json.loads(payload)
                if request.get("type") != "response.create":
                    case.log("other_client_event", kind=request.get("type"))
                    continue
                warmup = request.get("generate") is False
                case.log("response_create_received", warmup=warmup, request_bytes=len(payload))
                response_id = "resp_warmup" if warmup else "resp_synthetic"
                created = {"type": "response.created", "response": {"id": response_id, "status": "in_progress", "model": "gpt-6-astra", "output": []}}
                self.send_json(created)
                case.log("response_created_sent", warmup=warmup)
                if not warmup:
                    start = time.monotonic()
                    next_beat = start + 0.15
                    while time.monotonic() - start < case.delay_s and not case.stop.is_set():
                        now = time.monotonic()
                        if now >= next_beat:
                            if case.heartbeat == "protocol_ping":
                                send_frame(self.connection, 9, b"synthetic-liveness")
                                case.log("ping_sent")
                            else:
                                self.send_json({"type": "keepalive"})
                                case.log("application_keepalive_sent")
                            next_beat = now + 0.2
                        ready, _, _ = select.select([self.connection], [], [], 0.03)
                        if ready:
                            op, data = recv_frame(self.connection)
                            if op == 10:
                                case.log("pong_received")
                            elif op == 8:
                                case.log("client_close_during_wait")
                                return
                            elif op == 9:
                                send_frame(self.connection, 10, data)
                                case.log("client_ping_received")
                            else:
                                case.log("unexpected_during_wait", opcode=op)
                    if case.stop.is_set():
                        return
                for event in response_events(response_id, warmup):
                    self.send_json(event)
                case.log("completion_sent", transport="websocket", warmup=warmup)
        except (EOFError, OSError, ValueError) as error:
            case.log("peer_connection_end", error_type=type(error).__name__)

    def send_json(self, event):
        send_frame(self.connection, 1, json.dumps(event, separators=(",", ":")).encode())

    def do_POST(self):
        case = self.server.case
        size = int(self.headers.get("Content-Length", "0"))
        if size > 2_000_000:
            self.send_error(413)
            return
        self.rfile.read(size)
        case.log("http_fallback_post", request_bytes=size)
        events = [{"type": "response.created", "response": {"id": "resp_http", "status": "in_progress", "output": []}}]
        events.extend(response_events("resp_http"))
        body = b"".join(b"data: " + json.dumps(event, separators=(",", ":")).encode() + b"\n\n" for event in events)
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
        self.wfile.flush()
        case.log("completion_sent", transport="http", warmup=False)


def run_case(executable, scratch, name, heartbeat, idle_ms):
    case = Case(name, heartbeat, idle_ms)
    home = scratch / name / "home"
    work = scratch / name / "work"
    home.mkdir(parents=True)
    work.mkdir()
    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Peer)
    server.daemon_threads = True
    server.case = case
    threading.Thread(target=server.serve_forever, daemon=True).start()
    port = server.server_address[1]
    config = f'''model = "gpt-6-astra"
model_provider = "local_idle_probe"
approval_policy = "never"
web_search = "disabled"
[features]
memories = false
plugins = false
shell_snapshot = false
[memories]
use_memories = false
generate_memories = false
[model_providers.local_idle_probe]
name = "Loopback synthetic idle probe"
base_url = "http://127.0.0.1:{port}/v1"
wire_api = "responses"
requires_openai_auth = false
supports_websockets = true
stream_idle_timeout_ms = {idle_ms}
stream_max_retries = 0
request_max_retries = 0
'''
    (home / "config.toml").write_text(config, encoding="utf-8")
    allowed = {"SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT", "PATH", "TEMP", "TMP", "USERPROFILE", "APPDATA", "LOCALAPPDATA", "HOMEDRIVE", "HOMEPATH", "NUMBER_OF_PROCESSORS", "PROCESSOR_ARCHITECTURE"}
    env = {k: v for k, v in os.environ.items() if k.upper() in allowed}
    env.update(CODEX_HOME=str(home), NO_PROXY="127.0.0.1,localhost,::1")
    command = [str(executable), "exec", "--json", "--ephemeral", "--skip-git-repo-check", "--ignore-rules", "--color", "never", "--sandbox", "read-only", "--cd", str(work), "Reply with OK. Do not call any tools."]
    start = time.monotonic()
    timed_out = False
    try:
        result = subprocess.run(command, env=env, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=25, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
        stdout, stderr, code = result.stdout, result.stderr, result.returncode
    except subprocess.TimeoutExpired as error:
        timed_out = True
        stdout = error.stdout.decode("utf-8", "replace") if isinstance(error.stdout, bytes) else error.stdout or ""
        stderr = error.stderr.decode("utf-8", "replace") if isinstance(error.stderr, bytes) else error.stderr or ""
        code = None
    finally:
        case.stop.set()
        server.shutdown()
        server.server_close()
    combined = stdout + "\n" + stderr
    counters = {}
    for event in case.events:
        counters[event["event"]] = counters.get(event["event"], 0) + 1
    errors = [line for line in combined.splitlines() if "idle timeout" in line or "falling back" in line.lower() or '"type":"error"' in line]
    errors = [line.replace(str(home), "<isolated-home>").replace(str(work), "<isolated-work>") for line in errors]
    return {"name": name, "heartbeat": heartbeat, "configured_idle_ms": idle_ms,
            "synthetic_completion_delay_s": case.delay_s, "process_timeout": timed_out,
            "exit_code": code, "elapsed_s": round(time.monotonic() - start, 3),
            "idle_timeout_reported": "idle timeout waiting for websocket" in combined,
            "http_fallback_observed": counters.get("http_fallback_post", 0) > 0,
            "event_counts": counters, "selected_client_messages": errors,
            "events": case.events, "client_stdout": stdout.replace(str(home), "<isolated-home>").replace(str(work), "<isolated-work>"),
            "client_stderr": stderr.replace(str(home), "<isolated-home>").replace(str(work), "<isolated-work>")}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--codex", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    scratch = Path(tempfile.mkdtemp(prefix="codex-ws-idle-probe-"))
    print(json.dumps({"scratch_directory": str(scratch)}), flush=True)
    report = {"purpose": "Installed CLI versus loopback-only synthetic peer", "real_model_requests": 0,
              "production_configuration_modified": False,
              "codex_binary_sha256": hashlib.sha256(args.codex.read_bytes()).hexdigest(),
              "cases": []}
    for spec in [("ping_short_idle", "protocol_ping", 1500),
                 ("application_heartbeat_short_idle", "application_keepalive", 1500),
                 ("ping_long_idle", "protocol_ping", 6000)]:
        result = run_case(args.codex, scratch, *spec)
        report["cases"].append(result)
        args.output.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
        print(json.dumps({k: result[k] for k in ["name", "exit_code", "elapsed_s", "idle_timeout_reported", "http_fallback_observed", "event_counts"]}), flush=True)
    print(json.dumps({"output": str(args.output), "scratch_directory": str(scratch)}), flush=True)


if __name__ == "__main__":
    main()