Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions open-sse/executors/adapta-web.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import { sanitizeErrorMessage } from "../utils/error.ts";
const ADAPTA_APP_URL = "https://agent.adapta.one";
const ADAPTA_CLERK_URL = "https://clerk.agent.adapta.one";
const ADAPTA_STREAM_URL = `${ADAPTA_APP_URL}/api/chat/stream/v1`;
const ADAPTA_PUBLIC_STREAM_ERROR = `\n\n[Erro: ${sanitizeErrorMessage("Adapta upstream error")}]`;

// Default model ID in Adapta's internal system (corresponds to "ONE" / auto-select)
const DEFAULT_AI_MODEL_ID = 14;
Expand Down Expand Up @@ -308,10 +309,9 @@ function transformStream(adaptaStream: ReadableStream, model: string): ReadableS
if (event.id === "quick-response") continue;
// Real text ended — stream will send more events or close
} else if (type === "error") {
const errText = String(event.errorText ?? "Adapta upstream error");
ensureRole();
// Emit the error as content so the user sees it
chunk({ content: `\n\n[Erro: ${errText}]` });
// Keep upstream diagnostics private: the transformed SSE is a public HTTP 200 body.
chunk({ content: ADAPTA_PUBLIC_STREAM_ERROR });
finalize();
return;
} else if (type === "done" || type === "end") {
Expand Down
69 changes: 69 additions & 0 deletions tests/fixtures/adapta-web-stream-error-boundary.fixture.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
import assert from "node:assert/strict";
import test from "node:test";

assert.ok(process.env.DATA_DIR, "the subprocess fixture requires an isolated DATA_DIR");
assert.ok(
process.env.OMNIROUTE_PLUGINS_DIR,
"the subprocess fixture requires an isolated OMNIROUTE_PLUGINS_DIR"
);
assert.equal(process.env.HOME, undefined, "the subprocess must not inherit HOME");
assert.equal(process.env.CODEX_HOME, undefined, "the subprocess must not inherit CODEX_HOME");

const { AdaptaWebExecutor } = await import("../../open-sse/executors/adapta-web.ts");

const originalFetch = globalThis.fetch;

test.afterEach(() => {
globalThis.fetch = originalFetch;
});

test("terminates an upstream error event without exposing its text in the public SSE", async () => {
const hostileError =
"SQLSTATE 42P01 at /srv/omniroute/private.ts:91 — Authorization: Bearer secret-token";
const requestedUrls: string[] = [];
const logMessages: string[] = [];

globalThis.fetch = (async (input: RequestInfo | URL) => {
const url = String(input);
requestedUrls.push(url);

if (url.endsWith("/v1/client")) {
return Response.json({
response: { sessions: [{ id: "session-stream-error", status: "active" }] },
});
}

if (url.includes("/tokens")) {
return Response.json({ jwt: "eyJ.test-session.jwt" });
}

return new Response(`data: ${JSON.stringify({ type: "error", errorText: hostileError })}\n\n`, {
status: 200,
headers: { "Content-Type": "text/event-stream" },
});
}) as typeof fetch;

const executor = new AdaptaWebExecutor();
const result = await executor.execute({
model: "adapta-one",
body: { messages: [{ role: "user", content: "hello" }] },
stream: true,
credentials: { apiKey: "__client=unique-stream-error-cookie" },
signal: null,
log: {
info: (_tag, message) => logMessages.push(message),
warn: (_tag, message) => logMessages.push(message),
},
});

assert.equal(result.response.status, 200);
assert.equal(result.response.headers.get("content-type"), "text/event-stream");

const publicSse = await result.response.text();
assert.equal(requestedUrls.length, 3);
assert.match(publicSse, /"content":"\\n\\n\[Erro: Adapta upstream error\]"/);
assert.match(publicSse, /"finish_reason":"stop"/);
assert.match(publicSse, /data: \[DONE\]/);
assert.doesNotMatch(publicSse, /SQLSTATE|\/srv\/omniroute|secret-token/);
assert.doesNotMatch(logMessages.join("\n"), /SQLSTATE|\/srv\/omniroute|secret-token/);
});
97 changes: 97 additions & 0 deletions tests/unit/executor-adapta-web-stream-error.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import { fileURLToPath } from "node:url";
import test from "node:test";

const FIXTURE = fileURLToPath(
new URL("../fixtures/adapta-web-stream-error-boundary.fixture.ts", import.meta.url)
);
const REPO_ROOT = fileURLToPath(new URL("../..", import.meta.url));
const SYNTHETIC_API_KEY_SECRET =
"adapta-stream-boundary-test-secret-00000000000000000000000000000000";

type FixtureResult = {
code: number | null;
signal: NodeJS.Signals | null;
stdout: string;
stderr: string;
};

function runIsolatedFixture(testRoot: string): Promise<FixtureResult> {
const dataDir = path.join(testRoot, "data");
const pluginsDir = path.join(testRoot, "plugins");
fs.mkdirSync(dataDir, { recursive: true });
fs.mkdirSync(pluginsDir, { recursive: true });

const childEnv: NodeJS.ProcessEnv = {
PATH: process.env.PATH,
NODE_PATH: process.env.NODE_PATH,
LANG: process.env.LANG ?? "C.UTF-8",
LC_ALL: process.env.LC_ALL,
TZ: process.env.TZ ?? "UTC",
TMPDIR: process.env.TMPDIR ?? os.tmpdir(),
NODE_ENV: "test",
APP_LOG_TO_FILE: "false",
API_KEY_SECRET: SYNTHETIC_API_KEY_SECRET,
DATA_DIR: dataDir,
OMNIROUTE_PLUGINS_DIR: pluginsDir,
};
delete childEnv.NODE_TEST_CONTEXT;

return new Promise((resolve, reject) => {
const child = spawn(process.execPath, ["--import", "tsx/esm", "--test", FIXTURE], {
cwd: REPO_ROOT,
env: childEnv,
stdio: ["ignore", "pipe", "pipe"],
timeout: 90_000,
});
let stdout = "";
let stderr = "";
child.stdout.setEncoding("utf8").on("data", (chunk: string) => {
stdout += chunk;
});
child.stderr.setEncoding("utf8").on("data", (chunk: string) => {
stderr += chunk;
});
child.once("error", reject);
child.once("close", (code, signal) => resolve({ code, signal, stdout, stderr }));
});
}

test("Adapta stream errors are sanitized in an isolated executor fixture", async () => {
const testRoot = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-adapta-boundary-parent-"));
const originalDataDir = process.env.DATA_DIR;
const originalPluginsDir = process.env.OMNIROUTE_PLUGINS_DIR;
const originalFetch = globalThis.fetch;
const eventBusOwner = globalThis as { __omnirouteEventBus?: unknown };
const originalEventBus = eventBusOwner.__omnirouteEventBus;

try {
const result = await runIsolatedFixture(testRoot);
assert.equal(
result.code,
0,
`isolated Adapta fixture failed (signal=${result.signal ?? "none"})\n` +
`stdout:\n${result.stdout}\nstderr:\n${result.stderr}`
);
assert.equal(result.signal, null);
assert.match(result.stdout, /ℹ tests 1/);
assert.match(result.stdout, /ℹ pass 1/);
assert.match(result.stdout, /ℹ fail 0/);
assert.doesNotMatch(result.stdout + result.stderr, new RegExp(SYNTHETIC_API_KEY_SECRET));

assert.equal(process.env.DATA_DIR, originalDataDir);
assert.equal(process.env.OMNIROUTE_PLUGINS_DIR, originalPluginsDir);
assert.equal(globalThis.fetch, originalFetch);
assert.equal(
eventBusOwner.__omnirouteEventBus,
originalEventBus,
"the subprocess fixture must not replace the parent event bus singleton"
);
} finally {
fs.rmSync(testRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
}
});
31 changes: 19 additions & 12 deletions tests/unit/executor-adapta-web.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,18 +36,25 @@ describe("AdaptaWebExecutor", () => {
});

it("execute returns proper result shape on auth failure", async () => {
const executor = new mod.AdaptaWebExecutor();
const result = await executor.execute({
model: "adapta-one",
body: { messages: [{ role: "user", content: "hi" }] },
stream: false,
credentials: { apiKey: "invalid-jwt" },
signal: null,
});
assert.ok(result.response instanceof Response);
assert.ok(typeof result.url === "string");
assert.ok(typeof result.headers === "object");
assert.ok(result.transformedBody !== undefined);
const originalFetch = globalThis.fetch;
globalThis.fetch = (async () => new Response(null, { status: 401 })) as typeof fetch;

try {
const executor = new mod.AdaptaWebExecutor();
const result = await executor.execute({
model: "adapta-one",
body: { messages: [{ role: "user", content: "hi" }] },
stream: false,
credentials: { apiKey: "invalid-jwt" },
signal: null,
});
assert.ok(result.response instanceof Response);
assert.ok(typeof result.url === "string");
assert.ok(typeof result.headers === "object");
assert.ok(result.transformedBody !== undefined);
} finally {
globalThis.fetch = originalFetch;
}
});

it("testConnection returns false for invalid credentials", async () => {
Expand Down
Loading