Merge pull request #13 from gitcommit90/fix/bound-preoutput-inspection

Bound pre-output stream inspection

Author: gitcommit90

docs/architecture.md

@@ -85,7 +85,7 @@ The router supports:
 - `fallback`: members are attempted in their configured order.
 - `round-robin`: each request rotates the starting member, then retains fallback behavior through the remaining members.
 
-Named routes and OAuth account pools continue through every untried target until an upstream returns a usable `2xx` response. Any non-`2xx` status advances fallback regardless of error type. A `2xx` response that contains an immediate stream error, ends without usable output, has no response body, contains invalid JSON, or carries an explicit error payload also advances fallback. Failure classification controls account cooldown locks and diagnostics; it never stops routing. Capability failures do not create account cooldown locks. The per-member timeout defaults to 60 seconds. Caller cancellation stops immediately, and a stream cannot be transparently rerouted after output has already reached the client.
+Named routes and OAuth account pools continue through every untried target until an upstream returns a usable `2xx` response. Any non-`2xx` status advances fallback regardless of error type. A `2xx` response that contains an immediate stream error, ends without usable output, exceeds the bounded pre-output inspection budget, has no response body, contains invalid JSON, or carries an explicit error payload also advances fallback. Failure classification controls account cooldown locks and diagnostics; it never stops routing. Capability failures do not create account cooldown locks. The per-member timeout defaults to 60 seconds. Caller cancellation stops immediately, and a stream cannot be transparently rerouted after output has already reached the client.
 
 OAuth providers add an account-pool layer beneath model routing. Accounts receive monotonic, never-reused aliases (`oauth1`, `oauth2`, ...). Model discovery advertises one canonical pooled id such as `chatgpt/gpt-5.4`; account-qualified ids such as `chatgpt/oauth2/gpt-5.4` and legacy stored-account ids remain resolvable but are not advertised. Quota failures create an account-wide lock using provider reset hints when available; authentication and transient failures use shorter model-scoped cooldowns. Early streaming quota events are inspected before the client stream starts so fallback can still occur. Selection, failure, fallback, locked-account skips, and terminal exhaustion are written as structured logs.
 

package-lock.json

@@ -1,7 +1,7 @@
 {
   "name": "@gitcommit90/rerouted",
   "productName": "ReRouted",
-  "version": "0.5.1",
+  "version": "0.5.2",
   "description": "A local AI router for connected accounts, models, named routes, and automatic fallback.",
   "author": "gitcommit90",
   "license": "MIT",

package.json

@@ -23,6 +23,7 @@ const COOLDOWN_MS = {
   auth: 2 * 60_000,
   transient: 30_000,
 };
+const PREOUTPUT_INSPECTION_BYTES = 64 * 1024;
 
 function createRequestLog(max = 50) {
   const items = [];
@@ -578,7 +579,7 @@ function rebuildResponseWithPrelude(response, chunks, reader) {
   });
 }
 
-async function inspectEarlyResponsesSse(response) {
+async function inspectEarlyResponsesSse(response, maxBytes = PREOUTPUT_INSPECTION_BYTES) {
   if (!response?.ok || !response.body?.getReader) return { response, failure: null };
   const contentType = String(response.headers?.get?.("content-type") || "").toLowerCase();
   if (contentType && !contentType.includes("text/event-stream")) {
@@ -598,24 +599,55 @@ async function inspectEarlyResponsesSse(response) {
   const decoder = new TextDecoder();
   const chunks = [];
   let text = "";
+  let pendingEvents = "";
+  let bytes = 0;
   let done = false;
   while (true) {
     const next = await reader.read();
     done = next.done;
     if (done) break;
     chunks.push(next.value);
-    text += decoder.decode(next.value, { stream: true });
-    const failure = parseEarlyResponsesFailure(text, response);
-    if (failure) {
-      await reader.cancel().catch(() => {});
-      return { response: null, failure };
+    bytes += next.value?.byteLength || 0;
+    const decoded = decoder.decode(next.value, { stream: true });
+    text += decoded;
+    pendingEvents += decoded;
+
+    let completeEnd = 0;
+    const boundary = /\r?\n\r?\n/g;
+    for (let match = boundary.exec(pendingEvents); match; match = boundary.exec(pendingEvents)) {
+      completeEnd = boundary.lastIndex;
+    }
+    if (completeEnd) {
+      const completeEvents = pendingEvents.slice(0, completeEnd);
+      pendingEvents = pendingEvents.slice(completeEnd);
+      const failure = parseEarlyResponsesFailure(completeEvents, response);
+      if (failure) {
+        await reader.cancel().catch(() => {});
+        return { response: null, failure };
+      }
+      if (hasProductiveResponsesEvent(completeEvents)) break;
     }
     // Metadata events such as response.created can precede a quota error.
     // Hold the stream until actual output (or completion) proves the account usable.
-    if (hasProductiveResponsesEvent(text)) break;
+    if (bytes >= maxBytes) {
+      await reader.cancel().catch(() => {});
+      return {
+        response: null,
+        failure: {
+          status: 502,
+          error: `Upstream exceeded the ${maxBytes}-byte inspection budget before producing a usable response`,
+          resetAt: null,
+        },
+      };
+    }
   }
 
   if (done) {
+    const trailing = decoder.decode();
+    text += trailing;
+    pendingEvents += trailing;
+    const trailingFailure = parseEarlyResponsesFailure(pendingEvents, response);
+    if (trailingFailure) return { response: null, failure: trailingFailure };
     try {
       const payload = JSON.parse(text);
       if (isUsableChatCompletion(payload)) {
@@ -624,7 +656,7 @@ async function inspectEarlyResponsesSse(response) {
     } catch {
       /* the completed body may be SSE */
     }
-    if (!hasProductiveResponsesEvent(text)) {
+    if (!hasProductiveResponsesEvent(pendingEvents)) {
       return {
         response: null,
         failure: {

src/lib/router.js

@@ -1365,6 +1365,77 @@ describe("same-provider OAuth account fallback", () => {
     assert.deepEqual(calls, ["stream-1.test", "stream-2.test", "stream-3.test"]);
   });
 
+  it("bounds pre-output stream inspection and reroutes an oversized metadata preamble", async () => {
+    const store = createStore(tmpConfig());
+    const providers = ["Slow preamble", "Backup"].map((name, index) => ({
+      id: `prov_bounded_${index + 1}`,
+      type: "openai-compat",
+      name,
+      baseUrl: `https://bounded-${index + 1}.test/v1`,
+      apiKey: `bounded-key-${index + 1}`,
+      enabled: true,
+      models: [{ id: `bounded-model-${index + 1}`, name, enabled: true }],
+    }));
+    store.seed({
+      providers,
+      combos: [
+        {
+          id: "combo_bounded",
+          strategy: "fallback",
+          members: providers.map((provider, index) => ({
+            providerId: provider.id,
+            model: `bounded-model-${index + 1}`,
+          })),
+        },
+      ],
+    });
+    const calls = [];
+    let canceled = false;
+    const router = createRouter({
+      store,
+      logger: captureLogger(),
+      fetchImpl: async (url) => {
+        calls.push(new URL(url).hostname);
+        if (url.includes("bounded-1")) {
+          return new Response(
+            new ReadableStream({
+              start(controller) {
+                controller.enqueue(
+                  new TextEncoder().encode(
+                    `data: ${JSON.stringify({ type: "response.created", padding: "x".repeat(70 * 1024) })}\n\n`
+                  )
+                );
+              },
+              cancel() {
+                canceled = true;
+              },
+            }),
+            { status: 200, headers: { "Content-Type": "text/event-stream" } }
+          );
+        }
+        return new Response(
+          `data: ${JSON.stringify({ choices: [{ delta: { content: "bounded fallback" } }] })}\n\ndata: [DONE]\n\n`,
+          { status: 200, headers: { "Content-Type": "text/event-stream" } }
+        );
+      },
+    });
+
+    const result = await router.chatCompletions({
+      body: {
+        model: "combo_bounded",
+        messages: [{ role: "user", content: "hello" }],
+        stream: true,
+      },
+    });
+    const chunks = [];
+    await result.streamPipe({ write: (chunk) => chunks.push(String(chunk)) });
+
+    assert.equal(result.ok, true, JSON.stringify(result.error));
+    assert.equal(canceled, true);
+    assert.match(chunks.join(""), /bounded fallback/);
+    assert.deepEqual(calls, ["bounded-1.test", "bounded-2.test"]);
+  });
+
   it("records streaming usage after the final SSE event instead of an early zero-token success", async () => {
     const store = createStore(tmpConfig());
     store.seed({ providers: [oauthAccount("prov_a", "token-a", 100)] });