From 97d3cd0b3ab38a3dcf96556fe4a03fb6237e610b Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 21 Aug 2026 15:37:19 -0400 Subject: [PATCH] refactor: simplify server client helpers (#43976) --- packages/client/src/effect/service.ts | 12 +- packages/client/src/promise/service.ts | 10 +- packages/protocol/src/simulation.ts | 7 +- packages/sdk-next/src/logging.ts | 11 +- packages/sdk-next/test/logging.test.ts | 20 + packages/server/src/handlers/message.ts | 24 +- packages/server/src/handlers/model.ts | 38 +- packages/server/src/handlers/permission.ts | 22 +- .../server/src/handlers/plugin-readiness.ts | 13 + packages/server/src/handlers/session-error.ts | 20 + packages/server/src/handlers/session.ts | 352 ++++-------------- packages/server/src/handlers/websearch.ts | 29 +- packages/server/test/handler-policy.test.ts | 61 +++ 13 files changed, 239 insertions(+), 380 deletions(-) create mode 100644 packages/sdk-next/test/logging.test.ts create mode 100644 packages/server/src/handlers/plugin-readiness.ts create mode 100644 packages/server/src/handlers/session-error.ts create mode 100644 packages/server/test/handler-policy.test.ts diff --git a/packages/client/src/effect/service.ts b/packages/client/src/effect/service.ts index 53ee19fd87..33508c87b0 100644 --- a/packages/client/src/effect/service.ts +++ b/packages/client/src/effect/service.ts @@ -28,7 +28,10 @@ export type Info = import("../service.js").Info // Never spawns; escalation to ensure() is the caller's policy. /** Discover a healthy, compatible local service without starting one. */ export const discover = Effect.fn("service.discover")(function* (options: DiscoverOptions = {}) { - return (yield* discoverLocal(options))?.endpoint + const found = (yield* registered(options.file)).service + if (found?.state !== "ready") return undefined + if (!matchesVersion(found.version, options)) return undefined + return found.endpoint }) /** Recognize an authenticated compatible service bound to an expected URL, including while it starts or fails. */ @@ -42,13 +45,6 @@ export const incumbent = Effect.fn("service.incumbent")(function* ( return { endpoint: found.endpoint, state: found.state } }) -const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) { - const found = (yield* registered(options.file)).service - if (found?.state !== "ready") return undefined - if (!matchesVersion(found.version, options)) return undefined - return found -}) - // Idempotent ensure-running: reuses a healthy compatible server, replaces a // version-mismatched one, and otherwise spawns small contenders until a server // becomes discoverable. A contender is never killed merely for slow startup. diff --git a/packages/client/src/promise/service.ts b/packages/client/src/promise/service.ts index a0f2ad5176..1441cdd88b 100644 --- a/packages/client/src/promise/service.ts +++ b/packages/client/src/promise/service.ts @@ -22,14 +22,10 @@ export * from "../service.js" /** Discover a healthy, compatible local service without starting one. */ export async function discover(options: DiscoverOptions = {}) { - return (await discoverLocal(options))?.endpoint -} - -async function discoverLocal(options: DiscoverOptions) { const found = (await registered(options.file)).service if (found?.state !== "ready") return undefined if (!matchesVersion(found.version, options)) return undefined - return found + return found.endpoint } /** Ensure a healthy, compatible local service is running. */ @@ -144,10 +140,6 @@ type LocalService = { readonly legacy: boolean } -async function probe(info: Info, allowLegacy = false): Promise { - return (await probeResult(info, allowLegacy)).service -} - async function probeResult(info: Info, allowLegacy = false, timeout = defaultEnsureTiming.requestTimeout) { const endpoint = { url: info.url, diff --git a/packages/protocol/src/simulation.ts b/packages/protocol/src/simulation.ts index cde2027214..c3a657a2b7 100644 --- a/packages/protocol/src/simulation.ts +++ b/packages/protocol/src/simulation.ts @@ -444,14 +444,15 @@ export namespace Backend { ]) export type ToolContent = Schema.Schema.Type + const ProviderSafeName = /^[A-Za-z][A-Za-z0-9_-]{0,63}$/ const ToolName = Schema.NonEmptyString.check( Schema.makeFilter((name) => - /^[A-Za-z][A-Za-z0-9_-]{0,63}$/.test(name) ? undefined : "simulated tool names must be provider-safe", + ProviderSafeName.test(name) ? undefined : "simulated tool names must be provider-safe", ), ) const ToolNamespace = Schema.NonEmptyString.check( Schema.makeFilter((namespace) => - namespace.split(".").every((segment) => /^[A-Za-z][A-Za-z0-9_-]{0,63}$/.test(segment)) + namespace.split(".").every((segment) => ProviderSafeName.test(segment)) ? undefined : "simulated tool namespaces must contain provider-safe segments", ), @@ -476,7 +477,7 @@ export namespace Backend { tools: Schema.Array(ToolRegistration).check( Schema.makeFilter((tools) => { const names = tools.map(exposedToolName) - if (names.some((name) => !/^[A-Za-z][A-Za-z0-9_-]{0,63}$/.test(name))) + if (names.some((name) => !ProviderSafeName.test(name))) return "simulated tool names including namespaces must be provider-safe" if (new Set(names).size !== names.length) return "simulated tool registrations must have unique exposed names" if ( diff --git a/packages/sdk-next/src/logging.ts b/packages/sdk-next/src/logging.ts index ba4b89b9ba..2838d97222 100644 --- a/packages/sdk-next/src/logging.ts +++ b/packages/sdk-next/src/logging.ts @@ -24,10 +24,17 @@ const levels: Record["logLevel"]> = { error: "Error", fatal: "Fatal", } +const levelNames = new Map["logLevel"], LogLevel>([ + [levels.trace, "trace"], + [levels.debug, "debug"], + [levels.info, "info"], + [levels.warn, "warn"], + [levels.error, "error"], + [levels.fatal, "fatal"], +]) function normalizeLevel(level: Logger.Options["logLevel"]): LogLevel | undefined { - const output = Object.fromEntries(Object.entries(levels).map(([name, effect]) => [effect, name])) - return output[level] as LogLevel + return levelNames.get(level) } export function layer(log?: LogOptions) { diff --git a/packages/sdk-next/test/logging.test.ts b/packages/sdk-next/test/logging.test.ts new file mode 100644 index 0000000000..0dae94939d --- /dev/null +++ b/packages/sdk-next/test/logging.test.ts @@ -0,0 +1,20 @@ +import { expect, test } from "bun:test" +import { Effect } from "effect" +import type { LogEntry } from "../src/logging" +import { layer } from "../src/logging" + +test("maps Effect log levels to SDK log levels", async () => { + const entries: LogEntry[] = [] + await Effect.runPromise( + Effect.gen(function* () { + yield* Effect.logTrace("trace") + yield* Effect.logDebug("debug") + yield* Effect.logInfo("info") + yield* Effect.logWarning("warn") + yield* Effect.logError("error") + yield* Effect.logFatal("fatal") + }).pipe(Effect.provide(layer({ level: "trace", emit: (entry) => entries.push(entry) }))), + ) + + expect(entries.map((entry) => entry.level)).toEqual(["trace", "debug", "info", "warn", "error", "fatal"]) +}) diff --git a/packages/server/src/handlers/message.ts b/packages/server/src/handlers/message.ts index 8bff2191b0..6339e6f942 100644 --- a/packages/server/src/handlers/message.ts +++ b/packages/server/src/handlers/message.ts @@ -3,7 +3,8 @@ import { Session } from "@opencode-ai/core/session" import { Effect, Schema } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" import { Api } from "../api" -import { InvalidCursorError, SessionNotFoundError, UnknownError } from "@opencode-ai/protocol/errors" +import { InvalidCursorError } from "@opencode-ai/protocol/errors" +import { failedMessageDecode, missingSession } from "./session-error" const DefaultMessagesLimit = 50 @@ -46,25 +47,8 @@ export const MessageHandler = HttpApiBuilder.group(Api, "server.message", (handl cursor: decoded ? { id: decoded.id, direction: decoded.direction } : undefined, }) .pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), - Effect.catchTag("Session.MessageDecodeError", (error) => { - const ref = `err_${crypto.randomUUID().slice(0, 8)}` - return Effect.logError("failed to decode session message").pipe( - Effect.annotateLogs({ ref, sessionID: error.sessionID, messageID: error.messageID }), - Effect.andThen( - Effect.fail( - new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref }), - ), - ), - ) - }), + Effect.catchTag("Session.NotFoundError", missingSession), + Effect.catchTag("Session.MessageDecodeError", failedMessageDecode), ) const first = messages[0] const last = messages.at(-1) diff --git a/packages/server/src/handlers/model.ts b/packages/server/src/handlers/model.ts index 6ad09fa96b..3b5c28e0ad 100644 --- a/packages/server/src/handlers/model.ts +++ b/packages/server/src/handlers/model.ts @@ -1,10 +1,18 @@ import { Catalog } from "@opencode-ai/core/catalog" -import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor" import { ServiceUnavailableError } from "@opencode-ai/protocol/errors" import { Effect } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" import { Api } from "../api" import { response } from "../location" +import { pluginReadiness } from "./plugin-readiness" + +const flushPlugins = pluginReadiness( + () => + new ServiceUnavailableError({ + message: "Model catalog initialization timed out", + service: "model.catalog", + }), +) export const ModelHandler = HttpApiBuilder.group(Api, "server.model", (handlers) => Effect.gen(function* () { @@ -12,19 +20,7 @@ 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", - }), - ), - }), - ) + yield* flushPlugins const catalog = yield* Catalog.Service return yield* response(catalog.model.available()) }), @@ -32,19 +28,7 @@ export const ModelHandler = HttpApiBuilder.group(Api, "server.model", (handlers) .handle( "model.default", 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", - }), - ), - }), - ) + yield* flushPlugins const catalog = yield* Catalog.Service return yield* response(catalog.model.default()) }), diff --git a/packages/server/src/handlers/permission.ts b/packages/server/src/handlers/permission.ts index 53aea5b310..0d5df461b6 100644 --- a/packages/server/src/handlers/permission.ts +++ b/packages/server/src/handlers/permission.ts @@ -13,6 +13,16 @@ function missingRequest(id: Permission.ID) { export const PermissionHandler = HttpApiBuilder.group(Api, "server.permission", (handlers) => Effect.gen(function* () { + const requireOwnedRequest = Effect.fnUntraced(function* ( + sessionID: Permission.Request["sessionID"], + requestID: Permission.ID, + ) { + const permission = yield* Permission.Service + const request = yield* permission.get(requestID) + if (!request || request.sessionID !== sessionID) return yield* missingRequest(requestID) + return { permission, request } + }) + return handlers .handle( "permission.request.list", @@ -60,19 +70,15 @@ export const PermissionHandler = HttpApiBuilder.group(Api, "server.permission", .handle( "session.permission.get", Effect.fn(function* (ctx) { - const permission = yield* Permission.Service - const request = yield* permission.get(ctx.params.requestID) - if (!request || request.sessionID !== ctx.params.sessionID) return yield* missingRequest(ctx.params.requestID) - return { data: request } + const owned = yield* requireOwnedRequest(ctx.params.sessionID, ctx.params.requestID) + return { data: owned.request } }), ) .handle( "session.permission.reply", Effect.fn(function* (ctx) { - const permission = yield* Permission.Service - const request = yield* permission.get(ctx.params.requestID) - if (!request || request.sessionID !== ctx.params.sessionID) return yield* missingRequest(ctx.params.requestID) - yield* permission + const owned = yield* requireOwnedRequest(ctx.params.sessionID, ctx.params.requestID) + yield* owned.permission .reply({ requestID: ctx.params.requestID, reply: ctx.payload.reply, message: ctx.payload.message }) .pipe(Effect.catchTag("Permission.NotFoundError", () => missingRequest(ctx.params.requestID))) return HttpApiSchema.NoContent.make() diff --git a/packages/server/src/handlers/plugin-readiness.ts b/packages/server/src/handlers/plugin-readiness.ts new file mode 100644 index 0000000000..0852b31702 --- /dev/null +++ b/packages/server/src/handlers/plugin-readiness.ts @@ -0,0 +1,13 @@ +import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor" +import { ServiceUnavailableError } from "@opencode-ai/protocol/errors" +import { Effect } from "effect" + +export function pluginReadiness(error: () => ServiceUnavailableError) { + return PluginSupervisor.Service.pipe( + Effect.flatMap((plugins) => plugins.flush), + Effect.timeoutOrElse({ + duration: "5 seconds", + orElse: () => Effect.fail(error()), + }), + ) +} diff --git a/packages/server/src/handlers/session-error.ts b/packages/server/src/handlers/session-error.ts new file mode 100644 index 0000000000..65be0f03ce --- /dev/null +++ b/packages/server/src/handlers/session-error.ts @@ -0,0 +1,20 @@ +import { Session } from "@opencode-ai/core/session" +import { SessionNotFoundError, UnknownError } from "@opencode-ai/protocol/errors" +import { Effect } from "effect" + +export function missingSession(error: Session.NotFoundError) { + return new SessionNotFoundError({ + sessionID: error.sessionID, + message: `Session not found: ${error.sessionID}`, + }) +} + +export function failedMessageDecode(error: Session.MessageDecodeError) { + const ref = `err_${crypto.randomUUID().slice(0, 8)}` + return Effect.logError("failed to decode session message").pipe( + Effect.annotateLogs({ ref, sessionID: error.sessionID, messageID: error.messageID }), + Effect.andThen( + Effect.fail(new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref })), + ), + ) +} diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index e5e71eef31..d4a82e1288 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -14,11 +14,11 @@ import { MessageNotFoundError, ServiceUnavailableError, SessionBusyError, - SessionNotFoundError, SkillNotFoundError, UnknownError, } from "@opencode-ai/protocol/errors" import { AbsolutePath } from "@opencode-ai/core/schema" +import { failedMessageDecode, missingSession } from "./session-error" const DefaultSessionsLimit = 50 @@ -26,16 +26,14 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl Effect.gen(function* () { const session = yield* Session.Service const transfer = yield* SessionTransfer.Service + const busySession = (error: Session.BusyError) => + new SessionBusyError({ + sessionID: error.sessionID, + message: `Session is busy: ${error.sessionID}`, + }) const pendingMutation = (effect: ReturnType, conflict: string) => effect.pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag( "Session.InboxConflictError", (error) => new ConflictError({ resource: error.inboxID, message: `${conflict}: ${error.inboxID}` }), @@ -130,27 +128,12 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl "session.export", Effect.fn(function* (ctx) { return { - data: yield* transfer.export({ sessionID: ctx.params.sessionID, sanitize: ctx.query.sanitize }).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), + data: yield* transfer + .export({ sessionID: ctx.params.sessionID, sanitize: ctx.query.sanitize }) + .pipe( + Effect.catchTag("Session.NotFoundError", missingSession), + Effect.catchTag("Session.MessageDecodeError", failedMessageDecode), ), - Effect.catchTag("Session.MessageDecodeError", (error) => { - const ref = `err_${crypto.randomUUID().slice(0, 8)}` - return Effect.logError("failed to decode session message").pipe( - Effect.annotateLogs({ ref, sessionID: error.sessionID, messageID: error.messageID }), - Effect.andThen( - Effect.fail( - new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref }), - ), - ), - ) - }), - ), } }), ) @@ -167,48 +150,25 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl "session.get", Effect.fn(function* (ctx) { return { - data: yield* session.get(ctx.params.sessionID).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), + data: yield* session + .get(ctx.params.sessionID) + .pipe(Effect.catchTag("Session.NotFoundError", missingSession)), } }), ) .handle( "session.remove", Effect.fn(function* (ctx) { - yield* session.remove(ctx.params.sessionID).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ) + yield* session.remove(ctx.params.sessionID).pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return HttpApiSchema.NoContent.make() }), ) .handle( "session.environment", Effect.fn(function* (ctx) { - yield* session.environment({ sessionID: ctx.params.sessionID, variables: ctx.payload.variables }).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ) + yield* session + .environment({ sessionID: ctx.params.sessionID, variables: ctx.payload.variables }) + .pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return HttpApiSchema.NoContent.make() }), ) @@ -217,14 +177,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl Effect.fn(function* (ctx) { return { data: yield* session.fork({ sessionID: ctx.params.sessionID, boundary: ctx.payload.boundary }).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag( "Session.MessageNotFoundError", (error) => @@ -245,48 +198,27 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.switchAgent", Effect.fn(function* (ctx) { - yield* session.switchAgent({ sessionID: ctx.params.sessionID, agent: ctx.payload.agent }).pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), - ) + yield* session + .switchAgent({ sessionID: ctx.params.sessionID, agent: ctx.payload.agent }) + .pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return HttpApiSchema.NoContent.make() }), ) .handle( "session.switchModel", Effect.fn(function* (ctx) { - yield* session.switchModel({ sessionID: ctx.params.sessionID, model: ctx.payload.model }).pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), - ) + yield* session + .switchModel({ sessionID: ctx.params.sessionID, model: ctx.payload.model }) + .pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return HttpApiSchema.NoContent.make() }), ) .handle( "session.rename", Effect.fn(function* (ctx) { - yield* session.rename({ sessionID: ctx.params.sessionID, title: ctx.payload.title }).pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), - ) + yield* session + .rename({ sessionID: ctx.params.sessionID, title: ctx.payload.title }) + .pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return HttpApiSchema.NoContent.make() }), ) @@ -301,14 +233,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl delivery: ctx.payload.delivery, }) .pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag("Session.DestinationNotFoundError", (error) => Effect.fail(new InvalidRequestError({ message: `Directory does not exist: ${error.directory}` })), ), @@ -336,14 +261,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl resume: ctx.payload.resume, }) .pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag("Session.PromptConflictError", (error) => Effect.fail( new ConflictError({ @@ -381,14 +299,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl resume: ctx.payload.resume, }) .pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag("Command.NotFoundError", (error) => Effect.fail( new CommandNotFoundError({ @@ -434,14 +345,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl resume: ctx.payload.resume, }) .pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag("Session.SkillNotFoundError", (error) => Effect.fail(new SkillNotFoundError({ skill: error.skill, message: `Skill not found: ${error.skill}` })), ), @@ -463,14 +367,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl resume: ctx.payload.resume, }) .pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag("Session.SyntheticConflictError", (error) => Effect.fail( new ConflictError({ @@ -488,16 +385,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl Effect.fn(function* (ctx) { yield* session .shell({ sessionID: ctx.params.sessionID, id: ctx.payload.id, command: ctx.payload.command }) - .pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), - ) + .pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return HttpApiSchema.NoContent.make() }), ) @@ -508,14 +396,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl data: yield* session .compact({ sessionID: ctx.params.sessionID, id: ctx.payload.id, delivery: ctx.payload.delivery }) .pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag("Session.CompactionConflictError", (error) => Effect.fail( new ConflictError({ @@ -531,16 +412,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.wait", Effect.fn(function* (ctx) { - yield* session.wait(ctx.params.sessionID).pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), - ) + yield* session.wait(ctx.params.sessionID).pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return HttpApiSchema.NoContent.make() }), ) @@ -554,14 +426,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl }) return { data: yield* session.revert.stage({ ...ctx.params, ...ctx.payload }).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), + Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag( "Session.MessageNotFoundError", (error) => @@ -571,14 +436,7 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl message: `Message not found: ${error.messageID}`, }), ), - Effect.catchTag( - "Session.BusyError", - (error) => - new SessionBusyError({ - sessionID: error.sessionID, - message: `Session is busy: ${error.sessionID}`, - }), - ), + Effect.catchTag("Session.BusyError", busySession), Effect.catchTag("Snapshot.Error", (error) => { const ref = `err_${crypto.randomUUID().slice(0, 8)}` return Effect.logError("failed to stage session revert", { cause: error }).pipe( @@ -601,22 +459,8 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl Effect.fn(function* (ctx) { yield* Effect.log("session.revert.clear", { sessionID: ctx.params.sessionID }) yield* session.revert.clear(ctx.params.sessionID).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - Effect.catchTag( - "Session.BusyError", - (error) => - new SessionBusyError({ - sessionID: error.sessionID, - message: `Session is busy: ${error.sessionID}`, - }), - ), + Effect.catchTag("Session.NotFoundError", missingSession), + Effect.catchTag("Session.BusyError", busySession), Effect.catchTag("Snapshot.Error", (error) => { const ref = `err_${crypto.randomUUID().slice(0, 8)}` return Effect.logError("failed to clear session revert", { cause: error }).pipe( @@ -638,24 +482,12 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl "session.revert.commit", Effect.fn(function* (ctx) { yield* Effect.log("session.revert.commit", { sessionID: ctx.params.sessionID }) - yield* session.revert.commit(ctx.params.sessionID).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - Effect.catchTag( - "Session.BusyError", - (error) => - new SessionBusyError({ - sessionID: error.sessionID, - message: `Session is busy: ${error.sessionID}`, - }), - ), - ) + yield* session.revert + .commit(ctx.params.sessionID) + .pipe( + Effect.catchTag("Session.NotFoundError", missingSession), + Effect.catchTag("Session.BusyError", busySession), + ) return HttpApiSchema.NoContent.make() }), ) @@ -663,27 +495,12 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl "session.context", Effect.fn(function* (ctx) { return { - data: yield* session.context(ctx.params.sessionID).pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), + data: yield* session + .context(ctx.params.sessionID) + .pipe( + Effect.catchTag("Session.NotFoundError", missingSession), + Effect.catchTag("Session.MessageDecodeError", failedMessageDecode), ), - Effect.catchTag("Session.MessageDecodeError", (error) => { - const ref = `err_${crypto.randomUUID().slice(0, 8)}` - return Effect.logError("failed to decode session message").pipe( - Effect.annotateLogs({ ref, sessionID: error.sessionID, messageID: error.messageID }), - Effect.andThen( - Effect.fail( - new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref }), - ), - ), - ) - }), - ), } }), ) @@ -691,16 +508,9 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl "session.inbox.list", Effect.fn(function* (ctx) { return { - data: yield* session.inbox(ctx.params.sessionID).pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), - ), + data: yield* session + .inbox(ctx.params.sessionID) + .pipe(Effect.catchTag("Session.NotFoundError", missingSession)), } }), ) @@ -757,32 +567,22 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.generate", Effect.fn(function* (ctx) { - const text = yield* session.generate({ sessionID: ctx.params.sessionID, prompt: ctx.payload.prompt }).pipe( - Effect.mapError((error) => - error._tag === "Session.NotFoundError" - ? new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }) - : new ServiceUnavailableError({ message: error.message, service: "session generation" }), - ), - ) + const text = yield* session + .generate({ sessionID: ctx.params.sessionID, prompt: ctx.payload.prompt }) + .pipe( + Effect.mapError((error) => + error._tag === "Session.NotFoundError" + ? missingSession(error) + : new ServiceUnavailableError({ message: error.message, service: "session generation" }), + ), + ) return { data: { text } } }), ) .handle( "session.log", Effect.fn(function* (ctx) { - yield* session.get(ctx.params.sessionID).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ) + yield* session.get(ctx.params.sessionID).pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return session .log({ sessionID: ctx.params.sessionID, after: ctx.query.after, follow: ctx.query.follow }) .pipe(Stream.orDie) @@ -798,32 +598,14 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.background", Effect.fn(function* (ctx) { - yield* session.background(ctx.params.sessionID).pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ), - ) + yield* session.background(ctx.params.sessionID).pipe(Effect.catchTag("Session.NotFoundError", missingSession)) return HttpApiSchema.NoContent.make() }), ) .handle( "session.message", Effect.fn(function* (ctx) { - yield* session.get(ctx.params.sessionID).pipe( - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ) + yield* session.get(ctx.params.sessionID).pipe(Effect.catchTag("Session.NotFoundError", missingSession)) const message = yield* session.message(ctx.params) if (message) return { data: message } return yield* new MessageNotFoundError({ diff --git a/packages/server/src/handlers/websearch.ts b/packages/server/src/handlers/websearch.ts index 8baa385c70..4245c4da6c 100644 --- a/packages/server/src/handlers/websearch.ts +++ b/packages/server/src/handlers/websearch.ts @@ -1,33 +1,26 @@ -import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor" import { WebSearch } from "@opencode-ai/core/websearch" import { InvalidRequestError, ServiceUnavailableError } from "@opencode-ai/protocol/errors" import { Effect } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" import { Api } from "../api" import { response } from "../location" +import { pluginReadiness } from "./plugin-readiness" + +const awaitPlugins = pluginReadiness( + () => + new ServiceUnavailableError({ + message: "Web search provider initialization timed out", + service: "websearch", + }), +).pipe(Effect.withSpan("server.websearch.awaitPlugins")) export const WebSearchHandler = HttpApiBuilder.group(Api, "server.websearch", (handlers) => Effect.gen(function* () { - const awaitPlugins = Effect.fn("server.websearch.awaitPlugins")(function* () { - const plugins = yield* PluginSupervisor.Service - yield* plugins.flush.pipe( - Effect.timeoutOrElse({ - duration: "5 seconds", - orElse: () => - Effect.fail( - new ServiceUnavailableError({ - message: "Web search provider initialization timed out", - service: "websearch", - }), - ), - }), - ) - }) return handlers .handle( "websearch.providers", Effect.fn("server.websearch.providers")(function* () { - yield* awaitPlugins() + yield* awaitPlugins const websearch = yield* WebSearch.Service return yield* response(websearch.providers()) }), @@ -35,7 +28,7 @@ export const WebSearchHandler = HttpApiBuilder.group(Api, "server.websearch", (h .handle( "websearch.query", Effect.fn("server.websearch.query")(function* (request) { - yield* awaitPlugins() + yield* awaitPlugins const websearch = yield* WebSearch.Service return yield* response( websearch.query(request.payload).pipe( diff --git a/packages/server/test/handler-policy.test.ts b/packages/server/test/handler-policy.test.ts new file mode 100644 index 0000000000..22dd3daca8 --- /dev/null +++ b/packages/server/test/handler-policy.test.ts @@ -0,0 +1,61 @@ +import { Session } from "@opencode-ai/core/session" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor" +import { SessionNotFoundError, ServiceUnavailableError, UnknownError } from "@opencode-ai/protocol/errors" +import { expect, test } from "bun:test" +import { Effect, Layer, Logger, References } from "effect" +import { pluginReadiness } from "../src/handlers/plugin-readiness" +import { failedMessageDecode, missingSession } from "../src/handlers/session-error" + +test("yieldable session errors preserve the handler failure policy", async () => { + const sessionID = Session.ID.create() + const error = await Effect.runPromise( + Effect.fail(new Session.NotFoundError({ sessionID })).pipe( + Effect.catchTag("Session.NotFoundError", missingSession), + Effect.flip, + ), + ) + + expect(error).toBeInstanceOf(SessionNotFoundError) + expect(error).toMatchObject({ sessionID, message: `Session not found: ${sessionID}` }) +}) + +test("message decode policy preserves its reference and log annotations", async () => { + const sessionID = Session.ID.create() + const messageID = SessionMessage.ID.create() + const messages: unknown[] = [] + const annotations: Array> = [] + const logger = Logger.make((options) => { + messages.push(options.message) + annotations.push({ ...options.fiber.getRef(References.CurrentLogAnnotations) }) + }) + const error = await Effect.runPromise( + Effect.fail(new Session.MessageDecodeError({ sessionID, messageID })).pipe( + Effect.catchTag("Session.MessageDecodeError", failedMessageDecode), + Effect.flip, + Effect.provide(Logger.layer([logger], { mergeWithExisting: false })), + ), + ) + + expect(error).toBeInstanceOf(UnknownError) + expect(error.message).toBe("Unexpected server error. Check server logs for details.") + expect(error.ref).toMatch(/^err_[0-9a-f]{8}$/) + expect(messages).toEqual([["failed to decode session message"]]) + expect(annotations).toEqual([{ ref: error.ref, sessionID, messageID }]) +}) + +test("plugin readiness stays lazy and resolves the supervisor for every execution", async () => { + let flushes = 0 + const readiness = pluginReadiness( + () => new ServiceUnavailableError({ message: "initialization timed out", service: "test" }), + ) + const layer = Layer.succeed(PluginSupervisor.Service, { + flush: Effect.sync(() => { + flushes++ + }), + }) + + expect(flushes).toBe(0) + await Effect.runPromise(Effect.all([readiness, readiness], { concurrency: 1 }).pipe(Effect.provide(layer))) + expect(flushes).toBe(2) +})