Compare commits

...

4 Commits

4 changed files with 224 additions and 2 deletions

View File

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

View File

@@ -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,6 +482,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
@@ -488,9 +492,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,

View File

@@ -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");
});
});

View File

@@ -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}`
);
});