refactor(sse): classify SSE critical-path empty catches + add CONTRIBUTING convention (#8142)

CONTRIBUTING.md

@@ -228,6 +228,31 @@ Current test status: **122 unit test files** covering:
 - **Zod validation** — Use Zod v4 schemas for all API input validation
 - **Naming**: Files = camelCase/kebab-case, components = PascalCase, constants = UPPER_SNAKE
 
+### Error handling / empty catch blocks
+
+Never leave a `catch` unexplained. Classify it into one of two buckets (operationalizes
+the hard rule "never silently swallow errors in SSE streams"):
+
+- **Intentional (our own best-effort cleanup/telemetry)** — a failure here is expected and
+  harmless; add a one-line rationale comment, no logging (logging on every request is the
+  noise this convention avoids).
+
+  ```ts
+  } catch {} // closing an already-closed controller after client disconnect is expected
+  ```
+
+- **Should log (external/caller-supplied code, or the swallow changes control flow)** — keep
+  the catch (never let it break the stream) but emit a contextual `console.debug`/`warn` so the
+  failure is discoverable.
+
+  ```ts
+  } catch (e) {
+    console.debug("[STREAM] onFailure callback error:", e);
+  }
+  ```
+
+See `open-sse/utils/stream.ts` and `open-sse/utils/streamHandler.ts` for applied examples.
+
 ---
 
 ## Project Structure

changelog.d/features/8142-empty-catch-convention.md

@@ -0,0 +1 @@
+- refactor(sse): classify SSE critical-path empty catches + add CONTRIBUTING convention (#8142)

open-sse/utils/stream.ts

@@ -641,9 +641,7 @@ function restoreClaudePassthroughToolUseName(parsed: JsonRecord, toolNameMap: un
 // to avoid shared state issues with concurrent streams (TextDecoder with {stream:true}
 // maintains internal buffering state between decode() calls).
 
-/**
- * Stream modes
- */
+// Stream modes
 const STREAM_MODE = {
   TRANSLATE: "translate", // Full translation between formats
   PASSTHROUGH: "passthrough", // No translation, normalize output, extract usage
@@ -813,7 +811,7 @@ export function createSSEStream(options: StreamOptions = {}) {
   let pendingToolFinishTime: number | null = null;
   try {
     pendingToolFinishTime = consumeToolFinishTime(sessionId);
-  } catch {}
+  } catch {} // best-effort read of optional timing state — absence is normal
 
   // Guard against duplicate [DONE] events — ensures exactly one per stream
   let doneSent = false;
@@ -1830,7 +1828,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                             toolTs ? now - toolTs : null,
                             lastChunkTs ? now - lastChunkTs : null
                           );
-                        } catch {}
+                        } catch {} // best-effort telemetry — must never break the stream
                         pendingToolFinishTime = null;
                       }
                     }
@@ -1867,7 +1865,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                     toolFinishTime = now;
                     try {
                       markToolFinish(sessionId);
-                    } catch {}
+                    } catch {} // best-effort bookkeeping write — a miss just skips latency correlation
                   }
 
                   // T18: Normalize finish_reason to 'tool_calls' if tool calls were used
@@ -1943,7 +1941,9 @@ export function createSSEStream(options: StreamOptions = {}) {
               if (onFailure) {
                 try {
                   failureHandled = onFailure(failurePayload) === true;
-                } catch {}
+                } catch (e) {
+                  console.debug(`[STREAM] onFailure callback error:`, e);
+                }
               }
               clearIdleTimer();
               if (!failureHandled) {
@@ -1995,7 +1995,7 @@ export function createSSEStream(options: StreamOptions = {}) {
             toolFinishTime = now;
             try {
               markToolFinish(sessionId);
-            } catch {}
+            } catch {} // best-effort bookkeeping write — a miss just skips latency correlation
           }
 
           // Track content length and accumulate for call log (from raw provider chunk, so content is never missed)
@@ -2109,7 +2109,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                   toolTs ? now - toolTs : null,
                   lastChunkTs ? now - lastChunkTs : null
                 );
-              } catch {}
+              } catch {} // best-effort telemetry — must never break the stream
               pendingToolFinishTime = null;
             }
           }

open-sse/utils/streamHandler.ts

@@ -247,7 +247,10 @@ export function createStreamController({
     try {
       trackPendingRequest(model || "", provider || "", connectionId ?? null, false);
     } catch (e) {
-      console.error(`[${getTimeString()}] [streamHandler] trackPendingRequest decrement failed — counter may drift`, e);
+      console.error(
+        `[${getTimeString()}] [streamHandler] trackPendingRequest decrement failed — counter may drift`,
+        e
+      );
     }
   };
 
@@ -583,7 +586,9 @@ export function createDisconnectAwareStream(transformStream, streamController) {
 
           try {
             controller.close();
-          } catch {}
+          } catch {
+            // Closing an already-closed/aborted controller after client disconnect is expected.
+          }
         }
       },
 

tests/unit/stream-onfailure-callback-logging.test.ts

@@ -0,0 +1,119 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+// #8142 — the empty `catch {}` around the caller-supplied `onFailure(failurePayload)`
+// callback in createSSEStream (open-sse/utils/stream.ts) silently swallowed a throw
+// from consumer code, leaving `failureHandled` false with zero diagnostic trace.
+// Fix: keep the catch (a buggy failure handler must never break the stream) but emit
+// a contextual console.debug so the swallowed error is discoverable.
+
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-8142-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+const core = await import("../../src/lib/db/core.ts");
+const { createSSEStream } = await import("../../open-sse/utils/stream.ts");
+const { FORMATS } = await import("../../open-sse/translator/formats.ts");
+
+const enc = new TextEncoder();
+
+async function readTransformed(chunks: string[], options: Record<string, unknown>) {
+  const source = new ReadableStream<Uint8Array>({
+    start(c) {
+      for (const chunk of chunks) c.enqueue(enc.encode(chunk));
+      c.close();
+    },
+  });
+  return new Response(
+    source.pipeThrough(createSSEStream(options as Parameters<typeof createSSEStream>[0]))
+  ).text();
+}
+
+test.after(() => {
+  core.resetDbInstance();
+  if (fs.existsSync(TEST_DATA_DIR)) {
+    for (const entry of fs.readdirSync(TEST_DATA_DIR)) {
+      fs.rmSync(path.join(TEST_DATA_DIR, entry), { recursive: true, force: true });
+    }
+  }
+});
+
+// An OpenAI-Responses-style `response.failed` passthrough event reliably drives
+// createSSEStream into the `failurePayload` branch (~L1941-1953) that invokes the
+// caller-supplied `onFailure` callback — reused here as the trigger.
+function responseFailedChunk(message: string) {
+  return `data: ${JSON.stringify({
+    type: "response.failed",
+    response: { error: { message, code: "server_error" } },
+  })}\n\n`;
+}
+
+test("#8142 onFailure throwing does not crash the stream and is logged", async () => {
+  const originalDebug = console.debug;
+  const debugCalls: unknown[][] = [];
+  console.debug = (...args: unknown[]) => {
+    debugCalls.push(args);
+  };
+  try {
+    await assert.rejects(
+      readTransformed([responseFailedChunk("boom upstream")], {
+        mode: "passthrough",
+        sourceFormat: FORMATS.OPENAI,
+        provider: "openai",
+        model: "gpt-test",
+        body: { messages: [{ role: "user", content: "hello" }] },
+        onFailure() {
+          throw new Error("boom from consumer onFailure handler");
+        },
+      }),
+      /boom upstream/i,
+      "stream must still reject normally — the callback throw must not corrupt control flow"
+    );
+  } finally {
+    console.debug = originalDebug;
+  }
+  const loggedOnFailureThrow = debugCalls.some((args) =>
+    args.some((a) => typeof a === "string" && /onFailure/i.test(a))
+  );
+  assert.ok(
+    loggedOnFailureThrow,
+    "a console.debug call referencing onFailure must be emitted when the callback throws"
+  );
+});
+
+test("#8142 regression: onFailure returning normally logs nothing and behaves identically", async () => {
+  const originalDebug = console.debug;
+  const debugCalls: unknown[][] = [];
+  console.debug = (...args: unknown[]) => {
+    debugCalls.push(args);
+  };
+  let failurePayload: Record<string, unknown> | null = null;
+  try {
+    await assert.rejects(
+      readTransformed([responseFailedChunk("boom upstream normal")], {
+        mode: "passthrough",
+        sourceFormat: FORMATS.OPENAI,
+        provider: "openai",
+        model: "gpt-test",
+        body: { messages: [{ role: "user", content: "hello" }] },
+        onFailure(p: Record<string, unknown>) {
+          failurePayload = p;
+          return true;
+        },
+      }),
+      /boom upstream normal/i
+    );
+  } finally {
+    console.debug = originalDebug;
+  }
+  assert.ok(failurePayload, "onFailure callback must still be invoked normally");
+  const loggedOnFailureThrow = debugCalls.some((args) =>
+    args.some((a) => typeof a === "string" && /onFailure/i.test(a))
+  );
+  assert.equal(
+    loggedOnFailureThrow,
+    false,
+    "the happy path (no throw) must not emit the onFailure-throw debug log — behavior-free regression guard"
+  );
+});