#!/usr/bin/env python3 """Offline SSE timing lab. Python 3 standard library; loopback only, no real API. Run: python3 claude-stream-probe.py --scenario all Synthetic outcomes are demonstrations, not evidence about a production service. No CLI option accepts an endpoint, key, model, or prompt. No automatic retries. """ import argparse from collections import Counter import http.client from http.server import BaseHTTPRequestHandler, HTTPServer import json import math import socket import threading import time SCENARIOS = ( "normal", "sse-error", "truncated", "idle-gap", "delayed-first-event", "ping-deadline", "fragmented-multiline", ) KNOWN_EVENTS = { "message_start", "content_block_start", "content_block_delta", "content_block_stop", "message_delta", "message_stop", "ping", "error", } MAX_EVENT_BYTES = 65536 MAX_STREAM_BYTES = 1048576 MAX_EVENTS = 10000 def frame(kind, extra=None): data = {"type": kind} if extra: data.update(extra) return ("event: " + kind + "\ndata: " + json.dumps(data) + "\n\n").encode() class LimitExceeded(Exception): pass class SSEParser: """Keep only the current bounded frame; never expose event data.""" def __init__(self): self.buffer = b"" self.lines = [] self.size = 0 def feed(self, chunk): self.buffer += chunk while b"\n" in self.buffer: line, self.buffer = self.buffer.split(b"\n", 1) self.size += len(line) + 1 if self.size > MAX_EVENT_BYTES: raise LimitExceeded() line = line.rstrip(b"\r") if line: self.lines.append(line) continue name, data = "message", [] for field in self.lines: if field.startswith(b"event:"): name = field[6:].lstrip(b" ").decode("utf-8", "replace") elif field.startswith(b"data:"): data.append(field[5:].lstrip(b" ")) self.lines, self.size = [], 0 if not data: continue try: payload = json.loads(b"\n".join(data)) except (ValueError, UnicodeDecodeError): yield "invalid_event" continue if not isinstance(payload, dict) or payload.get("type") != name: yield "invalid_event" else: yield name if name in KNOWN_EVENTS else "other" if len(self.buffer) + self.size > MAX_EVENT_BYTES: raise LimitExceeded() class SyntheticServer(HTTPServer): """One ephemeral local server; accepts exactly one HTTP request.""" def __init__(self, scenario, idle_timeout, total_timeout): self.scenario = scenario self.idle_timeout = idle_timeout self.total_timeout = total_timeout self.requests = 0 self.request_valid = False super().__init__(("127.0.0.1", 0), SyntheticHandler) self.timeout = 2 class SyntheticHandler(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_POST(self): server = self.server server.requests += 1 raw = self.rfile.read(int(self.headers.get("Content-Length", "0"))) server.request_valid = self.path == "/v1/messages" and json.loads(raw)["stream"] is True if server.scenario == "redirect": self.send_response(307) self.send_header("Location", "https://example.invalid/private?secret=synthetic") self.end_headers() return self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.end_headers() try: self.emit_stream() except (BrokenPipeError, ConnectionResetError, OSError): pass def emit_stream(self): server = self.server scenario = server.scenario def send(data): self.wfile.write(data) self.wfile.flush() if scenario == "delayed-first-event": time.sleep(server.idle_timeout * 1.5) if scenario == "ping-deadline": end = time.monotonic() + server.total_timeout * 1.5 while time.monotonic() < end: send(frame("ping")) time.sleep(min(server.idle_timeout / 4, 0.02)) return if scenario == "oversized": send(b"data: " + b"x" * (MAX_EVENT_BYTES + 1)) return if scenario == "fragmented-multiline": payload = ( b": comment\r\nevent: message_start\r\ndata: {\r\n" b'data: "type": "message_start"}\r\n\r\n' + frame("content_block_start") + frame("content_block_delta", {"delta": {"text": "SYNTHETIC_OUTPUT_PRIVATE"}}) + frame("content_block_stop") + frame("message_delta") + frame("message_stop") ) for index in range(0, len(payload), 3): send(payload[index:index + 3]) return send(frame("message_start")) send(frame("content_block_start")) send(frame("content_block_delta", {"delta": {"text": "SYNTHETIC_OUTPUT_PRIVATE"}})) if scenario == "sse-error": send(frame("error", {"error": {"message": "SYNTHETIC_ERROR_PRIVATE"}})) elif scenario == "idle-gap": time.sleep(server.idle_timeout * 1.5) send(frame("message_stop")) elif scenario != "truncated": send(frame("content_block_stop")) send(frame("message_delta")) send(frame("message_stop")) def observe(port, idle_timeout, total_timeout): """Single local POST; report whitelist metadata only. TTFB approximates headers.""" start = time.monotonic() result = { "status": None, "outcome": "transport_error", "ttfb_ms": None, "first_event_ms": None, "event_count": 0, "events": {}, "event_times_ms": [], "elapsed_ms": None, } connection = http.client.HTTPConnection("127.0.0.1", port, timeout=min(idle_timeout, total_timeout)) parser, counts, received = SSEParser(), Counter(), 0 def milliseconds(): return round((time.monotonic() - start) * 1000, 2) try: body = json.dumps({"model": "synthetic-not-a-model", "stream": True, "max_tokens": 1, "messages": [{"role": "user", "content": "SYNTHETIC_PROMPT_PRIVATE"}]}) connection.request("POST", "/v1/messages", body=body, headers={ "Content-Type": "application/json", "x-api-key": "SYNTHETIC_KEY_PRIVATE", }) transport = connection.sock response = connection.getresponse() result["status"] = response.status result["ttfb_ms"] = milliseconds() if response.status != 200: result["outcome"] = "http_error" return result last_byte = time.monotonic() while True: now = time.monotonic() total_left = total_timeout - (now - start) idle_left = idle_timeout - (now - last_byte) if total_left <= 0 or idle_left <= 0: result["outcome"] = "total_timeout" if total_left <= idle_left else "idle_timeout" break # Keep a reference: HTTP/1.0 transfers socket ownership to response. transport.settimeout(min(total_left, idle_left)) try: chunk = response.read1(4096) except socket.timeout: result["outcome"] = "total_timeout" if total_left <= idle_left else "idle_timeout" break if not chunk: result["outcome"] = "truncated" break last_byte = time.monotonic() if last_byte - start >= total_timeout: result["outcome"] = "total_timeout" break received += len(chunk) if received > MAX_STREAM_BYTES: raise LimitExceeded() for kind in parser.feed(chunk): if result["event_count"] >= MAX_EVENTS: raise LimitExceeded() if result["first_event_ms"] is None: result["first_event_ms"] = milliseconds() counts[kind] += 1 result["event_count"] += 1 result["event_times_ms"].append(milliseconds()) if kind in ("message_stop", "error", "invalid_event"): result["outcome"] = "sse_error" if kind == "error" else kind return result except LimitExceeded: result["outcome"] = "limit_exceeded" except (OSError, http.client.HTTPException): result["outcome"] = "transport_error" finally: connection.close() result["events"] = dict(counts) result["elapsed_ms"] = milliseconds() return result def run_scenario(scenario, idle_timeout=0.2, total_timeout=0.6): with SyntheticServer(scenario, idle_timeout, total_timeout) as server: worker = threading.Thread(target=server.handle_request, daemon=True) worker.start() result = observe(server.server_port, idle_timeout, total_timeout) worker.join(timeout=min(idle_timeout * 2, 1)) return {"case": scenario, **result} def timeout_value(value): try: parsed = float(value) except ValueError: raise argparse.ArgumentTypeError("timeout must be a finite number between 0.01 and 10") from None if not math.isfinite(parsed) or not 0.01 <= parsed <= 10: raise argparse.ArgumentTypeError("timeout must be a finite number between 0.01 and 10") return parsed def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--scenario", choices=(*SCENARIOS, "all"), default="normal") parser.add_argument("--idle-timeout", type=timeout_value, default=0.2, help="no-body-byte deadline in seconds (default: 0.2)") parser.add_argument("--total-timeout", type=timeout_value, default=0.6, help="whole-request deadline in seconds (default: 0.6)") args = parser.parse_args(argv) for scenario in SCENARIOS if args.scenario == "all" else (args.scenario,): print(json.dumps(run_scenario(scenario, args.idle_timeout, args.total_timeout), sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())