diff --git a/packages/ai/src/protocols/openai-responses-channel.ts b/packages/ai/src/protocols/openai-responses-channel.ts index f8287d7def6..89f848eaca8 100644 --- a/packages/ai/src/protocols/openai-responses-channel.ts +++ b/packages/ai/src/protocols/openai-responses-channel.ts @@ -1,6 +1,6 @@ import { AIError, TransportReason } from "../schema/index.js" import type { ChannelCheckpoint, ChannelObservation, WebSocketChannelDriver } from "../route/transport/index.js" -import { Effect, Schema } from "effect" +import { Effect, Option, Schema } from "effect" import * as ProviderShared from "./shared.js" import { OpenResponses } from "./open-responses.js" @@ -47,6 +47,11 @@ const canonical = (value: unknown): string => { .join(",")}}` } +const json = (value: unknown) => { + if (typeof value !== "string") return value + return Option.getOrElse(Schema.decodeUnknownOption(ProviderShared.Json)(value), () => value) +} + const comparable = (value: unknown) => { if (!ProviderShared.isRecord(value)) return value if (value.type === "message" && value.role === "assistant") @@ -60,12 +65,11 @@ const comparable = (value: unknown) => { type: value.type, call_id: value.call_id, name: value.name, - arguments: value.arguments, + arguments: json(value.arguments), } if (value.type === "reasoning") return { type: value.type, - ...(value.id === undefined ? {} : { id: value.id }), summary: value.summary, encrypted_content: value.encrypted_content, } @@ -87,7 +91,8 @@ const incremental = ( if (canonical(invariant(request)) !== canonical(invariant(checkpoint.request))) return undefined const baseline = [...previousInput, ...checkpoint.output] if (input.length <= baseline.length) return undefined - if (!baseline.every((item, index) => canonical(comparable(item)) === canonical(input[index]))) return undefined + if (!baseline.every((item, index) => canonical(comparable(item)) === canonical(comparable(input[index])))) + return undefined return input.slice(baseline.length) } diff --git a/packages/ai/test/provider/openai-responses.test.ts b/packages/ai/test/provider/openai-responses.test.ts index 6910dd2a2a2..bb0036db5bc 100644 --- a/packages/ai/test/provider/openai-responses.test.ts +++ b/packages/ai/test/provider/openai-responses.test.ts @@ -413,7 +413,7 @@ describe("OpenAI Responses route", () => { status: "completed", call_id: "call_1", name: "weather", - arguments: '{"city":"Paris"}', + arguments: '{ "city": "Paris" }', }, }), ) @@ -484,6 +484,53 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("continues store-false reasoning without replaying the output-only item ID", () => + Effect.gen(function* () { + const firstInput = [{ role: "user", content: [{ type: "input_text", text: "Think" }] }] + const request = { type: "response.create", model: "gpt-5.2", store: false, input: firstInput } + const first = continuationDriver(request) + const create = yield* first.create(undefined) + yield* first.observe( + create, + ProviderShared.encodeJson({ + type: "response.output_item.done", + item: { + type: "reasoning", + id: "rs_1", + summary: [{ type: "summary_text", text: "Thought" }], + encrypted_content: "encrypted", + }, + }), + ) + const saved = checkpoint( + yield* first.observe( + create, + ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_1" } }), + ), + ) + const next = continuationDriver({ + ...request, + input: [ + ...firstInput, + { + type: "reasoning", + summary: [{ type: "summary_text", text: "Thought" }], + encrypted_content: "encrypted", + }, + { role: "user", content: [{ type: "input_text", text: "Continue" }] }, + ], + }) + + const continued = yield* next.create(saved) + + expect(continued.mode).toBe("incremental") + expect(ProviderShared.decodeJson(continued.message)).toMatchObject({ + previous_response_id: "resp_1", + input: [{ role: "user", content: [{ type: "input_text", text: "Continue" }] }], + }) + }), + ) + it.effect("uses a full request when any non-input invariant changes", () => Effect.gen(function* () { const request = { diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index 5865ed25ba6..a4d6ba2b180 100644 --- a/packages/core/src/session/model-transport.ts +++ b/packages/core/src/session/model-transport.ts @@ -12,7 +12,7 @@ import { } from "@opencode-ai/ai/route" import { AIError, TransportReason, type TransportOperation } from "@opencode-ai/ai" import { Hash } from "@opencode-ai/util/hash" -import { Cause, Clock, Context, Effect, Fiber, Layer, Queue, Scope, Semaphore, Stream } from "effect" +import { Cause, Clock, Context, Effect, Fiber, Layer, Metric, Queue, Scope, Semaphore, Stream } from "effect" import { Socket } from "effect/unstable/socket" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { SessionSchema } from "./schema.js" @@ -21,6 +21,12 @@ import { webSocketConstructor } from "../effect/app-node-platform.js" const ROTATE_AFTER_MS = 55 * 60 * 1000 const INBOUND_CAPACITY = 128 const IDLE_TIMEOUT = "5 minutes" +const events = Metric.counter("opencode_session_websocket_events_total", { + description: "Session WebSocket lifecycle events", + incremental: true, +}) +const metric = (event: string, attributes: Record = {}) => + Metric.update(events.pipe(Metric.withAttributes({ event, ...attributes })), 1) type Delivery = "queued" | "connecting" | "ready" | "send-attempted" | "provider-observed" | "terminal" @@ -144,6 +150,7 @@ export const makeLayer = (connector: WebSocketConnector) => }), ), ) + yield* metric("close") }) const poison = Effect.fn("SessionModelTransport.poison")(function* ( @@ -156,6 +163,11 @@ export const makeLayer = (connector: WebSocketConnector) => if (channel.closing) return channel.closing = true if (channel.active) Queue.failCauseUnsafe(channel.active.queue, Cause.fail(error)) + yield* metric( + error.reason._tag === "Transport" && error.reason.code === "queue-overflow" + ? "queue_overflow" + : "protocol_failure", + ) yield* channel.connection.close }) @@ -166,7 +178,9 @@ export const makeLayer = (connector: WebSocketConnector) => ) { return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { - const connection = yield* restore(connector.open(exchange.connect)) + const connection = yield* restore( + connector.open(exchange.connect).pipe(Effect.withSpan("SessionModelTransport.connect")), + ) if (owner.closed) { yield* connection.close return yield* transportError("open", "Session WebSocket owner closed while connecting", { @@ -237,6 +251,7 @@ export const makeLayer = (connector: WebSocketConnector) => sessionTransport: "websocket", phase: "connect", }) + yield* metric("connect") return channel }), ) @@ -278,6 +293,8 @@ export const makeLayer = (connector: WebSocketConnector) => phase: "connect", reason: rotation, }) + yield* metric("rotation", { reason: rotation }) + yield* metric("reconnect") yield* closeChannel(owner, current) } @@ -287,6 +304,7 @@ export const makeLayer = (connector: WebSocketConnector) => sessionTransport: "websocket", phase: "connect", }) + if (owner.channel) yield* metric("reuse") const channel = owner.channel ? owner.channel : yield* open(owner, exchange, key).pipe( @@ -298,7 +316,11 @@ export const makeLayer = (connector: WebSocketConnector) => phase: "connect", delivery: "not-sent", code: error.reason._tag === "Transport" ? error.reason.code : error.reason._tag, - }).pipe(Effect.andThen(Effect.succeed(undefined))), + }).pipe( + Effect.andThen(metric("connect_failure")), + Effect.andThen(metric("fallback")), + Effect.andThen(Effect.succeed(undefined)), + ), ), ) if (!channel) return fallback(exchange) @@ -318,6 +340,7 @@ export const makeLayer = (connector: WebSocketConnector) => channel.active = active lifecycle.delivery = "send-attempted" const sent = yield* channel.connection.sendText(create.message).pipe( + Effect.withSpan("SessionModelTransport.send"), Effect.onInterrupt(() => closeChannel(owner, channel)), Effect.result, ) @@ -325,9 +348,14 @@ export const makeLayer = (connector: WebSocketConnector) => const failure = sent.failure const notSent = failure.reason._tag === "Transport" && failure.reason.delivery === "not-sent" yield* closeChannel(owner, channel) - if (notSent) return fallback(exchange) + if (notSent) { + yield* metric("fallback") + return fallback(exchange) + } + yield* metric("ambiguous_delivery") return yield* annotate(failure, { phase: "send", delivery: "ambiguous" }) } + yield* metric("send") let terminal: ChannelObservation | undefined const token = {} @@ -353,6 +381,7 @@ export const makeLayer = (connector: WebSocketConnector) => terminal = observation lifecycle.delivery = "terminal" staged = observation.type === "completed" ? observation.checkpoint : undefined + if (staged) channel.pending = { token, checkpoint: staged } if (observation.type !== "completed" || !staged) channel.checkpoint = undefined }), ), @@ -364,11 +393,13 @@ export const makeLayer = (connector: WebSocketConnector) => const pending = yield* Queue.size(active.queue) yield* Queue.shutdown(active.queue) if (terminal && pending === 0) { - if (staged) channel.pending = { token, checkpoint: staged } + yield* metric("terminal", { type: terminal.type }) + if (terminal.type === "rejected") yield* metric("rejection", { recovery: terminal.recovery }) if (terminal.type === "rejected" && terminal.recovery === "rotate-and-retry-full") yield* closeChannel(owner, channel) return } + yield* metric("cancellation") channel.checkpoint = undefined channel.pending = undefined const error = terminal diff --git a/packages/core/test/lib/websocket-server.ts b/packages/core/test/lib/websocket-server.ts new file mode 100644 index 00000000000..f1897759c0f --- /dev/null +++ b/packages/core/test/lib/websocket-server.ts @@ -0,0 +1,71 @@ +import { Buffer } from "node:buffer" +import { Effect } from "effect" + +interface ConnectionData { + readonly id: number +} + +export interface WebSocketServerState { + readonly headers: Array> + readonly messages: string[] + opens: number + closes: number + pongs: number +} + +export interface WebSocketServerFixture { + readonly url: string + readonly state: WebSocketServerState +} + +export interface WebSocketServerOptions { + readonly upgrade?: (request: Request) => boolean + readonly open?: (socket: Bun.ServerWebSocket) => void + readonly message?: (socket: Bun.ServerWebSocket, message: string | Buffer) => void +} + +export const makeWebSocketServer = (options: WebSocketServerOptions = {}) => + Effect.acquireRelease( + Effect.sync(() => { + const state: WebSocketServerState = { headers: [], messages: [], opens: 0, closes: 0, pongs: 0 } + let connection = 0 + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request, server) { + state.headers.push(Object.fromEntries(request.headers.entries())) + if ((options.upgrade?.(request) ?? true) && server.upgrade(request, { data: { id: connection++ } })) + return undefined + return new Response("WebSocket upgrade required", { + status: 426, + headers: { "x-upgrade-rejected": "true" }, + }) + }, + websocket: { + open(socket) { + state.opens++ + options.open?.(socket) + }, + message(socket, message) { + const text = typeof message === "string" ? message : message.toString() + state.messages.push(text) + options.message?.(socket, message) + }, + close() { + state.closes++ + }, + pong() { + state.pongs++ + }, + }, + }) + return { + server, + fixture: { + url: `${server.url.toString().replace(/^http/, "ws")}responses`, + state, + } satisfies WebSocketServerFixture, + } + }), + ({ server }) => Effect.promise(() => server.stop(true)), + ).pipe(Effect.map((item) => item.fixture)) diff --git a/packages/core/test/session-model-transport-live.test.ts b/packages/core/test/session-model-transport-live.test.ts new file mode 100644 index 00000000000..a3574a4f0da --- /dev/null +++ b/packages/core/test/session-model-transport-live.test.ts @@ -0,0 +1,318 @@ +import { describe, expect, test } from "bun:test" +import { NodeSocket } from "@effect/platform-node" +import { AIError, LLM, Message } from "@opencode-ai/ai" +import { + LLMClient, + RequestExecutor, + WebSocketTransport, + type ChannelObservation, + type WebSocketChannelExchange, +} from "@opencode-ai/ai/route" +import { configure } from "@opencode-ai/ai/providers/openai" +import { SessionModelTransport } from "@opencode-ai/core/session/model-transport" +import { Session } from "@opencode-ai/schema/session" +import { Effect, Fiber, Layer, Stream } from "effect" +import { Headers } from "effect/unstable/http" +import { Socket } from "effect/unstable/socket" +import { makeWebSocketServer, type WebSocketServerFixture, type WebSocketServerOptions } from "./lib/websocket-server" + +const sessionID = Session.ID.make("ses_live_websocket") + +const exchange = (server: WebSocketServerFixture, id: string): WebSocketChannelExchange => ({ + id, + connect: { + url: server.url, + headers: Headers.fromInput({ authorization: "Bearer local-secret", "x-handshake": "visible" }), + }, + fallback: () => Stream.die("Unexpected HTTP fallback"), + driver: { + create: () => Effect.succeed({ message: id, mode: "full" }), + observe: (_create, frame): Effect.Effect => + Effect.succeed({ type: "completed", frame }), + }, +}) + +const withServer = ( + options: WebSocketServerOptions, + effect: (server: WebSocketServerFixture) => Effect.Effect, +) => + Effect.runPromise( + Effect.gen(function* () { + const constructor = yield* Socket.WebSocketConstructor + const server = yield* makeWebSocketServer(options) + return yield* effect(server).pipe( + Effect.provide( + SessionModelTransport.makeLayer({ + open: (input) => + WebSocketTransport.open(input).pipe(Effect.provideService(Socket.WebSocketConstructor, constructor)), + }), + ), + ) + }).pipe(Effect.scoped, Effect.provide(NodeSocket.layerWebSocketConstructorWS)), + ) + +const collect = (transport: SessionModelTransport.Interface, item: WebSocketChannelExchange) => + Effect.gen(function* () { + const execution = yield* transport.bind(sessionID).execute(item) + return Array.from(yield* Stream.runCollect(execution.frames.pipe(Stream.onEnd(execution.complete)))) + }).pipe(Effect.scoped) + +const waitFor = (predicate: () => boolean, remaining = 100): Effect.Effect => { + if (predicate()) return Effect.void + if (remaining === 0) return Effect.die("Timed out waiting for local WebSocket server") + return Effect.sleep("5 millis").pipe(Effect.andThen(Effect.suspend(() => waitFor(predicate, remaining - 1)))) +} + +describe("SessionModelTransport local WebSocket server", () => { + test("continues a real Responses connection with only the appended input", async () => { + const requests: Array> = [] + await withServer( + { + message: (socket, message) => { + const request = JSON.parse(message.toString()) + requests.push(request) + const index = requests.length + const id = `msg_${index}` + const text = index === 1 ? "Hello" : "Brief" + socket.send(JSON.stringify({ type: "response.created", response: { id: `resp_${index}` } })) + socket.send(JSON.stringify({ type: "response.output_item.added", item: { type: "message", id } })) + socket.send(JSON.stringify({ type: "response.output_text.delta", item_id: id, delta: text })) + socket.send(JSON.stringify({ type: "response.output_text.done", item_id: id, text })) + socket.send( + JSON.stringify({ + type: "response.output_item.done", + item: { + type: "message", + id, + status: "completed", + role: "assistant", + content: [{ type: "output_text", text }], + }, + }), + ) + socket.send(JSON.stringify({ type: "response.completed", response: { id: `resp_${index}` } })) + }, + }, + (server) => + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + const executor = transport.bind(sessionID) + const model = configure({ + baseURL: server.url.replace(/^ws/, "http").replace(/responses$/, ""), + apiKey: "local", + }).responses("gpt-5.2") + const client = LLMClient.Service + const layer = LLMClient.layer.pipe( + Layer.provide( + Layer.succeed( + RequestExecutor.Service, + RequestExecutor.Service.of({ execute: () => Effect.die("Unexpected HTTP request") }), + ), + ), + ) + + const first = yield* client + .use((llm) => llm.generate(LLM.request({ model, prompt: "First" }), { webSocket: executor })) + .pipe(Effect.provide(layer)) + const second = yield* client + .use((llm) => + llm.generate( + LLM.request({ + model, + messages: [Message.user("First"), Message.assistant("Hello"), Message.user("Be brief")], + }), + { webSocket: executor }, + ), + ) + .pipe(Effect.provide(layer)) + + expect(first.text).toBe("Hello") + expect(second.text).toBe("Brief") + expect(server.state.opens).toBe(1) + expect(requests[1]).toMatchObject({ + previous_response_id: "resp_1", + input: [{ role: "user", content: [{ type: "input_text", text: "Be brief" }] }], + }) + }), + ) + }) + + test("clears a rejected continuation and keeps one provider request per attempt", async () => { + const requests: Array> = [] + await withServer( + { + message: (socket, message) => { + requests.push(JSON.parse(message.toString())) + const index = requests.length + if (index === 2) { + socket.send( + JSON.stringify({ + type: "error", + error: { code: "previous_response_not_found", message: "Missing response" }, + }), + ) + return + } + const id = `msg_${index}` + const text = index === 1 ? "Hello" : "Recovered" + socket.send(JSON.stringify({ type: "response.created", response: { id: `resp_${index}` } })) + socket.send(JSON.stringify({ type: "response.output_item.added", item: { type: "message", id } })) + socket.send(JSON.stringify({ type: "response.output_text.delta", item_id: id, delta: text })) + socket.send(JSON.stringify({ type: "response.output_text.done", item_id: id, text })) + socket.send( + JSON.stringify({ + type: "response.output_item.done", + item: { + type: "message", + id, + role: "assistant", + content: [{ type: "output_text", text }], + }, + }), + ) + socket.send(JSON.stringify({ type: "response.completed", response: { id: `resp_${index}` } })) + }, + }, + (server) => + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + const executor = transport.bind(sessionID) + const model = configure({ + baseURL: server.url.replace(/^ws/, "http").replace(/responses$/, ""), + apiKey: "local", + }).responses("gpt-5.2") + const layer = LLMClient.layer.pipe( + Layer.provide( + Layer.succeed( + RequestExecutor.Service, + RequestExecutor.Service.of({ execute: () => Effect.die("Unexpected HTTP request") }), + ), + ), + ) + const request = LLM.request({ + model, + messages: [Message.user("First"), Message.assistant("Hello"), Message.user("Continue")], + }) + + yield* LLMClient.Service.use((llm) => + llm.generate(LLM.request({ model, prompt: "First" }), { webSocket: executor }), + ).pipe(Effect.provide(layer)) + const rejected = yield* LLMClient.Service.use((llm) => llm.generate(request, { webSocket: executor })).pipe( + Effect.provide(layer), + Effect.flip, + ) + const recovered = yield* LLMClient.Service.use((llm) => llm.generate(request, { webSocket: executor })).pipe( + Effect.provide(layer), + ) + + expect(rejected.reason).toMatchObject({ + _tag: "Transport", + delivery: "rejected", + recovery: "retry-full", + }) + expect(recovered.text).toBe("Recovered") + expect(requests).toHaveLength(3) + expect(requests[1]).toHaveProperty("previous_response_id", "resp_1") + expect(requests[2]).not.toHaveProperty("previous_response_id") + expect(server.state.opens).toBe(1) + }), + ) + }) + + // The browser-compatible client surface cannot originate ping frames, so the server sends one and verifies pong. + test("reuses one real connection with handshake headers and ping/pong", async () => { + await withServer( + { + open: (socket) => socket.ping("health"), + message: (socket, message) => socket.send(`completed:${message.toString()}`), + }, + (server) => + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + + expect(yield* collect(transport, exchange(server, "first"))).toEqual(["completed:first"]) + expect(yield* collect(transport, exchange(server, "second"))).toEqual(["completed:second"]) + yield* waitFor(() => server.state.pongs === 1) + + expect(server.state.opens).toBe(1) + expect(server.state.messages).toEqual(["first", "second"]) + expect(server.state.headers[0]).toMatchObject({ + authorization: "Bearer local-secret", + "x-handshake": "visible", + }) + }), + ) + }) + + test("closes a real active connection on cancellation", async () => { + await withServer({ message: () => {} }, (server) => + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + const running = yield* collect(transport, exchange(server, "blocked")).pipe( + Effect.forkChild({ startImmediately: true }), + ) + yield* waitFor(() => server.state.messages.length === 1) + + yield* Fiber.interrupt(running) + yield* waitFor(() => server.state.closes === 1) + + expect(server.state.messages).toEqual(["blocked"]) + }), + ) + }) + + test("poisons a real connection after an unsupported binary frame", async () => { + await withServer({ message: (socket) => socket.sendBinary(new Uint8Array([1, 2, 3])) }, (server) => + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + + const result = yield* Effect.result(collect(transport, exchange(server, "binary"))) + yield* waitFor(() => server.state.closes === 1) + + expect(result).toMatchObject({ + _tag: "Failure", + failure: { reason: { _tag: "Transport", code: "message", delivery: "accepted" } }, + }) + }), + ) + }) + + test("closes a real connection after an oversized frame", async () => { + await withServer({ message: (socket) => socket.send("x".repeat(16 * 1024 * 1024 + 1)) }, (server) => + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + + const result = yield* Effect.result(collect(transport, exchange(server, "oversized"))) + yield* waitFor(() => server.state.closes === 1) + + expect(result).toMatchObject({ + _tag: "Failure", + failure: { reason: { _tag: "Transport", code: "message-too-large", delivery: "ambiguous" } }, + }) + }), + ) + }) + + // Effect's browser-compatible constructor does not expose upgrade response bodies or headers. + // The real 426 fixture therefore pins the observable contract: a not-sent connect failure and one HTTP fallback. + test("falls back once after a real rejected upgrade", async () => { + let fallbacks = 0 + await withServer({ upgrade: () => false }, (server) => + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + const item = exchange(server, "fallback") + const result = yield* collect(transport, { + ...item, + fallback: () => { + fallbacks++ + return Stream.make("http") + }, + }) + + expect(result).toEqual(["http"]) + expect(fallbacks).toBe(1) + expect(server.state.opens).toBe(0) + }), + ) + }) +}) diff --git a/packages/core/test/session-model-transport.test.ts b/packages/core/test/session-model-transport.test.ts index 876fd89dfce..b4177068d7a 100644 --- a/packages/core/test/session-model-transport.test.ts +++ b/packages/core/test/session-model-transport.test.ts @@ -8,7 +8,7 @@ import type { } from "@opencode-ai/ai/route" import { SessionModelTransport } from "@opencode-ai/core/session/model-transport" import { Session } from "@opencode-ai/schema/session" -import { Deferred, Effect, Fiber, Queue, Stream } from "effect" +import { Deferred, Effect, Fiber, Metric, Queue, Stream } from "effect" import { TestClock } from "effect/testing" import { Headers } from "effect/unstable/http" @@ -717,4 +717,28 @@ describe("SessionModelTransport", () => { expect(fixture.connections[0]?.closed).toBe(1) }) + + test("records metadata-only lifecycle metrics", async () => { + const fixture = automatic() + + await run( + fixture.connector, + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + const executor = transport.bind(session) + yield* collect(executor, exchange("first", { headers: { authorization: "secret-one" } })) + yield* collect(executor, exchange("second", { headers: { authorization: "secret-one" } })) + yield* collect(executor, exchange("third", { headers: { authorization: "secret-two" } })) + + const snapshots = yield* Metric.snapshot + const lifecycle = snapshots.filter((item) => item.id === "opencode_session_websocket_events_total") + const names = new Set(lifecycle.map((item) => item.attributes?.event)) + expect(Array.from(names)).toEqual( + expect.arrayContaining(["connect", "reuse", "rotation", "reconnect", "send", "terminal"]), + ) + expect(JSON.stringify(lifecycle)).not.toContain("secret-one") + expect(JSON.stringify(lifecycle)).not.toContain("secret-two") + }), + ) + }) }) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 43ee1debfc9..1e6d502de8a 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -32,8 +32,11 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { Session } from "@opencode-ai/core/session" import { Snapshot } from "@opencode-ai/core/snapshot" import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionContext } from "@opencode-ai/core/session/context" import { SessionInbox } from "@opencode-ai/core/session/inbox" import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionModelRequest } from "@opencode-ai/core/session/model-request" +import { SessionModelTransport } from "@opencode-ai/core/session/model-transport" import { Money } from "@opencode-ai/schema/money" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" @@ -60,6 +63,7 @@ import { SessionTable, } from "@opencode-ai/core/session/sql" import { InstructionEntry } from "@opencode-ai/core/session/instruction-entry" +import { InstructionState } from "@opencode-ai/core/session/instruction-state" import { SessionStore } from "@opencode-ai/core/session/store" import { Instructions } from "@opencode-ai/core/instructions/index" import { InstructionBuiltIns } from "@opencode-ai/core/instructions/builtins" @@ -72,6 +76,7 @@ import { Location } from "@opencode-ai/core/location" import { Provider } from "@opencode-ai/core/provider" import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Schema, Scope, Stream } from "effect" import { TestClock } from "effect/testing" +import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http" import { asc, desc, eq } from "drizzle-orm" import { testEffect } from "./lib/effect" import { permissionLayer } from "./lib/permission" @@ -132,6 +137,15 @@ const testLLM = TestLLM.layer({ }), }) const client = TestLLM.clientLayer +const closedTransports: Session.ID[] = [] +const modelTransport = Layer.succeed( + SessionModelTransport.Service, + SessionModelTransport.Service.of({ + bind: () => ({ execute: () => Effect.die("Unexpected WebSocket execution") }), + close: (sessionID) => Effect.sync(() => closedTransports.push(sessionID)), + closeAll: Effect.void, + }), +) const model = LanguageModel.make({ id: "fake-model", provider: "fake", route: OpenAIChat.route }) const defaultSystem = PROMPT_DEFAULT const replacementModel = LanguageModel.make({ id: "replacement", provider: "fake", route: OpenAIChat.route }) @@ -373,6 +387,7 @@ const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [ [Config.node, config], [McpInstructions.node, mcpInstructions], [PluginSupervisor.node, pluginSupervisor], + [SessionModelTransport.node, modelTransport], ]) const execution = Layer.effect( SessionExecution.Service, @@ -426,6 +441,8 @@ const it = testEffect( ReferenceInstructions.node, Config.node, Snapshot.node, + SessionContext.node, + SessionModelRequest.node, SessionRunnerLLM.node, SessionExecution.node, Session.node, @@ -445,6 +462,7 @@ const it = testEffect( [SessionExecution.node, execution], [Config.node, config], [PluginSupervisor.node, pluginSupervisor], + [SessionModelTransport.node, modelTransport], ], ).pipe(Layer.provideMerge(testLLM)), ) @@ -491,6 +509,7 @@ const setup = Effect.gen(function* () { requests = (yield* TestLLM.Service).requests authorizations.length = 0 executions.length = 0 + closedTransports.length = 0 systemBaseline = "Initial context" systemRemoved = false systemUnavailable = false @@ -961,6 +980,48 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("forces HTTP and triggers active request and response hooks once", () => + Effect.gen(function* () { + yield* setup + const hooks = yield* PluginHooks.Service + let requestTriggers = 0 + let responseTriggers = 0 + yield* hooks.register("session", "http.request", (event) => + Effect.sync(() => { + requestTriggers++ + event.request.headers.set("x-request-hook", "active") + }), + ) + yield* hooks.register("session", "http.response", (event) => + Effect.sync(() => { + responseTriggers++ + event.response.headers.set("x-response-hook", "active") + }), + ) + const context = yield* SessionContext.Service + const modelRequests = yield* SessionModelRequest.Service + const selected = yield* context.select(sessionID) + const database = yield* Database.Service + const bus = yield* Bus.Service + yield* InstructionState.prepare(database.db, bus, selected.instructions, sessionID) + const prepared = yield* modelRequests.prepare({ + context: yield* context.load(selected), + step: 1, + }) + const http = prepared.options.http ?? (yield* Effect.die("Expected Session HTTP middleware")) + + const response = yield* http(HttpClientRequest.post("https://provider.test/responses"), (request) => { + expect(request.headers["x-request-hook"]).toBe("active") + return Effect.succeed(HttpClientResponse.fromWeb(request, new Response("network"))) + }) + + expect(prepared.webSocketEligible).toBe(false) + expect(response.headers["x-response-hook"]).toBe("active") + expect(requestTriggers).toBe(1) + expect(responseTriggers).toBe(1) + }), + ) + it.effect("executes a tool renamed by a session context hook", () => Effect.gen(function* () { const session = yield* setup @@ -1274,6 +1335,7 @@ describe("SessionRunnerLLM", () => { expect((yield* session.get(sessionID)).location.directory).toBe(AbsolutePath.make("/moved")) expect(yield* session.inbox(sessionID)).toEqual([]) expect(requests).toEqual([]) + expect(closedTransports).toEqual([sessionID]) expect( (yield* db .select({ type: EventTable.type }) diff --git a/packages/www/content/docs/(Configure)/providers.mdx b/packages/www/content/docs/(Configure)/providers.mdx index 218b67614c7..b8b33eda117 100644 --- a/packages/www/content/docs/(Configure)/providers.mdx +++ b/packages/www/content/docs/(Configure)/providers.mdx @@ -95,6 +95,26 @@ models, and connection continue to apply: `settings` is package-specific. A field only has an effect when the selected package supports it. +### Experimental OpenAI Responses WebSocket + +OpenAI Responses can reuse one Session-bound WebSocket connection across sequential model calls. This transport is +experimental and disabled by default. To test it, set the flag before starting or restarting the OpenCode service: + +```bash +export OPENCODE_EXPERIMENTAL_OPENAI_RESPONSES_WEBSOCKET=true +opencode2 service restart +``` + +The flag applies only to the `openai` provider using the native `openai-responses` route. Azure, xAI, +OpenAI-compatible providers, and other protocols continue to use HTTP. A Session HTTP plugin hook also forces HTTP so +the hook is never bypassed. + +OpenCode reuses a connection only while its endpoint and effective handshake headers remain unchanged. It rotates +connections before the provider lifetime limit and never replays a request after ambiguous delivery. HTTP fallback is +limited to failures proven to occur before the provider received the request. + +Unset the variable or set it to `false`, then restart the service, to return to HTTP-only execution. + ### Headers and body Headers and body fields can be set at provider, model, or variant scope: