refactor: architecture audit fixes (P0-P2 stability & reliability)

P0: curl header parse timeout (30s), kill curl on stream disconnect
via AbortController, cap error response body at 1MB.

P1: acquireLock TTL (5min auto-release), atomic config hot-reload,
encapsulate 429 counting in AccountPool, graceful shutdown with
server.close() + 5s drain period.

P2: send error SSE event on stream failure, structured JSON logger,
O(1) session LRU eviction via Map insertion order.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/auth/account-pool.ts

@@ -31,9 +31,12 @@ import type {
 const ACCOUNTS_FILE = resolve(process.cwd(), "data", "accounts.json");
 const LEGACY_AUTH_FILE = resolve(process.cwd(), "data", "auth.json");
 
+// P1-4: Lock TTL — auto-release locks older than this
+const ACQUIRE_LOCK_TTL_MS = 5 * 60 * 1000; // 5 minutes
+
 export class AccountPool {
   private accounts: Map<string, AccountEntry> = new Map();
-  private acquireLocks: Set<string> = new Set();
+  private acquireLocks: Map<string, number> = new Map(); // entryId → timestamp
   private roundRobinIndex = 0;
   private persistTimer: ReturnType<typeof setTimeout> | null = null;
 
@@ -61,12 +64,21 @@ export class AccountPool {
   acquire(): AcquiredAccount | null {
     const config = getConfig();
     const now = new Date();
+    const nowMs = now.getTime();
 
     // Update statuses before selecting
     for (const entry of this.accounts.values()) {
       this.refreshStatus(entry, now);
     }
 
+    // P1-4: Auto-release stale locks (older than TTL)
+    for (const [id, lockedAt] of this.acquireLocks) {
+      if (nowMs - lockedAt > ACQUIRE_LOCK_TTL_MS) {
+        console.warn(`[AccountPool] Auto-releasing stale lock for ${id} (locked ${Math.round((nowMs - lockedAt) / 1000)}s ago)`);
+        this.acquireLocks.delete(id);
+      }
+    }
+
     // Filter available accounts
     const available = [...this.accounts.values()].filter(
       (a) => a.status === "active" && !this.acquireLocks.has(a.id),
@@ -91,7 +103,7 @@ export class AccountPool {
       selected = available[0];
     }
 
-    this.acquireLocks.add(selected.id);
+    this.acquireLocks.set(selected.id, Date.now());
     return {
       entryId: selected.id,
       token: selected.token,
@@ -121,18 +133,31 @@ export class AccountPool {
 
   /**
    * Mark an account as rate-limited after a 429.
+   * P1-6: countRequest option to track 429s as usage without exposing entry internals.
    */
-  markRateLimited(entryId: string, retryAfterSec?: number): void {
+  markRateLimited(
+    entryId: string,
+    options?: { retryAfterSec?: number; countRequest?: boolean },
+  ): void {
     this.acquireLocks.delete(entryId);
     const entry = this.accounts.get(entryId);
     if (!entry) return;
 
     const config = getConfig();
-    const backoff = jitter(retryAfterSec ?? config.auth.rate_limit_backoff_seconds, 0.2);
+    const backoff = jitter(
+      options?.retryAfterSec ?? config.auth.rate_limit_backoff_seconds,
+      0.2,
+    );
     const until = new Date(Date.now() + backoff * 1000);
 
     entry.status = "rate_limited";
     entry.usage.rate_limit_until = until.toISOString();
+
+    if (options?.countRequest) {
+      entry.usage.request_count++;
+      entry.usage.last_used = new Date().toISOString();
+    }
+
     this.schedulePersist();
   }
 

src/config.ts

@@ -134,16 +134,25 @@ export function mutateClientConfig(patch: Partial<AppConfig["client"]>): void {
   Object.assign(_config.client, patch);
 }
 
-/** Reload config from disk (hot-reload after full-update). */
+/** Reload config from disk (hot-reload after full-update).
+ *  P1-5: Load to temp first, then swap atomically to avoid null window. */
 export function reloadConfig(configDir?: string): AppConfig {
-  _config = null;
-  return loadConfig(configDir);
+  const dir = configDir ?? resolve(process.cwd(), "config");
+  const raw = loadYaml(resolve(dir, "default.yaml")) as Record<string, unknown>;
+  applyEnvOverrides(raw);
+  const fresh = ConfigSchema.parse(raw);
+  _config = fresh;
+  return _config;
 }
 
-/** Reload fingerprint from disk (hot-reload after full-update). */
+/** Reload fingerprint from disk (hot-reload after full-update).
+ *  P1-5: Load to temp first, then swap atomically. */
 export function reloadFingerprint(configDir?: string): FingerprintConfig {
-  _fingerprint = null;
-  return loadFingerprint(configDir);
+  const dir = configDir ?? resolve(process.cwd(), "config");
+  const raw = loadYaml(resolve(dir, "fingerprint.yaml"));
+  const fresh = FingerprintSchema.parse(raw);
+  _fingerprint = fresh;
+  return _fingerprint;
 }
 
 /** Reload both config and fingerprint from disk. */

src/index.ts

@@ -94,35 +94,54 @@ async function main() {
   // Start background update checker
   startUpdateChecker();
 
-  serve({
+  const server = serve({
     fetch: app.fetch,
     hostname: host,
     port,
   });
 
-  // Graceful shutdown with timeout protection
+  // P1-7: Graceful shutdown — stop accepting, drain, then cleanup
   let shutdownCalled = false;
+  const DRAIN_TIMEOUT_MS = 5_000;
   const shutdown = () => {
     if (shutdownCalled) return;
     shutdownCalled = true;
-    console.log("\n[Shutdown] Cleaning up...");
+    console.log("\n[Shutdown] Stopping new connections...");
+
     const forceExit = setTimeout(() => {
       console.error("[Shutdown] Timeout after 10s — forcing exit");
       process.exit(1);
     }, 10_000);
     if (forceExit.unref) forceExit.unref();
 
-    try {
-      stopUpdateChecker();
-      refreshScheduler.destroy();  // Cancel timers first
-      sessionManager.destroy();
-      cookieJar.destroy();         // Flush cookies
-      accountPool.destroy();       // Flush accounts
-    } catch (err) {
-      console.error("[Shutdown] Error during cleanup:", err instanceof Error ? err.message : err);
+    // 1. Stop accepting new connections
+    server.close(() => {
+      console.log("[Shutdown] Server closed, cleaning up resources...");
+      cleanup();
+    });
+
+    // 2. Grace period for active streams, then force cleanup
+    setTimeout(() => {
+      console.log("[Shutdown] Drain timeout reached, cleaning up...");
+      cleanup();
+    }, DRAIN_TIMEOUT_MS);
+
+    let cleanupDone = false;
+    function cleanup() {
+      if (cleanupDone) return;
+      cleanupDone = true;
+      try {
+        stopUpdateChecker();
+        refreshScheduler.destroy();
+        sessionManager.destroy();
+        cookieJar.destroy();
+        accountPool.destroy();
+      } catch (err) {
+        console.error("[Shutdown] Error during cleanup:", err instanceof Error ? err.message : err);
+      }
+      clearTimeout(forceExit);
+      process.exit(0);
     }
-    clearTimeout(forceExit);
-    process.exit(0);
   };
 
   process.on("SIGINT", shutdown);

src/middleware/logger.ts

@@ -1,4 +1,5 @@
 import type { Context, Next } from "hono";
+import { log } from "../utils/logger.js";
 
 export async function logger(c: Context, next: Next): Promise<void> {
   const start = Date.now();
@@ -6,11 +7,11 @@ export async function logger(c: Context, next: Next): Promise<void> {
   const path = c.req.path;
   const rid = c.get("requestId") ?? "-";
 
-  console.log(`→ ${method} ${path} [${rid}]`);
+  log.info(`→ ${method} ${path}`, { rid, method, path });
 
   await next();
 
   const ms = Date.now() - start;
   const status = c.res.status;
-  console.log(`← ${method} ${path} ${status} ${ms}ms [${rid}]`);
+  log.info(`← ${method} ${path} ${status} ${ms}ms`, { rid, method, path, status, ms });
 }

src/proxy/codex-api.ts

@@ -150,6 +150,16 @@ export class CodexApi {
       let headersParsed = false;
       let bodyController: ReadableStreamDefaultController<Uint8Array> | null = null;
 
+      // P0-1: Header parse timeout — kill curl if headers aren't received within 30s
+      const HEADER_TIMEOUT_MS = 30_000;
+      const headerTimer = setTimeout(() => {
+        if (!headersParsed) {
+          child.kill("SIGTERM");
+          reject(new CodexApiError(0, `curl header parse timeout after ${HEADER_TIMEOUT_MS}ms`));
+        }
+      }, HEADER_TIMEOUT_MS);
+      if (headerTimer.unref) headerTimer.unref();
+
       const bodyStream = new ReadableStream<Uint8Array>({
         start(c) {
           bodyController = c;
@@ -171,6 +181,7 @@ export class CodexApi {
         if (separatorIdx === -1) return;
 
         headersParsed = true;
+        clearTimeout(headerTimer);
         const headerBlock = headerBuf.subarray(0, separatorIdx).toString("utf-8");
         const remaining = headerBuf.subarray(separatorIdx + 4);
 
@@ -200,6 +211,7 @@ export class CodexApi {
       });
 
       child.on("close", (code) => {
+        clearTimeout(headerTimer);
         if (signal) {
           signal.removeEventListener("abort", onAbort);
         }
@@ -210,6 +222,7 @@ export class CodexApi {
       });
 
       child.on("error", (err) => {
+        clearTimeout(headerTimer);
         if (signal) {
           signal.removeEventListener("abort", onAbort);
         }
@@ -297,13 +310,26 @@ export class CodexApi {
     this.captureCookiesFromCurl(curlRes.setCookieHeaders);
 
     if (curlRes.status < 200 || curlRes.status >= 300) {
-      // Read the body for error details
+      // Read the body for error details (P0-3: cap at 1MB to prevent memory spikes)
+      const MAX_ERROR_BODY = 1024 * 1024; // 1MB
       const reader = curlRes.body.getReader();
       const chunks: Uint8Array[] = [];
+      let totalSize = 0;
       while (true) {
         const { done, value } = await reader.read();
         if (done) break;
-        chunks.push(value);
+        totalSize += value.byteLength;
+        if (totalSize <= MAX_ERROR_BODY) {
+          chunks.push(value);
+        } else {
+          // Truncate: push only the part that fits
+          const overshoot = totalSize - MAX_ERROR_BODY;
+          if (value.byteLength > overshoot) {
+            chunks.push(value.subarray(0, value.byteLength - overshoot));
+          }
+          reader.cancel();
+          break;
+        }
       }
       const errorBody = Buffer.concat(chunks).toString("utf-8");
       throw new CodexApiError(curlRes.status, errorBody);

src/routes/shared/proxy-handler.ts

@@ -89,10 +89,13 @@ export async function handleProxyRequest(
 
   let usageInfo: { input_tokens: number; output_tokens: number } | undefined;
 
+  // P0-2: AbortController to kill curl when client disconnects
+  const abortController = new AbortController();
+
   try {
     // 3. Retry + send to Codex
     const rawResponse = await withRetry(
-      () => codexApi.createResponse(req.codexRequest),
+      () => codexApi.createResponse(req.codexRequest, abortController.signal),
       { tag: fmt.tag },
     );
 
@@ -126,7 +129,16 @@ export async function handleProxyRequest(
           )) {
             await s.write(chunk);
           }
+        } catch (err) {
+          // P2-8: Send error SSE event to client before closing
+          try {
+            const errMsg = err instanceof Error ? err.message : "Stream interrupted";
+            await s.write(`data: ${JSON.stringify({ error: { message: errMsg, type: "stream_error" } })}\n\n`);
+          } catch { /* client already gone */ }
+          throw err;
         } finally {
+          // P0-2: Kill curl subprocess if still running
+          abortController.abort();
           accountPool.release(entryId, usageInfo);
         }
       });
@@ -156,15 +168,8 @@ export async function handleProxyRequest(
         err.message,
       );
       if (err.status === 429) {
-        accountPool.markRateLimited(entryId);
-        // Note: markRateLimited releases the lock but does not increment
-        // request_count. We intentionally count 429s as requests for
-        // accurate load tracking across accounts.
-        const entry = accountPool.getEntry(entryId);
-        if (entry) {
-          entry.usage.request_count++;
-          entry.usage.last_used = new Date().toISOString();
-        }
+        // P1-6: Count 429s as requests via encapsulated API (no direct entry mutation)
+        accountPool.markRateLimited(entryId, { countRequest: true });
         c.status(429);
         return c.json(fmt.format429(err.message));
       }

src/session/manager.ts

@@ -71,16 +71,9 @@ export class SessionManager {
     messages: Array<{ role: string; content: string }>,
   ): void {
     const hash = this.hashMessages(messages);
-    // Evict oldest session if at capacity
+    // P2-10: O(1) LRU eviction using Map insertion order
     if (this.sessions.size >= MAX_SESSIONS) {
-      let oldestKey: string | null = null;
-      let oldestTime = Infinity;
-      for (const [key, s] of this.sessions) {
-        if (s.createdAt < oldestTime) {
-          oldestTime = s.createdAt;
-          oldestKey = key;
-        }
-      }
+      const oldestKey = this.sessions.keys().next().value;
       if (oldestKey) this.sessions.delete(oldestKey);
     }
     this.sessions.set(taskId, {

src/utils/logger.ts

@@ -0,0 +1,66 @@
+/**
+ * Structured logger — JSON in production, readable in development.
+ *
+ * Usage:
+ *   import { log } from "../utils/logger.js";
+ *   log.info("Request received", { method: "POST", path: "/v1/chat/completions" });
+ *   log.warn("Stale lock", { entryId: "abc" });
+ *   log.error("Curl failed", { code: 1, stderr: "..." });
+ */
+
+type LogLevel = "debug" | "info" | "warn" | "error";
+
+const LEVEL_PRIORITY: Record<LogLevel, number> = {
+  debug: 0,
+  info: 1,
+  warn: 2,
+  error: 3,
+};
+
+const isProduction = process.env.NODE_ENV === "production";
+const minLevel: LogLevel = (process.env.LOG_LEVEL as LogLevel) ?? (isProduction ? "info" : "debug");
+
+function shouldLog(level: LogLevel): boolean {
+  return LEVEL_PRIORITY[level] >= LEVEL_PRIORITY[minLevel];
+}
+
+function emit(level: LogLevel, message: string, extra?: Record<string, unknown>): void {
+  if (!shouldLog(level)) return;
+
+  if (isProduction) {
+    // JSON structured output for container/log aggregator consumption
+    const entry: Record<string, unknown> = {
+      ts: new Date().toISOString(),
+      level,
+      msg: message,
+      ...extra,
+    };
+    const line = JSON.stringify(entry);
+    if (level === "error") {
+      process.stderr.write(line + "\n");
+    } else {
+      process.stdout.write(line + "\n");
+    }
+  } else {
+    // Human-readable for development
+    const prefix = `[${level.toUpperCase()}]`;
+    const parts: unknown[] = [prefix, message];
+    if (extra && Object.keys(extra).length > 0) {
+      parts.push(extra);
+    }
+    if (level === "error") {
+      console.error(...parts);
+    } else if (level === "warn") {
+      console.warn(...parts);
+    } else {
+      console.log(...parts);
+    }
+  }
+}
+
+export const log = {
+  debug: (msg: string, extra?: Record<string, unknown>) => emit("debug", msg, extra),
+  info: (msg: string, extra?: Record<string, unknown>) => emit("info", msg, extra),
+  warn: (msg: string, extra?: Record<string, unknown>) => emit("warn", msg, extra),
+  error: (msg: string, extra?: Record<string, unknown>) => emit("error", msg, extra),
+};