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 { makeOpenApiHttpApiTestIntegrationConfig, 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(      makeOpenApiHttpApiTestIntegrationConfig(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/);    }),  );   it.effect("advertises an NDJSON operation's output as the array invoke returns", () =>    Effect.gen(function* () {      // The spec declares the schema of ONE line (Vercel's runtime-logs      // convention); the runtime returns an array of parsed lines. The      // advertised output schema must describe that array on BOTH serve      // paths (the add-time compile and the stored-binding rebuild) or      // agents write code against a shape that never comes back.      const lineSchema = {        type: "object",        properties: { level: { type: "string" }, message: { type: "string" } },      };      const server = yield* startRawServer((res) => {        res.writeHead(200, { "content-type": "application/stream+json" });        res.end(          `${JSON.stringify({ level: "info", message: "a" })}\n` +            `${JSON.stringify({ level: "error", message: "b" })}\n`,        );      });       const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() }));      yield* executor.openapi.addSpec({        spec: {          kind: "blob",          value: JSON.stringify({            openapi: "3.0.3",            info: { title: "Runtime Logs", version: "1.0.0" },            servers: [{ url: server.baseUrl }],            paths: {              "/logs": {                get: {                  operationId: "getLogs",                  tags: ["logs"],                  responses: {                    "200": {                      description: "Log lines",                      content: { "application/stream+json": { schema: lineSchema } },                    },                  },                },              },            },          }),        },        slug: "ndjson_logs",        baseUrl: server.baseUrl,      });      yield* executor.connections.create({        owner: "org",        name: ConnectionName.make("main"),        integration: IntegrationSlug.make("ndjson_logs"),        template: AuthTemplateSlug.make("apiKey"),        value: "token",      });      const address = ToolAddress.make("tools.ndjson_logs.org.main.logs.getLogs");       const schema = yield* executor.tools.schema(address);      const outputSchema = schema?.outputSchema as Record<string, unknown>;      expect(outputSchema.type).toBe("array");      expect(outputSchema.items).toEqual(lineSchema);       // Force a catalog rebuild (the stored-binding serve path) to re-derive      // the same array wrap from the persisted content type.      yield* executor.connections.refresh({        owner: "org",        integration: IntegrationSlug.make("ndjson_logs"),        name: ConnectionName.make("main"),      });      const rebuilt = yield* executor.tools.schema(address);      const rebuiltOutput = rebuilt?.outputSchema as Record<string, unknown>;      expect(rebuiltOutput.type).toBe("array");      expect(rebuiltOutput.items).toEqual(lineSchema);       // And the runtime value is exactly that array.      const result = unwrapInvocation<unknown[]>(yield* executor.execute(address, {}));      expect(result.data).toEqual([        { level: "info", message: "a" },        { level: "error", message: "b" },      ]);    }),  );});