Compare commits

..
25 changed files with 445 additions and 365 deletions
@@ -18,7 +18,6 @@ const WebSocketResponseCreate = Schema.StructWithRest(Schema.Struct({ type: Sche
])
const decodeMessage = ProviderShared.validateWith(Schema.decodeUnknownEffect(WebSocketResponseCreate))
const encodeMessage = Schema.encodeSync(Schema.fromJsonString(WebSocketResponseCreate))
const decodeEvent = Schema.decodeUnknownEffect(OpenResponses.protocol.stream.event)
export interface Options {
readonly id: string
@@ -27,6 +26,7 @@ export interface Options {
readonly enabled?: (url: string) => boolean
readonly url?: (url: string) => string
readonly headers?: (headers: Headers.Headers) => Headers.Headers
readonly continuation?: OpenResponsesContinuation.Shape
}
export interface Prepared {
@@ -60,7 +60,7 @@ const driver = (options: Options, body: string): WebSocketChannelDriver => {
}),
observe: (_create, frame) =>
Effect.gen(function* () {
const event = yield* decodeEvent(frame).pipe(
const event = yield* OpenResponses.decodeChannelEvent(frame).pipe(
Effect.mapError((cause) =>
ProviderShared.eventError(options.id, `Invalid ${options.name} WebSocket event`, frame, cause),
),
@@ -163,6 +163,7 @@ export const transport = <Body>(options: Options): Transport<Body, Prepared, str
request: create.request,
message: create.message,
base,
continuation: options.continuation,
}),
}
})
@@ -6,7 +6,6 @@ import { OpenResponses } from "./open-responses.js"
const PROTOCOL = "open-responses.websocket.v1"
const VERSION = 1
const decodeEvent = Schema.decodeUnknownEffect(OpenResponses.protocol.stream.event)
interface CheckpointValue {
readonly version: typeof VERSION
@@ -15,12 +14,19 @@ interface CheckpointValue {
readonly output: ReadonlyArray<unknown>
}
/**
* Fields to send next to `previous_response_id` on an incremental step, or undefined to send the step in full.
* Whether omitted fields carry over from the continued response is provider behavior the route must know.
*/
export type Shape = (request: Readonly<Record<string, unknown>>) => Readonly<Record<string, unknown>> | undefined
export interface DriverInput {
readonly id: string
readonly name: string
readonly request: Readonly<Record<string, unknown>>
readonly message: string
readonly base: WebSocketChannelDriver
readonly continuation?: Shape
}
const checkpointValue = (checkpoint: ChannelCheckpoint | undefined): CheckpointValue | undefined => {
@@ -127,22 +133,26 @@ const rejected = (
export const driver = (input: DriverInput): WebSocketChannelDriver => {
const { previous_response_id: _previousResponseID, ...request } = input.request
const shape = input.continuation ?? ((fields: Readonly<Record<string, unknown>>) => fields)
let output: OpenResponses.StreamItem[] = []
return {
create: (checkpoint) =>
Effect.sync(() => {
output = []
const previous = checkpointValue(checkpoint)
const delta = previous ? incremental(request, previous) : undefined
if (!previous || !delta) return { message: ProviderShared.encodeJson(request), mode: "full" as const }
// Ask the route first: diffing the whole history is wasted when it declines the continuation.
const fields = previous ? shape(request) : undefined
const delta = previous && fields ? incremental(request, previous) : undefined
if (!previous || !fields || !delta)
return { message: ProviderShared.encodeJson(request), mode: "full" as const }
return {
message: ProviderShared.encodeJson({ ...request, input: delta, previous_response_id: previous.responseID }),
message: ProviderShared.encodeJson({ ...fields, input: delta, previous_response_id: previous.responseID }),
mode: "incremental" as const,
}
}),
observe: (create, frame) =>
Effect.gen(function* () {
const event = yield* decodeEvent(frame).pipe(
const event = yield* OpenResponses.decodeChannelEvent(frame).pipe(
Effect.mapError((cause) =>
ProviderShared.eventError(input.id, `Invalid ${input.name} WebSocket event`, frame, cause),
),
@@ -195,4 +205,4 @@ export const driver = (input: DriverInput): WebSocketChannelDriver => {
}
}
export const OpenResponsesContinuation = { driver } as const
export * as OpenResponsesContinuation from "./open-responses-continuation.js"
@@ -405,6 +405,24 @@ export const Event = Schema.StructWithRest(
export type Event = Schema.Schema.Type<typeof Event>
export type NormalizedEvent = Event & { readonly item?: OutputItem | null }
const decodeEventValue = Schema.decodeUnknownEffect(Event)
const decodeFrame = Schema.decodeUnknownEffect(ProviderShared.Json)
/**
* Decodes one WebSocket frame. xAI answers a rejected `response.create` with `{ "error": { "message", "type" } }` and no
* event type; that envelope reads as an error event so the failure classifies instead of failing decoding.
*/
export const decodeChannelEvent = (frame: string) =>
decodeFrame(frame).pipe(
Effect.flatMap((value) =>
decodeEventValue(
ProviderShared.isRecord(value) && value.type === undefined && ProviderShared.isRecord(value.error)
? { ...value, type: "error" }
: value,
),
),
)
export interface ProviderAdapter {
readonly id: string
readonly name: string
+4
View File
@@ -41,6 +41,10 @@ const responsesRoute = Route.make({
id: "openai-responses",
name: "xAI Responses",
rotateAfterMs: RESPONSES_WEBSOCKET_ROTATE_AFTER_MS,
// xAI continues a chain only from stored responses: with `store: false` (the route default) `previous_response_id`
// fails with "Response with id=… not found", so those steps are sent in full over the reused connection. It also
// rejects `instructions` next to `previous_response_id` and keeps the instructions of the response it continues.
continuation: ({ instructions: _instructions, ...request }) => (request.store === false ? undefined : request),
}),
defaults: { providerOptions: { store: false, include: ["reasoning.encrypted_content"] } },
})
@@ -90,7 +90,11 @@ const classifyingChannelDriver = (message: string): WebSocketChannelDriver => {
}
}
const continuationDriver = (request: Readonly<Record<string, unknown>>, base = baseChannelDriver) => {
const continuationDriver = (
request: Readonly<Record<string, unknown>>,
base = baseChannelDriver,
continuation?: OpenResponsesContinuation.Shape,
) => {
const message = ProviderShared.encodeJson(request)
return OpenResponsesContinuation.driver({
id: "openai-responses",
@@ -98,6 +102,7 @@ const continuationDriver = (request: Readonly<Record<string, unknown>>, base = b
request,
message,
base: base(message),
continuation,
})
}
@@ -921,6 +926,58 @@ describe("OpenAI Responses route", () => {
type: "provider-failure",
error: { reason: { _tag: "InvalidRequest", classification: "context-overflow" } },
})
// A retryable failure stays one: the runner retries it, and the transport has already dropped the
// checkpoint, so that retry is a full send. xAI reports every rejection this way.
const internal = ProviderShared.encodeJson({
type: "error",
error: { type: "api_error", message: "gRPC error: Response with id=resp_1 not found" },
})
expect(yield* second.observe(yield* second.create(saved), internal)).toMatchObject({
type: "provider-failure",
error: { reason: { _tag: "ProviderInternal" } },
})
}),
)
it.effect("shapes the incremental send with the route continuation", () =>
Effect.gen(function* () {
const firstRequest = {
type: "response.create",
model: "grok-4.6",
store: true,
instructions: "You are terse.",
input: [{ role: "user", content: [{ type: "input_text", text: "First" }] }],
}
const secondRequest = {
...firstRequest,
input: [...firstRequest.input, { role: "user", content: [{ type: "input_text", text: "Second" }] }],
}
const saved = checkpoint(
yield* continuationDriver(firstRequest).observe(
yield* continuationDriver(firstRequest).create(undefined),
ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_1" } }),
),
)
const trimmed = yield* continuationDriver(
secondRequest,
baseChannelDriver,
({ instructions: _, ...rest }) => rest,
).create(saved)
expect(trimmed.mode).toBe("incremental")
expect(JSON.parse(trimmed.message)).toEqual({
type: "response.create",
model: "grok-4.6",
store: true,
previous_response_id: "resp_1",
input: [{ role: "user", content: [{ type: "input_text", text: "Second" }] }],
})
// Declining the continuation sends the step in full and never sends a previous_response_id.
const declined = yield* continuationDriver(secondRequest, baseChannelDriver, () => undefined).create(saved)
expect(declined.mode).toBe("full")
expect(JSON.parse(declined.message)).toEqual(secondRequest)
}),
)
+110 -2
View File
@@ -1,11 +1,18 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Effect, Layer, Stream } from "effect"
import { LLM, LLMEvent, Message } from "../../src/index.js"
import { XAI } from "../../src/providers.js"
import { OpenResponses } from "../../src/protocols/open-responses.js"
import { OpenAIResponses } from "../../src/protocols/openai-responses.js"
import * as ProviderShared from "../../src/protocols/shared.js"
import { XAIResponses } from "../../src/protocols/xai-responses.js"
import { LLMClient } from "../../src/route.js"
import {
LLMClient,
RequestExecutor,
WebSocketTransport,
type ChannelCheckpoint,
type WebSocketChannelDriver,
} from "../../src/route.js"
import { compileRequest } from "../../src/route/client.js"
import { it } from "../lib/effect.js"
import { fixedResponse } from "../lib/http.js"
@@ -13,6 +20,35 @@ import { sseEvents } from "../lib/sse.js"
const model = XAI.configure({ apiKey: "test", baseURL: "https://api.x.ai/v1" }).responses("grok-4.6")
/** Runs a request through the WebSocket transport and hands back its channel driver; the HTTP fallback answers. */
const channelDriver = (request: ReturnType<typeof LLM.request>) =>
Effect.gen(function* () {
let driver: WebSocketChannelDriver | undefined
yield* LLMClient.generate(request, {
webSocket: {
execute: (exchange) =>
Effect.sync(() => {
driver = exchange.driver
return { frames: exchange.fallback(), complete: Effect.void }
}),
},
}).pipe(Effect.provide(fixedResponse(sseEvents({ type: "response.completed", response: { id: "http" } }))))
if (!driver) throw new Error("Expected a WebSocket channel driver")
return driver
})
const completed = (driver: WebSocketChannelDriver, id: string) =>
Effect.gen(function* () {
const create = yield* driver.create(undefined)
yield* driver.observe(create, ProviderShared.encodeJson({ type: "response.created", response: { id } }))
const observation = yield* driver.observe(
create,
ProviderShared.encodeJson({ type: "response.completed", response: { id } }),
)
if (observation.type !== "completed" || !observation.checkpoint) throw new Error("Expected a checkpoint")
return observation.checkpoint
})
describe("xAI Responses route", () => {
it.effect("composes the Open Responses baseline with xAI extensions", () =>
Effect.gen(function* () {
@@ -162,6 +198,78 @@ describe("xAI Responses route", () => {
}),
)
it.effect("classifies xAI's untyped WebSocket error envelope", () =>
Effect.gen(function* () {
// xAI answers a rejected response.create with an error envelope that carries no event type.
const envelope = ProviderShared.encodeJson({
error: {
message:
'Request validation error: {"code":"400","error":"Argument not supported: instructions and previous_response_id together"}',
type: "api_error",
},
})
const webSocket = WebSocketTransport.makeDirect({
open: () =>
Effect.succeed({ sendText: () => Effect.void, messages: Stream.make(envelope), close: Effect.void }),
})
const error = yield* LLMClient.generate(LLM.request({ model, prompt: "Hello" }), { webSocket }).pipe(
Effect.provide(
LLMClient.layer.pipe(
Layer.provide(
Layer.succeed(
RequestExecutor.Service,
RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
),
),
),
),
Effect.flip,
)
expect(error.reason._tag).toBe("ProviderInternal")
expect(error.message).toContain("Argument not supported: instructions and previous_response_id together")
expect(error.reason.body).toBe(envelope)
}),
)
it.effect("continues stored responses without instructions and sends unstored steps in full", () =>
Effect.gen(function* () {
const step = (store: boolean, ...prompts: string[]) =>
LLM.request({
model,
system: "You are terse.",
messages: prompts.map((prompt) => Message.user(prompt)),
providerOptions: { store },
})
const send = (store: boolean, checkpoint: ChannelCheckpoint) =>
channelDriver(step(store, "First", "Second")).pipe(Effect.flatMap((driver) => driver.create(checkpoint)))
const stored = yield* send(true, yield* completed(yield* channelDriver(step(true, "First")), "resp_1"))
expect(stored.mode).toBe("incremental")
expect(JSON.parse(stored.message)).toEqual({
type: "response.create",
model: "grok-4.6",
store: true,
include: ["reasoning.encrypted_content"],
previous_response_id: "resp_1",
input: [{ role: "user", content: [{ type: "input_text", text: "Second" }] }],
})
// The connection cache only serves stored responses, so the default store: false never chains.
const unstored = yield* send(false, yield* completed(yield* channelDriver(step(false, "First")), "resp_1"))
expect(unstored.mode).toBe("full")
expect(JSON.parse(unstored.message)).toMatchObject({
instructions: "You are terse.",
store: false,
input: [
{ role: "user", content: [{ type: "input_text", text: "First" }] },
{ role: "user", content: [{ type: "input_text", text: "Second" }] },
],
})
expect(JSON.parse(unstored.message).previous_response_id).toBeUndefined()
}),
)
it.effect("parses xAI hosted tool items", () =>
Effect.gen(function* () {
const item = { type: "x_search_call", id: "x_search_1", status: "completed", action: { query: "news" } }
@@ -0,0 +1,74 @@
import { expect, test } from "@playwright/test"
import { sessionID, setupTimeline, userMessage } from "../performance/timeline-stability/fixture"
test("keeps a submitted prompt in place while its optimistic rows are measured", async ({ page }) => {
await setupTimeline(page, { messages: [userMessage()], seedHistory: true })
const release = Promise.withResolvers<void>()
await page.route(`**/api/session/${sessionID}/prompt`, async (route) => {
if (route.request().method() !== "POST") return route.fallback()
await release.promise
return route.fallback()
})
const editor = page.locator('[data-component="composer"]').getByRole("textbox")
await expect(editor).toBeEditable()
await editor.fill("Observe optimistic prompt spacing.")
await expect
.poll(() =>
page.locator("[data-timeline-virtual-content]").evaluate((element) => {
const root = element.parentElement!
return root.scrollHeight - root.clientHeight - root.scrollTop
}),
)
.toBe(0)
const observation = await page.evaluateHandle(() => {
const frames: { prompt?: number; working: boolean }[] = []
let frame = 0
const sample = () => {
const prompt = [...document.querySelectorAll<HTMLElement>('[data-timeline-row="UserMessage"]')].find((row) =>
row.textContent?.includes("Observe optimistic prompt spacing."),
)
frames.push({
...(prompt ? { prompt: prompt.getBoundingClientRect().y } : {}),
working: !!document.querySelector('[data-component="session-working"]'),
})
frame = requestAnimationFrame(sample)
}
frame = requestAnimationFrame(sample)
return {
stop: () => {
cancelAnimationFrame(frame)
return frames
},
}
})
const requested = page.waitForRequest(
(request) => request.method() === "POST" && new URL(request.url()).pathname === `/api/session/${sessionID}/prompt`,
)
try {
await editor.press("Enter")
await requested
const prompt = page
.locator('[data-timeline-row="UserMessage"]')
.filter({ hasText: "Observe optimistic prompt spacing." })
await expect(prompt).toBeInViewport()
await expect(page.locator('[data-component="session-working"]')).toBeVisible()
await expect
.poll(() =>
page.locator("[data-timeline-virtual-content]").evaluate((element) => {
const root = element.parentElement!
return root.scrollHeight - root.clientHeight - root.scrollTop
}),
)
.toBe(0)
const frames = await observation.evaluate((value) => value.stop())
expect(frames.some((frame) => frame.working && frame.prompt === undefined)).toBe(false)
const positions = frames.flatMap((frame) => (frame.prompt === undefined ? [] : [frame.prompt]))
expect(positions.length).toBeGreaterThan(0)
expect(new Set(positions).size).toBe(1)
} finally {
release.resolve()
await observation.dispose()
}
})
@@ -122,7 +122,15 @@ test("renders compaction progress, summary, and outcome in order", async ({ page
)
await expect(compaction.getByRole("heading", { name: "Checkpoint" })).toBeVisible()
await expect(compaction).toContainText("Streamed implementation details.")
await expect(compaction.getByRole("status").getByLabel("Compacting", { exact: true })).toBeVisible()
const running = compaction.getByRole("status").getByLabel("Compacting", { exact: true })
await expect(running).toBeVisible()
await expect
.poll(async () => {
const summary = await compaction.locator('[data-component="text-part"]').boundingBox()
const status = await running.boundingBox()
return !!summary && !!status && status.y >= summary.y + summary.height
})
.toBe(true)
await expect(compaction.getByText("Session compacted", { exact: true })).toHaveCount(0)
await timeline.send(
+16 -14
View File
@@ -88,8 +88,12 @@ export function createComposerSubmit(input: ComposerSubmitInput) {
if (value.mode === "normal" && !command) {
session.handoff?.set(handoffMessage(value))
const optimisticBusy = !input.adapter.working()
if (optimisticBusy) session.data.session.setStatus(session.id, "running")
const sending = sendPrompt(session, value, input.adapter.controls().model.selection.trackSessionCommit).then(
if (optimisticBusy && input.adapter.kind === "new-session")
session.data.session.setStatus(session.id, "running")
const sending = sendPrompt(session, value, input.adapter.controls().model.selection.trackSessionCommit, () => {
if (optimisticBusy && input.adapter.kind === "active-session")
session.data.session.setStatus(session.id, "running")
}).then(
() => ({ ok: true as const }),
(error) => ({ ok: false as const, error }),
)
@@ -122,15 +126,9 @@ export function createComposerSubmit(input: ComposerSubmitInput) {
if (command) {
clearSubmission(input, submission)
// Commands always steer: the server applies a command's configured
// agent and model immediately at admission, so queueing one would
// reconfigure the turn it is supposed to wait behind.
void sendCommand(
session,
{ ...value, delivery: "steer" },
command,
input.adapter.controls().model.selection.trackSessionCommit,
).catch((error) => failSubmission(input, session, "command", error, restore, value.id))
void sendCommand(session, value, command, input.adapter.controls().model.selection.trackSessionCommit).catch(
(error) => failSubmission(input, session, "command", error, restore, value.id),
)
return
}
} finally {
@@ -322,7 +320,8 @@ async function sendCommand(
track?: ModelSelection["trackSessionCommit"],
) {
const request = await buildSubmissionRequest(session, value)
await applySelection(session, value.selection, track)
// Like queued prompts, queued commands must not apply the composer's selection to active work.
if (value.delivery === "steer") await applySelection(session, value.selection, track)
await session.api.command({
sessionID: session.id,
command: command.command,
@@ -359,7 +358,8 @@ async function applySelection(
async function sendPrompt(
session: ComposerSession,
value: ComposerSubmission,
track?: ModelSelection["trackSessionCommit"],
track: ModelSelection["trackSessionCommit"] | undefined,
onAdmit: () => void,
) {
const request = await buildSubmissionRequest(session, value)
// Switching agent or model reconfigures the session immediately, and with it
@@ -389,7 +389,9 @@ async function sendPrompt(
},
},
}
await session.data.session.prompt(admission).catch(() => session.data.session.prompt(admission))
const sending = session.data.session.prompt(admission).catch(() => session.data.session.prompt(admission))
onAdmit()
await sending
}
async function buildSubmissionRequest(session: ComposerSession, value: ComposerSubmission) {
@@ -515,6 +515,19 @@ export function createTimelineVirtualizer(input: Input) {
<div
ref={(value) => {
element = value
if (row()._tag !== "UserMessage" || !addedKeys.has(rowProps.rowKey) || !input.pinned() || coldPending)
return
// The optimistic row can paint before ResizeObserver corrects the tail estimates.
// Measure the mounted tail and pin it in this render's microtask instead.
queueMicrotask(() => {
if (!input.pinned() || !virtualContent?.isConnected) return
virtualizer.elementsCache.forEach((item) => {
if (item.isConnected) virtualizer.resizeItem(virtualizer.indexFromElement(item), item.offsetHeight)
})
virtualizer.resizeItem(item().index, element.offsetHeight)
virtualContent.style.height = `${virtualizer.getTotalSize()}px`
virtualizer.scrollToEnd()
})
}}
data-index={item().index}
style={{ "min-height": ready() ? undefined : `${initialItem.size}px` }}
-3
View File
@@ -369,9 +369,6 @@ ultimate source of truth.
- [x] `atob` and `btoa` with forgiving-base64 decoding and WebIDL string conversion; invalid input throws an Error
named `InvalidCharacterError`, since there is no `DOMException`.
- [x] `crypto.randomUUID()`.
- [x] `structuredClone` over the data model: objects, arrays with holes, Date, RegExp (`lastIndex` reset), Map, Set,
URL, URLSearchParams, and Errors (name, message, and cause only); shared references stay shared within one
clone; functions, promises, and tool references throw an Error named `DataCloneError`.
- [ ] `crypto.getRandomValues` and `crypto.subtle`, `TextEncoder`/`TextDecoder`, and `Blob`: these need a binary
value type, which the JSON-like data model does not have yet.
+1 -1
View File
@@ -138,6 +138,6 @@ const copy = (value: unknown, label: string, mode: Mode, depth: number, seen: Se
// Own data property regardless of the target's prototype, so a "__proto__" key on a host object or
// array never reaches the Object.prototype setter.
export const define = (target: object, key: string, value: unknown): void => {
const define = (target: object, key: string, value: unknown): void => {
Object.defineProperty(target, key, { value, enumerable: true, writable: true, configurable: true })
}
+1 -2
View File
@@ -11,7 +11,7 @@ import { regexpGlobal } from "../stdlib/regexp.js"
import { stringGlobal } from "../stdlib/string.js"
import { uriGlobal, urlGlobal, urlSearchParamsGlobal } from "../stdlib/url.js"
import { coercion, errorConstructors } from "../stdlib/value.js"
import { atobGlobal, btoaGlobal, cryptoGlobal, structuredCloneGlobal } from "../stdlib/web.js"
import { atobGlobal, btoaGlobal, cryptoGlobal } from "../stdlib/web.js"
import { ToolReference } from "../tool-runtime.js"
import { errorGlobal } from "./errors.js"
import { HostFunction } from "./host.js"
@@ -75,6 +75,5 @@ export const globals = <R>(host: Host<R>): ReadonlyArray<readonly [string, unkno
["atob", atobGlobal],
["btoa", btoaGlobal],
["crypto", cryptoGlobal],
["structuredClone", structuredCloneGlobal],
...[...errorConstructors].map((name) => [name, errorGlobal(name, host.runner)] as const),
]
+2 -58
View File
@@ -1,9 +1,6 @@
import { define, type SafeObject } from "../data.js"
import { HostNamespace, sync } from "../interpreter/host.js"
import { type AstNode, InterpreterRuntimeError } from "../interpreter/model.js"
import { describeValue, isRuntimeReference } from "../interpreter/references.js"
import { Values } from "../values.js"
import { coerceToString, createErrorValue, errorBrandName } from "./value.js"
import { InterpreterRuntimeError } from "../interpreter/model.js"
import { coerceToString } from "./value.js"
// WebIDL DOMString conversion: a missing argument is a TypeError, anything else stringifies.
const base64 = (name: "atob" | "btoa") =>
@@ -25,56 +22,3 @@ export const btoaGlobal = base64("btoa")
export const cryptoGlobal = new HostNamespace("crypto", {
randomUUID: sync("crypto.randomUUID", () => crypto.randomUUID()),
})
// HTML structured clone over the data model: wrappers are copied, shared references stay shared within
// one clone, Errors keep only name, message, and cause, and RegExp lastIndex resets like the spec.
const cloneValue = (value: unknown, seen: Map<object, unknown>, node: AstNode): unknown => {
if (value === null || typeof value !== "object") return value
if (value instanceof Values.Promise || (isRuntimeReference(value) && !Values.isValue(value))) {
throw new InterpreterRuntimeError(`${describeValue(value)} could not be cloned.`, node).as("DataCloneError")
}
const existing = seen.get(value)
if (existing !== undefined) return existing
const remember = <T extends object>(copied: T): T => {
seen.set(value, copied)
return copied
}
if (value instanceof Values.Date) return remember(new Values.Date(value.time))
if (value instanceof Values.RegExp) return remember(new Values.RegExp(value.regex.source, value.regex.flags))
if (value instanceof Values.URL) return remember(new Values.URL(new URL(value.url.href)))
if (value instanceof Values.URLSearchParams) {
return remember(new Values.URLSearchParams(new URLSearchParams(value.params)))
}
if (value instanceof Values.Map) {
const copied = remember(new Values.Map())
for (const [key, item] of value.map) copied.map.set(cloneValue(key, seen, node), cloneValue(item, seen, node))
return copied
}
if (value instanceof Values.Set) {
const copied = remember(new Values.Set())
for (const item of value.set) copied.set.add(cloneValue(item, seen, node))
return copied
}
if (Array.isArray(value)) {
const copied = remember(new Array<unknown>(value.length))
for (const [key, item] of Object.entries(value)) define(copied, key, cloneValue(item, seen, node))
return copied
}
const brand = errorBrandName(value)
if (brand !== undefined) {
const error = value as { name?: unknown; message?: unknown; cause?: unknown }
const copied = remember(createErrorValue(brand, coerceToString(error.message)))
if (Object.hasOwn(value, "cause")) copied.cause = cloneValue(error.cause, seen, node)
return copied
}
const copied = remember(Object.create(null) as SafeObject)
for (const [key, item] of Object.entries(value)) define(copied, key, cloneValue(item, seen, node))
return copied
}
export const structuredCloneGlobal = sync("structuredClone", (args, node) => {
if (args.length === 0) {
throw new InterpreterRuntimeError("structuredClone requires a value to clone.", node).as("TypeError")
}
return cloneValue(args[0], new Map(), node)
})
@@ -1,245 +0,0 @@
/**
* Portions adapted from web-platform-tests at revision 863077959ca8c1a7ceecfbe2534b75d2527b9013:
* - html/webappapis/structured-clone/structured-clone-battery-of-tests.js
*
* Copyright © web-platform-tests contributors. Governed by the 3-Clause BSD license in LICENSE.wpt.
*
* The battery's `check(description, input, compare)` shape and its `compare_*` helpers are kept, run
* inside the interpreter. Ported: primitives, Array/Object of primitives, Date, RegExp, Error, sparse
* arrays, identical (shared) property values, and the index-property-plus-length object. Not portable:
* boxed primitives, BigInt, Blob/File/ImageData/ArrayBuffer/typed arrays (no binary values), circular
* references (rejected at insertion here), property descriptors and prototype properties (no
* defineProperty or prototypes), and the throwing-getter case (no getters). `assert_throws_dom` for
* `DataCloneError` becomes an `error.name` check.
*/
import { describe, expect, test } from "bun:test"
import { Effect } from "effect"
import { CodeMode } from "../src/index.js"
const value = async (code: string) => {
const result = await Effect.runPromise(CodeMode.execute({ code, tools: {} }))
if (!result.ok) throw new Error(`expected success, got ${result.error.kind}: ${result.error.message}`)
return result.value
}
// The WPT harness, minus async: assertions push a failure description instead of throwing so one run
// reports every failing check.
const harness = `
const failures = []
const assert_equals = (a, b, m) => { if (!Object.is(a, b) && !(a !== a && b !== b)) failures.push((m ?? "") + ": " + String(a) + " !== " + String(b)) }
const assert_not_equals = (a, b, m) => { if (a === b) failures.push((m ?? "") + ": unexpectedly identical") }
const assert_true = (a, m) => { if (a !== true) failures.push((m ?? "") + ": not true") }
const assert_false = (a, m) => { if (a !== false) failures.push((m ?? "") + ": not false") }
let current = ""
function check(description, input, callback) {
current = description
const newInput = typeof input === "function" ? input() : input
const copy = structuredClone(newInput)
const before = failures.length
callback(copy, newInput)
for (let i = before; i < failures.length; i++) failures[i] = description + " — " + failures[i]
}
function compare_primitive(actual, input) { assert_equals(actual, input) }
function compare_Array(callback) {
return function (actual, input) {
assert_true(Array.isArray(actual), "instanceof Array")
assert_not_equals(actual, input)
assert_equals(actual.length, input.length, "length")
callback(actual, input)
}
}
function compare_Object(callback) {
return function (actual, input) {
assert_true(actual instanceof Object, "instanceof Object")
assert_false(Array.isArray(actual), "instanceof Array")
assert_not_equals(actual, input)
callback(actual, input)
}
}
function enumerate_props(compare_func) {
return function (actual, input) { for (const x in input) compare_func(actual[x], input[x]) }
}
`
describe("structuredClone WPT battery", () => {
test("primitives, and arrays and objects of primitives", async () => {
expect(
await value(`
${harness}
check('primitive undefined', undefined, compare_primitive)
check('primitive null', null, compare_primitive)
check('primitive true', true, compare_primitive)
check('primitive false', false, compare_primitive)
check('primitive string, empty string', '', compare_primitive)
check('primitive string, lone high surrogate', '\\uD800', compare_primitive)
check('primitive string, lone low surrogate', '\\uDC00', compare_primitive)
check('primitive string, NUL', '\\u0000', compare_primitive)
check('primitive string, astral character', '\\uDBFF\\uDFFD', compare_primitive)
check('primitive number, 0.2', 0.2, compare_primitive)
check('primitive number, 0', 0, compare_primitive)
check('primitive number, -0', -0, compare_primitive)
check('primitive number, NaN', NaN, compare_primitive)
check('primitive number, Infinity', Infinity, compare_primitive)
check('primitive number, -Infinity', -Infinity, compare_primitive)
check('primitive number, 9007199254740992', 9007199254740992, compare_primitive)
check('primitive number, -9007199254740992', -9007199254740992, compare_primitive)
check('primitive number, 9007199254740994', 9007199254740994, compare_primitive)
check('primitive number, -9007199254740994', -9007199254740994, compare_primitive)
check('Array primitives', [undefined, null, true, false, '', '\\uD800', '\\uDC00', '\\u0000', '\\uDBFF\\uDFFD',
0.2, 0, -0, NaN, Infinity, -Infinity, 9007199254740992, -9007199254740992, 9007199254740994, -9007199254740994],
compare_Array(enumerate_props(compare_primitive)))
check('Object primitives', { 'undefined': undefined, 'null': null, 'true': true, 'false': false, 'empty': '',
'high surrogate': '\\uD800', 'low surrogate': '\\uDC00', 'nul': '\\u0000', 'astral': '\\uDBFF\\uDFFD',
'0.2': 0.2, '0': 0, '-0': -0, 'NaN': NaN, 'Infinity': Infinity, '-Infinity': -Infinity,
'9007199254740992': 9007199254740992, '-9007199254740992': -9007199254740992,
'9007199254740994': 9007199254740994, '-9007199254740994': -9007199254740994 },
compare_Object(enumerate_props(compare_primitive)))
return failures
`),
).toEqual([])
})
test("Date", async () => {
expect(
await value(`
${harness}
function compare_Date(actual, input) {
assert_true(actual instanceof Date, 'instanceof Date')
assert_equals(Number(actual), Number(input), 'converted to primitive')
assert_not_equals(actual, input)
}
check('Date 0', new Date(0), compare_Date)
check('Date -0', new Date(-0), compare_Date)
check('Date -8.64e15', new Date(-8.64e15), compare_Date)
check('Date 8.64e15', new Date(8.64e15), compare_Date)
check('Array Date objects', [new Date(0), new Date(-0), new Date(-8.64e15), new Date(8.64e15)],
compare_Array(enumerate_props(compare_Date)))
check('Object Date objects', { '0': new Date(0), '-0': new Date(-0), '-8.64e15': new Date(-8.64e15), '8.64e15': new Date(8.64e15) },
compare_Object(enumerate_props(compare_Date)))
return failures
`),
).toEqual([])
})
test("RegExp: flags copied, lastIndex reset, source escaped", async () => {
expect(
await value(`
${harness}
function compare_RegExp(expected_source) {
return function (actual, input) {
assert_true(actual instanceof RegExp, 'instanceof RegExp')
assert_equals(actual.global, input.global, 'global')
assert_equals(actual.ignoreCase, input.ignoreCase, 'ignoreCase')
assert_equals(actual.multiline, input.multiline, 'multiline')
assert_equals(actual.source, expected_source, 'source')
assert_equals(actual.sticky, input.sticky, 'sticky')
assert_equals(actual.unicode, input.unicode, 'unicode')
assert_equals(actual.lastIndex, 0, 'lastIndex')
assert_not_equals(actual, input)
}
}
function func_RegExp_flags_lastIndex() {
const r = /foo/gim
r.lastIndex = 2
return r
}
function func_RegExp_sticky() { return new RegExp('foo', 'y') }
function func_RegExp_unicode() { return new RegExp('foo', 'u') }
check('RegExp flags and lastIndex', func_RegExp_flags_lastIndex, compare_RegExp('foo'))
check('RegExp sticky flag', func_RegExp_sticky, compare_RegExp('foo'))
check('RegExp unicode flag', func_RegExp_unicode, compare_RegExp('foo'))
check('RegExp empty', new RegExp(''), compare_RegExp('(?:)'))
check('RegExp slash', new RegExp('/'), compare_RegExp('\\\\/'))
check('RegExp new line', new RegExp('\\n'), compare_RegExp('\\\\n'))
check('Array RegExp object, RegExp flags and lastIndex', [func_RegExp_flags_lastIndex()], compare_Array(enumerate_props(compare_RegExp('foo'))))
check('Array RegExp object, RegExp sticky flag', function () { return [func_RegExp_sticky()] }, compare_Array(enumerate_props(compare_RegExp('foo'))))
check('Array RegExp object, RegExp unicode flag', function () { return [func_RegExp_unicode()] }, compare_Array(enumerate_props(compare_RegExp('foo'))))
check('Array RegExp object, RegExp empty', [new RegExp('')], compare_Array(enumerate_props(compare_RegExp('(?:)'))))
check('Array RegExp object, RegExp slash', [new RegExp('/')], compare_Array(enumerate_props(compare_RegExp('\\\\/'))))
check('Array RegExp object, RegExp new line', [new RegExp('\\n')], compare_Array(enumerate_props(compare_RegExp('\\\\n'))))
check('Object RegExp object, RegExp flags and lastIndex', { 'x': func_RegExp_flags_lastIndex() }, compare_Object(enumerate_props(compare_RegExp('foo'))))
check('Object RegExp object, RegExp sticky flag', function () { return { 'x': func_RegExp_sticky() } }, compare_Object(enumerate_props(compare_RegExp('foo'))))
check('Object RegExp object, RegExp unicode flag', function () { return { 'x': func_RegExp_unicode() } }, compare_Object(enumerate_props(compare_RegExp('foo'))))
check('Object RegExp object, RegExp empty', { 'x': new RegExp('') }, compare_Object(enumerate_props(compare_RegExp('(?:)'))))
check('Object RegExp object, RegExp slash', { 'x': new RegExp('/') }, compare_Object(enumerate_props(compare_RegExp('\\\\/'))))
check('Object RegExp object, RegExp new line', { 'x': new RegExp('\\n') }, compare_Object(enumerate_props(compare_RegExp('\\\\n'))))
return failures
`),
).toEqual([])
})
test("Error: name and message kept, custom properties dropped", async () => {
expect(
await value(`
${harness}
function compare_Error(actual, input) {
assert_true(actual instanceof Error, "Checking instanceof")
assert_equals(actual.name, input.name, "Checking name")
assert_equals(Object.hasOwn(actual, "message"), Object.hasOwn(input, "message"), "Checking message existence")
assert_equals(actual.message, input.message, "Checking message")
assert_equals(actual.foo, undefined, "Checking for absence of custom property")
}
check('Empty Error object', new Error(), compare_Error)
for (const constructor of [Error, RangeError, ReferenceError, SyntaxError, TypeError, URIError]) {
check(constructor.name, () => {
const error = new constructor("Error message here")
error.foo = "testing"
return error
}, compare_Error)
}
return failures
`),
).toEqual([])
})
test("sparse arrays, index-property objects, and identical property values", async () => {
expect(
await value(`
${harness}
check('Array sparse', new Array(10), compare_Array(enumerate_props(compare_primitive)))
check('Object with index property and length', { '0': 'foo', 'length': 1 }, compare_Object(enumerate_props(compare_primitive)))
function check_identical_property_values(prop1, prop2) {
return function (actual) { assert_equals(actual[prop1], actual[prop2]) }
}
check('Array with identical property values', function () {
const obj = {}
return [obj, obj]
}, compare_Array(check_identical_property_values('0', '1')))
check('Object with identical property values', function () {
const obj = {}
return { 'x': obj, 'y': obj }
}, compare_Object(check_identical_property_values('x', 'y')))
return failures
`),
).toEqual([])
})
})
describe("structuredClone beyond the WPT battery", () => {
test("Map, Set, URL, and URLSearchParams are copied, with shared references preserved across containers", async () => {
expect(
await value(`
const shared = { n: 1 }
const input = { m: new Map([[shared, shared]]), s: new Set([shared]), u: new URL("https://a.b/c?d=1") }
const copy = structuredClone(input)
const [[key, item]] = [...copy.m]
copy.u.searchParams.set("d", "2")
return [
copy.m !== input.m, key !== shared, key === item, key === [...copy.s][0],
copy.u !== input.u, input.u.href, copy.u.href,
]
`),
).toEqual([true, true, true, true, true, "https://a.b/c?d=1", "https://a.b/c?d=2"])
})
test("functions, promises, and tool references throw DataCloneError", async () => {
expect(
await value(`
return [() => 1, Promise.resolve(1), tools, Math, { nested: [() => 1] }].map((input) => {
try { structuredClone(input); return "cloned" } catch (error) { return error.name }
})
`),
).toEqual(Array(5).fill("DataCloneError"))
expect(await value(`try { structuredClone() } catch (error) { return error.name }`)).toBe("TypeError")
})
})
+19 -9
View File
@@ -1,6 +1,7 @@
export * as Generate from "./generate.js"
import { LLM, LLMClient, AIError } from "@opencode/ai"
import { SessionID } from "@opencode/schema/session-id"
import { Context, Effect, Layer, Schema } from "effect"
import { makeLocationNode } from "@opencode/util/effect/app-node"
import { llmClient } from "./effect/app-node-platform.js"
@@ -58,15 +59,24 @@ export const layer = Layer.effect(
? `Model unavailable: ${input.model.providerID}/${input.model.id}`
: "No model specified and no supported model is available",
})
const response = yield* llm.generate(LLM.request({ model: resolved.model, prompt: input.prompt })).pipe(
Effect.mapError(
(error: AIError) =>
new UnavailableError({
message: error.message,
service: resolved.ref.providerID,
}),
),
)
const response = yield* llm
.generate(
LLM.request({
model: resolved.model,
prompt: input.prompt,
// Gateways require session attribution even for a stateless call; no Session is stored.
http: { headers: { "x-opencode-session": SessionID.create() } },
}),
)
.pipe(
Effect.mapError(
(error: AIError) =>
new UnavailableError({
message: error.message,
service: resolved.ref.providerID,
}),
),
)
return response.text
})
+1
View File
@@ -101,6 +101,7 @@ export const XAIPlugin = define({
for (const model of provider.models.values()) {
catalog.model.update(providerID, model.id, (draft) => {
draft.capabilities.responsesWebsockets = true
draft.websocket = true
})
}
})
+56 -1
View File
@@ -1,5 +1,6 @@
import { expect } from "bun:test"
import { LanguageModel } from "@opencode/ai"
import { LanguageModel, LLMClient } from "@opencode/ai"
import { RequestExecutor } from "@opencode/ai/route"
import { OpenAIChat } from "@opencode/ai/protocols"
import { TestLLM } from "@opencode/ai/testing"
import { AISDK } from "@opencode/core/aisdk"
@@ -11,6 +12,7 @@ import { ID, Info, Ref } from "@opencode/core/model"
import { Provider } from "@opencode/core/provider"
import { Npm } from "@opencode/util/npm"
import { Effect, Layer } from "effect"
import { HttpClient, HttpClientResponse } from "effect/unstable/http"
import { testEffect } from "./lib/effect"
const selected = Info.make({
@@ -98,3 +100,56 @@ resolverIt.effect("resolves dynamic models with their catalog metadata", () =>
})
}),
)
testEffect(Layer.empty).effect("attributes each stateless completion without creating a stored session", () =>
Effect.gen(function* () {
const sessions: string[] = []
const http = Layer.succeed(
HttpClient.HttpClient,
HttpClient.make((request) =>
Effect.sync(() => {
const session = request.headers["x-opencode-session"]
if (!session)
return HttpClientResponse.fromWeb(
request,
Response.json(
{
error: { type: "MissingSessionID", message: "Session ID is required" },
},
{ status: 400 },
),
)
sessions.push(session)
return HttpClientResponse.fromWeb(
request,
new Response(
`data: ${JSON.stringify({
id: "completion",
object: "chat.completion.chunk",
created: 1,
model: "gemini",
choices: [{ index: 0, delta: { content: "OK" }, finish_reason: "stop" }],
})}\n\ndata: [DONE]\n\n`,
{ headers: { "content-type": "text/event-stream" } },
),
)
}),
),
)
const native = LLMClient.layer.pipe(Layer.provide(RequestExecutor.layer.pipe(Layer.provide(http))))
yield* Effect.gen(function* () {
const generate = yield* Generate.Service
for (let index = 0; index < 2; index++) {
expect(
yield* generate.text({
prompt: "Return exactly OK",
model: Ref.make({ providerID: selected.providerID, id: selected.id }),
}),
).toBe("OK")
}
}).pipe(Effect.provide(Generate.layer.pipe(Layer.provide(Layer.merge(resolver, native)))))
expect(sessions).toHaveLength(2)
expect(sessions[0]).toStartWith("ses_")
expect(sessions[1]).not.toBe(sessions[0])
}),
)
@@ -69,7 +69,7 @@ describe("XAIPlugin", () => {
}),
)
it.effect("keeps xAI Responses WebSockets opt-in", () =>
it.effect("enables xAI Responses WebSockets", () =>
Effect.gen(function* () {
const catalog = yield* Catalog.Service
const providerID = Provider.ID.make("xai")
@@ -84,7 +84,7 @@ describe("XAIPlugin", () => {
const model = yield* catalog.model.get(providerID, Model.ID.make("grok-4.6"))
expect(model?.capabilities.responsesWebsockets).toBe(true)
expect(model?.websocket).toBeUndefined()
expect(model?.websocket).toBe(true)
}),
)
})
+10 -1
View File
@@ -1944,7 +1944,16 @@ describe("SessionRunnerLLM", () => {
expect(yield* entries.list(sessionID)).toEqual([{ key: "nullable", value: null }])
})
scenario("rejects API instruction entries larger than 8KB", function* () {
scenario("accepts API instruction entries up to 256 KiB", function* () {
const entries = yield* InstructionEntry.Service
const value = "x".repeat(InstructionEntry.MaxValueBytes - 2)
yield* entries.put({ sessionID, key: "large", value })
expect(yield* entries.list(sessionID)).toEqual([{ key: "large", value }])
})
scenario("rejects API instruction entries larger than 256 KiB", function* () {
const entries = yield* InstructionEntry.Service
const exit = yield* entries
+1 -1
View File
@@ -27,7 +27,7 @@ export const Snapshot = Schema.Array(
).annotate({ identifier: "InstructionEntry.Snapshot" })
export type Snapshot = typeof Snapshot.Type
export const MaxValueBytes = 8 * 1024
export const MaxValueBytes = 256 * 1024
export class ValueTooLargeError extends Schema.TaggedError<ValueTooLargeError>()(
"InstructionEntryValueTooLargeError",
@@ -427,17 +427,6 @@ export function SessionCompactionMessage(props: { message: SessionMessageCompact
<div class="py-2">
<TimelineSeparator label={i18n.t("ui.messagePart.compaction.started")} />
</div>
<Show when={props.message.status === "running"}>
<div role="status" class="py-2">
<BasicTool
icon="archive"
trigger={{ title: i18n.t("ui.messagePart.compaction.running") }}
status="running"
locked
hideDetails
/>
</div>
</Show>
<Show when={summary().trim()}>
<div data-component="text-part" data-timeline-part-id={props.message.id}>
<div data-slot="text-part-body">
@@ -449,6 +438,17 @@ export function SessionCompactionMessage(props: { message: SessionMessageCompact
</div>
</div>
</Show>
<Show when={props.message.status === "running"}>
<div role="status" class="py-2">
<BasicTool
icon="archive"
trigger={{ title: i18n.t("ui.messagePart.compaction.running") }}
status="running"
locked
hideDetails
/>
</div>
</Show>
<Show when={props.message.status !== "running"}>
<div class="py-2">
<TimelineSeparator label={label()} />
+2 -1
View File
@@ -51,7 +51,8 @@ export function createClipboardAdapter(clipboard: CoreClipboardService): OwnedCl
throw new Error(`Unexpected clipboard MIME type: ${result.representation.mimeType}`)
},
async write(text) {
const result = await clipboard.writeText(text, {
// OpenTUI rejects NUL before any destination; host clipboard text cannot contain it.
const result = await clipboard.writeText(text.replaceAll("\0", ""), {
destination: "all-available",
selection: "clipboard",
})
+13
View File
@@ -102,6 +102,19 @@ test("uses all available routes but skips the process host remotely", async () =
expect(writes).toEqual({ host: 0, terminal: 1 })
})
test("removes NUL characters before writing", async () => {
const writes: string[] = []
const clipboard = createClipboardAdapter(
coreClipboard({
onWrite: (text) => writes.push(text),
}),
)
expect(await clipboard.write("before\0after")).toBeUndefined()
expect(await clipboard.write("clean")).toBeUndefined()
expect(writes).toEqual(["beforeafter", "clean"])
})
test("rejects only when no clipboard route accepted the write", async () => {
const writes: [string, ClipboardWriteOptions][] = []
const failure = new Error("native clipboard failed")
+5 -4
View File
@@ -110,10 +110,11 @@ role for other Foundry models. If a request fails because the token belongs to a
## WebSocket transport
OpenAI and supported Azure Responses models keep one WebSocket connection open per session and send each step over it
instead of opening a new HTTP request. While the request prefix is unchanged, consecutive steps only transmit what was
added since the previous response, which cuts upload volume on long sessions. Provider compaction runs over the same
connection.
OpenAI, xAI, and supported Azure Responses models keep one WebSocket connection open per session and send each step
over it instead of opening a new HTTP request. While the request prefix is unchanged, consecutive steps only transmit
what was added since the previous response, which cuts upload volume on long sessions. OpenAI provider compaction runs
over the same connection. xAI continues a chain only from stored responses, so with its default `store: false` each step
is sent in full over the reused connection.
The connection is transparent. When the provider closes the socket, the next step reconnects; when a connection cannot
be opened at all, the session continues over HTTP. Plugins that register `http.request` or `http.response` hooks for a