import { describe, expect, it } from "@effect/vitest";import { Effect, Exit, Schema, Stream } from "effect";import { FetchHttpClient } from "effect/unstable/http";import { HttpApi, HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi";import { createServer, type ServerResponse } from "node:http";import type { AddressInfo, Socket } from "node:net"; import { createExecutor, AuthTemplateSlug, ConnectionName, IntegrationSlug, ToolAddress,} from "@executor-js/sdk";import { makeTestConfig, memoryCredentialsPlugin } from "@executor-js/sdk/testing";import { makeOpenApiHttpApiTestSourceConfig, unwrapInvocation } from "../testing"; import { collectStreamingBody, STREAM_MAX_BYTES } from "./invoke";import { openApiPlugin } from "./plugin"; const testPlugins = () => [openApiPlugin({ httpClientLayer: FetchHttpClient.layer }), memoryCredentialsPlugin()] as const; const STREAM_TOOL = "logs.getLogs"; const LogsGroup = HttpApiGroup.make("logs").add( HttpApiEndpoint.get("getLogs", "/logs", { success: Schema.Unknown, }),); const StreamingApi = HttpApi.make("streamingResponseTest") .add(LogsGroup) .annotateMerge(OpenApi.annotations({ title: "StreamingResponseTest", version: "1.0.0" })); const buildExecutor = (baseUrl: string) => Effect.gen(function* () { const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() })); yield* executor.openapi.addSpec( makeOpenApiHttpApiTestSourceConfig(StreamingApi, { slug: "streaming", baseUrl }), ); yield* executor.connections.create({ owner: "org", name: ConnectionName.make("main"), integration: IntegrationSlug.make("streaming"), template: AuthTemplateSlug.make("apiKey"), value: "token", }); return { executor, address: ToolAddress.make(`tools.streaming.org.main.${STREAM_TOOL}`), }; }); const startRawServer = (handler: (res: ServerResponse) => void) => Effect.acquireRelease( Effect.callback<{ baseUrl: string; close: () => void }>((resume) => { const sockets = new Set<Socket>(); const server = createServer((req, res) => { if (req.url?.split("?")[0] !== "/logs") { res.writeHead(200, { "content-type": "application/json" }); res.end(JSON.stringify({ ok: true })); return; } handler(res); }); server.on("connection", (socket) => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); }); server.listen(0, "127.0.0.1", () => { const port = (server.address() as AddressInfo).port; resume( Effect.succeed({ baseUrl: `http://127.0.0.1:${port}`, close: () => { for (const socket of sockets) socket.destroy(); server.close(); }, }), ); }); }), (server) => Effect.sync(() => server.close()), ); const encoder = new TextEncoder(); const delayedReadableStream = ( chunks: readonly Uint8Array[], delayMs: number, options: { close: boolean },) => Stream.fromReadableStream({ evaluate: () => { const timers: ReturnType<typeof setTimeout>[] = []; return new ReadableStream<Uint8Array>({ start: (controller) => { chunks.forEach((chunk, index) => { timers.push( setTimeout(() => { controller.enqueue(chunk); if (options.close && index === chunks.length - 1) { controller.close(); } }, index * delayMs), ); }); }, cancel: () => { for (const timer of timers) clearTimeout(timer); }, }); }, onError: (error) => error, }); describe("OpenAPI streaming responses", () => { it.effect("returns from an oversized NDJSON stream with truncation metadata", () => Effect.gen(function* () { const rows = Array.from( { length: 6 }, (_, index) => `${JSON.stringify({ index, phase: "immediate" })}\n`, ).join(""); const rowBytes = encoder.encode(rows); const partialBytes = encoder.encode(`{"partial":"${"x".repeat(STREAM_MAX_BYTES)}"`); const result = yield* collectStreamingBody( Stream.make(rowBytes, partialBytes), "application/stream+json", ); expect(result.headers["x-executor-stream"]).toBe("truncated"); expect(result.headers["x-executor-stream-bytes"]).toBe(String(STREAM_MAX_BYTES)); expect(Array.isArray(result.data)).toBe(true); const data = result.data as unknown[]; expect(data.slice(0, 6)).toEqual( Array.from({ length: 6 }, (_, index) => ({ index, phase: "immediate" })), ); }), ); it.effect("collects every chunk from a delayed ReadableStream", () => Effect.gen(function* () { const chunks = [ encoder.encode(`${JSON.stringify({ id: 1 })}\n`), encoder.encode(`${JSON.stringify({ id: 2 })}\n`), encoder.encode(`${JSON.stringify({ id: 3 })}\n`), ]; const expectedBytes = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0); const result = yield* collectStreamingBody( delayedReadableStream(chunks, 75, { close: true }), "application/x-ndjson", { maxBytes: STREAM_MAX_BYTES, maxMs: 1_000 }, ); console.log( JSON.stringify({ case: "delayed-readable-complete", bytes: result.headers["x-executor-stream-bytes"], rows: Array.isArray(result.data) ? result.data.length : 0, }), ); expect(result.headers["x-executor-stream"]).toBe("complete"); expect(result.headers["x-executor-stream-bytes"]).toBe(String(expectedBytes)); expect(result.data).toEqual([{ id: 1 }, { id: 2 }, { id: 3 }]); }), ); it("timeout truncation preserves all chunks received before interruption", async () => { await Effect.runPromise( Effect.gen(function* () { const chunks = [ encoder.encode(`${JSON.stringify({ id: 1 })}\n`), encoder.encode(`${JSON.stringify({ id: 2 })}\n`), ]; const expectedBytes = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0); const startedAt = Date.now(); const result = yield* collectStreamingBody( delayedReadableStream(chunks, 100, { close: false }), "application/stream+json", { maxBytes: STREAM_MAX_BYTES, maxMs: 500 }, ); const elapsedMs = Date.now() - startedAt; console.log( JSON.stringify({ case: "delayed-readable-timeout", elapsedMs, bytes: result.headers["x-executor-stream-bytes"], rows: Array.isArray(result.data) ? result.data.length : 0, }), ); expect(elapsedMs).toBeGreaterThanOrEqual(450); expect(elapsedMs).toBeLessThan(1_500); expect(result.headers["x-executor-stream"]).toBe("truncated"); expect(result.headers["x-executor-stream-bytes"]).toBe(String(expectedBytes)); expect(result.data).toEqual([{ id: 1 }, { id: 2 }]); }), ); }); it.effect("drops a trailing partial JSON line when the byte cap truncates", () => Effect.gen(function* () { let body = ""; for (let index = 0; Buffer.byteLength(body) < STREAM_MAX_BYTES - 2_000; index++) { body += `${JSON.stringify({ index, value: "x".repeat(512) })}\n`; } const encoded = encoder.encode(`${body}{"partial":"${"y".repeat(STREAM_MAX_BYTES)}"`); const result = yield* collectStreamingBody(Stream.make(encoded), "application/x-ndjson"); expect(result.headers["x-executor-stream"]).toBe("truncated"); expect(Array.isArray(result.data)).toBe(true); const data = result.data as unknown[]; expect(data.length).toBeGreaterThan(0); expect(data.every((row: unknown) => typeof row === "object" && row !== null)).toBe(true); expect(data.some((row: unknown) => "partial" in (row as Record<string, unknown>))).toBe( false, ); }), ); it.effect("parses a completed short NDJSON stream", () => Effect.gen(function* () { const server = yield* startRawServer((res) => { res.writeHead(200, { "content-type": "application/jsonl" }); res.end(`${JSON.stringify({ id: 1 })}\n${JSON.stringify({ id: 2 })}\n`); }); const { executor, address } = yield* buildExecutor(server.baseUrl); const result = unwrapInvocation<unknown[]>(yield* executor.execute(address, {})); expect(result.data).toEqual([{ id: 1 }, { id: 2 }]); expect(result.headers?.["x-executor-stream"]).toBe("complete"); }), ); it.effect("leaves ordinary JSON responses on the buffered path", () => Effect.gen(function* () { const server = yield* startRawServer((res) => { res.writeHead(200, { "content-type": "application/json" }); res.end(JSON.stringify({ ok: true })); }); const { executor, address } = yield* buildExecutor(server.baseUrl); const result = unwrapInvocation<Record<string, unknown>>( yield* executor.execute(address, {}), ); expect(result.data).toEqual({ ok: true }); expect(result.headers?.["x-executor-stream"]).toBeUndefined(); }), ); it.effect("leaves non-2xx JSON error responses on the buffered path", () => Effect.gen(function* () { const server = yield* startRawServer((res) => { res.writeHead(422, { "content-type": "application/json" }); res.end('{"error":{"field":"name","reason":"too_short"}}'); }); const { executor, address } = yield* buildExecutor(server.baseUrl); const exit = yield* executor.execute(address, {}).pipe(Effect.exit); const text = Exit.match(exit, { onFailure: (cause) => JSON.stringify(cause), onSuccess: (value) => JSON.stringify(value), }); expect(text).toMatch(/422|too_short|name|response|error/i); expect(text).not.toMatch(/x-executor-stream/); }), );});
import { describe, expect, it } from "@effect/vitest";
import { Effect, Exit, Schema, Stream } from "effect";
import { FetchHttpClient } from "effect/unstable/http";
import { HttpApi, HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi";
import { createServer, type ServerResponse } from "node:http";
import type { AddressInfo, Socket } from "node:net";
import {
createExecutor,
AuthTemplateSlug,
ConnectionName,
IntegrationSlug,
ToolAddress,
} from "@executor-js/sdk";
import { makeTestConfig, memoryCredentialsPlugin } from "@executor-js/sdk/testing";
import { makeOpenApiHttpApiTestSourceConfig, unwrapInvocation } from "../testing";
import { collectStreamingBody, STREAM_MAX_BYTES } from "./invoke";
import { openApiPlugin } from "./plugin";
const testPlugins = () =>
[openApiPlugin({ httpClientLayer: FetchHttpClient.layer }), memoryCredentialsPlugin()] as const;
const STREAM_TOOL = "logs.getLogs";
const LogsGroup = HttpApiGroup.make("logs").add(
HttpApiEndpoint.get("getLogs", "/logs", {
success: Schema.Unknown,
}),
);
const StreamingApi = HttpApi.make("streamingResponseTest")
.add(LogsGroup)
.annotateMerge(OpenApi.annotations({ title: "StreamingResponseTest", version: "1.0.0" }));
const buildExecutor = (baseUrl: string) =>
Effect.gen(function* () {
const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() }));
yield* executor.openapi.addSpec(
makeOpenApiHttpApiTestSourceConfig(StreamingApi, { slug: "streaming", baseUrl }),
);
yield* executor.connections.create({
owner: "org",
name: ConnectionName.make("main"),
integration: IntegrationSlug.make("streaming"),
template: AuthTemplateSlug.make("apiKey"),
value: "token",
});
return {
executor,
address: ToolAddress.make(`tools.streaming.org.main.${STREAM_TOOL}`),
};
});
const startRawServer = (handler: (res: ServerResponse) => void) =>
Effect.acquireRelease(
Effect.callback<{ baseUrl: string; close: () => void }>((resume) => {
const sockets = new Set<Socket>();
const server = createServer((req, res) => {
if (req.url?.split("?")[0] !== "/logs") {
res.writeHead(200, { "content-type": "application/json" });
res.end(JSON.stringify({ ok: true }));
return;
}
handler(res);
});
server.on("connection", (socket) => {
sockets.add(socket);
socket.on("close", () => sockets.delete(socket));
});
server.listen(0, "127.0.0.1", () => {
const port = (server.address() as AddressInfo).port;
resume(
Effect.succeed({
baseUrl: `http://127.0.0.1:${port}`,
close: () => {
for (const socket of sockets) socket.destroy();
server.close();
},
}),
);
});
}),
(server) => Effect.sync(() => server.close()),
);
const encoder = new TextEncoder();
const delayedReadableStream = (
chunks: readonly Uint8Array[],
delayMs: number,
options: { close: boolean },
) =>
Stream.fromReadableStream({
evaluate: () => {
const timers: ReturnType<typeof setTimeout>[] = [];
return new ReadableStream<Uint8Array>({
start: (controller) => {
chunks.forEach((chunk, index) => {
timers.push(
setTimeout(() => {
controller.enqueue(chunk);
if (options.close && index === chunks.length - 1) {
controller.close();
}
}, index * delayMs),
);
});
},
cancel: () => {
for (const timer of timers) clearTimeout(timer);
},
});
},
onError: (error) => error,
});
describe("OpenAPI streaming responses", () => {
it.effect("returns from an oversized NDJSON stream with truncation metadata", () =>
Effect.gen(function* () {
const rows = Array.from(
{ length: 6 },
(_, index) => `${JSON.stringify({ index, phase: "immediate" })}\n`,
).join("");
const rowBytes = encoder.encode(rows);
const partialBytes = encoder.encode(`{"partial":"${"x".repeat(STREAM_MAX_BYTES)}"`);
const result = yield* collectStreamingBody(
Stream.make(rowBytes, partialBytes),
"application/stream+json",
);
expect(result.headers["x-executor-stream"]).toBe("truncated");
expect(result.headers["x-executor-stream-bytes"]).toBe(String(STREAM_MAX_BYTES));
expect(Array.isArray(result.data)).toBe(true);
const data = result.data as unknown[];
expect(data.slice(0, 6)).toEqual(
Array.from({ length: 6 }, (_, index) => ({ index, phase: "immediate" })),
);
}),
);
it.effect("collects every chunk from a delayed ReadableStream", () =>
Effect.gen(function* () {
const chunks = [
encoder.encode(`${JSON.stringify({ id: 1 })}\n`),
encoder.encode(`${JSON.stringify({ id: 2 })}\n`),
encoder.encode(`${JSON.stringify({ id: 3 })}\n`),
];
const expectedBytes = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0);
const result = yield* collectStreamingBody(
delayedReadableStream(chunks, 75, { close: true }),
"application/x-ndjson",
{ maxBytes: STREAM_MAX_BYTES, maxMs: 1_000 },
);
console.log(
JSON.stringify({
case: "delayed-readable-complete",
bytes: result.headers["x-executor-stream-bytes"],
rows: Array.isArray(result.data) ? result.data.length : 0,
}),
);
expect(result.headers["x-executor-stream"]).toBe("complete");
expect(result.headers["x-executor-stream-bytes"]).toBe(String(expectedBytes));
expect(result.data).toEqual([{ id: 1 }, { id: 2 }, { id: 3 }]);
}),
);
it("timeout truncation preserves all chunks received before interruption", async () => {
await Effect.runPromise(
Effect.gen(function* () {
const chunks = [
encoder.encode(`${JSON.stringify({ id: 1 })}\n`),
encoder.encode(`${JSON.stringify({ id: 2 })}\n`),
];
const expectedBytes = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0);
const startedAt = Date.now();
const result = yield* collectStreamingBody(
delayedReadableStream(chunks, 100, { close: false }),
"application/stream+json",
{ maxBytes: STREAM_MAX_BYTES, maxMs: 500 },
);
const elapsedMs = Date.now() - startedAt;
console.log(
JSON.stringify({
case: "delayed-readable-timeout",
elapsedMs,
bytes: result.headers["x-executor-stream-bytes"],
rows: Array.isArray(result.data) ? result.data.length : 0,
}),
);
expect(elapsedMs).toBeGreaterThanOrEqual(450);
expect(elapsedMs).toBeLessThan(1_500);
expect(result.headers["x-executor-stream"]).toBe("truncated");
expect(result.headers["x-executor-stream-bytes"]).toBe(String(expectedBytes));
expect(result.data).toEqual([{ id: 1 }, { id: 2 }]);
}),
);
});
it.effect("drops a trailing partial JSON line when the byte cap truncates", () =>
Effect.gen(function* () {
let body = "";
for (let index = 0; Buffer.byteLength(body) < STREAM_MAX_BYTES - 2_000; index++) {
body += `${JSON.stringify({ index, value: "x".repeat(512) })}\n`;
}
const encoded = encoder.encode(`${body}{"partial":"${"y".repeat(STREAM_MAX_BYTES)}"`);
const result = yield* collectStreamingBody(Stream.make(encoded), "application/x-ndjson");
expect(result.headers["x-executor-stream"]).toBe("truncated");
expect(Array.isArray(result.data)).toBe(true);
const data = result.data as unknown[];
expect(data.length).toBeGreaterThan(0);
expect(data.every((row: unknown) => typeof row === "object" && row !== null)).toBe(true);
expect(data.some((row: unknown) => "partial" in (row as Record<string, unknown>))).toBe(
false,
);
}),
);
it.effect("parses a completed short NDJSON stream", () =>
Effect.gen(function* () {
const server = yield* startRawServer((res) => {
res.writeHead(200, { "content-type": "application/jsonl" });
res.end(`${JSON.stringify({ id: 1 })}\n${JSON.stringify({ id: 2 })}\n`);
});
const { executor, address } = yield* buildExecutor(server.baseUrl);
const result = unwrapInvocation<unknown[]>(yield* executor.execute(address, {}));
expect(result.data).toEqual([{ id: 1 }, { id: 2 }]);
expect(result.headers?.["x-executor-stream"]).toBe("complete");
}),
);
it.effect("leaves ordinary JSON responses on the buffered path", () =>
Effect.gen(function* () {
const server = yield* startRawServer((res) => {
res.writeHead(200, { "content-type": "application/json" });
res.end(JSON.stringify({ ok: true }));
});
const { executor, address } = yield* buildExecutor(server.baseUrl);
const result = unwrapInvocation<Record<string, unknown>>(
yield* executor.execute(address, {}),
);
expect(result.data).toEqual({ ok: true });
expect(result.headers?.["x-executor-stream"]).toBeUndefined();
}),
);
it.effect("leaves non-2xx JSON error responses on the buffered path", () =>
Effect.gen(function* () {
const server = yield* startRawServer((res) => {
res.writeHead(422, { "content-type": "application/json" });
res.end('{"error":{"field":"name","reason":"too_short"}}');
});
const { executor, address } = yield* buildExecutor(server.baseUrl);
const exit = yield* executor.execute(address, {}).pipe(Effect.exit);
const text = Exit.match(exit, {
onFailure: (cause) => JSON.stringify(cause),
onSuccess: (value) => JSON.stringify(value),
});
expect(text).toMatch(/422|too_short|name|response|error/i);
expect(text).not.toMatch(/x-executor-stream/);
}),
);
});