Something went wrong. Try again.
opencode v2 mirror
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864import { describe, expect } from "bun:test"import { Deferred, Effect, Fiber, Ref, Stream } from "effect"import { Headers, HttpClientError, HttpClientRequest } from "effect/unstable/http"import { LLM, AIError, HttpContext, InvalidProviderOutputError, TransportError } from "../src/index.js"import { LLMClient, RequestExecutor, WebSocketTransport, type WebSocketChannelExecutor } from "../src/route.js"import { route } from "../src/protocols/openai-chat.js"import { configure } from "../src/providers/openai.js"import { dynamicResponse, fixedResponse, systemError } from "./lib/http.js"import { deltaChunk } from "./lib/openai-chunks.js"import { sseEvents, sseRaw } from "./lib/sse.js"import { it } from "./lib/effect.js"
const request = HttpClientRequest.post("https://provider.test/v1/chat?api_key=secret&key=secret&debug=1").pipe( HttpClientRequest.setHeaders(Headers.fromInput({ authorization: "Bearer secret", "x-safe": "visible" })),)
const secretRequest = HttpClientRequest.post("https://provider.test/v1/chat?api_key=query-secret-123&debug=1").pipe( HttpClientRequest.setHeaders(Headers.fromInput({ authorization: "Bearer header-secret-456" })),)
const expectAIError = (error: unknown) => { expect(error).toBeInstanceOf(AIError) if (!(error instanceof AIError)) throw new Error("expected AIError") expect(error.reason).toBeInstanceOf(Error) expect(error.cause).toBe(error.reason) return error}
const largeProviderMessage = `Upstream request failed: ${"validation failed; ".repeat(1_000)}`
describe("RequestExecutor", () => { it.effect("preserves externally captured HTTP errors without inventing response context", () => Effect.sync(() => { const cause = new Error("upstream request failed") const body = '{"error":{"message":"Rate limited","trace":"original"}}' const error = RequestExecutor.httpFailure({ message: "Rate limited", url: request.url, status: 429, responseHeaders: { "Retry-After": "2", "X-Request-ID": "req_external" }, responseBody: body, cause, })
expect(error.message).toBe("Rate limited") expect(error.reason).toMatchObject({ _tag: "RateLimit", retryAfterMs: 2000 }) expect(error.reason.body).toBe(body) expect(error.reason.cause).toBe(cause) expect(error.reason.http).toEqual( new HttpContext({ url: request.url, status: 429, headers: { "retry-after": "2", "x-request-id": "req_external" }, }), ) expect(RequestExecutor.httpFailure({ message: "No response", url: request.url }).reason.http).toBeUndefined() expect(RequestExecutor.httpFailure({ message: "No URL", status: 500 }).reason.http).toBeUndefined() }), )
it.effect("retains the original body-read failure on an HTTP status error", () => Effect.gen(function* () { const cause = new Error("response body disconnected") const error = yield* Effect.gen(function* () { const executor = yield* RequestExecutor.Service return yield* executor.execute(request).pipe(Effect.flip) }).pipe( Effect.provide( fixedResponse( new ReadableStream({ start(controller) { controller.error(cause) }, }), { status: 503, headers: { "x-request-id": "req_failed_body" }, }, ), ), )
expect(error.reason._tag).toBe("ProviderInternal") expect(error.reason.cause).toBe(cause) expect(error.reason.body).toBeUndefined() expect(error.reason.http).toMatchObject({ status: 503, headers: { "x-request-id": "req_failed_body" } }) }), )
it.effect("parses response body failures at the executor seam", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* RequestExecutor.stream(executor, secretRequest).pipe(Stream.runDrain, Effect.flip)
expectAIError(error) expect(error.message).toBe("ECONNRESET: disconnected query-secret-123 header-secret-456") expect(error.reason.http).toMatchObject({ status: 200, url: secretRequest.url }) expect(error.reason.cause).toMatchObject({ code: "ECONNRESET" }) expect(error.reason).toMatchObject({ _tag: "Transport", transport: "http", operation: "read", code: "ECONNRESET", url: "https://provider.test/v1/chat?api_key=query-secret-123&debug=1", }) }).pipe( Effect.provide( fixedResponse( new ReadableStream({ start(controller) { controller.error(systemError("ECONNRESET", "disconnected query-secret-123 header-secret-456")) }, }), {}, ), ), ), )
it.effect("unwraps native transport failure causes", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* RequestExecutor.stream(executor, secretRequest).pipe(Stream.runDrain, Effect.flip)
expectAIError(error) expect(error.message).toBe("ECONNRESET: socket closed") expect(error.reason.cause).toBeInstanceOf(TypeError) expect(error.reason).toMatchObject({ _tag: "Transport", operation: "read", code: "ECONNRESET", }) }).pipe( Effect.provide( fixedResponse( new ReadableStream({ pull(controller) { controller.error(new TypeError("fetch failed", { cause: systemError("ECONNRESET", "socket closed") })) }, }), {}, ), ), ), )
it.effect("preserves middleware error messages", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor .execute(request, () => Effect.fail(new Error("plugin rejected request"))) .pipe(Effect.flip)
expectAIError(error) expect(error.message).toBe("plugin rejected request") expect(error.reason.cause).toBeInstanceOf(Error) expect(error.reason.http).toBeUndefined() }).pipe(Effect.provide(dynamicResponse(() => Effect.die(new Error("unexpected HTTP request"))))), )
it.effect("reports the request sent by middleware", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor .execute(request, (original, handler) => handler( original.pipe( HttpClientRequest.setUrl("https://proxy.test/v1/chat?api_key=proxy-secret"), HttpClientRequest.setHeader("authorization", "Bearer proxy-secret"), ), ), ) .pipe(Effect.flip)
expectAIError(error) expect(error.message).toBe("ECONNRESET: proxy disconnected proxy-secret") expect(error.reason.http).toBeUndefined() expect(error.reason).toMatchObject({ _tag: "Transport", url: "https://proxy.test/v1/chat?api_key=proxy-secret", }) }).pipe( Effect.provide( dynamicResponse((input) => Effect.fail( new HttpClientError.HttpClientError({ reason: new HttpClientError.TransportError({ request: input.request, cause: systemError("ECONNRESET", "proxy disconnected proxy-secret"), }), }), ), ), ), ), )
it.effect("classifies context overflow responses", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "InvalidRequest", classification: "context-overflow" }) }).pipe( Effect.provide( fixedResponse('{"error":{"code":"context_length_exceeded","message":"prompt too long"}}', { status: 400, }), ), ), )
it.effect("classifies generic HTTP 413 payload errors", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "InvalidRequest", classification: "payload-too-large", }) expect(error.reason.http?.status).toBe(413) }).pipe(Effect.provide(fixedResponse("request too large", { status: 413 }))), )
it.effect("classifies Anthropic request_too_large as context overflow", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "InvalidRequest", classification: "context-overflow", }) expect(error.reason.http?.status).toBe(413) }).pipe( Effect.provide( fixedResponse('{"error":{"type":"request_too_large","message":"Request exceeds the maximum size"}}', { status: 413, }), ), ), )
it.effect("does not classify ordinary invalid requests as context overflow", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "InvalidRequest" }) expect("classification" in error.reason ? error.reason.classification : undefined).toBeUndefined() expect(error.message).toBe("Provider request failed with HTTP 400") }).pipe(Effect.provide(fixedResponse("invalid parameter", { status: 400 }))), )
it.effect("preserves structured provider messages from large error bodies", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "InvalidRequest" }) expect(error.message).toBe(largeProviderMessage) expect(error.reason.body).toContain(largeProviderMessage) }).pipe( Effect.provide( fixedResponse( JSON.stringify({ model: "test-model", error: { type: "invalid_request", message: largeProviderMessage }, }), { status: 400 }, ), ), ), )
it.effect("falls back when structured provider messages are empty", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "InvalidRequest", }) expect(error.message).toBe("Provider request failed with HTTP 400") }).pipe(Effect.provide(fixedResponse('{"error":{"message":" "}}', { status: 400 }))), )
it.effect("classifies provider rate limits hidden behind HTTP 400", () => Effect.gen(function* () { const classify = (body: string) => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "RateLimit" }) }).pipe(Effect.provide(fixedResponse(body, { status: 400 })))
yield* classify("Request rate increased too quickly") yield* classify('{"type":"error","error":{"type":"too_many_requests"}}') yield* classify('{"type":"error","error":{"code":"rate_limit_exceeded"}}') }), )
it.effect("does not let server codes override a 4xx rejection", () => Effect.gen(function* () { const classify = (body: string) => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "InvalidRequest" }) }).pipe(Effect.provide(fixedResponse(body, { status: 400 })))
yield* classify('{"code":"resource_exhausted"}') yield* classify('{"error":{"type":"server_error","message":"Upstream request failed: Model is unavailable."}}') }), )
it.effect("returns complete diagnostics for rate limits", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "RateLimit", retryAfterMs: 0, rateLimit: { retryAfterMs: 0 }, http: { url: "https://provider.test/v1/chat?api_key=secret&key=secret&debug=1", status: 429, headers: { "retry-after-ms": "0", "x-request-id": "req_123", "x-api-key": "secret", }, }, }) expect(error.reason.body).toBe("rate limited") }).pipe( Effect.provide( fixedResponse("rate limited", { status: 429, headers: { "retry-after-ms": "0", "x-request-id": "req_123", "x-api-key": "secret" }, }), ), ), )
it.effect("preserves configured header names in diagnostics", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason.http?.headers["x-safe"]).toBe("response-secret") }).pipe( Effect.provide(fixedResponse("bad", { status: 400, headers: { "x-safe": "response-secret" } })), Effect.provideService(Headers.CurrentRedactedNames, ["x-safe"]), ), )
it.effect("extracts OpenAI-style rate-limit diagnostics", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "RateLimit" }) expect(error.reason._tag === "RateLimit" ? error.reason.rateLimit : undefined).toEqual({ retryAfterMs: 0, limit: { requests: "500", tokens: "30000" }, remaining: { requests: "499", tokens: "29900" }, reset: { requests: "1s", tokens: "10s" }, }) }).pipe( Effect.provide( fixedResponse("rate limited", { status: 429, headers: { "retry-after-ms": "0", "x-ratelimit-limit-requests": "500", "x-ratelimit-limit-tokens": "30000", "x-ratelimit-remaining-requests": "499", "x-ratelimit-remaining-tokens": "29900", "x-ratelimit-reset-requests": "1s", "x-ratelimit-reset-tokens": "10s", }, }), ), ), )
it.effect("extracts Anthropic-style rate-limit diagnostics", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "RateLimit" }) expect(error.reason._tag === "RateLimit" ? error.reason.rateLimit : undefined).toEqual({ retryAfterMs: 0, limit: { requests: "100", "input-tokens": "10000" }, remaining: { requests: "12", "input-tokens": "9000" }, reset: { requests: "2026-05-06T12:00:00Z", "input-tokens": "2026-05-06T12:00:10Z" }, }) }).pipe( Effect.provide( fixedResponse("rate limited", { status: 429, headers: { "retry-after-ms": "0", "anthropic-ratelimit-requests-limit": "100", "anthropic-ratelimit-requests-remaining": "12", "anthropic-ratelimit-requests-reset": "2026-05-06T12:00:00Z", "anthropic-ratelimit-input-tokens-limit": "10000", "anthropic-ratelimit-input-tokens-remaining": "9000", "anthropic-ratelimit-input-tokens-reset": "2026-05-06T12:00:10Z", }, }), ), ), )
it.effect("returns provider status failures without retrying", () => Effect.gen(function* () { const attempts = yield* Ref.make(0) const error = yield* Effect.gen(function* () { const executor = yield* RequestExecutor.Service return yield* executor.execute(request).pipe(Effect.flip) }).pipe( Effect.provide( dynamicResponse((input) => Effect.gen(function* () { const attempt = yield* Ref.getAndUpdate(attempts, (value) => value + 1) return attempt === 0 ? input.respond("busy", { status: 503, headers: { "retry-after-ms": "0" } }) : input.respond("ok", { status: 200 }) }), ), ), )
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "ProviderInternal" }) expect(error.reason.http?.status).toBe(503) expect(yield* Ref.get(attempts)).toBe(1) }), )
it.effect("marks 504 and 529 status responses as provider-internal", () => Effect.gen(function* () { const failWith = (status: number) => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "ProviderInternal" }) expect(error.reason.http?.status).toBe(status) }).pipe( Effect.provide( fixedResponse("provider failure", { status, headers: { "retry-after-ms": "0" }, }), ), )
yield* failWith(504) yield* failWith(529) }), )
it.effect("preserves large authentication error bodies", () => Effect.gen(function* () { const attempts = yield* Ref.make(0) const error = yield* Effect.gen(function* () { const executor = yield* RequestExecutor.Service return yield* executor.execute(request).pipe(Effect.flip) }).pipe( Effect.provide( dynamicResponse((input) => Effect.gen(function* () { const attempt = yield* Ref.getAndUpdate(attempts, (value) => value + 1) return attempt === 0 ? input.respond("x".repeat(20_000), { status: 401 }) : input.respond("should not retry", { status: 200 }) }), ), ), )
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "Authentication" }) expect(error.reason.body).toHaveLength(20_000) expect(yield* Ref.get(attempts)).toBe(1) }), )
it.effect("preserves response body fields", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(request).pipe(Effect.flip)
expectAIError(error) expect(error.reason.body).toBe('{"error":{"message":"bad","key":"body-secret","detail":"api_key=query-secret"}}') }).pipe( Effect.provide( fixedResponse('{"error":{"message":"bad","key":"body-secret","detail":"api_key=query-secret"}}', { status: 400, }), ), ), )
it.effect("preserves echoed request values in response bodies", () => Effect.gen(function* () { const executor = yield* RequestExecutor.Service const error = yield* executor.execute(secretRequest).pipe(Effect.flip)
expectAIError(error) expect(error.reason.body).toBe("provider echoed query-secret-123 and authorization header-secret-456") }).pipe( Effect.provide( fixedResponse("provider echoed query-secret-123 and authorization header-secret-456", { status: 400 }), ), ), )
it.effect("does not re-execute after a successful response reaches stream parsing", () => Effect.gen(function* () { const attempts = yield* Ref.make(0) const model = route.with({ endpoint: { baseURL: "https://api.openai.test/v1" } }).model({ id: "gpt-4o-mini" }) const error = yield* LLMClient.generate(LLM.request({ model, prompt: "Say hello." })).pipe( Effect.provide( dynamicResponse((input) => Ref.update(attempts, (value) => value + 1).pipe( Effect.as( input.respond( sseRaw( `data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}`, "data: not-json", ), { headers: { "content-type": "text/event-stream" } }, ), ), ), ), ), Effect.flip, )
expectAIError(error) expect(error.reason).toMatchObject({ _tag: "InvalidProviderOutput" }) expect(error.reason.body).toBe("not-json") expect(error.reason.cause).toBeDefined() expect(error.reason.http).toMatchObject({ status: 200, headers: { "content-type": "text/event-stream" } }) expect(yield* Ref.get(attempts)).toBe(1) }), )})
describe("WebSocket channel execution", () => { const model = configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responses("gpt-4.1-mini") const request = LLM.request({ model, prompt: "Say hello." }) const frames = [ JSON.stringify({ type: "response.output_item.added", item: { type: "message", id: "msg_1" } }), JSON.stringify({ type: "response.output_text.delta", item_id: "msg_1", delta: "Hi" }), JSON.stringify({ type: "response.completed", response: { id: "resp_1" } }), ]
it.effect("preserves close reasons and native event causes without fabricated HTTP metadata", () => Effect.gen(function* () { class TestSocket extends EventTarget { readyState = globalThis.WebSocket.OPEN send() {} close() {} } const socket = new TestSocket() const connection = yield* WebSocketTransport.fromWebSocket( // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion socket as unknown as globalThis.WebSocket, { url: "wss://provider.test/responses", headers: Headers.empty }, ) const event = new CloseEvent("close", { code: 1011, reason: "upstream trace: req_close" }) socket.dispatchEvent(event) const error = yield* connection.messages.pipe(Stream.runDrain, Effect.flip)
expect(error.reason).toMatchObject({ _tag: "Transport", code: "1011", phase: "close" }) expect(error.message).toBe("WebSocket closed with code 1011") expect(error.reason.body).toBe(event.reason) expect(error.reason.cause).toBe(event) expect(error.reason.http).toBeUndefined() yield* connection.close }), )
it.effect("preserves opening event errors and native send exceptions", () => Effect.gen(function* () { const cause = new Error("native send failed") class TestSocket extends EventTarget { readyState = globalThis.WebSocket.CONNECTING send() { throw cause } close() {} } const socket = new TestSocket() const open = WebSocketTransport.fromWebSocket( // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion socket as unknown as globalThis.WebSocket, { url: "wss://provider.test/responses", headers: Headers.empty }, ) const fiber = yield* open.pipe(Effect.flip, Effect.forkChild({ startImmediately: true })) const event = new ErrorEvent("error", { message: "handshake rejected", error: cause }) socket.dispatchEvent(event) const error = yield* Fiber.join(fiber) expect(error.reason.cause).toBe(cause) expect(error.message).toContain("handshake rejected") expect(error.reason.http).toBeUndefined()
socket.readyState = globalThis.WebSocket.OPEN const connection = yield* open const sent = yield* connection.sendText("create").pipe(Effect.flip) expect(sent.reason.cause).toBe(cause) expect(sent.message).toBe(cause.message) yield* connection.close }), )
it.effect("preserves raw driver failures and known upgrade metadata", () => Effect.gen(function* () { const cause = new Error("driver validation failed") const frame = '{ "error": "failed", "trace": "original" }' const http = new HttpContext({ url: "https://provider.test/responses", status: 101, headers: { upgrade: "websocket" }, }) const executor = WebSocketTransport.makeDirect({ open: () => Effect.succeed({ http, sendText: () => Effect.void, messages: Stream.make(frame), close: Effect.void, }), }) const execution = yield* executor.execute({ id: "exchange_error", connect: { url: "wss://provider.test/responses", headers: Headers.empty }, fallback: () => Stream.empty, driver: { create: () => Effect.succeed({ message: "create", mode: "full" }), observe: () => Effect.succeed({ type: "provider-failure", error: new AIError({ reason: new InvalidProviderOutputError({ message: "Driver failed", cause, body: "narrowed", }), }), }), }, }) const error = yield* execution.frames.pipe(Stream.runDrain, Effect.flip)
expect(error.message).toBe("Driver failed") expect(error.reason.body).toBe(frame) expect(error.reason.cause).toBe(cause) expect(error.reason.http).toBe(http) expect(execution.http).toBe(http) }), )
it.effect("retains diagnostic fields when annotating transport delivery", () => Effect.gen(function* () { const cause = new Error("connection closed") const executor = WebSocketTransport.makeDirect({ open: () => Effect.succeed({ sendText: () => Effect.void, messages: Stream.fail( new AIError({ reason: new TransportError({ message: "Socket closed", transport: "websocket", operation: "read", phase: "close", recovery: "retry-full", body: "server close detail", cause, }), }), ), close: Effect.void, }), }) const execution = yield* executor.execute({ id: "exchange_closed", connect: { url: "wss://provider.test/responses", headers: Headers.empty }, fallback: () => Stream.empty, driver: { create: () => Effect.succeed({ message: "create", mode: "full" }), observe: (_create, frame) => Effect.succeed({ type: "frame", frame }), }, }) const error = yield* execution.frames.pipe(Stream.runDrain, Effect.flip)
expect(error.message).toBe("Socket closed") expect(error.reason.body).toBe("server close detail") expect(error.reason.cause).toBe(cause) expect(error.reason).toMatchObject({ phase: "close", delivery: "ambiguous", recovery: "retry-full" }) expect(error.reason.http).toBeUndefined() }), )
it.effect("runs a channel driver through the direct executor", () => Effect.gen(function* () { const sent = yield* Ref.make("") const closed = yield* Ref.make(false) const observed = yield* Ref.make(0) const webSocket = WebSocketTransport.makeDirect({ open: () => Effect.succeed({ sendText: (message) => Ref.set(sent, message), messages: Stream.make("one", "done", "late"), close: Ref.set(closed, true), }), }) const received = yield* Effect.scoped( Effect.gen(function* () { const execution = yield* webSocket.execute({ id: "exchange_1", connect: { url: "wss://api.openai.test/v1/responses", headers: Headers.empty }, fallback: () => Stream.empty, driver: { create: () => Effect.succeed({ message: "create", mode: "full" }), observe: (_create, frame) => Ref.update(observed, (value) => value + 1).pipe( Effect.as( frame === "done" ? { type: "completed" as const, frame } : { type: "frame" as const, frame }, ), ), }, }) return yield* Stream.runCollect(execution.frames) }), )
expect(Array.from(received)).toEqual(["one", "done"]) expect(yield* Ref.get(sent)).toBe("create") expect(yield* Ref.get(observed)).toBe(2) expect(yield* Ref.get(closed)).toBe(true) }), )
it.effect("rejects a closed socket before attempting to send", () => Effect.gen(function* () { class ClosedBeforeSend extends EventTarget { readyState = globalThis.WebSocket.OPEN sends = 0 send() { this.sends++ } close() {} } const socket = new ClosedBeforeSend() const connection = yield* WebSocketTransport.fromWebSocket( // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion socket as unknown as globalThis.WebSocket, { url: "wss://api.openai.test/v1/responses", headers: Headers.empty }, ) socket.readyState = globalThis.WebSocket.CLOSED
const error = yield* connection.sendText("create").pipe(Effect.flip)
expect(error.reason).toMatchObject({ _tag: "Transport", phase: "send", delivery: "not-sent" }) expect(socket.sends).toBe(0) yield* connection.close }), )
it.effect("uses HTTP when no per-call WebSocket executor is provided", () => Effect.gen(function* () { const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(sseEvents(...frames))))
expect(response.text).toBe("Hi") }), )
it.effect("commits channel execution only after complete consumption", () => Effect.gen(function* () { const commits = yield* Ref.make(0) const executor = (input: Stream.Stream<string, AIError>): WebSocketChannelExecutor => ({ execute: () => Effect.succeed({ frames: input, complete: Ref.update(commits, (value) => value + 1), }), })
const response = yield* LLMClient.generate(request, { webSocket: executor(Stream.fromArray(frames)), }).pipe(Effect.provide(fixedResponse(""))) expect(response.text).toBe("Hi") expect(yield* Ref.get(commits)).toBe(1)
yield* LLMClient.generate(request, { webSocket: executor(Stream.make("not-json")) }).pipe( Effect.provide(fixedResponse("")), Effect.flip, ) expect(yield* Ref.get(commits)).toBe(1)
yield* LLMClient.stream(request, { webSocket: executor(Stream.fromArray(frames)) }).pipe( Stream.take(1), Stream.runDrain, Effect.provide(fixedResponse("")), ) expect(yield* Ref.get(commits)).toBe(1) }), )
it.effect("does not commit interrupted channel execution", () => Effect.gen(function* () { const commits = yield* Ref.make(0) const started = yield* Deferred.make<void>() const executor: WebSocketChannelExecutor = { execute: () => Effect.succeed({ frames: Stream.fromEffect( Deferred.succeed(started, undefined).pipe( Effect.as(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })), ), ).pipe(Stream.concat(Stream.never)), complete: Ref.update(commits, (value) => value + 1), }), } const fiber = yield* LLMClient.stream(request, { webSocket: executor }).pipe( Stream.runDrain, Effect.provide(fixedResponse("")), Effect.forkChild({ startImmediately: true }), )
yield* Deferred.await(started) yield* Fiber.interrupt(fiber)
expect(yield* Ref.get(commits)).toBe(0) }), )})