Compare commits

..

3 Commits

Author SHA1 Message Date
Aiden Cline d2f7145bf9 fix(ai): preserve stream transport failures 2026-08-11 16:21:21 +00:00
Kit Langton 23b0688a7f fix(server): await models before listing (#41783) 2026-08-11 12:18:07 -04:00
opencode-agent[bot] d0dd49bd84 chore: generate 2026-08-11 16:12:58 +00:00
23 changed files with 317 additions and 208 deletions
+65 -32
View File
@@ -1,4 +1,4 @@
import { Cause, Context, Effect, Layer, Option, Schema } from "effect"
import { Cause, Context, Effect, Layer, Option, Schema, Stream } from "effect"
import {
FetchHttpClient,
Headers,
@@ -16,12 +16,17 @@ import {
TransportReason,
} from "../schema"
import { classifyProviderFailure } from "../provider-error"
import { isRecord } from "../utils/record"
export interface Interface {
readonly execute: (
request: HttpClientRequest.HttpClientRequest,
middleware?: HttpMiddleware,
) => Effect.Effect<HttpClientResponse.HttpClientResponse, AIError>
readonly stream: (
request: HttpClientRequest.HttpClientRequest,
middleware?: HttpMiddleware,
) => Stream.Stream<Uint8Array, AIError>
}
export type HttpHandler = (
@@ -297,41 +302,51 @@ export const classifyHttpFailure = (input: {
})
}
const toHttpError = (redactedNames: ReadonlyArray<string | RegExp>) => (error: unknown) => {
const transportError = (input: {
readonly message: string
readonly kind?: string | undefined
readonly request?: HttpClientRequest.HttpClientRequest | undefined
}) =>
type HttpOperation = "request" | "read"
const httpError = (input: {
readonly error: unknown
readonly request: HttpClientRequest.HttpClientRequest
readonly operation: HttpOperation
readonly redactedNames: ReadonlyArray<string | RegExp>
}) => {
const transportError = (failure: { readonly message: string; readonly code?: string | undefined }) =>
new AIError({
module: "RequestExecutor",
method: "execute",
method: input.operation,
reason: new TransportReason({
message: input.message,
kind: input.kind,
url: input.request ? redactUrl(input.request.url) : undefined,
http: input.request ? new HttpContext({ request: requestDetails(input.request, redactedNames) }) : undefined,
message: failure.message,
transport: "http",
operation: input.operation,
code: failure.code,
url: redactUrl(input.request.url),
http: new HttpContext({ request: requestDetails(input.request, input.redactedNames) }),
}),
})
if (Cause.isTimeoutError(error)) {
return transportError({ message: error.message, kind: "Timeout" })
}
if (!HttpClientError.isHttpClientError(error)) {
return transportError({ message: error instanceof Error ? error.message : "HTTP transport failed" })
}
const request = "request" in error ? error.request : undefined
if (error.reason._tag === "TransportError") {
const source =
HttpClientError.isHttpClientError(input.error) && "cause" in input.error.reason
? input.error.reason.cause
: input.error
const code = isRecord(source) && typeof source.code === "string" ? source.code : undefined
const request = HttpClientError.isHttpClientError(input.error) ? input.error.request : input.request
const raw = source instanceof Error ? source.message : input.error instanceof Error ? input.error.message : undefined
const detail = raw ? redactBody(raw, secretValues(request)) : undefined
const message = code && detail && !detail.includes(code) ? `${code}: ${detail}` : detail
if (Cause.isTimeoutError(input.error) || Cause.isTimeoutError(source))
return transportError({ message: message ?? "HTTP transport timed out", code: code ?? "Timeout" })
if (!HttpClientError.isHttpClientError(input.error))
return transportError({ message: message ?? "HTTP transport failed", code })
if (input.error.reason._tag === "TransportError") {
return transportError({
message: error.reason.description ?? "HTTP transport failed",
kind: error.reason._tag,
request,
message: message ?? input.error.reason.description ?? "HTTP transport failed",
code: code ?? input.error.reason._tag,
})
}
return transportError({
message: `HTTP transport failed: ${error.reason._tag}`,
kind: error.reason._tag,
request,
message: message ?? `HTTP transport failed: ${input.error.reason._tag}`,
code: code ?? input.error.reason._tag,
})
}
@@ -339,23 +354,41 @@ export const layer: Layer.Layer<Service, never, HttpClient.HttpClient> = Layer.e
Service,
Effect.gen(function* () {
const http = yield* HttpClient.HttpClient
const executeOnce = (request: HttpClientRequest.HttpClientRequest, middleware?: HttpMiddleware) =>
const execute = (
request: HttpClientRequest.HttpClientRequest,
middleware: HttpMiddleware | undefined,
redactedNames: ReadonlyArray<string | RegExp>,
) =>
Effect.gen(function* () {
const redactedNames = yield* Headers.CurrentRedactedNames
if (!middleware)
return yield* http
.execute(request)
.pipe(Effect.mapError(toHttpError(redactedNames)), Effect.flatMap(statusError(request, redactedNames)))
return yield* http.execute(request).pipe(
Effect.mapError((error) => httpError({ error, request, operation: "request", redactedNames })),
Effect.flatMap(statusError(request, redactedNames)),
)
const response = yield* middleware(request, (input) =>
http
.execute(input)
.pipe(Effect.mapError((cause) => (cause instanceof Error ? cause : new Error(String(cause))))),
).pipe(Effect.mapError(toHttpError(redactedNames)))
).pipe(Effect.mapError((error) => httpError({ error, request, operation: "request", redactedNames })))
return yield* statusError(response.request, redactedNames)(response)
})
const executeOnce = (request: HttpClientRequest.HttpClientRequest, middleware?: HttpMiddleware) =>
Effect.flatMap(Headers.CurrentRedactedNames, (redactedNames) => execute(request, middleware, redactedNames))
return Service.of({
execute: executeOnce,
stream: (request, middleware) =>
Stream.unwrap(
Effect.gen(function* () {
const redactedNames = yield* Headers.CurrentRedactedNames
const response = yield* execute(request, middleware, redactedNames)
return response.stream.pipe(
Stream.mapError((error) =>
httpError({ error, request: response.request, operation: "read", redactedNames }),
),
)
}),
),
})
}),
)
+3 -21
View File
@@ -1,4 +1,4 @@
import { Effect, Stream } from "effect"
import { Effect } from "effect"
import { Headers, HttpClientRequest } from "effect/unstable/http"
import { Auth } from "../auth"
import { render as renderEndpoint } from "../endpoint"
@@ -86,26 +86,8 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
middleware: prepareInput.middleware,
}
}),
frames: (prepared, request, runtime) =>
Stream.unwrap(
runtime.http
.execute(prepared.request, prepared.middleware)
.pipe(
Effect.map((response) =>
prepared.framing.frame(
response.stream.pipe(
Stream.mapError((error) =>
ProviderShared.eventError(
`${request.model.provider}/${request.model.route.id}`,
`Failed to read ${request.model.provider}/${request.model.route.id} stream`,
ProviderShared.errorText(error),
),
),
),
),
),
),
),
frames: (prepared, _request, runtime) =>
prepared.framing.frame(runtime.http.stream(prepared.request, prepared.middleware)),
})
export const sseJson = {
+36 -13
View File
@@ -1,6 +1,6 @@
import { Cause, Context, Effect, Layer, Queue, Stream } from "effect"
import { Headers } from "effect/unstable/http"
import { AIError, TransportReason } from "../../schema"
import { AIError, TransportReason, type TransportOperation } from "../../schema"
import * as HttpTransport from "./http"
import type { Transport } from "./index"
@@ -29,12 +29,18 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/AI
const transportError = (
method: string,
message: string,
input: { readonly url?: string; readonly kind?: string } = {},
input: { readonly operation: TransportOperation; readonly url?: string; readonly code?: string },
) =>
new AIError({
module: "WebSocketExecutor",
method,
reason: new TransportReason({ message, url: input.url, kind: input.kind }),
reason: new TransportReason({
message,
transport: "websocket",
operation: input.operation,
url: input.url,
code: input.code,
}),
})
const eventMessage = (event: Event) => {
@@ -55,7 +61,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
return Effect.fail(
transportError("open", `WebSocket closed before opening (state ${ws.readyState})`, {
url: input.url,
kind: "open",
operation: "request",
code: "closed",
}),
)
}
@@ -79,7 +86,10 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
cleanup()
resume(
Effect.fail(
transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, { url: input.url, kind: "open" }),
transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, {
url: input.url,
operation: "request",
}),
),
)
}
@@ -89,7 +99,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
Effect.fail(
transportError("open", `WebSocket closed before opening with code ${event.code}`, {
url: input.url,
kind: "open",
operation: "request",
code: String(event.code),
}),
),
)
@@ -118,7 +129,8 @@ const webSocketUrl = (value: string) =>
catch: (error) =>
transportError("prepare", error instanceof Error ? error.message : "Invalid WebSocket URL", {
url: value,
kind: "websocket",
operation: "request",
code: "invalid-url",
}),
})
@@ -129,7 +141,7 @@ export const open = (input: WebSocketRequest) =>
catch: (error) =>
transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
url: input.url,
kind: "open",
operation: "request",
}),
}).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input)))
@@ -150,7 +162,10 @@ export const fromWebSocket = (
Queue.failCauseUnsafe(
messages,
Cause.fail(
transportError("message", "Unsupported WebSocket message payload", { url: input.url, kind: "message" }),
transportError("message", "Unsupported WebSocket message payload", {
url: input.url,
operation: "read",
}),
),
)
}
@@ -158,7 +173,10 @@ export const fromWebSocket = (
Queue.failCauseUnsafe(
messages,
Cause.fail(
transportError("message", `WebSocket error: ${eventMessage(event)}`, { url: input.url, kind: "message" }),
transportError("message", `WebSocket error: ${eventMessage(event)}`, {
url: input.url,
operation: "read",
}),
),
)
}
@@ -167,7 +185,11 @@ export const fromWebSocket = (
Queue.failCauseUnsafe(
messages,
Cause.fail(
transportError("message", `WebSocket closed with code ${event.code}`, { url: input.url, kind: "close" }),
transportError("message", `WebSocket closed with code ${event.code}`, {
url: input.url,
operation: "read",
code: String(event.code),
}),
),
)
}
@@ -188,7 +210,7 @@ export const fromWebSocket = (
catch: (error) =>
transportError("sendText", error instanceof Error ? error.message : "Failed to send WebSocket message", {
url: input.url,
kind: "write",
operation: "write",
}),
}),
messages: Stream.fromQueue(messages),
@@ -243,7 +265,8 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
return Stream.fail(
transportError("json", "WebSocket JSON transport requires WebSocketExecutor.Service", {
url: prepared.url,
kind: "websocket",
operation: "request",
code: "unavailable",
}),
)
}
+9 -1
View File
@@ -92,10 +92,18 @@ export class ProviderInternalReason extends Schema.Class<ProviderInternalReason>
http: Schema.optional(HttpContext),
}) {}
export const TransportType = Schema.Literals(["http", "websocket"])
export type TransportType = typeof TransportType.Type
export const TransportOperation = Schema.Literals(["request", "read", "write"])
export type TransportOperation = typeof TransportOperation.Type
export class TransportReason extends Schema.Class<TransportReason>("AI.Error.Transport")({
_tag: Schema.tag("Transport"),
message: Schema.String,
kind: Schema.optional(Schema.String),
transport: TransportType,
operation: TransportOperation,
code: Schema.optional(Schema.String),
url: Schema.optional(Schema.String),
http: Schema.optional(HttpContext),
}) {}
+31 -2
View File
@@ -1,10 +1,10 @@
import { describe, expect } from "bun:test"
import { Effect, Layer, Ref } from "effect"
import { Effect, Layer, Ref, Stream } from "effect"
import { Headers, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { LLM, AIError } from "../src"
import { LLMClient, RequestExecutor } from "../src/route"
import * as OpenAIChat from "../src/protocols/openai-chat"
import { dynamicResponse } from "./lib/http"
import { dynamicResponse, systemError } from "./lib/http"
import { deltaChunk } from "./lib/openai-chunks"
import { sseRaw } from "./lib/sse"
import { it } from "./lib/effect"
@@ -67,6 +67,35 @@ const expectAIError = (error: unknown) => {
const errorHttp = (error: AIError) => ("http" in error.reason ? error.reason.http : undefined)
describe("RequestExecutor", () => {
it.effect("parses response body failures at the executor seam", () =>
Effect.gen(function* () {
const executor = yield* RequestExecutor.Service
const error = yield* executor.stream(secretRequest).pipe(Stream.runDrain, Effect.flip)
expectAIError(error)
expect(error.reason).toMatchObject({
_tag: "Transport",
message: "ECONNRESET: disconnected <redacted> <redacted>",
transport: "http",
operation: "read",
code: "ECONNRESET",
url: "https://provider.test/v1/chat?api_key=%3Credacted%3E&debug=1",
})
}).pipe(
Effect.provide(
responsesLayer([
new Response(
new ReadableStream({
start(controller) {
controller.error(systemError("ECONNRESET", "disconnected query-secret-123 header-secret-456"))
},
}),
),
]),
),
),
)
it.effect("preserves middleware error messages", () =>
Effect.gen(function* () {
const executor = yield* RequestExecutor.Service
+8 -2
View File
@@ -34,6 +34,12 @@ const handlerLayer = (handler: Handler): Layer.Layer<HttpClient.HttpClient> =>
export type RuntimeEnv = RequestExecutorService | WebSocketExecutorService | LLMClientService
export interface SystemError extends Error {
readonly code: string
}
export const systemError = (code: string, message: string): SystemError => Object.assign(new Error(message), { code })
export const runtimeLayer = (layer: Layer.Layer<HttpClient.HttpClient>): Layer.Layer<RuntimeEnv> => {
const requestExecutorLayer = RequestExecutor.layer.pipe(Layer.provide(layer))
const deps = Layer.mergeAll(requestExecutorLayer, WebSocketExecutor.layer)
@@ -63,14 +69,14 @@ export const dynamicResponse = (handler: Handler) => runtimeLayer(handlerLayer(h
* Layer that emits the supplied SSE chunks and then aborts mid-stream. Used to
* exercise transport errors that surface during parsing.
*/
export const truncatedStream = (chunks: ReadonlyArray<string>) =>
export const truncatedStream = (chunks: ReadonlyArray<string>, error: Error = new Error("connection reset")) =>
dynamicResponse((input) =>
Effect.sync(() => {
const encoder = new TextEncoder()
const stream = new ReadableStream({
start(controller) {
for (const chunk of chunks) controller.enqueue(encoder.encode(chunk))
controller.error(new Error("connection reset"))
controller.error(error)
},
})
return input.respond(stream, { headers: SSE_HEADERS })
+13 -5
View File
@@ -22,7 +22,7 @@ import { ProviderShared } from "../../src/protocols/shared"
import { Auth, LLMClient } from "../../src/route"
import { compileRequest } from "../../src/route/client"
import { it } from "../lib/effect"
import { dynamicResponse, fixedResponse, truncatedStream } from "../lib/http"
import { dynamicResponse, fixedResponse, systemError, truncatedStream } from "../lib/http"
import { deltaChunk, usageChunk } from "../lib/openai-chunks"
import { sseEvents } from "../lib/sse"
@@ -1221,12 +1221,20 @@ describe("OpenAI Chat route", () => {
it.effect("surfaces transport errors that occur mid-stream", () =>
Effect.gen(function* () {
const layer = truncatedStream([
`data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`,
])
const layer = truncatedStream(
[`data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`],
systemError("ECONNRESET", "socket closed unexpectedly"),
)
const error = yield* LLMClient.generate(request).pipe(Effect.provide(layer), Effect.flip)
expect(error.message).toContain("Failed to read openai/openai-chat stream")
expect(error.reason).toMatchObject({
_tag: "Transport",
message: "ECONNRESET: socket closed unexpectedly",
transport: "http",
operation: "read",
code: "ECONNRESET",
url: "https://api.openai.test/v1/chat/completions",
})
}),
)
@@ -243,6 +243,7 @@ describe("OpenAI Responses route", () => {
RequestExecutor.Service,
RequestExecutor.Service.of({
execute: () => Effect.die("unexpected HTTP request"),
stream: () => Stream.die("unexpected HTTP request"),
}),
),
Layer.succeed(
-21
View File
@@ -310,27 +310,6 @@ test("unrelated managed port occupancy reports an actionable conflict", async ()
}
}, 30_000)
test("managed service startup reports an actionable port conflict", async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-managed-conflict-"))
const registration = path.join(root, "state", "opencode", "service-local.json")
const message =
"Managed service port 49374 on 127.0.0.1 is already in use by another process. " +
"Configure another port with `opencode service set port <port>` and start the service again."
try {
await expect(
Effect.runPromise(
Service.ensure({
file: registration,
command: [process.execPath, "-e", `console.error(${JSON.stringify(message)}); process.exit(1)`],
}).pipe(Effect.provide(NodeFileSystem.layer)),
),
).rejects.toThrow(message)
} finally {
await fs.rm(root, { recursive: true, force: true })
}
}, 30_000)
test("unresponsive managed port occupancy reports a bounded conflict", async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-unresponsive-conflict-"))
const recognizing = Promise.withResolvers<void>()
+28 -8
View File
@@ -1,14 +1,9 @@
import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
import { Effect, FileSystem, Option, Schedule, Schema } from "effect"
import { spawn, type ChildProcess } from "node:child_process"
import { homedir } from "node:os"
import { join } from "node:path"
import type { DiscoverOptions, Endpoint, EnsureOptions, StopOptions } from "../service.js"
import {
contenderFailure,
contenderFinished,
type ServiceContender,
spawnServiceContender,
} from "../service-contender.js"
export * from "../service.js"
/** Contents of the local service registration file. */
@@ -22,6 +17,11 @@ export type Info = import("../service.js").Info
// is all a client needs to connect. The daemon's own configuration (port,
// persisted password) is CLI-owned and never read here.
type Contender = {
readonly child: ChildProcess
readonly error: () => Error | undefined
}
// Read-only lookup: registration file plus health check and version gate.
// Never spawns; escalation to ensure() is the caller's policy.
/** Discover a healthy, compatible local service without starting one. */
@@ -52,7 +52,7 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) {
// becomes discoverable. A contender is never killed merely for slow startup.
/** Ensure a healthy, compatible local service is running. */
export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOptions = {}) {
const contenders = new Set<ServiceContender>()
const contenders = new Set<Contender>()
let timeouts: { readonly info: Info; readonly count: number } | undefined
let announced = false
let lastSpawn = 0
@@ -68,7 +68,13 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
if (command === undefined) return yield* Effect.fail(new Error("Missing service command"))
return yield* Effect.try({
try: () => {
return spawnServiceContender(command, args)
const child = spawn(command, args, { detached: true, stdio: "ignore" })
let error: Error | undefined
child.once("error", (cause) => {
error = new Error("Failed to start server", { cause })
})
child.unref()
return { child, error: () => error }
},
catch: (cause) => new Error("Failed to start server", { cause }),
})
@@ -127,6 +133,20 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
return found.value.endpoint
})
function contenderFailure(contender: Contender) {
const error = contender.error()
if (error !== undefined) return error
if (contender.child.exitCode !== null && contender.child.exitCode !== 0)
return new Error(`Server process exited with code ${contender.child.exitCode}`)
if (contender.child.signalCode !== null)
return new Error(`Server process terminated by ${contender.child.signalCode}`)
return undefined
}
function contenderFinished(contender: Contender) {
return contender.error() !== undefined || contender.child.exitCode !== null || contender.child.signalCode !== null
}
/** Stop the registered local service. */
export const stop = Effect.fn("service.stop")(function* (options: StopOptions = {}) {
const existing = yield* find(options)
+28 -8
View File
@@ -1,13 +1,8 @@
import { readFile } from "node:fs/promises"
import { spawn, type ChildProcess } from "node:child_process"
import { homedir } from "node:os"
import { join } from "node:path"
import type { DiscoverOptions, Endpoint, Info, EnsureOptions, StopOptions } from "../service.js"
import {
contenderFailure,
contenderFinished,
type ServiceContender,
spawnServiceContender,
} from "../service-contender.js"
import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js"
export * from "../service.js"
@@ -18,6 +13,11 @@ export * from "../service.js"
// intentionally implemented with Node APIs so Promise clients do not need
// Effect or @effect/platform-node at runtime.
type Contender = {
readonly child: ChildProcess
readonly error: () => Error | undefined
}
/** Discover a healthy, compatible local service without starting one. */
export async function discover(options: DiscoverOptions = {}) {
return (await discoverLocal(options))?.endpoint
@@ -33,7 +33,7 @@ async function discoverLocal(options: DiscoverOptions) {
/** Ensure a healthy, compatible local service is running. */
export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
const deadline = Date.now() + 120_000
const contenders = new Set<ServiceContender>()
const contenders = new Set<Contender>()
let timeouts: { readonly info: Info; readonly count: number } | undefined
let announced = false
let lastSpawn = 0
@@ -48,7 +48,13 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
const [command, ...args] = options.command ?? ["opencode", "serve", "--service"]
if (command === undefined) throw new Error("Missing service command")
try {
return spawnServiceContender(command, args)
const child = spawn(command, args, { detached: true, stdio: "ignore" })
let error: Error | undefined
child.once("error", (cause) => {
error = new Error("Failed to start server", { cause })
})
child.unref()
return { child, error: () => error }
} catch (cause) {
throw new Error("Failed to start server", { cause })
}
@@ -101,6 +107,20 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
}
}
function contenderFailure(contender: Contender) {
const error = contender.error()
if (error !== undefined) return error
if (contender.child.exitCode !== null && contender.child.exitCode !== 0)
return new Error(`Server process exited with code ${contender.child.exitCode}`)
if (contender.child.signalCode !== null)
return new Error(`Server process terminated by ${contender.child.signalCode}`)
return undefined
}
function contenderFinished(contender: Contender) {
return contender.error() !== undefined || contender.child.exitCode !== null || contender.child.signalCode !== null
}
/** Stop the registered local service. */
export async function stop(options: StopOptions = {}) {
const existing = await find(options)
-48
View File
@@ -1,48 +0,0 @@
import { spawn, type ChildProcess } from "node:child_process"
export type ServiceContender = {
readonly child: ChildProcess
readonly error: () => Error | undefined
readonly closed: () => boolean
readonly stderr: () => string
}
const stderrLimit = 8 * 1024
export function spawnServiceContender(command: string, args: ReadonlyArray<string>): ServiceContender {
const child = spawn(command, args, { detached: true, stdio: ["ignore", "ignore", "pipe"] })
let error: Error | undefined
let closed = false
let stderr = Buffer.alloc(0)
child.stderr?.on("data", (chunk: Buffer) => {
stderr = Buffer.concat([stderr, chunk]).subarray(-stderrLimit)
})
if (child.stderr !== null && "unref" in child.stderr && typeof child.stderr.unref === "function")
child.stderr.unref()
child.once("error", (cause) => {
error = new Error("Failed to start server", { cause })
})
child.once("close", () => {
closed = true
})
child.unref()
return { child, error: () => error, closed: () => closed, stderr: () => stderr.toString("utf8").trim() }
}
export function contenderFailure(contender: ServiceContender) {
const error = contender.error()
if (error !== undefined) return error
if (contender.child.exitCode !== null && contender.child.exitCode !== 0)
return startupError(`Server process exited with code ${contender.child.exitCode}`, contender.stderr())
if (contender.child.signalCode !== null)
return startupError(`Server process terminated by ${contender.child.signalCode}`, contender.stderr())
return undefined
}
export function contenderFinished(contender: ServiceContender) {
return contender.error() !== undefined || contender.closed()
}
function startupError(message: string, stderr: string) {
return new Error(stderr ? `${message}\n${stderr}` : message)
}
-4
View File
@@ -3,10 +3,6 @@ import { appendFile, rename, writeFile } from "node:fs/promises"
const [registration, mode, delay] = process.argv.slice(2)
if (registration === undefined || mode === undefined) throw new Error("Missing service fixture arguments")
if (mode === "failed") process.exit(1)
if (mode === "stderr-failed") {
process.stderr.write("x".repeat(16_384) + "\nactionable startup failure\n")
process.exit(1)
}
if (mode === "record-start") {
await writeFile(registration + ".started", "")
process.exit(1)
@@ -70,21 +70,6 @@ test("reports a failed registered service", async () => {
)
})
test("reports a bounded contender stderr tail with native promises", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const error = await Service.ensure({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "stderr-failed"],
}).catch((error: unknown) => error)
expect(error).toBeInstanceOf(Error)
if (!(error instanceof Error)) throw error
expect(error.message).toContain("actionable startup failure")
expect(error.message.length).toBeLessThan(9_000)
}, 10_000)
test("evicts an unresponsive registered service before starting its replacement", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
-17
View File
@@ -197,23 +197,6 @@ test("reports a contender that fails to start", async () => {
).rejects.toThrow("Server process exited with code 1")
}, 10_000)
test("reports a bounded contender stderr tail", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const error = await run(
Service.ensure({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "stderr-failed"],
}),
).catch((error: unknown) => error)
expect(error).toBeInstanceOf(Error)
if (!(error instanceof Error)) throw error
expect(error.message).toContain("actionable startup failure")
expect(error.message.length).toBeLessThan(9_000)
}, 10_000)
test("reports a contender terminated by a signal", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
+3 -1
View File
@@ -763,7 +763,9 @@ function apiCallErrorReason(error: APICallError) {
if (error.statusCode !== undefined || !error.isRetryable) return reason
return new TransportReason({
message: reason.message,
kind: error.name,
transport: "http",
operation: "request",
code: error.name,
url: error.url,
http: "http" in reason ? reason.http : undefined,
})
+1 -3
View File
@@ -21,9 +21,7 @@ Usage notes:
- If you recommend a specific option, make that the first option in the list and add "(Recommended)" at the end of the label`
export const Input = Schema.Struct({
questions: Schema.Array(Question.Prompt)
.check(Schema.isNonEmpty())
.annotate({ description: "Questions to ask" }),
questions: Schema.Array(Question.Prompt).check(Schema.isNonEmpty()).annotate({ description: "Questions to ask" }),
})
export const Output = Schema.Struct({
+11 -3
View File
@@ -9,7 +9,7 @@ import { LLM, AIError, LLMEvent, Message, isContextOverflowFailure } from "@open
import { LLMClient, RequestExecutor } from "@opencode-ai/ai/route"
import { compileRequest } from "@opencode-ai/ai/route/client"
import { expect } from "bun:test"
import { Effect, Layer } from "effect"
import { Effect, Layer, Stream } from "effect"
import { testEffect } from "./lib/effect"
const it = testEffect(AISDK.locationLayer)
@@ -49,7 +49,10 @@ const client = LLMClient.layer.pipe(
Layer.provide(
Layer.succeed(
RequestExecutor.Service,
RequestExecutor.Service.of({ execute: () => Effect.die("Unexpected HTTP request") }),
RequestExecutor.Service.of({
execute: () => Effect.die("Unexpected HTTP request"),
stream: () => Stream.die("Unexpected HTTP request"),
}),
),
),
)
@@ -542,7 +545,12 @@ it.effect("retries status-less AI SDK transport failures", () =>
isRetryable: true,
}),
)
expect(error.reason).toMatchObject({ _tag: "Transport", kind: "AI_APICallError" })
expect(error.reason).toMatchObject({
_tag: "Transport",
transport: "http",
operation: "request",
code: "AI_APICallError",
})
expect(SessionRunnerRetry.isRetryable(error)).toBeTrue()
expect("http" in error.reason ? error.reason.http?.request.url : undefined).toBe("https://api.example.com/chat")
}),
+4 -2
View File
@@ -39,7 +39,9 @@ describe("toSessionError", () => {
)
expect(toSessionError(llm(new QuotaExceededReason({ message: "quota" }))).type).toBe("provider.quota")
expect(toSessionError(llm(new ContentPolicyReason({ message: "blocked" }))).type).toBe("provider.content-filter")
expect(toSessionError(llm(new TransportReason({ message: "transport" }))).type).toBe("provider.transport")
expect(
toSessionError(llm(new TransportReason({ message: "transport", transport: "http", operation: "request" }))).type,
).toBe("provider.transport")
expect(toSessionError(llm(new ProviderInternalReason({ message: "internal", status: 500 }))).type).toBe(
"provider.internal",
)
@@ -111,7 +113,7 @@ describe("toSessionError", () => {
const eligible = [
llm(new RateLimitReason({ message: "rate" })),
llm(new ProviderInternalReason({ message: "internal", status: 500 })),
llm(new TransportReason({ message: "transport" })),
llm(new TransportReason({ message: "transport", transport: "http", operation: "request" })),
]
const ineligible = [
llm(new AuthenticationReason({ message: "auth", kind: "invalid" })),
+1 -1
View File
@@ -31,7 +31,7 @@ describe("SessionExecution lifecycle", () => {
new AIError({
module: "test",
method: "stream",
reason: new TransportReason({ message: "Disconnected" }),
reason: new TransportReason({ message: "Disconnected", transport: "http", operation: "request" }),
}),
),
),
+5 -1
View File
@@ -515,7 +515,11 @@ const providerUnavailable = () =>
new AIError({
module: "test",
method: "stream",
reason: new TransportReason({ message: "Provider unavailable" }),
reason: new TransportReason({
message: "Provider unavailable",
transport: "http",
operation: "request",
}),
})
const incompleteStream = () =>
+13
View File
@@ -12,6 +12,19 @@ export const ModelHandler = HttpApiBuilder.group(Api, "server.model", (handlers)
.handle(
"model.list",
Effect.fn(function* () {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush.pipe(
Effect.timeoutOrElse({
duration: "5 seconds",
orElse: () =>
Effect.fail(
new ServiceUnavailableError({
message: "Model catalog initialization timed out",
service: "model.catalog",
}),
),
}),
)
const catalog = yield* Catalog.Service
return yield* response(catalog.model.available())
}),
+57
View File
@@ -0,0 +1,57 @@
import fs from "node:fs/promises"
import path from "node:path"
import { expect } from "bun:test"
import { Effect } from "effect"
import { HttpServer } from "effect/unstable/http"
import { tmpdir } from "../../core/test/fixture/tmpdir"
import { it } from "../../core/test/lib/effect"
import { ServerProcess } from "../src/process"
it.live("waits for plugin initialization before listing models", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir("opencode-model-endpoint-")),
(tmp) =>
Effect.gen(function* () {
yield* Effect.promise(() =>
fs.writeFile(
path.join(tmp.path, "opencode.json"),
JSON.stringify({
providers: {
custom: {
package: "aisdk:@ai-sdk/openai-compatible",
settings: { apiKey: "secret" },
models: { chat: {} },
},
},
}),
),
)
const server = yield* ServerProcess.start<never, never>({
hostname: "127.0.0.1",
port: 0,
password: "secret",
app: { version: "test-version" },
database: { path: ":memory:" },
config: { directory: tmp.path },
fs: { filewatcher: false },
})
const url = new URL("/api/model", HttpServer.formatAddress(server.address))
url.searchParams.set("location[directory]", tmp.path)
const response = yield* Effect.promise(() =>
fetch(url, { headers: { authorization: `Basic ${btoa("opencode:secret")}` } }),
)
expect(response.status).toBe(200)
const body: unknown = yield* Effect.promise(() => response.json())
if (!isRecord(body) || !Array.isArray(body["data"])) throw new Error("Expected a model list response")
expect(
body["data"].some((model) => isRecord(model) && model["providerID"] === "custom" && model["id"] === "chat"),
).toBeTrue()
}),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),
)
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
}