diff --git a/README.md b/README.md index b39fbcb..745638c 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,8 @@ CURI is a local, privacy-first dashboard for Codex usage and OpenAI-compatible relay reliability. It reads local JSONL files, stores only aggregate metadata in SQLite, and serves a loopback-only dashboard. +It includes the local retry relay. The relay and dashboard can run together, so CURI is both the observer and the local request boundary. + ## What it shows - latest quota windows from `token_count.rate_limits` (unknown windows stay unknown) @@ -10,6 +12,7 @@ CURI is a local, privacy-first dashboard for Codex usage and OpenAI-compatible r - daily trend filtering by observed model and project - tool calls grouped as Shell, MCP, Browser/search and Other - structured relay events: status, attempts, latency, terminal state and requested/reported model differences +- a local OpenAI-compatible relay with transport/temporary-error retries and safe SSE reconnects - coverage dates and the last scan time CURI does not read `auth.json`, request bodies, prompts, response text or API keys. It sends no telemetry. @@ -20,11 +23,13 @@ Python 3.10+ is enough. ```bash python curi.py doctor -python curi.py serve +python curi.py serve --upstream https://api.example.com/v1 ``` Open . CURI scans `~/.codex/sessions` every three seconds. Override paths when needed: +The relay listens on `http://127.0.0.1:8080/v1`; point Codex's API base URL at that address and keep the CURI process running. Your existing API key remains in Codex and is forwarded to the configured upstream; CURI never stores it. + ```bash python curi.py serve \ --codex-home ~/.codex \ @@ -38,7 +43,15 @@ Run a one-shot scan and inspect JSON: python curi.py scan # use `doctor --json` for machine-readable diagnostics ``` -The relay side is intentionally an input contract. A relay (including Steady Relay or your own proxy) can append one JSON object per line: +To run only the relay: + +```bash +python curi.py relay --upstream https://api.example.com/v1 +``` + +The relay retries connection failures, timeouts, `408/425/429/5xx`, and recognized capacity/usage-limit SSE failures before real output or tool-call data reaches Codex. Once output is committed, it closes the incomplete stream instead of replaying a request that could duplicate text or a tool call. `--buffer-until-success` enables the stronger mode that holds SSE in memory until `response.completed`; its per-attempt limit is 64 MiB. + +The relay appends one metadata-only JSON object per request to `~/.curi/relay-events.jsonl`: ```json {"schema_version":1,"timestamp":"2026-09-25T12:00:00Z","request_id":"req-1","requested_model":"model-a","reported_model":"model-a","status":200,"attempts":2,"first_byte_ms":420,"duration_ms":3800,"error_class":null,"stream_terminal":"response.completed"} @@ -53,11 +66,11 @@ python -m unittest -v python -m py_compile curi.py ``` -The project deliberately has no runtime dependencies. The dashboard is served by Python's standard library. The scanner uses file offsets and resumes safely after a restart; a truncated or rewritten JSONL file is rescanned from the beginning. +The project deliberately has no runtime dependencies. The dashboard and relay use Python's standard library. The scanner uses file offsets and resumes safely after a restart; a truncated or rewritten JSONL file is rescanned from the beginning. ## Design boundaries -CURI observes local events; it does not automatically route between providers, run probes, evaluate answer quality, or copy Codex credentials. Relay retry safety remains the relay's responsibility: retries are only safe before real output or tool-call data has been committed to a client. +CURI does not automatically route between providers, run probes, evaluate answer quality, or copy Codex credentials. It keeps the relay and monitor in one project but they remain separate local roles: the relay handles forwarding/retry, while the monitor parses local usage and relay events. The integration direction was informed by [Steady Relay](https://github.com/937204197/steady-relay) and [Codex Model Watch](https://github.com/ysh1112/codex-model-watch). See [NOTICE.md](NOTICE.md) for attribution and license notes. diff --git a/curi.py b/curi.py index 48455ae..87bce68 100644 --- a/curi.py +++ b/curi.py @@ -21,6 +21,8 @@ from pathlib import Path from typing import Any +from relay import RelayConfig, create_server + SCHEMA = """ CREATE TABLE IF NOT EXISTS files( @@ -396,6 +398,15 @@ def scan_loop() -> None: store.scan(args.codex_home, args.relay_events, args.archive_dir) time.sleep(max(1, args.interval)) threading.Thread(target=scan_loop, daemon=True).start() + relay_server = None + if args.upstream: + relay_server = create_server(RelayConfig( + upstream=args.upstream, host=args.relay_host, port=args.relay_port, + max_retries=args.max_retries, backoff_seconds=args.retry_backoff, + request_timeout=args.request_timeout, event_path=args.relay_events, + buffer_until_success=args.buffer_until_success)) + threading.Thread(target=relay_server.serve_forever, daemon=True).start() + print(f"CURI relay listening at http://{args.relay_host}:{relay_server.server_port}/v1") server = ThreadingHTTPServer(("127.0.0.1", args.port), Handler) print(f"CURI listening at http://127.0.0.1:{args.port} (loopback only)") try: @@ -404,6 +415,9 @@ def scan_loop() -> None: pass finally: server.server_close() + if relay_server is not None: + relay_server.shutdown() + relay_server.server_close() def parser() -> argparse.ArgumentParser: @@ -416,6 +430,22 @@ def common(s: argparse.ArgumentParser) -> None: s.add_argument("--db", default=os.getenv("CURI_DB", str(Path.home() / ".curi" / "curi.sqlite3"))) s = sub.add_parser("serve", help="scan and serve the local dashboard") common(s); s.add_argument("--port", type=int, default=8792); s.add_argument("--interval", type=int, default=3) + s.add_argument("--upstream", default=os.getenv("UPSTREAM_BASE_URL", ""), help="also start the local retry relay") + s.add_argument("--relay-host", default="127.0.0.1") + s.add_argument("--relay-port", type=int, default=8080) + s.add_argument("--max-retries", type=int, default=3) + s.add_argument("--retry-backoff", type=float, default=0.5) + s.add_argument("--request-timeout", type=float, default=120.0) + s.add_argument("--buffer-until-success", action="store_true", help="buffer SSE until response.completed") + s = sub.add_parser("relay", help="start the local OpenAI-compatible retry relay") + s.add_argument("--upstream", default=os.getenv("UPSTREAM_BASE_URL", ""), required=False) + s.add_argument("--host", default="127.0.0.1") + s.add_argument("--port", type=int, default=8080) + s.add_argument("--relay-events", default=os.getenv("CURI_RELAY_EVENTS", str(Path.home() / ".curi" / "relay-events.jsonl"))) + s.add_argument("--max-retries", type=int, default=3) + s.add_argument("--retry-backoff", type=float, default=0.5) + s.add_argument("--request-timeout", type=float, default=120.0) + s.add_argument("--buffer-until-success", action="store_true", help="buffer SSE until response.completed") s = sub.add_parser("scan", help="scan local JSONL once and print a summary") common(s); s.add_argument("--days", type=int, default=0) s = sub.add_parser("doctor", help="check local paths without reading credentials") @@ -429,6 +459,13 @@ def main(argv: list[str] | None = None) -> int: ok, checks = doctor(args.codex_home, args.relay_events, args.db, args.archive_dir) print(json.dumps({"ok": ok, "checks": checks}, ensure_ascii=False, indent=2) if args.json else "\n".join(f"{'OK' if x['ok'] else 'MISSING'} {x['name']}: {x['path']}" for x in checks)) return 0 if ok else 1 + if args.command == "relay": + from relay import serve as serve_relay + serve_relay(RelayConfig(upstream=args.upstream, host=args.host, port=args.port, + max_retries=args.max_retries, backoff_seconds=args.retry_backoff, + request_timeout=args.request_timeout, event_path=args.relay_events, + buffer_until_success=args.buffer_until_success)) + return 0 store = Store(args.db) stats = store.scan(args.codex_home, args.relay_events, args.archive_dir) if args.command == "scan": diff --git a/relay.py b/relay.py new file mode 100644 index 0000000..8208625 --- /dev/null +++ b/relay.py @@ -0,0 +1,370 @@ +#!/usr/bin/env python3 +"""CURI's local OpenAI-compatible relay. + +The relay retries transport errors and temporary upstream failures before any +real SSE output is committed to the client. It records metadata-only JSONL +events for the CURI dashboard; request and response bodies are never logged. +""" +from __future__ import annotations + +import json +import os +import random +import ssl +import threading +import time +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, BinaryIO +from urllib.error import HTTPError, URLError +from urllib.parse import urlsplit, urlunsplit +from urllib.request import Request, urlopen + + +RETRY_STATUS = {408, 425, 429} | set(range(500, 600)) +HOP_BY_HOP = {"connection", "keep-alive", "proxy-authenticate", "proxy-authorization", + "te", "trailer", "transfer-encoding", "upgrade", "host"} +STRUCTURAL_EVENTS = {"response.created", "response.queued", "response.in_progress", + "response.metadata", "response.output_item.added", "response.content_part.added"} +TERMINAL_EVENTS = {"response.completed", "response.failed", "error"} +MAX_BUFFERED_SSE_BYTES = 64 * 1024 * 1024 + + +@dataclass(frozen=True) +class RelayConfig: + upstream: str + host: str = "127.0.0.1" + port: int = 8080 + max_retries: int = 3 + backoff_seconds: float = 0.5 + request_timeout: float = 120.0 + event_path: str = os.path.expanduser("~/.curi/relay-events.jsonl") + buffer_until_success: bool = False + + +class EventWriter: + def __init__(self, path: str): + self.path = os.path.expanduser(path) + self.lock = threading.Lock() + + def write(self, event: dict[str, Any]) -> None: + os.makedirs(os.path.dirname(os.path.abspath(self.path)), exist_ok=True) + payload = {"schema_version": 1, "timestamp": iso_now(), **event} + with self.lock, open(self.path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n") + + +def iso_now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def safe_upstream_url(value: str) -> str: + try: + parsed = urlsplit(value) + if not parsed.scheme or not parsed.hostname: + return "" + host = parsed.hostname + if ":" in host and not host.startswith("["): + host = f"[{host}]" + netloc = host if parsed.port is None else f"{host}:{parsed.port}" + return urlunsplit((parsed.scheme, netloc, parsed.path, "", "")) + except ValueError: + return "" + + +def upstream_url(base: str, path: str) -> str: + base = base.rstrip("/") + if base.endswith("/v1") and path == "/v1": + return base + if base.endswith("/v1") and path.startswith("/v1/"): + path = path[3:] + return base + (path if path.startswith("/") else "/" + path) + + +def requested_model(body: bytes) -> str: + try: + value = json.loads(body).get("model", "") + return str(value)[:120] if value is not None else "" + except (TypeError, ValueError, AttributeError): + return "" + + +def retry_delay(config: RelayConfig, attempt: int, retry_after: str | None = None) -> float: + if retry_after: + try: + return min(60.0, max(0.0, float(retry_after))) + except ValueError: + pass + ceiling = min(60.0, max(0.0, config.backoff_seconds) * (2 ** attempt)) + return ceiling / 2 + random.random() * ceiling / 2 if ceiling else 0.0 + + +def classify_error(value: Any) -> str | None: + text = str(value or "").lower() + if not text: + return None + if "capacity" in text or "overloaded" in text or "server_unavailable" in text: + return "capacity" + if "usage_limit" in text or "usage limit" in text or "limit reached" in text: + return "usage_limit" + if "rate limit" in text or "429" in text: + return "rate_limit" + return "error" + + +def retryable_error(value: Any) -> bool: + return classify_error(value) in {"capacity", "usage_limit", "rate_limit"} + + +def frame_parts(frame: bytes) -> tuple[str, str]: + event = "" + data: list[str] = [] + for line in frame.decode("utf-8", errors="replace").splitlines(): + if line.startswith("event:"): + event = line[6:].strip() + elif line.startswith("data:"): + data.append(line[5:].lstrip()) + raw = "\n".join(data) + if not event and raw: + try: + event = str(json.loads(raw).get("type", "")) + except (ValueError, AttributeError): + pass + return event, raw + + +def response_model(raw: str) -> str: + try: + value = json.loads(raw) + except (TypeError, ValueError): + return "" + if not isinstance(value, dict): + return "" + response = value.get("response") if isinstance(value.get("response"), dict) else {} + return str(response.get("model") or value.get("model") or "")[:120] + + +def frame_commits_output(event: str, raw: str) -> bool: + if event in STRUCTURAL_EVENTS or event in {"", "response.failed", "error"}: + return False + if event == "response.completed": + return False + return True + + +class PreCommitStreamError(Exception): + def __init__(self, reason: str, error_class: str | None = None): + super().__init__(reason) + self.error_class = error_class + + +class RelayHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.0" + config: RelayConfig + events: EventWriter + + def send_response(self, code: int, message: str | None = None) -> None: + self.log_request(code) + self.send_response_only(code, message) + + def do_GET(self) -> None: + if self.path in ("/healthz", "/readyz"): + self.send_bytes(200, b'{"status":"ok"}', "application/json") + else: + self.proxy_request() + + def do_POST(self) -> None: self.proxy_request() + def do_HEAD(self) -> None: self.proxy_request() + def do_OPTIONS(self) -> None: self.proxy_request() + def do_PUT(self) -> None: self.proxy_request() + def do_PATCH(self) -> None: self.proxy_request() + def do_DELETE(self) -> None: self.proxy_request() + + def send_bytes(self, status: int, data: bytes, content_type: str) -> None: + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(data))) + self.send_header("Cache-Control", "no-cache") + self.end_headers() + if self.command != "HEAD": + self.wfile.write(data) + + def send_headers(self, response: Any, streaming: bool = False) -> None: + self.send_response(getattr(response, "status", 200)) + for key, value in response.headers.items(): + lowered = key.lower() + if lowered in HOP_BY_HOP or (streaming and lowered == "content-length"): + continue + self.send_header(key, value) + self.send_header("Cache-Control", "no-cache") + self.end_headers() + + def proxy_request(self) -> None: + request_id = "req-" + uuid.uuid4().hex[:12] + started = time.monotonic() + if not self.path.startswith("/v1"): + self.send_bytes(404, b'{"error":"path must start with /v1"}', "application/json") + return + try: + length = int(self.headers.get("Content-Length", "0") or 0) + except ValueError: + length = 0 + body = self.rfile.read(length) if length else b"" + model = requested_model(body) + target = upstream_url(self.config.upstream, self.path) + headers = {key: value for key, value in self.headers.items() if key.lower() not in HOP_BY_HOP} + final: dict[str, Any] = {"request_id": request_id, "requested_model": model, + "reported_model": None, "status": None, "error_class": None, + "attempts": 0, "first_byte_ms": None, "duration_ms": None, + "stream_terminal": None} + for attempt in range(self.config.max_retries + 1): + final["attempts"] = attempt + 1 + req = Request(target, data=body if "Content-Length" in self.headers else None, + headers=headers, method=self.command) + try: + with urlopen(req, timeout=self.config.request_timeout, context=ssl.create_default_context()) as response: + content_type = (response.headers.get("Content-Type") or "").lower() + if "text/event-stream" in content_type and self.command != "HEAD": + self.forward_sse(response, final, started) + else: + data = response.read() + final["reported_model"] = response_model(data.decode("utf-8", errors="replace")) or None + self.send_headers(response) + if self.command != "HEAD": + self.wfile.write(data) + final["status"] = getattr(response, "status", 200) + if final["stream_terminal"] in (None, "response.completed"): + final["error_class"] = None + final["duration_ms"] = round((time.monotonic() - started) * 1000) + self.events.write(final) + return + except HTTPError as exc: + detail = "" + try: + detail = exc.read(2048).decode("utf-8", errors="replace") + except Exception: + pass + final["status"] = exc.code + final["error_class"] = classify_error(detail) or ("rate_limit" if exc.code == 429 else "error") + if exc.code in RETRY_STATUS and attempt < self.config.max_retries: + time.sleep(retry_delay(self.config, attempt, exc.headers.get("Retry-After") if exc.headers else None)) + continue + payload = detail.encode("utf-8") or json.dumps({"error": "upstream request failed", "status": exc.code}).encode() + self.send_bytes(exc.code, payload, "application/json") + break + except PreCommitStreamError as exc: + final["error_class"] = exc.error_class or "error" + if attempt < self.config.max_retries: + time.sleep(retry_delay(self.config, attempt)) + continue + self.send_bytes(502, json.dumps({"error": "upstream stream failed", "detail": str(exc)}).encode(), "application/json") + final["status"] = 502 + break + except (URLError, TimeoutError, OSError, ValueError) as exc: + final["error_class"] = classify_error(exc) or "transport" + if attempt < self.config.max_retries: + time.sleep(retry_delay(self.config, attempt)) + continue + self.send_bytes(502, json.dumps({"error": "upstream unavailable"}).encode(), "application/json") + final["status"] = 502 + break + final["duration_ms"] = round((time.monotonic() - started) * 1000) + self.events.write(final) + + def forward_sse(self, response: Any, final: dict[str, Any], started: float) -> None: + pending: list[bytes] = [] + buffer = b"" + committed = False + terminal = "" + first_byte_at: float | None = None + while True: + try: + chunk = response.read(64 * 1024) + except (OSError, TimeoutError, URLError) as exc: + if not committed: + raise PreCommitStreamError(str(exc), classify_error(exc)) + final["stream_terminal"] = "incomplete" + return + if not chunk: + break + if first_byte_at is None: + first_byte_at = time.monotonic() + final["first_byte_ms"] = round((first_byte_at - started) * 1000) + buffer += chunk.replace(b"\r\n", b"\n") + while b"\n\n" in buffer: + frame, buffer = buffer.split(b"\n\n", 1) + wire = frame + b"\n\n" + event, raw = frame_parts(wire) + reported = response_model(raw) + if reported: + final["reported_model"] = reported + pending.append(wire) + if self.config.buffer_until_success and sum(len(item) for item in pending) > MAX_BUFFERED_SSE_BYTES: + raise PreCommitStreamError("buffered SSE response exceeded 64 MiB") + if event in TERMINAL_EVENTS: + terminal = event + final["error_class"] = classify_error(raw) if event != "response.completed" else final["error_class"] + if event != "response.completed" and not committed and retryable_error(raw): + raise PreCommitStreamError(raw[:200], final["error_class"]) + ready_to_commit = event == "response.completed" or (event in TERMINAL_EVENTS and not retryable_error(raw)) + if not self.config.buffer_until_success: + ready_to_commit = ready_to_commit or frame_commits_output(event, raw) + if not committed and ready_to_commit: + self.send_headers(response, streaming=True) + committed = True + for item in pending: + self.wfile.write(item) + self.wfile.flush() + pending.clear() + elif committed: + self.wfile.write(wire) + self.wfile.flush() + if buffer: + if not committed: + raise PreCommitStreamError("upstream ended with an incomplete SSE frame") + if not terminal: + if not committed: + raise PreCommitStreamError("upstream closed before a terminal SSE event") + final["stream_terminal"] = "incomplete" + else: + final["stream_terminal"] = terminal + if not committed: + self.send_headers(response, streaming=True) + for item in pending: + self.wfile.write(item) + self.wfile.flush() + + def log_message(self, fmt: str, *args: Any) -> None: + return + + +def create_server(config: RelayConfig) -> ThreadingHTTPServer: + if not config.upstream: + raise ValueError("an upstream URL is required") + parsed = urlsplit(config.upstream) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("upstream must be an http or https URL") + if config.max_retries < 0 or config.request_timeout <= 0 or config.backoff_seconds < 0: + raise ValueError("retry and timeout settings must be non-negative; timeout must be positive") + writer = EventWriter(config.event_path) + handler = type("ConfiguredRelayHandler", (RelayHandler,), {"config": config, "events": writer}) + server = ThreadingHTTPServer((config.host, config.port), handler) + return server + + +def serve(config: RelayConfig) -> None: + server = create_server(config) + print(f"CURI relay listening at http://{config.host}:{server.server_port}/v1 -> {safe_upstream_url(config.upstream)}") + print(f"relay events: {os.path.abspath(os.path.expanduser(config.event_path))}") + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + server.server_close() + + +if __name__ == "__main__": + serve(RelayConfig(upstream=os.environ.get("UPSTREAM_BASE_URL", ""))) diff --git a/test_curi.py b/test_curi.py index a7b9a84..59980ef 100644 --- a/test_curi.py +++ b/test_curi.py @@ -3,10 +3,13 @@ import unittest from pathlib import Path -from curi import Store, doctor +from curi import Store, doctor, parser class CuriScanTests(unittest.TestCase): + def test_default_dashboard_port(self): + self.assertEqual(parser().parse_args(["serve"]).port, 8792) + def test_incremental_scan_counts_usage_tools_quota_and_relay(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) diff --git a/test_relay.py b/test_relay.py new file mode 100644 index 0000000..d780f4e --- /dev/null +++ b/test_relay.py @@ -0,0 +1,121 @@ +import json +import tempfile +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.request import Request, urlopen + +from relay import RelayConfig, create_server + + +class FakeUpstream(BaseHTTPRequestHandler): + attempts = 0 + mode = "retry" + + def do_POST(self): + type(self).attempts += 1 + if self.mode == "retry" and self.attempts == 1: + self.send_response(503) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error":"server overloaded"}') + return + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"model":"served-model","ok":true}') + + def do_GET(self): + type(self).attempts += 1 + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.end_headers() + self.wfile.write(b"event: response.created\ndata: {\"type\":\"response.created\"}\n\n") + self.wfile.flush() + if self.mode == "sse_retry" and self.attempts == 1: + return + self.wfile.write(b"event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"model\":\"served-model\"}\n\n") + if self.mode == "buffer_retry" and self.attempts == 1: + self.wfile.flush() + return + self.wfile.write(b"event: response.completed\ndata: {\"type\":\"response.completed\"}\n\n") + self.wfile.flush() + + def log_message(self, *args): + return + + +def start_upstream(): + FakeUpstream.attempts = 0 + server = ThreadingHTTPServer(("127.0.0.1", 0), FakeUpstream) + threading.Thread(target=server.serve_forever, daemon=True).start() + return server + + +class RelayTests(unittest.TestCase): + def test_retries_temporary_http_failure_and_writes_event(self): + FakeUpstream.mode = "retry" + upstream = start_upstream() + with tempfile.TemporaryDirectory() as tmp: + event_path = Path(tmp) / "relay.jsonl" + relay = create_server(RelayConfig(f"http://127.0.0.1:{upstream.server_port}", port=0, + max_retries=2, backoff_seconds=0, request_timeout=2, + event_path=str(event_path))) + threading.Thread(target=relay.serve_forever, daemon=True).start() + try: + request = Request(f"http://127.0.0.1:{relay.server_port}/v1/responses", + data=b'{"model":"requested-model"}', + headers={"Content-Type": "application/json"}, method="POST") + with urlopen(request, timeout=3) as response: + self.assertEqual(response.status, 200) + self.assertIn(b"served-model", response.read()) + event = json.loads(event_path.read_text(encoding="utf-8").splitlines()[-1]) + self.assertEqual(FakeUpstream.attempts, 2) + self.assertEqual(event["attempts"], 2) + self.assertEqual(event["requested_model"], "requested-model") + self.assertEqual(event["reported_model"], "served-model") + finally: + relay.shutdown(); relay.server_close() + upstream.shutdown(); upstream.server_close() + + def test_retries_sse_disconnect_before_real_output(self): + FakeUpstream.mode = "sse_retry" + upstream = start_upstream() + with tempfile.TemporaryDirectory() as tmp: + event_path = Path(tmp) / "relay.jsonl" + relay = create_server(RelayConfig(f"http://127.0.0.1:{upstream.server_port}", port=0, + max_retries=2, backoff_seconds=0, request_timeout=2, + event_path=str(event_path))) + threading.Thread(target=relay.serve_forever, daemon=True).start() + try: + with urlopen(f"http://127.0.0.1:{relay.server_port}/v1/events", timeout=3) as response: + body = response.read() + self.assertIn(b"response.output_text.delta", body) + event = json.loads(event_path.read_text(encoding="utf-8").splitlines()[-1]) + self.assertEqual(FakeUpstream.attempts, 2) + self.assertEqual(event["attempts"], 2) + self.assertEqual(event["stream_terminal"], "response.completed") + finally: + relay.shutdown(); relay.server_close() + upstream.shutdown(); upstream.server_close() + + def test_buffer_until_success_retries_after_output_disconnect(self): + FakeUpstream.mode = "buffer_retry" + upstream = start_upstream() + with tempfile.TemporaryDirectory() as tmp: + relay = create_server(RelayConfig(f"http://127.0.0.1:{upstream.server_port}", port=0, + max_retries=2, backoff_seconds=0, request_timeout=2, + event_path=str(Path(tmp) / "relay.jsonl"), buffer_until_success=True)) + threading.Thread(target=relay.serve_forever, daemon=True).start() + try: + with urlopen(f"http://127.0.0.1:{relay.server_port}/v1/events", timeout=3) as response: + self.assertIn(b"response.completed", response.read()) + self.assertEqual(FakeUpstream.attempts, 2) + finally: + relay.shutdown(); relay.server_close() + upstream.shutdown(); upstream.server_close() + + +if __name__ == "__main__": + unittest.main()