diff --git a/changelog.d/fixes/adapta-nonstream-sse-error.md b/changelog.d/fixes/adapta-nonstream-sse-error.md new file mode 100644 index 0000000000..fb34658a4b --- /dev/null +++ b/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. diff --git a/open-sse/executors/adapta-web.ts b/open-sse/executors/adapta-web.ts index 83969ed927..619b952c3c 100644 --- a/open-sse/executors/adapta-web.ts +++ b/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"; @@ -480,9 +480,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 }); @@ -494,6 +495,9 @@ 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 @@ -501,9 +505,24 @@ export class AdaptaWebExecutor extends BaseExecutor { } } } 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, diff --git a/tests/fixtures/adapta-web-nonstream-error-boundary.fixture.ts b/tests/fixtures/adapta-web-nonstream-error-boundary.fixture.ts new file mode 100644 index 0000000000..2e58eb87ea --- /dev/null +++ b/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 => { + 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"); + }); +}); diff --git a/tests/unit/adapta-web-nonstream-error-boundary.test.ts b/tests/unit/adapta-web-nonstream-error-boundary.test.ts new file mode 100644 index 0000000000..ef5c5894f4 --- /dev/null +++ b/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}` + ); +});