v3.0.0: ThreadingHTTPServer, dynamic ports, health gating, atomic config, safe cleanup, buffered SSE, batched stats, graceful shutdown

CHANGELOG.md

@@ -1,5 +1,34 @@
 # Changelog
 
+## v3.0.0 (2026-05-20)
+
+**Major architectural overhaul — Phase 0 + Phase 1 of engineering roadmap**
+
+### Proxy (translate-proxy.py)
+- **ThreadingHTTPServer** — serves concurrent requests (no more blocking)
+- **Thread-safe shared state** — OrderedDict response store with locks, Crof state lock, stats lock
+- **Batched + atomic stats writes** — stats buffered in memory, flushed every 5s via `os.replace()`
+- **Graceful shutdown** — SIGTERM/SIGINT drain active connections (up to 5s), reject new with 503
+- **Progressive upstream timeouts** — based on input size and tools (60-300s instead of flat 180s)
+- **Lazy JSON parsing** — skip parsing SSE events unless they contain `response.completed`
+- **Buffered SSE writes** — flush every 30ms, on urgent events, or at 4KB (reduces syscalls)
+- **`/health` endpoint** — returns backend, target, models, BGP route count
+- **Consolidated imports** — all at top, no more missing import crashes
+- **`main()` entry point** — runtime init moved out of module level
+- **TCP_NODELAY** — on all streaming paths (from v2.7.0)
+- **Anthropic prompt caching** — `cache_control: ephemeral` on system prompts (from v2.7.0)
+
+### Launcher (codex-launcher-gui)
+- **Dynamic port allocation** — `_pick_free_port()` picks random free port, no more 8080 conflicts
+- **Proxy health gating** — Codex will NOT launch if proxy fails health check within 15s
+- **Error dialogs** — clear GTK error dialog when proxy startup fails
+- **Atomic config backup/restore** — temp file + `os.replace()`, no more corrupted config.toml
+- **Config transactions** — recovery from interrupted sessions on next startup
+- **Safe cleanup (PID registry)** — only kills processes launched by the app (pids.json)
+- **Proxy stderr piped to log** — real-time proxy logs in launcher UI
+- **Bearer token** — Codex config uses `codex-launcher-local` instead of real API key
+- **Usage Dashboard v2** — OpenUsage-inspired dark theme with status pills, KPI strip, model bars (from v2.7.0)
+
 ## v2.7.0 (2026-05-20)
 
 - **Usage Dashboard redesigned** (inspired by OpenUsage design patterns)

src/codex-launcher-gui

@@ -5,7 +5,7 @@ import gi
 gi.require_version("Gtk", "3.0")
 from gi.repository import Gtk, GLib
 import subprocess, os, signal, sys, threading, time, json, urllib.request, tempfile, shutil
-import hashlib
+import hashlib, socket, contextlib
 from pathlib import Path
 
 HOME = Path.home()
@@ -435,11 +435,51 @@ def import_profile_bundle(path):
 
 def backup_config():
     if CONFIG.exists():
-        shutil.copy2(str(CONFIG), str(CONFIG_BAK))
+        tmp = CONFIG_BAK.with_suffix(".tmp")
+        shutil.copy2(str(CONFIG), str(tmp))
+        os.replace(str(tmp), str(CONFIG_BAK))
 
 def restore_config():
     if CONFIG_BAK.exists():
-        CONFIG_BAK.rename(CONFIG)
+        tmp = CONFIG.with_suffix(".tmp")
+        shutil.copy2(str(CONFIG_BAK), str(tmp))
+        os.replace(str(tmp), str(CONFIG))
+
+def write_secure_text(path, text):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp.write_text(text, encoding="utf-8")
+    os.chmod(str(tmp), 0o600)
+    os.replace(str(tmp), str(path))
+
+CONFIG_TXN = HOME / ".codex/config.toml.launcher-txn.json"
+
+def begin_config_transaction(reason):
+    txn = {"started_at": time.time(), "reason": reason,
+           "config_existed": CONFIG.exists(), "backup_path": str(CONFIG_BAK)}
+    if CONFIG.exists():
+        backup_config()
+    CONFIG_TXN.parent.mkdir(parents=True, exist_ok=True)
+    CONFIG_TXN.write_text(json.dumps(txn, indent=2))
+
+def end_config_transaction():
+    CONFIG_TXN.unlink(missing_ok=True)
+
+def recover_config_if_needed(logfn=None):
+    if not CONFIG_TXN.exists():
+        return
+    try:
+        txn = json.loads(CONFIG_TXN.read_text())
+        if txn.get("config_existed") and CONFIG_BAK.exists():
+            restore_config()
+            if logfn:
+                logfn("Recovered Codex config from interrupted session.")
+        elif CONFIG.exists():
+            CONFIG.unlink()
+            if logfn:
+                logfn("Removed generated config from interrupted session.")
+    finally:
+        CONFIG_TXN.unlink(missing_ok=True)
 
 def write_config_for_native(endpoint, selected_model):
     """Write config for native OpenAI (no proxy needed)."""
@@ -470,8 +510,7 @@ def _toml_safe(val):
     val = str(val).replace('"', '\\"')
     return val.split('\n', 1)[0].strip()
 
-def write_config_for_translated(endpoint, selected_model):
-    """Write config pointing at local proxy."""
+def write_config_for_translated(endpoint, selected_model, proxy_port=8080):
     backup_config()
     model_catalog = _gen_model_catalog(endpoint, selected_model)
     mc_path = PROXY_CONFIG_DIR / f"models-{safe_name(endpoint['name'])}.json"
@@ -484,8 +523,8 @@ def write_config_for_translated(endpoint, selected_model):
         f'model_catalog_json = "{mc_path}"\n',
         f'\n[model_providers."{endpoint["name"]}"]\n',
         f'name = "{_toml_safe(endpoint["name"])}"\n',
-        f'base_url = "http://127.0.0.1:8080"\n',
-        f'experimental_bearer_token = "{_toml_safe(endpoint["api_key"])}"\n',
+        f'base_url = "http://127.0.0.1:{proxy_port}"\n',
+        f'experimental_bearer_token = "codex-launcher-local"\n',
         f'\n[profiles."{endpoint["name"]}"]\n',
         f'model_provider = "{_toml_safe(endpoint["name"])}"\n',
         f'model = "{_toml_safe(selected_model)}"\n',
@@ -493,7 +532,7 @@ def write_config_for_translated(endpoint, selected_model):
         f'service_tier = "fast"\n',
         f'approvals_reviewer = "user"\n',
     ]
-    CONFIG.write_text("".join(lines))
+    write_secure_text(CONFIG, "".join(lines))
 
 def _gen_model_catalog(endpoint, selected_model=None):
     default_model = selected_model or endpoint.get("default_model")
@@ -533,13 +572,66 @@ def _gen_model_catalog(endpoint, selected_model=None):
 # ═══════════════════════════════════════════════════════════════════
 
 _proxy_proc = None
+_proxy_port = None
+
+PID_REGISTRY = HOME / ".cache" / "codex-launcher" / "pids.json"
+
+def _pick_free_port():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+def _load_pid_registry():
+    if PID_REGISTRY.exists():
+        try:
+            return json.loads(PID_REGISTRY.read_text())
+        except Exception:
+            pass
+    return {}
+
+def _save_pid_registry(data):
+    PID_REGISTRY.parent.mkdir(parents=True, exist_ok=True)
+    tmp = PID_REGISTRY.with_suffix(".tmp")
+    tmp.write_text(json.dumps(data, indent=2))
+    os.replace(str(tmp), str(PID_REGISTRY))
+
+def _register_pgid(kind, pid):
+    data = _load_pid_registry()
+    try:
+        pgid = os.getpgid(pid)
+    except ProcessLookupError:
+        return
+    data[kind] = {"pid": pid, "pgid": pgid, "ts": time.time()}
+    _save_pid_registry(data)
+
+def safe_cleanup_owned(logfn=None):
+    data = _load_pid_registry()
+    changed = False
+    for kind, meta in list(data.items()):
+        pgid = meta.get("pgid")
+        if not pgid:
+            continue
+        try:
+            os.killpg(pgid, signal.SIGTERM)
+            if logfn:
+                logfn(f"Stopped {kind} (pgid {pgid})")
+            changed = True
+        except ProcessLookupError:
+            changed = True
+        except Exception as e:
+            if logfn:
+                logfn(f"Could not stop {kind}: {e}")
+    if changed:
+        _save_pid_registry({})
 
 def _start_proxy_for(endpoint, logfn):
-    global _proxy_proc
+    global _proxy_proc, _proxy_port
     _stop_proxy()
+    port = _pick_free_port()
+    _proxy_port = port
 
     pcfg = {
-        "port": 8080,
+        "port": port,
         "backend_type": endpoint["backend_type"],
         "target_url": normalize_base_url(endpoint["base_url"]),
         "api_key": endpoint["api_key"],
@@ -550,26 +642,49 @@ def _start_proxy_for(endpoint, logfn):
         "models": [{"id": m, "object": "model", "created": 1700000000, "owned_by": endpoint["name"]}
                    for m in endpoint.get("models", [])],
     }
-    pcfg_path = PROXY_CONFIG_DIR / f"proxy-{safe_name(endpoint['name'])}.json"
+    pcfg_path = PROXY_CONFIG_DIR / f"proxy-{safe_name(endpoint['name'])}-{port}.json"
     pcfg_path.parent.mkdir(parents=True, exist_ok=True)
     pcfg_path.write_text(json.dumps(pcfg, indent=2))
-    _start_proxy_with_config(pcfg_path, logfn)
+    _start_proxy_with_config(pcfg_path, port, logfn)
+    return port
 
-def _start_proxy_with_config(pcfg_path, logfn):
+def _start_proxy_with_config(pcfg_path, port, logfn):
     global _proxy_proc
     _proxy_proc = subprocess.Popen(
         ["python3", str(PROXY), "--config", str(pcfg_path)],
         stdout=subprocess.DEVNULL,
+        stderr=subprocess.PIPE,
         preexec_fn=os.setsid,
+        text=True,
     )
-    for _ in range(30):
-        try:
-            urllib.request.urlopen("http://127.0.0.1:8080/v1/models", timeout=2)
-            logfn("Proxy ready on port 8080")
+    _register_pgid("proxy", _proxy_proc.pid)
+
+    def _pipe_stderr():
+        if not _proxy_proc.stderr:
             return
+        for line in _proxy_proc.stderr:
+            GLib.idle_add(logfn, f"[proxy] {line.rstrip()}")
+    threading.Thread(target=_pipe_stderr, daemon=True).start()
+
+    deadline = time.time() + 15
+    last_err = None
+    while time.time() < deadline:
+        if _proxy_proc.poll() is not None:
+            raise RuntimeError(f"Proxy exited early with code {_proxy_proc.returncode}")
+        try:
+            urllib.request.urlopen(f"http://127.0.0.1:{port}/v1/models", timeout=2)
+            logfn(f"Proxy ready on port {port}")
+            return
+        except Exception as e:
+            last_err = e
+            time.sleep(0.3)
+    try:
+        os.killpg(os.getpgid(_proxy_proc.pid), signal.SIGTERM)
+        _proxy_proc.wait(timeout=3)
     except Exception:
-            time.sleep(0.5)
-    logfn("WARNING: proxy may not have started in time")
+        with contextlib.suppress(Exception):
+            os.killpg(os.getpgid(_proxy_proc.pid), signal.SIGKILL)
+    raise RuntimeError(f"Proxy failed health check on port {port}: {last_err}")
 
 def _stop_proxy():
     global _proxy_proc
@@ -583,8 +698,8 @@ def _stop_proxy():
             pass
         _proxy_proc = None
 
-def _run_cleanup():
-    subprocess.run(["bash", str(CLEANUP)], capture_output=True, timeout=30)
+def _run_cleanup(logfn=None):
+    safe_cleanup_owned(logfn)
 
 def _last_log_lines(n=15):
     try:
@@ -640,6 +755,7 @@ class LauncherWin(Gtk.Window):
         self.set_position(Gtk.WindowPosition.CENTER)
         self._proc = None
         self._endpoints_data = load_endpoints()
+        recover_config_if_needed()
 
         vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
         self.add(vbox)
@@ -647,7 +763,7 @@ class LauncherWin(Gtk.Window):
         # header row
         hdr = Gtk.Box(spacing=8)
         vbox.pack_start(hdr, False, False, 0)
-        lbl = Gtk.Label(label="<b>Codex Launcher v2.7.0</b>")
+        lbl = Gtk.Label(label="<b>Codex Launcher v3.0.0</b>")
         lbl.set_use_markup(True)
         hdr.pack_start(lbl, False, False, 0)
         changelog_btn = Gtk.Button(label="Changelog")
@@ -1207,17 +1323,24 @@ class LauncherWin(Gtk.Window):
     def _run(self, ep, model, target):
         try:
             self.log("Cleaning up stale processes…")
-            _run_cleanup()
+            _run_cleanup(self.log)
+            recover_config_if_needed(self.log)
 
             needs_proxy = ep["backend_type"] != "native"
 
             if needs_proxy:
                 self.log("Starting translation proxy…")
-                _start_proxy_for(ep, self.log)
-                self.log(f"Configuring Codex for {ep['name']} (proxied)…")
-                write_config_for_translated(ep, model)
+                try:
+                    proxy_port = _start_proxy_for(ep, self.log)
+                except RuntimeError as e:
+                    GLib.idle_add(self._show_error_dialog, "Proxy startup failed", str(e))
+                    return
+                self.log(f"Configuring Codex for {ep['name']} (proxied on :{proxy_port})…")
+                begin_config_transaction(f"launch:{ep['name']}")
+                write_config_for_translated(ep, model, proxy_port)
             else:
                 self.log(f"Configuring Codex for {ep['name']} (native)…")
+                begin_config_transaction(f"launch:{ep['name']}")
                 write_config_for_native(ep, model)
 
             if target == "desktop":
@@ -1230,15 +1353,18 @@ class LauncherWin(Gtk.Window):
         finally:
             _stop_proxy()
             restore_config()
+            end_config_transaction()
             self._set_busy(False)
             self.log("Ready.")
 
     def _run_bgp(self, pool, model, target):
         try:
             self.log("Cleaning up stale processes…")
-            _run_cleanup()
+            _run_cleanup(self.log)
+            recover_config_if_needed(self.log)
 
-            self.log(f"Starting BGP proxy with {len(pool.get('routes', []))} routes…")
+            port = _pick_free_port()
+            self.log(f"Starting BGP proxy with {len(pool.get('routes', []))} routes on :{port}…")
             bgp_ep = {
                 "name": pool["name"],
                 "backend_type": "openai-compat",
@@ -1248,19 +1374,24 @@ class LauncherWin(Gtk.Window):
                 "models": list(dict.fromkeys(r.get("model", model) for r in pool.get("routes", []))),
             }
             pcfg = {
-                "port": 8080,
+                "port": port,
                 "backend_type": "openai-compat",
                 "target_url": "http://bgp.placeholder",
                 "api_key": "",
                 "bgp_routes": pool.get("routes", []),
                 "models": [{"id": m, "object": "model", "created": 1700000000, "owned_by": "bgp"} for m in bgp_ep["models"]],
             }
-            pcfg_path = PROXY_CONFIG_DIR / f"proxy-{safe_name(pool['name'])}.json"
+            pcfg_path = PROXY_CONFIG_DIR / f"proxy-{safe_name(pool['name'])}-{port}.json"
             pcfg_path.parent.mkdir(parents=True, exist_ok=True)
             pcfg_path.write_text(json.dumps(pcfg, indent=2))
-            _start_proxy_with_config(pcfg_path, self.log)
+            try:
+                _start_proxy_with_config(pcfg_path, port, self.log)
+            except RuntimeError as e:
+                GLib.idle_add(self._show_error_dialog, "BGP proxy startup failed", str(e))
+                return
 
-            write_config_for_translated(bgp_ep, model)
+            begin_config_transaction(f"launch:bgp:{pool['name']}")
+            write_config_for_translated(bgp_ep, model, port)
 
             if target == "desktop":
                 self._launch_desktop(bgp_ep, model)
@@ -1272,17 +1403,19 @@ class LauncherWin(Gtk.Window):
         finally:
             _stop_proxy()
             restore_config()
+            end_config_transaction()
             self._set_busy(False)
             self.log("Ready.")
 
     def _run_codex_default(self, target):
         try:
             self.log("Cleaning up stale processes…")
-            _run_cleanup()
+            _run_cleanup(self.log)
             _stop_proxy()
+            recover_config_if_needed(self.log)
 
             self.log("Resetting config to Codex defaults (OAuth)…")
-            backup_config()
+            begin_config_transaction("launch:default")
             if CONFIG.exists():
                 CONFIG.unlink()
 
@@ -1294,9 +1427,19 @@ class LauncherWin(Gtk.Window):
             self.log(f"ERROR: {e}")
         finally:
             restore_config()
+            end_config_transaction()
             self._set_busy(False)
             self.log("Ready.")
 
+    def _show_error_dialog(self, title, message):
+        dialog = Gtk.MessageDialog(
+            transient_for=self, flags=0,
+            message_type=Gtk.MessageType.ERROR,
+            buttons=Gtk.ButtonsType.CLOSE, text=str(title))
+        dialog.format_secondary_text(str(message))
+        dialog.run()
+        dialog.destroy()
+
     def _launch_desktop(self, ep, model):
         args = [str(START_SH)]
         if ep["backend_type"] != "native":
@@ -1454,8 +1597,9 @@ class LauncherWin(Gtk.Window):
                 pass
             self._proc = None
         _stop_proxy()
-        _run_cleanup()
+        _run_cleanup(self.log)
         restore_config()
+        end_config_transaction()
         LOG_DIR.mkdir(parents=True, exist_ok=True)
         LAUNCH_LOG.unlink(missing_ok=True)
         self.log("Cleanup complete")

src/translate-proxy.py

@@ -11,7 +11,8 @@ Usage:
   python3 translate-proxy.py --backend openai-compat --target-url https://... --api-key sk-...
 """
 
-import json, http.server, urllib.request, time, uuid, os, sys, argparse, threading, socket
+import json, http.server, socketserver, urllib.request, urllib.parse, urllib.error
+import time, uuid, os, sys, argparse, threading, socket, collections, contextlib, signal
 
 # ═══════════════════════════════════════════════════════════════════
 # Config
@@ -74,24 +75,63 @@ def load_config():
 
     return cfg
 
-CONFIG = load_config()
-PORT = CONFIG["port"]
-BACKEND = CONFIG["backend_type"]
-TARGET_URL = CONFIG["target_url"].rstrip("/")
-API_KEY = CONFIG["api_key"]
-OAUTH_PROVIDER = CONFIG.get("oauth_provider", "")
-MODELS = CONFIG["models"]
-CC_VERSION = CONFIG.get("cc_version", "")
-REASONING_ENABLED = CONFIG.get("reasoning_enabled", True)
-REASONING_EFFORT = CONFIG.get("reasoning_effort", "medium")
-BGP_ROUTES = CONFIG.get("bgp_routes", [])
-BGP_MODELS = []
-for _r in BGP_ROUTES:
+CONFIG = None
+PORT = 8080
+BACKEND = "openai-compat"
+TARGET_URL = ""
+API_KEY = ""
+OAUTH_PROVIDER = ""
+MODELS = []
+CC_VERSION = ""
+REASONING_ENABLED = True
+REASONING_EFFORT = "medium"
+BGP_ROUTES = []
+SERVER = None
+
+_LOG_DIR = os.path.join(os.path.expanduser("~"), ".cache", "codex-proxy")
+os.makedirs(_LOG_DIR, exist_ok=True)
+_stats_path = os.path.join(_LOG_DIR, "usage-stats.json")
+_stats_lock = threading.Lock()
+_stats_pending = []
+_stats_flush_timer = None
+_STATS_FLUSH_INTERVAL = 5.0
+
+_response_store = collections.OrderedDict()
+_response_store_lock = threading.Lock()
+_MAX_STORED = 50
+
+_crof_lock = threading.Lock()
+
+_shutdown_requested = False
+_active_connections = 0
+_active_connections_lock = threading.Lock()
+
+_pool = uuid.uuid4().hex[:8]
+
+def _init_runtime():
+    global CONFIG, PORT, BACKEND, TARGET_URL, API_KEY, OAUTH_PROVIDER
+    global MODELS, CC_VERSION, REASONING_ENABLED, REASONING_EFFORT, BGP_ROUTES
+
+    CONFIG = load_config()
+    PORT = CONFIG["port"]
+    BACKEND = CONFIG["backend_type"]
+    TARGET_URL = CONFIG["target_url"].rstrip("/")
+    API_KEY = CONFIG["api_key"]
+    OAUTH_PROVIDER = CONFIG.get("oauth_provider", "")
+    MODELS = CONFIG["models"]
+    CC_VERSION = CONFIG.get("cc_version", "")
+    REASONING_ENABLED = CONFIG.get("reasoning_enabled", True)
+    REASONING_EFFORT = CONFIG.get("reasoning_effort", "medium")
+    BGP_ROUTES = CONFIG.get("bgp_routes", [])
+
+    bgp_models = []
+    for _r in BGP_ROUTES:
         for _m in _r.get("models", [{"id": _r.get("model", "unknown")}]):
-        if _m.get("id", _m) not in BGP_MODELS:
-            BGP_MODELS.append(_m.get("id", _m) if isinstance(_m, dict) else _m)
-if BGP_ROUTES and not MODELS:
-    MODELS = [{"id": m, "object": "model", "created": 1700000000, "owned_by": "bgp"} for m in BGP_MODELS]
+            mid = _m.get("id", _m) if isinstance(_m, dict) else _m
+            if mid not in bgp_models:
+                bgp_models.append(mid)
+    if BGP_ROUTES and not MODELS:
+        MODELS = [{"id": m, "object": "model", "created": 1700000000, "owned_by": "bgp"} for m in bgp_models]
         CONFIG["models"] = MODELS
 
 def _refresh_oauth_token():
@@ -138,14 +178,6 @@ def _refresh_oauth_token_for(api_key, oauth_provider):
 
 _pool = uuid.uuid4().hex[:8]
 
-_response_store = {}
-_MAX_STORED = 50
-
-_LOG_DIR = os.path.join(os.path.expanduser("~"), ".cache", "codex-proxy")
-os.makedirs(_LOG_DIR, exist_ok=True)
-_stats_path = os.path.join(_LOG_DIR, "usage-stats.json")
-_stats_lock = threading.Lock()
-
 def _load_stats():
     try:
         if os.path.exists(_stats_path):
@@ -154,46 +186,78 @@ def _load_stats():
         pass
     return {"providers": {}, "updated": None}
 
-def _record_usage(provider, model, success, duration_s, tokens_in=0, tokens_out=0, error_type=None):
+def _atomic_write_json(path, obj):
+    tmp = path + ".tmp"
+    with open(tmp, "w") as f:
+        json.dump(obj, f, indent=2, ensure_ascii=False)
+    os.replace(tmp, path)
+
+def _flush_stats():
+    global _stats_flush_timer
     with _stats_lock:
+        batch = list(_stats_pending)
+        _stats_pending.clear()
+        _stats_flush_timer = None
+    if not batch:
+        return
     stats = _load_stats()
+    for entry in batch:
+        provider = entry["provider"]
+        model = entry["model"]
         p = stats["providers"].setdefault(provider, {
             "total_requests": 0, "successes": 0, "failures": 0,
             "total_tokens_in": 0, "total_tokens_out": 0,
             "total_duration_s": 0.0, "models": {}, "last_used": None, "last_error": None,
         })
         p["total_requests"] += 1
-        p["total_tokens_in"] += tokens_in
-        p["total_tokens_out"] += tokens_out
-        p["total_duration_s"] += duration_s
-        p["last_used"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
-        if success:
+        p["total_tokens_in"] += entry["tokens_in"]
+        p["total_tokens_out"] += entry["tokens_out"]
+        p["total_duration_s"] += entry["duration_s"]
+        p["last_used"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(entry["ts"]))
+        if entry["success"]:
             p["successes"] += 1
         else:
             p["failures"] += 1
-            p["last_error"] = error_type or "unknown"
+            p["last_error"] = entry.get("error_type") or "unknown"
         m = p["models"].setdefault(model, {"requests": 0, "tokens_in": 0, "tokens_out": 0})
         m["requests"] += 1
-        m["tokens_in"] += tokens_in
-        m["tokens_out"] += tokens_out
+        m["tokens_in"] += entry["tokens_in"]
+        m["tokens_out"] += entry["tokens_out"]
     stats["updated"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
-        with open(_stats_path, "w") as f:
-            json.dump(stats, f, indent=2)
+    _atomic_write_json(_stats_path, stats)
+
+def _record_usage(provider, model, success, duration_s, tokens_in=0, tokens_out=0, error_type=None):
+    global _stats_flush_timer
+    entry = {
+        "provider": provider or "unknown", "model": model or "unknown",
+        "success": bool(success), "duration_s": float(duration_s or 0),
+        "tokens_in": int(tokens_in or 0), "tokens_out": int(tokens_out or 0),
+        "error_type": error_type, "ts": time.time(),
+    }
+    with _stats_lock:
+        _stats_pending.append(entry)
+        if _stats_flush_timer is None:
+            _stats_flush_timer = threading.Timer(_STATS_FLUSH_INTERVAL, _flush_stats)
+            _stats_flush_timer.daemon = True
+            _stats_flush_timer.start()
 
 def store_response(resp_id, input_data, output_items):
     if not resp_id:
         return
-    _response_store[resp_id] = {"input": input_data, "output": output_items}
-    if len(_response_store) > _MAX_STORED:
-        oldest = list(_response_store.keys())[0]
-        del _response_store[oldest]
+    with _response_store_lock:
+        _response_store[resp_id] = {"input": input_data, "output": output_items, "ts": time.time()}
+        while len(_response_store) > _MAX_STORED:
+            _response_store.popitem(last=False)
 
 def resolve_previous_response(body):
     prev_id = body.get("previous_response_id")
     input_data = body.get("input", "")
-    if not prev_id or prev_id not in _response_store:
+    if not prev_id:
+        return input_data
+    with _response_store_lock:
+        stored = _response_store.get(prev_id)
+    if not stored:
         return input_data
-    stored = _response_store[prev_id]
     prev_input = stored["input"]
     prev_output = stored["output"]
     new_input = input_data if isinstance(input_data, list) else []
@@ -983,17 +1047,59 @@ def _log_resp(resp_id, status, output):
     except Exception:
         pass
 
+class ConnectionTracker:
+    def __enter__(self):
+        global _active_connections
+        with _active_connections_lock:
+            _active_connections += 1
+    def __exit__(self, *a):
+        global _active_connections
+        with _active_connections_lock:
+            _active_connections -= 1
+
+def _handle_shutdown_signal(signum, frame):
+    global _shutdown_requested
+    _shutdown_requested = True
+    print("[proxy] shutdown requested; draining connections", file=sys.stderr)
+    def _drain():
+        deadline = time.time() + 5
+        while time.time() < deadline:
+            with _active_connections_lock:
+                if _active_connections == 0:
+                    break
+            time.sleep(0.1)
+        if SERVER is not None:
+            SERVER.shutdown()
+    threading.Thread(target=_drain, daemon=True).start()
+
+def _upstream_timeout(body, stream):
+    input_data = body.get("input", "")
+    n_items = len(input_data) if isinstance(input_data, list) else 1
+    has_tools = bool(body.get("tools"))
+    if stream:
+        return min((180 if has_tools else 120) + n_items * 2, 300)
+    return min(60 + n_items * 2, 120)
+
 class Handler(http.server.BaseHTTPRequestHandler):
     protocol_version = "HTTP/1.1"
 
     def do_GET(self):
         if self.path in ("/v1/models", "/models"):
             self.send_json(200, {"object": "list", "data": MODELS})
+        elif self.path in ("/health", "/v1/health"):
+            self.send_json(200, {"ok": True, "backend": BACKEND,
+                                 "target_url": TARGET_URL,
+                                 "models": [m.get("id") for m in MODELS],
+                                 "bgp_routes": len(BGP_ROUTES)})
         else:
             self.send_error(404)
 
     def do_POST(self):
+        if _shutdown_requested:
+            return self.send_json(503, {"error": {"type": "proxy_shutting_down",
+                                                   "message": "Proxy is shutting down"}})
         if self.path in ("/v1/responses", "/responses"):
+            with ConnectionTracker():
                 self._handle()
         else:
             self.send_error(404)
@@ -1082,7 +1188,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
             for attempt in range(max_retries + 1):
                 req = urllib.request.Request(target, data=chat_body_b, headers=fwd)
                 try:
-                    upstream = urllib.request.urlopen(req, timeout=180)
+                    upstream = urllib.request.urlopen(req, timeout=_upstream_timeout(body, stream))
                 except urllib.error.HTTPError as e:
                     err_body = e.read().decode()
                     if e.code in (429, 502, 503) and attempt < max_retries:
@@ -1163,7 +1269,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
             route_ok = False
             for attempt in range(3):
                 try:
-                    upstream = urllib.request.urlopen(req, timeout=180)
+                    upstream = urllib.request.urlopen(req, timeout=_upstream_timeout(body, stream))
                     print(f"[bgp] route '{route.get('name', r_url)}' connected OK", file=sys.stderr)
                     self._forward_oa_compat(upstream, stream, r_model, chat_body, body, input_data, fwd, target)
                     return
@@ -1284,7 +1390,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
 
     def _forward_oa_compat_retry(self, req, model, chat_body, body, input_data):
         try:
-            upstream = urllib.request.urlopen(req, timeout=180)
+            upstream = urllib.request.urlopen(req, timeout=_upstream_timeout(body, True))
         except Exception as e:
             print(f"[crof-adaptive] retry failed: {e}", file=sys.stderr)
             return
@@ -1427,7 +1533,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
 
         if stream:
             try:
-                upstream = urllib.request.urlopen(req, timeout=180)
+                upstream = urllib.request.urlopen(req, timeout=_upstream_timeout(body, True))
             except urllib.error.HTTPError as e:
                 err = e.read().decode()
                 return self.send_json(e.code, {"error": {"type": "upstream_error", "message": err}})
@@ -1461,7 +1567,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
                 store_response(last_resp_id, body.get("input", ""), last_output)
         else:
             try:
-                upstream = urllib.request.urlopen(req, timeout=180)
+                upstream = urllib.request.urlopen(req, timeout=_upstream_timeout(body, False))
             except urllib.error.HTTPError as e:
                 err = e.read().decode()
                 return self.send_json(e.code, {"error": {"type": "upstream_error", "message": err}})
@@ -1478,7 +1584,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
 
     def _forward(self, req, stream, model, nonstream_fn, stream_fn, input_data=None):
         try:
-            upstream = urllib.request.urlopen(req, timeout=180)
+            upstream = urllib.request.urlopen(req, timeout=_upstream_timeout({}, stream))
         except urllib.error.HTTPError as e:
             err = e.read().decode()
             return self.send_json(e.code, {"error": {"type": "upstream_error", "message": err}})
@@ -1533,17 +1639,54 @@ class Handler(http.server.BaseHTTPRequestHandler):
         self.end_headers()
         self.wfile.write(body)
 
+    def stream_buffered_events(self, event_iter, flush_interval=0.03, max_bytes=4096):
+        buf = bytearray()
+        last_flush = time.monotonic()
+        def _flush():
+            nonlocal buf, last_flush
+            if buf:
+                self.wfile.write(buf)
+                self.wfile.flush()
+                buf.clear()
+                last_flush = time.monotonic()
+        for event in event_iter:
+            encoded = event.encode("utf-8") if isinstance(event, str) else event
+            buf.extend(encoded)
+            urgent = ("response.completed" in event or "response.output_text.done" in event
+                      or "response.output_item.done" in event
+                      or "function_call_arguments.done" in event)
+            if urgent or len(buf) >= max_bytes or time.monotonic() - last_flush >= flush_interval:
+                _flush()
+        _flush()
+
     def log_message(self, fmt, *args):
         msg = fmt % args if args else fmt
         print(f"[translate-proxy] {BACKEND} {msg}", file=sys.stderr)
 
-if __name__ == "__main__":
-    class ReusableHTTPServer(http.server.HTTPServer):
+def main():
+    global SERVER
+    _init_runtime()
+    signal.signal(signal.SIGTERM, _handle_shutdown_signal)
+    signal.signal(signal.SIGINT, _handle_shutdown_signal)
+    try:
+        from http.server import ThreadingHTTPServer as _BaseSrv
+    except ImportError:
+        class _BaseSrv(socketserver.ThreadingMixIn, http.server.HTTPServer):
+            daemon_threads = True
+    class ReusableHTTPServer(_BaseSrv):
         allow_reuse_address = True
-    server = ReusableHTTPServer(("127.0.0.1", PORT), Handler)
+        daemon_threads = True
+        request_queue_size = 64
+    SERVER = ReusableHTTPServer(("127.0.0.1", PORT), Handler)
     print(f"translate-proxy ({BACKEND}) listening on http://127.0.0.1:{PORT}", flush=True)
     print(f"Target: {TARGET_URL}", flush=True)
     print(f"Models: {[m['id'] for m in MODELS]}", flush=True)
     if BGP_ROUTES:
         print(f"BGP routes: {len(BGP_ROUTES)} ({[r.get('name','?') for r in BGP_ROUTES]})", flush=True)
-    server.serve_forever()
+    try:
+        SERVER.serve_forever()
+    finally:
+        _flush_stats()
+
+if __name__ == "__main__":
+    main()