fix(sse): surface Adapta non-stream SSE errors

changelog.d/fixes/adapta-nonstream-sse-error.md

@@ -0,0 +1 @@
+- **fix(sse):** Treat Adapta Web `type:error` SSE events as sanitized non-stream failures instead of empty HTTP 200 completions.

open-sse/executors/adapta-web.ts

@@ -1,6 +1,6 @@
 import { BaseExecutor, type ExecuteInput } from "./base.ts";
 import { prepareToolMessages, buildToolAwareResult } from "../translator/webTools.ts";
-import { sanitizeErrorMessage } from "../utils/error.ts";
+import { buildErrorBody, sanitizeErrorMessage } from "../utils/error.ts";
 
 const ADAPTA_APP_URL = "https://agent.adapta.one";
 const ADAPTA_CLERK_URL = "https://clerk.agent.adapta.one";
@@ -467,9 +467,10 @@ export class AdaptaWebExecutor extends BaseExecutor {
     const reader = resp.body!.getReader();
     let buf = "";
     let fullText = "";
+    let upstreamErrorMessage: string | null = null;
 
     try {
-      while (true) {
+      readLoop: while (true) {
         const { done, value } = await reader.read();
         if (done) break;
         buf += decoder.decode(value, { stream: true });
@@ -481,16 +482,34 @@ export class AdaptaWebExecutor extends BaseExecutor {
             const ev = JSON.parse(line.slice(6));
             if (ev.type === "text-delta" && ev.id !== "quick-response") {
               fullText += String(ev.delta ?? "");
+            } else if (ev.type === "error") {
+              upstreamErrorMessage = "Adapta upstream error";
+              break readLoop;
             }
           } catch {
             // skip
           }
         }
       }
     } finally {
+      if (upstreamErrorMessage) {
+        void reader.cancel("Adapta upstream SSE error").catch(() => undefined);
+      }
       reader.releaseLock();
     }
 
+    if (upstreamErrorMessage) {
+      return {
+        response: new Response(JSON.stringify(buildErrorBody(502, upstreamErrorMessage)), {
+          status: 502,
+          headers: { "Content-Type": "application/json" },
+        }),
+        url: ADAPTA_STREAM_URL,
+        headers,
+        transformedBody: requestPayload,
+      };
+    }
+
     if (hasTools) {
       const { content, toolCalls, finishReason } = buildToolAwareResult(
         fullText,

tests/fixtures/adapta-web-nonstream-error-boundary.fixture.ts

@@ -0,0 +1,129 @@
+// This suite owns process-wide DATA_DIR, plugin, fetch, and DB state. It must run only inside
+// the subprocess launched by tests/unit/adapta-web-nonstream-error-boundary.test.ts.
+import assert from "node:assert/strict";
+import { mkdirSync, mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { after, afterEach, describe, it } from "node:test";
+
+const TEST_DATA_DIR = mkdtempSync(join(tmpdir(), "omniroute-adapta-nonstream-error-"));
+const TEST_PLUGINS_DIR = join(TEST_DATA_DIR, "plugins");
+mkdirSync(TEST_PLUGINS_DIR, { recursive: true });
+process.env.DATA_DIR = TEST_DATA_DIR;
+process.env.OMNIROUTE_PLUGINS_DIR = TEST_PLUGINS_DIR;
+
+const originalFetch = globalThis.fetch;
+const { AdaptaWebExecutor } = await import("../../open-sse/executors/adapta-web.ts");
+
+interface ErrorEnvelope {
+  error?: {
+    message?: string;
+    type?: string;
+    code?: string;
+  };
+  choices?: unknown[];
+}
+
+interface CompletionEnvelope {
+  choices?: Array<{
+    message?: {
+      content?: string;
+    };
+    finish_reason?: string;
+  }>;
+}
+
+function installAdaptaFetch(upstreamBody: string): void {
+  const mockFetch = async (input: string | URL | Request): Promise<Response> => {
+    const url =
+      typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url;
+
+    if (url === "https://clerk.agent.adapta.one/v1/client") {
+      return Response.json({ response: { sessions: [{ id: "sess-fixture", status: "active" }] } });
+    }
+
+    if (url === "https://clerk.agent.adapta.one/v1/client/sessions/sess-fixture/tokens") {
+      return Response.json({ jwt: "eyJ.fixture.signature" });
+    }
+
+    if (url === "https://agent.adapta.one/api/chat/stream/v1") {
+      return new Response(upstreamBody, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      });
+    }
+
+    throw new Error(`Unexpected test fetch URL: ${url}`);
+  };
+
+  globalThis.fetch = mockFetch as typeof fetch;
+}
+
+afterEach(() => {
+  globalThis.fetch = originalFetch;
+});
+
+after(async () => {
+  const { resetDbInstance } = await import("../../src/lib/db/core.ts");
+  resetDbInstance();
+  rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
+});
+
+describe("Adapta Web non-stream error boundary", () => {
+  it("returns a sanitized 502 when an HTTP 200 SSE body contains type:error", async () => {
+    installAdaptaFetch(
+      `data: ${JSON.stringify({
+        type: "error",
+        errorText:
+          "SQLSTATE 42P01 private detail at /srv/omniroute/open-sse/executors/adapta-web.ts:481:9 Authorization: Bearer secret-token\n    at secret (/srv/omniroute/internal.ts:1:1)",
+      })}\n\n`
+    );
+
+    const executor = new AdaptaWebExecutor();
+    const result = await executor.execute({
+      model: "adapta-one",
+      body: { messages: [{ role: "user", content: "hello" }] },
+      stream: false,
+      credentials: { apiKey: "fixture-client-error" },
+      signal: null,
+    });
+
+    assert.equal(result.response.status, 502);
+    const payload = (await result.response.json()) as ErrorEnvelope;
+    assert.equal(payload.error?.type, "server_error");
+    assert.equal(payload.error?.code, "bad_gateway");
+    assert.equal(payload.error?.message, "Adapta upstream error");
+    assert.ok(!payload.error?.message?.includes("SQLSTATE"));
+    assert.ok(!payload.error?.message?.includes("private detail"));
+    assert.ok(!payload.error?.message?.includes("/srv/omniroute"));
+    assert.ok(!payload.error?.message?.includes("secret-token"));
+    assert.ok(!payload.error?.message?.includes("\n"));
+    assert.equal(payload.choices, undefined);
+  });
+
+  it("preserves a normal non-stream completion assembled from text-delta events", async () => {
+    installAdaptaFetch(
+      [
+        `data: ${JSON.stringify({ type: "text-delta", id: "quick-response", delta: "Loading" })}`,
+        `data: ${JSON.stringify({ type: "text-delta", id: "answer", delta: "Hello" })}`,
+        `data: ${JSON.stringify({ type: "text-delta", id: "answer", delta: " world" })}`,
+        `data: ${JSON.stringify({ type: "done" })}`,
+        "",
+      ].join("\n\n")
+    );
+
+    const executor = new AdaptaWebExecutor();
+    const result = await executor.execute({
+      model: "adapta-one",
+      body: { messages: [{ role: "user", content: "hello" }] },
+      stream: false,
+      credentials: { apiKey: "fixture-client-success" },
+      signal: null,
+    });
+
+    assert.equal(result.response.status, 200);
+    const payload = (await result.response.json()) as CompletionEnvelope;
+    assert.equal(payload.choices?.[0]?.message?.content, "Hello world");
+    assert.equal(payload.choices?.[0]?.finish_reason, "stop");
+  });
+});

tests/unit/adapta-web-nonstream-error-boundary.test.ts

@@ -0,0 +1,73 @@
+import assert from "node:assert/strict";
+import { spawnSync } from "node:child_process";
+import { fileURLToPath } from "node:url";
+import test from "node:test";
+
+const REPO_ROOT = fileURLToPath(new URL("../..", import.meta.url));
+const FIXTURE = fileURLToPath(
+  new URL("../fixtures/adapta-web-nonstream-error-boundary.fixture.ts", import.meta.url)
+);
+
+const CHILD_RUNTIME_ENV_KEYS = [
+  "PATH",
+  "TMPDIR",
+  "TMP",
+  "TEMP",
+  "SystemRoot",
+  "ComSpec",
+  "PATHEXT",
+  "LANG",
+  "LC_ALL",
+  "TZ",
+] as const;
+
+function buildFixtureEnv(): NodeJS.ProcessEnv {
+  const env: NodeJS.ProcessEnv = {
+    NODE_ENV: "test",
+    APP_LOG_TO_FILE: "false",
+    API_KEY_SECRET: "adapta-boundary-fixture-api-key-secret-20260902",
+    DISABLE_SQLITE_AUTO_BACKUP: "true",
+  };
+
+  for (const key of CHILD_RUNTIME_ENV_KEYS) {
+    const value = process.env[key];
+    if (value !== undefined) env[key] = value;
+  }
+
+  // A nested test runner must receive its own context instead of inheriting the parent's.
+  delete env.NODE_TEST_CONTEXT;
+  return env;
+}
+
+test("Adapta Web non-stream error boundaries pass in an isolated process", () => {
+  const result = spawnSync(
+    process.execPath,
+    [
+      "--import",
+      "tsx/esm",
+      "--import",
+      "./open-sse/utils/setupPolyfill.ts",
+      "--test",
+      "--test-force-exit",
+      FIXTURE,
+    ],
+    {
+      cwd: REPO_ROOT,
+      encoding: "utf8",
+      env: buildFixtureEnv(),
+      timeout: 60_000,
+    }
+  );
+
+  assert.ifError(result.error);
+  assert.equal(
+    result.signal,
+    null,
+    `isolated Adapta boundary fixture terminated by ${result.signal}\n${result.stdout}\n${result.stderr}`
+  );
+  assert.equal(
+    result.status,
+    0,
+    `isolated Adapta boundary fixture failed\n${result.stdout}\n${result.stderr}`
+  );
+});