Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 75 Packages Projects Releases Wiki Activity

refactor: simplify server client helpers (#43976)

Browse Source
This commit is contained in:
Kit Langton
2026-08-21 15:37:19 -04:00
committed by GitHub
parent 7fd1eee35a
commit 97d3cd0b3a
13 changed files with 239 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/client/src/effect/service.ts
+4 -8
View File
@@ -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.
packages/client/src/promise/service.ts
+1 -9
View File
@@ -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<LocalService | undefined> {
return (await probeResult(info, allowLegacy)).service
}
async function probeResult(info: Info, allowLegacy = false, timeout = defaultEnsureTiming.requestTimeout) {
const endpoint = {
url: info.url,
packages/protocol/src/simulation.ts
+4 -3
View File
@@ -444,14 +444,15 @@ export namespace Backend {
])
export type ToolContent = Schema.Schema.Type<typeof ToolContent>
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 (
packages/sdk-next/src/logging.ts
+9 -2
View File
@@ -24,10 +24,17 @@ const levels: Record<LogLevel, Logger.Options<unknown>["logLevel"]> = {
error: "Error",
fatal: "Fatal",
}
const levelNames = new Map<Logger.Options<unknown>["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<unknown>["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) {
packages/sdk-next/test/logging.test.ts
+20
View File
@@ -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"])
})
packages/server/src/handlers/message.ts
+4 -20
View File
@@ -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)
packages/server/src/handlers/model.ts
+11 -27
View File
@@ -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())
}),
packages/server/src/handlers/permission.ts
+14 -8
View File
@@ -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()
packages/server/src/handlers/plugin-readiness.ts
+13
View File
@@ -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()),
}),
)
}
packages/server/src/handlers/session-error.ts
+20
View File
@@ -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 })),
),
)
}
packages/server/src/handlers/session.ts
+67 -285
View File
@@ -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<typeof session.cancelInbox>, 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({
packages/server/src/handlers/websearch.ts
+11 -18
View File
@@ -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(
packages/server/test/handler-policy.test.ts
+61
View File
@@ -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<Record<string, unknown>> = []
const logger = Logger.make<unknown, void>((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)
})