ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: expose background service lifecycle (#36895)

Browse source
This commit is contained in:
Kit Langton 2026-07-14 16:38:22 -04:00 • committed by GitHub
commit ece2b16cdf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
36 changed files with 2421 additions and 293 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/cli/package.json
View file

@ -55,6 +55,7 @@
},
"devDependencies": {
"@opencode-ai/script": "workspace:*",
"@opencode-ai/protocol": "workspace:*",
"@tsconfig/bun": "catalog:",
"@types/bun": "catalog:",
"@types/semver": "catalog:",

143
packages/cli/script/service-smoke.ts Normal file
View file

@ -0,0 +1,143 @@
#!/usr/bin/env bun
import { Service } from "@opencode-ai/client/effect"
import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
import { Schema } from "effect"
import fs from "node:fs/promises"
import os from "node:os"
import path from "node:path"
const target = `cli-${process.platform === "win32" ? "windows" : process.platform}-${process.arch}`
const directory = path.join(import.meta.dir, "..", "dist", target, "bin")
const binary = path.join(directory, `opencode2${process.platform === "win32" ? ".exe" : ""}`)
if (!(await Bun.file(binary).exists())) throw new Error(`Missing compiled CLI in ${directory}`)
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-smoke-"))
const env = {
...process.env,
HOME: root,
USERPROFILE: root,
OPENCODE_DB: path.join(root, "opencode.db"),
OPENCODE_TEST_HOME: root,
XDG_CACHE_HOME: path.join(root, "cache"),
XDG_CONFIG_HOME: path.join(root, "config"),
XDG_DATA_HOME: path.join(root, "data"),
XDG_STATE_HOME: path.join(root, "state"),
}
const processes: Array<ReturnType<typeof Bun.spawn>> = []
const errors: Array<Promise<string>> = []
let failure: unknown
try {
spawnService()
spawnService()
const registration = await waitForRegistration()
const info = await Schema.decodeUnknownPromise(Service.Info)(await Bun.file(registration).json())
if (info.id === undefined || info.password === undefined) throw new Error("Registration is missing service identity")
const credential = btoa(`opencode:${info.password}`)
const headers = { authorization: "Basic " + credential }
const token = encodeURIComponent(credential)
const health = await waitForReady(info.url, headers)
if (health.pid !== info.pid || health.instanceID !== info.id)
throw new Error("Health identity does not match registration")
const tokenHealth = await fetch(
new URL(`/api/health?auth_token=${token}`, info.url),
{ signal: AbortSignal.timeout(5_000) },
)
if (tokenHealth.status !== 200) throw new Error("Compiled service rejected query authentication")
const tokenOpenApi = await fetch(
new URL(`/openapi.json?auth_token=${token}`, info.url),
{ signal: AbortSignal.timeout(5_000) },
)
if (tokenOpenApi.status !== 200) throw new Error("Compiled application rejected query authentication")
const unauthorizedHealth = await fetch(new URL("/api/health", info.url), {
signal: AbortSignal.timeout(5_000),
})
if (unauthorizedHealth.status !== 401) throw new Error("Compiled service exposed health without authentication")
const unauthorizedOpenApi = await fetch(new URL("/openapi.json", info.url), {
signal: AbortSignal.timeout(5_000),
})
if (unauthorizedOpenApi.status !== 401) throw new Error("Compiled service exposed application routes without authentication")
const unauthorizedStop = await fetch(new URL("/api/service/stop", info.url), {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ instanceID: info.id }),
signal: AbortSignal.timeout(5_000),
})
if (unauthorizedStop.status !== 401) throw new Error("Compiled service accepted unauthenticated stop")
const winner = processes.find((process) => process.pid === info.pid)
const loser = processes.find((process) => process.pid !== info.pid)
if (!winner || !loser) throw new Error("Compiled contenders did not elect one registered owner")
if (!(await exitsWithin(loser, 10_000))) throw new Error("Losing compiled contender did not exit")
const stopped = await Schema.decodeUnknownPromise(ServiceStatus.StopResponse)(
await fetch(new URL("/api/service/stop", info.url), {
method: "POST",
headers: { ...headers, "content-type": "application/json" },
body: JSON.stringify({ instanceID: info.id, targetVersion: "smoke-next" }),
signal: AbortSignal.timeout(5_000),
}).then((response) => response.json()),
)
if (!stopped.accepted) throw new Error("Compiled service rejected exact-instance stop")
if (!(await exitsWithin(winner, 10_000))) throw new Error("Compiled service did not stop")
for (let attempt = 0; attempt < 200 && (await Bun.file(registration).exists()); attempt++) await Bun.sleep(25)
if (await Bun.file(registration).exists()) throw new Error("Compiled service registration was not removed")
} catch (cause) {
failure = cause
} finally {
processes.forEach((process) => process.kill())
await Promise.all(processes.map((process) => process.exited))
}
const output = await Promise.all(errors)
await fs.rm(root, { recursive: true, force: true })
if (failure)
throw new Error(output.filter(Boolean).join("\n") || "Compiled service lifecycle smoke test failed", {
cause: failure,
})
function spawnService() {
const process = Bun.spawn([binary, "serve", "--service"], { env, stdout: "ignore", stderr: "pipe" })
processes.push(process)
errors.push(new Response(process.stderr).text())
return process
}
async function waitForRegistration() {
const directory = path.join(root, "state", "opencode")
for (let attempt = 0; attempt < 400; attempt++) {
const files = await fs.readdir(directory).catch(() => [])
const file = files.find(
(file) => file === "service.json" || (file.startsWith("service-") && file.endsWith(".json")),
)
if (file) return path.join(directory, file)
await Bun.sleep(25)
}
throw new Error("Compiled service did not publish registration")
}
async function waitForReady(url: string, headers: HeadersInit) {
const deadline = Date.now() + 20_000
while (Date.now() < deadline) {
const health = await fetch(new URL("/api/health", url), {
headers,
signal: AbortSignal.timeout(1_000),
})
.then((response) => response.json())
.then(Schema.decodeUnknownPromise(ServiceStatus.Health))
.catch(() => undefined)
if (health === undefined) {
await Bun.sleep(25)
continue
}
if (health.status.type === "ready") return health
if (health.status.type === "failed") throw new Error(health.status.message)
await Bun.sleep(25)
}
throw new Error("Compiled service did not become ready")
}
function exitsWithin(process: Bun.Subprocess, milliseconds: number) {
return Promise.race([process.exited.then(() => true), Bun.sleep(milliseconds).then(() => false)])
}

2
packages/cli/src/commands/handlers/service/restart.ts
View file

@ -9,7 +9,7 @@ export default Runtime.handler(
Commands.commands.service.commands.restart,
Effect.fn("cli.service.restart")(function* () {
const options = yield* ServiceConfig.options()
yield* Service.stop(options)
yield* Service.stop(options, { targetVersion: options.version })
const transport = yield* Service.start(options)
process.stdout.write(transport.url + EOL)
}),

10
packages/cli/src/commands/handlers/service/status.ts
View file

@ -8,7 +8,13 @@ import { ServiceConfig } from "../../../services/service-config"
export default Runtime.handler(
Commands.commands.service.commands.status,
Effect.fn("cli.service.status")(function* () {
const found = yield* Service.discover(yield* ServiceConfig.options())
process.stdout.write((found ? found.url : "stopped") + EOL)
const options = yield* ServiceConfig.options()
const status = yield* Service.status(options)
if (status.type !== "ready") {
process.stdout.write(status.type + EOL)
return
}
const found = yield* Service.discover({ ...options, version: undefined })
process.stdout.write((found?.url ?? status.type) + EOL)
}),
)

112
packages/cli/src/server-process.ts
View file

@ -7,11 +7,10 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node"
import { Global } from "@opencode-ai/core/global"
import { InstallationVersion } from "@opencode-ai/core/installation/version"
import { AppProcess } from "@opencode-ai/core/process"
import { EffectFlock } from "@opencode-ai/core/util/effect-flock"
import { start } from "@opencode-ai/server/process"
import { ProcessLock } from "@opencode-ai/core/util/process-lock"
import { randomBytes, randomUUID } from "node:crypto"
import path from "node:path"
import { Effect, Exit, FileSystem, Logger, Option, Redacted, Schedule, Schema, Scope } from "effect"
import { Effect, FileSystem, Logger, Option, Redacted, Schedule, Schema } from "effect"
import { HttpServer } from "effect/unstable/http"
import { Env } from "./env"
import { ServiceConfig } from "./services/service-config"
@ -28,7 +27,7 @@ export type Options = {
export const run = Effect.fn("cli.server-process.run")((options: Options) =>
processEffect(options).pipe(
Effect.provide(Updater.layer),
Effect.provide(AppNodeBuilder.build(LayerNode.group([Global.node, AppProcess.node, EffectFlock.node]))),
Effect.provide(AppNodeBuilder.build(LayerNode.group([Global.node, AppProcess.node]))),
Effect.provide(NodeServices.layer),
),
)
@ -38,15 +37,15 @@ const processEffect = Effect.fnUntraced(function* (options: Options) {
return yield* Effect.scoped(
Effect.gen(function* () {
const serviceOptions = options.mode === "service" ? yield* ServiceConfig.options() : undefined
const lockScope = serviceOptions === undefined ? undefined : yield* acquireServiceLock(serviceOptions.file)
if (
serviceOptions !== undefined &&
lockScope !== undefined &&
(yield* Service.discover(serviceOptions)) !== undefined
) {
yield* Scope.close(lockScope, Exit.void)
return
if (serviceOptions !== undefined) {
const acquired = yield* ProcessLock.acquire(serviceOptions.file + ".lock").pipe(
Effect.as(true),
Effect.catchTag("ProcessLockHeldError", () => Effect.succeed(false)),
)
if (!acquired) return yield* Effect.void
if ((yield* Service.discover(serviceOptions)) !== undefined) return yield* Effect.void
}
const { start } = yield* Effect.promise(() => import("@opencode-ai/server/process"))
const environmentPassword = yield* Env.password
// Keep the lease credential out of the environment inherited by tools.
if (options.mode === "stdio") {
@ -61,77 +60,82 @@ const processEffect = Effect.fnUntraced(function* (options: Options) {
? Redacted.value(environmentPassword)
: randomBytes(32).toString("base64url")
if (!password) return yield* Effect.fail(new Error("Missing server password"))
const address = yield* start({
const instanceID = randomUUID()
const server = yield* start({
hostname: options.hostname ?? config.hostname ?? "127.0.0.1",
port: Option.fromNullishOr(options.port ?? config.port),
password,
restartContinuity: options.mode === "service",
instanceID,
service:
serviceOptions === undefined
? undefined
: { onListen: (address) => register(address, password, instanceID, serviceOptions.file) },
}).pipe(Effect.provide(Logger.layer([], { mergeWithExisting: false })))
if (lockScope !== undefined) {
yield* register(address, password)
yield* Scope.close(lockScope, Exit.void)
}
const url = HttpServer.formatAddress(address)
const url = HttpServer.formatAddress(server.address)
console.log(options.mode === "stdio" ? JSON.stringify({ url }) : `server listening on ${url}`)
if (options.mode === "default" && !environmentPassword) console.log(`server password ${password}`)
const updater = yield* Updater.Service
yield* updater.check().pipe(Effect.schedule(Schedule.spaced("10 minutes")), Effect.forkScoped)
return yield* options.mode === "stdio" ? waitForStdinClose() : Effect.never
return yield* options.mode === "service"
? server.shutdown
: options.mode === "stdio"
? waitForStdinClose()
: Effect.never
}).pipe(Effect.annotateLogs({ role: "server" })),
)
})
const acquireServiceLock = Effect.fnUntraced(function* (file: string) {
const flock = yield* EffectFlock.Service
const scope = yield* Scope.make()
yield* Effect.addFinalizer((exit) => Scope.close(scope, exit))
yield* flock
.acquire(`service:${file}`, undefined, { staleMs: 3_000, timeoutMs: 3_000 })
.pipe(Effect.provideService(Scope.Scope, scope))
return scope
})
// The latest atomic registration wins. A displaced process notices the new id,
// exits, and cannot remove its successor's registration from its finalizer.
const infoJson = Schema.fromJsonString(Service.Info)
const encodeInfo = Schema.encodeEffect(infoJson)
const decodeInfo = Schema.decodeUnknownEffect(infoJson)
const register = Effect.fnUntraced(function* (address: HttpServer.Address, password: string) {
const register = Effect.fnUntraced(function* (
address: HttpServer.Address,
password: string,
id: string,
file: string,
) {
const fs = yield* FileSystem.FileSystem
const options = yield* ServiceConfig.options()
const id = randomUUID()
const temp = options.file + "." + id + ".tmp"
yield* fs.makeDirectory(path.dirname(options.file), { recursive: true })
const encoded = yield* encodeInfo({
const temp = file + "." + id + ".tmp"
yield* fs.makeDirectory(path.dirname(file), { recursive: true })
const info = {
id,
version: InstallationVersion,
url: HttpServer.formatAddress(address),
pid: process.pid,
password,
})
yield* fs.writeFileString(temp, encoded, { mode: 0o600 })
yield* fs.rename(temp, options.file)
const currentID = fs.readFileString(options.file).pipe(
}
const encoded = yield* encodeInfo(info)
const publish = fs.writeFileString(temp, encoded, { mode: 0o600 }).pipe(Effect.andThen(fs.rename(temp, file)))
yield* publish
const current = fs.readFileString(file).pipe(
Effect.flatMap(decodeInfo),
Effect.map((info) => info.id),
Effect.orElseSucceed(() => undefined),
)
yield* currentID.pipe(
Effect.flatMap((current) =>
current === id
? Effect.void
: Effect.try({ try: () => process.kill(process.pid, "SIGTERM"), catch: (cause) => cause }).pipe(Effect.ignore),
),
Effect.repeat(Schedule.spaced("10 seconds")),
Effect.forkScoped,
)
const assertRegistration = Effect.gen(function* () {
const found = yield* current
if (
found !== undefined &&
found.id === info.id &&
found.version === info.version &&
found.url === info.url &&
found.pid === info.pid &&
found.password === info.password
)
return
yield* publish
})
yield* Effect.addFinalizer(() =>
currentID.pipe(
Effect.flatMap((current) => (current === id ? fs.remove(options.file) : Effect.void)),
current.pipe(
Effect.flatMap((current) => (current?.id === id ? fs.remove(file) : Effect.void)),
Effect.ignore,
),
)
yield* assertRegistration.pipe(
Effect.catchCause((cause) => Effect.logWarning("failed to reassert service registration", { cause })),
Effect.repeat(Schedule.spaced("5 seconds")),
Effect.forkScoped,
)
})
function waitForStdinClose() {

24
packages/cli/src/services/server.ts
View file

@ -16,7 +16,7 @@ export type Args = {
export type Resolved = {
readonly endpoint: Service.Endpoint
readonly reconnect?: (attempt: number) => Promise<Service.Endpoint>
readonly reconnect?: (onStatus: (status: Service.Status) => void, signal: AbortSignal) => Promise<Service.Endpoint>
readonly reload?: () => Promise<void>
}
@ -27,9 +27,7 @@ export const resolve = Effect.fn("cli.server.resolve")(function* (args: Args) {
const password = yield* Env.password
const endpoint = {
url: args.server,
auth: password
? { type: "basic" as const, username: "opencode", password: Redacted.value(password) }
: undefined,
auth: password ? { type: "basic" as const, username: "opencode", password: Redacted.value(password) } : undefined,
} satisfies Service.Endpoint
const client = OpenCode.make({ baseUrl: endpoint.url, headers: Service.headers(endpoint) })
const health = yield* Effect.tryPromise({
@ -51,19 +49,14 @@ export const resolve = Effect.fn("cli.server.resolve")(function* (args: Args) {
const reconnectOptions = { ...options, version: undefined }
return {
endpoint,
reconnect: (attempt) =>
Effect.runPromise(
Effect.gen(function* () {
if (attempt > 3) return yield* Service.start(reconnectOptions)
const endpoint = yield* Service.discover(reconnectOptions)
if (endpoint !== undefined) return endpoint
return yield* Effect.fail(new Error("Background server is unavailable"))
}).pipe(Effect.provide(NodeFileSystem.layer)),
),
reconnect: (onStatus, signal) =>
Effect.runPromise(Service.start({ ...reconnectOptions, onStatus }).pipe(Effect.provide(NodeFileSystem.layer)), {
signal,
}),
reload: () =>
Effect.runPromise(
Effect.gen(function* () {
yield* Service.stop(options)
yield* Service.stop(options, { targetVersion: options.version })
yield* Service.start(options)
}).pipe(Effect.provide(NodeFileSystem.layer)),
),
@ -80,7 +73,8 @@ const resolveManaged = Effect.fnUntraced(function* (
const compatible = yield* Service.discover(options)
if (compatible !== undefined) return compatible
const existing = yield* Service.discover({ ...options, version: undefined })
if (existing !== undefined) return yield* Effect.fail(new Error("Background server version does not match this client"))
if (existing !== undefined)
return yield* Effect.fail(new Error("Background server version does not match this client"))
return yield* Service.start(options)
})

58
packages/cli/src/services/service-config.ts
View file

@ -1,7 +1,8 @@
import { Global } from "@opencode-ai/core/global"
import { InstallationChannel, InstallationVersion } from "@opencode-ai/core/installation/version"
import { Hash } from "@opencode-ai/core/util/hash"
import { Service } from "@opencode-ai/client/effect"
import { Effect, FileSystem, Schema } from "effect"
import { Effect, FileSystem, Option, Schema } from "effect"
import { randomBytes } from "crypto"
import path from "path"
@ -20,25 +21,63 @@ const keys = ["hostname", "port", "password"] as const
type Key = (typeof keys)[number]
const decodeInfo = Schema.decodeUnknownEffect(Schema.fromJsonString(Info))
const decodeRegistration = Schema.decodeUnknownEffect(Schema.fromJsonString(Service.Info))
export function filename(channel = InstallationChannel) {
if (channel === "latest") return "service.json"
if (channel === "local") return "service-local.json"
return `service-${Hash.fast(channel)}.json`
}
export function versionBelongsToChannel(
version: string | undefined,
channel = InstallationChannel,
installedVersion = InstallationVersion,
) {
if (version === undefined) return false
if (version === installedVersion) return true
const prefix = `0.0.0-${channel}-`
if (!version.startsWith(prefix)) return false
return /^\d+(?:\.\d+)?$/.test(version.slice(prefix.length))
}
export const migrateRegistration = Effect.fnUntraced(function* (
legacy: string,
file: string,
channel = InstallationChannel,
installedVersion = InstallationVersion,
) {
if (channel === "latest" || channel === "local") return
const fs = yield* FileSystem.FileSystem
const text = yield* fs.readFileString(legacy).pipe(Effect.option)
if (Option.isNone(text)) return
const registration = yield* decodeRegistration(text.value).pipe(Effect.option)
if (Option.isNone(registration)) return
if (!versionBelongsToChannel(registration.value.version, channel, installedVersion)) return
yield* fs.writeFileString(file, text.value, { flag: "wx", mode: 0o600 }).pipe(Effect.ignore)
})
function configKey(key: string): Key {
if (keys.includes(key as Key)) return key as Key
if (key === "hostname" || key === "port" || key === "password") return key
throw new Error(`Unknown service config key: ${key}`)
}
const env = Effect.gen(function* () {
const paths = Effect.gen(function* () {
const fs = yield* FileSystem.FileSystem
const global = yield* Global.Service
const filename = InstallationChannel === "local" ? "service-local.json" : "service.json"
const name = filename()
const file = path.join(global.state, name)
return {
fs,
file: path.join(global.state, filename),
configFile: path.join(global.config, filename),
file,
legacyFile: path.join(global.state, "service.json"),
configFile: path.join(global.config, name),
}
})
export const options = Effect.fnUntraced(function* () {
const { file } = yield* env
const { file, legacyFile } = yield* paths
yield* migrateRegistration(legacyFile, file)
const compiled = path.basename(process.execPath).replace(/\.exe$/, "") !== "bun"
const entrypoint = compiled ? undefined : process.argv[1]
if (!compiled && entrypoint === undefined) return yield* Effect.fail(new Error("Failed to resolve CLI entrypoint"))
@ -50,7 +89,7 @@ export const options = Effect.fnUntraced(function* () {
})
export const read = Effect.fn("cli.service-config.read")(function* () {
const { fs, configFile } = yield* env
const { fs, configFile } = yield* paths
return yield* fs.readFileString(configFile).pipe(
Effect.flatMap(decodeInfo),
Effect.catch(() => Effect.succeed({} as Info)),
@ -58,7 +97,7 @@ export const read = Effect.fn("cli.service-config.read")(function* () {
})
const write = Effect.fn("cli.service-config.write")(function* (value: Info) {
const { fs, configFile } = yield* env
const { fs, configFile } = yield* paths
const temp = configFile + ".tmp"
yield* fs.makeDirectory(path.dirname(configFile), { recursive: true })
yield* fs.writeFileString(temp, JSON.stringify(value, null, 2) + "\n", { mode: 0o600 })
@ -93,6 +132,7 @@ export const get = Effect.fn("cli.service-config.get")(function* (key?: string)
return yield* password()
}
}
throw new Error(`Unknown service config key: ${key}`)
})
export const set = Effect.fn("cli.service-config.set")(function* (key: string, value: string) {

175
packages/cli/test/service.test.ts
View file

@ -35,6 +35,61 @@ test("local channel stores service config with the local service filename", asyn
}
})
test("service filenames isolate installation channels", () => {
expect(ServiceConfig.filename("latest")).toBe("service.json")
expect(ServiceConfig.filename("local")).toBe("service-local.json")
expect(ServiceConfig.filename("preview-a")).not.toBe(ServiceConfig.filename("preview-b"))
expect(ServiceConfig.filename("preview-a")).not.toBe(ServiceConfig.filename("latest"))
expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-1234", "preview-a")).toBe(true)
expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-1234.2", "preview-a")).toBe(true)
expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-other-1234", "preview-a")).toBe(false)
expect(ServiceConfig.versionBelongsToChannel("1.2.3", "preview-a")).toBe(false)
})
test("preview registration migration never moves stable discovery", async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-migration-"))
const legacy = path.join(root, "service.json")
const target = path.join(root, ServiceConfig.filename("preview-a"))
try {
await fs.writeFile(
legacy,
JSON.stringify({ id: "old-preview", version: "0.0.0-preview-a-1234", url: "http://localhost:4096", pid: 1 }),
)
await Effect.runPromise(
ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe(
Effect.provide(NodeFileSystem.layer),
),
)
expect(await Bun.file(legacy).exists()).toBe(true)
expect(await Bun.file(target).json()).toMatchObject({ id: "old-preview" })
await fs.rm(target)
await fs.writeFile(legacy, JSON.stringify({ id: "stable", version: "1.2.3", url: "http://localhost:4096", pid: 1 }))
await Effect.runPromise(
ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe(
Effect.provide(NodeFileSystem.layer),
),
)
expect(await Bun.file(legacy).exists()).toBe(true)
expect(await Bun.file(target).exists()).toBe(false)
await fs.writeFile(
legacy,
JSON.stringify({ id: "old-preview", version: "0.0.0-preview-a-1234", url: "http://localhost:4096", pid: 1 }),
)
await fs.writeFile(target, JSON.stringify({ id: "current-preview" }))
await Effect.runPromise(
ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe(
Effect.provide(NodeFileSystem.layer),
),
)
expect(await Bun.file(legacy).exists()).toBe(true)
expect(await Bun.file(target).json()).toMatchObject({ id: "current-preview" })
} finally {
await fs.rm(root, { recursive: true, force: true })
}
})
test("concurrent service processes elect one server", async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-election-"))
const database = path.join(root, "opencode.db")
@ -74,18 +129,55 @@ test("concurrent service processes elect one server", async () => {
}),
)
const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"]
const first = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
const second = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
const registration = path.join(root, "state", "opencode", "service-local.json")
const processes = Array.from({ length: 10 }, () => Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }))
try {
const registration = path.join(root, "state", "opencode", "service-local.json")
const info = await waitForInfo(registration)
const winner = info.pid === first.pid ? first : second
const loser = info.pid === first.pid ? second : first
const exited = await Promise.race([loser.exited.then(() => true), Bun.sleep(10_000).then(() => false)])
const winner = processes.find((process) => process.pid === info.pid)
const losers = processes.filter((process) => process.pid !== info.pid)
const exited = await Promise.all(
losers.map((process) => Promise.race([process.exited.then(() => true), Bun.sleep(10_000).then(() => false)])),
)
expect(exited).toBe(true)
expect(winner.exitCode).toBe(null)
expect(exited).toEqual(losers.map(() => true))
expect(winner?.exitCode).toBe(null)
expect(
await fetch(new URL("/api/health", info.url), {
headers: { authorization: "Basic " + btoa(`opencode:${info.password}`) },
}).then((response) => response.json()),
).toMatchObject({
healthy: true,
pid: info.pid,
instanceID: info.id,
status: { type: "ready" },
})
const blockedTemp = registration + "." + info.id + ".tmp"
await fs.mkdir(blockedTemp)
await fs.rm(registration)
await Bun.sleep(6_000)
expect(await Bun.file(registration).exists()).toBe(false)
await fs.rm(blockedTemp, { recursive: true })
const restored = await waitForInfo(registration)
expect(restored.id).toBe(info.id)
expect(restored.pid).toBe(info.pid)
await fs.writeFile(registration, "not-json")
const repaired = await waitForInfo(registration)
expect(repaired.id).toBe(info.id)
expect(repaired.pid).toBe(info.pid)
const contender = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
try {
const contenderExited = await Promise.race([
contender.exited.then(() => true),
Bun.sleep(10_000).then(() => false),
])
expect(contenderExited).toBe(true)
expect((await waitForInfo(registration)).id).toBe(info.id)
} finally {
contender.kill("SIGTERM")
await contender.exited
}
expect(
await withDatabase(
database,
@ -100,13 +192,70 @@ test("concurrent service processes elect one server", async () => {
),
).toEqual({ timeSuspended: null })
expect(await waitForExecutionStart(database, sessionID)).toBe(1)
await Effect.runPromise(
Service.stop({ file: registration }, { targetVersion: "next" }).pipe(Effect.provide(NodeFileSystem.layer)),
)
await winner?.exited
} finally {
first.kill("SIGTERM")
second.kill("SIGTERM")
await Promise.all([first.exited, second.exited])
processes.forEach((process) => process.kill("SIGTERM"))
await Promise.all(processes.map((process) => process.exited))
try {
expect(await Bun.file(registration).exists()).toBe(false)
} finally {
await fs.rm(root, { recursive: true, force: true })
}
}
}, 60_000)
test("a failed service stays registered and owns the lock until stopped", async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-failed-"))
const database = path.join(root, "database")
await fs.mkdir(database)
const env = {
...process.env,
HOME: root,
OPENCODE_DB: database,
OPENCODE_TEST_HOME: root,
XDG_CACHE_HOME: path.join(root, "cache"),
XDG_CONFIG_HOME: path.join(root, "config"),
XDG_DATA_HOME: path.join(root, "data"),
XDG_STATE_HOME: path.join(root, "state"),
}
const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"]
const registration = path.join(root, "state", "opencode", "service-local.json")
const owner = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
try {
const info = await waitForInfo(registration)
const status = await Effect.runPromise(
Service.status({ file: registration }).pipe(
Effect.filterOrFail((status) => status.type === "failed"),
Effect.retry(Schedule.spaced("50 millis").pipe(Schedule.both(Schedule.recurs(200)))),
Effect.provide(NodeFileSystem.layer),
),
)
expect(status).toEqual({
type: "failed",
version: info.version,
message: "The background service could not start.",
action: "Run `opencode service restart` after checking the service logs.",
})
expect(owner.exitCode).toBe(null)
const contender = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
expect(await Promise.race([contender.exited.then(() => true), Bun.sleep(10_000).then(() => false)])).toBe(true)
expect((await waitForInfo(registration)).id).toBe(info.id)
expect(owner.exitCode).toBe(null)
await Effect.runPromise(Service.stop({ file: registration }).pipe(Effect.provide(NodeFileSystem.layer)))
await owner.exited
expect(await Bun.file(registration).exists()).toBe(false)
} finally {
owner.kill("SIGTERM")
await owner.exited
await fs.rm(root, { recursive: true, force: true })
}
})
}, 30_000)
function withDatabase<A, E>(file: string, effect: Effect.Effect<A, E, Database.Service>) {
return Effect.runPromise(effect.pipe(Effect.provide(Database.layerFromPath(file)), Effect.scoped))
@ -143,7 +292,7 @@ function waitForExecutionStart(file: string, sessionID: SessionV2.ID) {
}
async function waitForInfo(file: string) {
for (let attempt = 0; attempt < 200; attempt++) {
for (let attempt = 0; attempt < 400; attempt++) {
const value = await Bun.file(file)
.json()
.catch(() => undefined)

9
packages/client/src/effect/api/api.ts
View file

@ -10,8 +10,17 @@ type StreamValue<A> = A extends Stream.Stream<infer Success, any, any> ? Success
export type Endpoint0_0Output = EffectValue<ReturnType<RawClient["server.health"]["health.get"]>>
export type HealthGetOperation<E = never> = () => Effect.Effect<Endpoint0_0Output, E>
type Endpoint0_1Request = Parameters<RawClient["server.health"]["health.stop"]>[0]
export type Endpoint0_1Input = {
readonly instanceID: Endpoint0_1Request["payload"]["instanceID"]
readonly targetVersion?: Endpoint0_1Request["payload"]["targetVersion"]
}
export type Endpoint0_1Output = EffectValue<ReturnType<RawClient["server.health"]["health.stop"]>>
export type HealthStopOperation<E = never> = (input: Endpoint0_1Input) => Effect.Effect<Endpoint0_1Output, E>
export interface HealthApi<E = never> {
readonly get: HealthGetOperation<E>
readonly stop: HealthStopOperation<E>
}
export type Endpoint1_0Output = EffectValue<ReturnType<RawClient["server.server"]["server.get"]>>

12
packages/client/src/effect/generated/client.ts
View file

@ -16,7 +16,17 @@ const mapClientError = <E>(error: E) =>
const Endpoint0_0 = (raw: RawClient["server.health"]) => () =>
raw["health.get"]({}).pipe(Effect.mapError(mapClientError))
const adaptGroup0 = (raw: RawClient["server.health"]) => ({ get: Endpoint0_0(raw) })
type Endpoint0_1Request = Parameters<RawClient["server.health"]["health.stop"]>[0]
type Endpoint0_1Input = {
readonly instanceID: Endpoint0_1Request["payload"]["instanceID"]
readonly targetVersion?: Endpoint0_1Request["payload"]["targetVersion"]
}
const Endpoint0_1 = (raw: RawClient["server.health"]) => (input: Endpoint0_1Input) =>
raw["health.stop"]({ payload: { instanceID: input["instanceID"], targetVersion: input["targetVersion"] } }).pipe(
Effect.mapError(mapClientError),
)
const adaptGroup0 = (raw: RawClient["server.health"]) => ({ get: Endpoint0_0(raw), stop: Endpoint0_1(raw) })
const Endpoint1_0 = (raw: RawClient["server.server"]) => () =>
raw["server.get"]({}).pipe(Effect.mapError(mapClientError))

268
packages/client/src/effect/service.ts
View file

@ -1,5 +1,6 @@
import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
import { Effect, FileSystem, Option, Schedule, Schema } from "effect"
import { spawn } from "node:child_process"
import { spawn, type ChildProcess } from "node:child_process"
import { homedir } from "node:os"
import { join } from "node:path"
@ -39,6 +40,23 @@ export type StartOptions = Options & {
// found, or a healthy service with a different version is being replaced.
// `existing` carries the registration of the service being replaced.
readonly onStart?: (reason: StartReason, existing?: Info) => void
readonly onStatus?: (status: Status) => void
}
export type Status =
| { readonly type: "missing" }
| { readonly type: "unreachable" }
| { readonly type: "unresponsive" }
| (ServiceStatus.State & { readonly version?: string })
export class FailedError extends Schema.TaggedErrorClass<FailedError>()("ServiceFailedError", {
message: Schema.String,
action: Schema.String,
}) {}
type Contender = {
readonly child: ChildProcess
readonly error: () => Error | undefined
}
// Read-only lookup: registration file plus health check and version gate.
@ -47,58 +65,138 @@ export const discover = Effect.fn("service.discover")(function* (options: Option
return (yield* discoverLocal(options))?.endpoint
})
export const status = Effect.fn("service.status")(function* (options: Options = {}) {
const result = yield* registered(options.file, true)
if (result.info === undefined) return { type: "missing" } satisfies Status
if (result.service === undefined) return { type: "unreachable" } satisfies Status
return publicStatus(result.service)
})
function publicStatus(service: LocalService): Status {
return { ...service.status, version: service.version }
}
const discoverLocal = Effect.fnUntraced(function* (options: Options) {
const info = yield* read(options.file)
if (info === undefined) return undefined
if (options.version !== undefined && info.version !== options.version) return undefined
return yield* probe(info, options.version)
const found = (yield* registered(options.file)).service
if (found?.status.type !== "ready") return undefined
if (options.version !== undefined && found.version !== options.version) return undefined
return found
})
// Idempotent ensure-running: reuses a healthy compatible server, replaces a
// version-mismatched one, and otherwise spawns the service command detached.
// version-mismatched one, and otherwise spawns small contenders until a server
// becomes discoverable. A contender is never killed merely for slow startup.
export const start = Effect.fn("service.start")(function* (options: StartOptions = {}) {
const compatible = yield* discover(options)
if (compatible !== undefined) return compatible
const existing = yield* find(options)
if (existing?.version !== undefined && (options.version === undefined || existing.version === options.version))
return existing.endpoint
yield* Effect.sync(() => options.onStart?.(existing === undefined ? "missing" : "version-mismatch", existing?.info))
if (existing !== undefined) yield* kill(existing.info, options).pipe(Effect.ignore)
const [command, ...args] = options.command ?? ["opencode", "serve", "--service"]
if (command === undefined) return yield* Effect.fail(new Error("Missing service command"))
const child = yield* Effect.try({
try: () => {
const child = spawn(command, args, { detached: true, stdio: "ignore" })
child.unref()
return child
},
catch: (cause) => new Error("Failed to start server", { cause }),
const contenders = new Set<Contender>()
let announced = false
let reported: Status | undefined
let lastSpawn = 0
let spawnDelay = 5_000
let ownerHeld = false
const announce = (reason: StartReason, existing?: Info) =>
Effect.sync(() => {
if (announced) return
announced = true
options.onStart?.(reason, existing)
})
const spawnContender = Effect.gen(function* () {
const [command, ...args] = options.command ?? ["opencode", "serve", "--service"]
if (command === undefined) return yield* Effect.fail(new Error("Missing service command"))
return yield* Effect.try({
try: () => {
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 }),
})
})
const found = yield* Effect.gen(function* () {
const registration = yield* registered(options.file, true)
const info = registration.info
const service = registration.service
const current: Status =
service === undefined ? { type: info === undefined ? "missing" : "unreachable" } : publicStatus(service)
const next = ownerHeld && service === undefined ? ({ type: "unresponsive" } satisfies Status) : current
yield* Effect.sync(() => {
if (sameStatus(reported, next)) return
reported = next
options.onStatus?.(next)
})
if (service !== undefined) {
ownerHeld = false
spawnDelay = 5_000
const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
if (compatible && service.status.type === "ready") return Option.some(service)
if (compatible && service.status.type === "failed")
return yield* new FailedError({ message: service.status.message, action: service.status.action })
if (compatible || service.status.type === "stopping") return Option.none<LocalService>()
yield* announce("version-mismatch", service.info)
yield* kill(service, options, options.version).pipe(Effect.ignore)
lastSpawn = 0
return Option.none<LocalService>()
} else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now()
return yield* discoverLocal(options).pipe(
Effect.flatMap((found) =>
found === undefined ? Effect.fail(new Error("Server is not ready")) : Effect.succeed(found),
),
Effect.retry(poll),
Effect.tap((found) =>
found.info.pid === child.pid
? Effect.void
: Effect.sync(() => {
child.kill("SIGTERM")
}),
),
Effect.map((found) => found.endpoint),
Effect.tapError(() => Effect.try({ try: () => child.kill("SIGTERM"), catch: () => undefined }).pipe(Effect.ignore)),
Effect.mapError(() => new Error("Failed to start server")),
)
const failure = [...contenders].map(contenderFailure).find((error): error is Error => error !== undefined)
if (failure !== undefined) return yield* Effect.fail(failure)
const finished = [...contenders].filter(contenderFinished)
if (finished.some((item) => item.child.exitCode === 0)) {
ownerHeld = true
spawnDelay = Math.min(spawnDelay * 2, 30_000)
}
finished.forEach((item) => contenders.delete(item))
// Keep one candidate plus one lock probe so a pre-lock stall cannot block recovery.
if (contenders.size < 2 && Date.now() - lastSpawn >= spawnDelay) {
yield* announce("missing")
contenders.add(yield* spawnContender)
lastSpawn = Date.now()
}
return Option.none<LocalService>()
}).pipe(Effect.repeat({ until: Option.isSome, schedule: Schedule.spaced("1 second") }))
return Option.getOrThrow(found).endpoint
})
export const stop = Effect.fn("service.stop")(function* (options: Options = {}) {
const fs = yield* FileSystem.FileSystem
function sameStatus(left: Status | undefined, right: Status) {
if (left?.type !== right.type) return false
if (right.type === "failed")
return (
left.type === "failed" &&
left.version === right.version &&
left.message === right.message &&
left.action === right.action
)
if (right.type === "stopping")
return left.type === "stopping" && left.version === right.version && left.targetVersion === right.targetVersion
if (right.type === "starting" || right.type === "ready")
return left.type === right.type && left.version === right.version
return true
}
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
}
export type StopMetadata = {
readonly targetVersion?: string
}
export const stop = Effect.fn("service.stop")(function* (options: Options = {}, metadata: StopMetadata = {}) {
const existing = yield* find(options)
if (existing !== undefined) yield* kill(existing.info, options)
yield* fs.remove(options.file ?? fallback()).pipe(Effect.ignore)
if (existing !== undefined) yield* kill(existing, options, metadata.targetVersion)
})
function fallback() {
@ -106,7 +204,7 @@ function fallback() {
return join(state, "opencode", "service.json")
}
export function headers(endpoint: Endpoint): RequestInit["headers"] {
export function headers(endpoint: Endpoint) {
if (endpoint.auth === undefined) return undefined
return { authorization: "Basic " + btoa(endpoint.auth.username + ":" + endpoint.auth.password) }
}
@ -121,9 +219,7 @@ export const Info = Schema.Struct({
export type Info = typeof Info.Type
const decode = Schema.decodeUnknownEffect(Schema.fromJsonString(Info))
const decodeHealth = Schema.decodeUnknownOption(
Schema.Struct({ healthy: Schema.Literal(true), version: Schema.String, pid: Schema.Int }),
)
const decodeHealth = Schema.decodeUnknownOption(ServiceStatus.Health)
const decodeLegacyHealth = Schema.decodeUnknownOption(Schema.Struct({ healthy: Schema.Literal(true) }))
// A missing or corrupt file means no valid info; callers treat both
@ -139,9 +235,11 @@ type LocalService = {
readonly info: Info
readonly endpoint: Endpoint
readonly version?: string
readonly status: ServiceStatus.State
readonly legacy: boolean
}
const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLegacy = false) {
const probe = Effect.fnUntraced(function* (info: Info, allowLegacy = false) {
const endpoint = {
url: info.url,
auth:
@ -155,14 +253,21 @@ const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLe
signal: AbortSignal.timeout(2_000),
}),
).pipe(Effect.option, Effect.map(Option.getOrUndefined))
if (response === undefined || !response.ok) return undefined
if (response === undefined) return undefined
const body = yield* Effect.tryPromise(() => response.json()).pipe(Effect.option, Effect.map(Option.getOrUndefined))
const health = decodeHealth(body)
if (Option.isSome(health)) {
if (health.value.pid !== info.pid) return undefined
if (info.version !== undefined && health.value.version !== info.version) return undefined
if (version !== undefined && health.value.version !== version) return undefined
return { info, endpoint, version: health.value.version } satisfies LocalService
if (info.id !== undefined && health.value.instanceID !== undefined && health.value.instanceID !== info.id)
return undefined
return {
info,
endpoint,
version: health.value.version,
status: health.value.status,
legacy: false,
} satisfies LocalService
}
if (
!allowLegacy ||
@ -170,18 +275,23 @@ const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLe
(typeof body === "object" && body !== null && ("version" in body || "pid" in body))
)
return undefined
return { info, endpoint } satisfies LocalService
return { info, endpoint, status: { type: "ready" }, legacy: true } satisfies LocalService
})
// Health-checked lookup without the version gate: lifecycle operations must be
const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = false) {
const info = yield* read(file)
if (info === undefined) return { info: undefined, service: undefined }
return { info, service: yield* probe(info, allowLegacy) }
})
// Health-checked lookup without the version gate: status operations must be
// able to see (and replace or stop) a server from a different version.
const find = Effect.fnUntraced(function* (options: Options) {
const info = yield* read(options.file)
if (info === undefined) return undefined
return yield* probe(info, undefined, true)
return (yield* registered(options.file, true)).service
})
// 50ms cadence bounded at ~5s, shared by stop escalation and start readiness.
// 50ms cadence bounded at ~5s, shared by stop escalation and each start
// discovery window.
const poll = Schedule.spaced("50 millis").pipe(Schedule.both(Schedule.recurs(100)))
const signal = (pid: number, name: NodeJS.Signals) =>
@ -199,20 +309,42 @@ function same(left: Info, right: Info) {
return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
}
const kill = Effect.fnUntraced(function* (info: Info, options: Options) {
// A stale registration may point at a PID that has since been reused by
// another process. Only signal the PID after authenticating the server.
const current = yield* find(options)
if (current === undefined || !same(current.info, info)) return
yield* signal(info.pid, "SIGTERM")
const done = yield* stopped(info.pid).pipe(Effect.retry(poll), Effect.option)
const kill = Effect.fnUntraced(function* (service: LocalService, options: Options, targetVersion?: string) {
const requested = yield* requestStop(service, targetVersion)
if (requested === "rejected") return
if (requested === "unsupported") {
// A stale registration may point at a reused PID. Authenticate again
// immediately before the legacy signal fallback.
const current = yield* find(options)
if (current === undefined || !same(current.info, service.info)) return
yield* signal(service.info.pid, "SIGTERM")
}
const done = yield* stopped(service.info.pid).pipe(Effect.retry(poll), Effect.option)
if (Option.isSome(done)) return
const latest = yield* find(options)
if (latest === undefined || !same(latest.info, info)) return
yield* signal(info.pid, "SIGKILL")
yield* stopped(info.pid).pipe(Effect.retry(poll))
if (latest === undefined || !same(latest.info, service.info)) return
yield* signal(service.info.pid, "SIGKILL")
yield* stopped(service.info.pid).pipe(Effect.retry(poll))
})
const decodeStopResponse = Schema.decodeUnknownOption(ServiceStatus.StopResponse)
const requestStop = Effect.fnUntraced(function* (service: LocalService, targetVersion?: string) {
if (service.info.id === undefined || service.legacy) return "unsupported" as const
const response = yield* Effect.tryPromise(() =>
fetch(new URL("/api/service/stop", service.info.url), {
method: "POST",
headers: { ...headers(service.endpoint), "content-type": "application/json" },
body: JSON.stringify({ instanceID: service.info.id, targetVersion }),
signal: AbortSignal.timeout(2_000),
}),
).pipe(Effect.option, Effect.map(Option.getOrUndefined))
if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const
const body = yield* Effect.tryPromise(() => response.json()).pipe(Effect.option, Effect.map(Option.getOrUndefined))
const decoded = decodeStopResponse(body)
if (!response.ok || Option.isNone(decoded) || !decoded.value.accepted) return "rejected" as const
return "accepted" as const
})
export * as Service from "./service.js"

14
packages/client/src/promise/generated/client.ts
View file

@ -1,5 +1,7 @@
import type {
HealthGetOutput,
HealthStopInput,
HealthStopOutput,
ServerGetOutput,
LocationGetInput,
LocationGetOutput,
@ -332,6 +334,18 @@ export function make(options: ClientOptions) {
{ method: "GET", path: `/api/health`, successStatus: 200, declaredStatuses: [401, 400], empty: false },
requestOptions,
),
stop: (input: HealthStopInput, requestOptions?: RequestOptions) =>
request<HealthStopOutput>(
{
method: "POST",
path: `/api/service/stop`,
body: { instanceID: input["instanceID"], targetVersion: input["targetVersion"] },
successStatus: 200,
declaredStatuses: [401, 400],
empty: false,
},
requestOptions,
),
},
server: {
get: (requestOptions?: RequestOptions) =>

25
packages/client/src/promise/generated/types.ts
View file

@ -1,5 +1,13 @@
export type JsonValue = null | boolean | number | string | Array<JsonValue> | { [key: string]: JsonValue }
export type ServiceStatus =
| { type: "starting" }
| { type: "ready" }
| { type: "stopping"; targetVersion?: string | null }
| { type: "failed"; message: string; action: string }
export type ServiceStopResponse = { accepted: boolean }
export type ModelRef = { id: string; providerID: string; variant?: string }
export type ProviderSettings = { [x: string]: JsonValue }
@ -498,6 +506,14 @@ export type VcsFileStatus = {
status: "added" | "deleted" | "modified"
}
export type ServiceHealth = {
healthy: true
version: string
pid: number
instanceID?: string | null
status?: ServiceStatus
}
export type SessionMessageModelSelected = {
id: string
metadata?: { [x: string]: JsonValue }
@ -2483,7 +2499,14 @@ export type ProjectCopyError = {
export const isProjectCopyError = (value: unknown): value is ProjectCopyError =>
typeof value === "object" && value !== null && "name" in value && value["name"] === "ProjectCopyError"
export type HealthGetOutput = { healthy: true; version: string; pid: number }
export type HealthGetOutput = ServiceHealth
export type HealthStopInput = {
readonly instanceID: { readonly instanceID: string; readonly targetVersion?: string | undefined }["instanceID"]
readonly targetVersion?: { readonly instanceID: string; readonly targetVersion?: string | undefined }["targetVersion"]
}
export type HealthStopOutput = ServiceStopResponse
export type ServerGetOutput = { urls: Array<string> }

12
packages/client/test/effect.test.ts
View file

@ -15,6 +15,18 @@ import {
const synced = { type: "log.synced" as const, aggregateID: "ses_test", seq: Event.Seq.make(1) }
test("health.get treats an old server as ready", async () => {
const httpClient = HttpClient.make((request) =>
Effect.succeed(HttpClientResponse.fromWeb(request, Response.json({ healthy: true, version: "old", pid: 123 }))),
)
const result = await Effect.gen(function* () {
const client = yield* OpenCode.make({ baseUrl: "http://localhost:3000" })
return yield* client.health.get()
}).pipe(Effect.provideService(HttpClient.HttpClient, httpClient), Effect.runPromise)
expect(result.status).toEqual({ type: "ready" })
})
test("session.get returns the decoded Effect projection", async () => {
const httpClient = HttpClient.make((request) =>
Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(session))),

64
packages/client/test/fixture/service.ts
View file

@ -1,13 +1,45 @@
import { rename, writeFile } from "node:fs/promises"
import { appendFile, rename, writeFile } from "node:fs/promises"
const [registration, mode] = process.argv.slice(2)
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 === "record-start") {
await writeFile(registration + ".started", "")
process.exit(1)
}
if (mode === "signal") process.kill(process.pid, process.platform === "win32" ? "SIGTERM" : "SIGKILL")
if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated") {
await appendFile(registration + ".starts", process.pid + "\n")
const owner = await writeFile(registration + ".owner", String(process.pid), { flag: "wx" })
.then(() => true)
.catch(() => false)
if (!owner) process.exit()
if (mode === "coordinated") {
while ((await Bun.file(registration + ".starts").text()).trim().split("\n").length < 2) await Bun.sleep(10)
} else await Bun.sleep(Number(delay))
if (mode === "delayed-failed") process.exit(1)
}
let requests = 0
const version = mode === "old" || mode === "reject-stop" ? "old" : "test"
const id = crypto.randomUUID()
const server = Bun.serve({
port: 0,
async fetch(request) {
if (new URL(request.url).pathname !== "/api/health") return new Response(null, { status: 404 })
const pathname = new URL(request.url).pathname
if (pathname === "/api/service/stop" && mode === "reject-stop") {
await writeFile(registration + ".stop-attempt", "")
return Response.json({ accepted: false })
}
if (pathname === "/api/service/stop" && mode === "graceful") {
const body = await request.json()
if (typeof body !== "object" || body === null || body.instanceID !== id) return Response.json({ accepted: false })
await writeFile(registration + ".stop", JSON.stringify(body))
setTimeout(shutdown, 25)
return Response.json({ accepted: true })
}
if (pathname !== "/api/health") return new Response(null, { status: 404 })
requests += 1
if (mode === "modern" && requests === 1) {
await writeFile(registration + ".first-request", "")
@ -15,15 +47,33 @@ const server = Bun.serve({
return new Response(null, { status: 503 })
}
if (mode === "legacy") return Response.json({ healthy: true })
return Response.json({ healthy: true, version: "test", pid: process.pid })
if (mode === "starting" && !(await Bun.file(registration + ".release").exists()))
return Response.json(
{ healthy: true, version, pid: process.pid, instanceID: id, status: { type: "starting" } },
{ status: 503 },
)
if (mode === "failed-owner")
return Response.json(
{
healthy: true,
version,
pid: process.pid,
instanceID: id,
status: { type: "failed", message: "Could not open the database.", action: "Check the service logs." },
},
{ status: 503 },
)
if (mode === "starting" || mode === "graceful" || mode === "reject-stop")
return Response.json({ healthy: true, version, pid: process.pid, instanceID: id, status: { type: "ready" } })
return Response.json({ healthy: true, version, pid: process.pid })
},
})
await writeFile(
registration + ".tmp",
JSON.stringify({
id: crypto.randomUUID(),
version: mode === "legacy" ? undefined : "test",
id,
version: mode === "legacy" ? undefined : version,
url: server.url.toString(),
pid: process.pid,
}),
@ -31,7 +81,7 @@ await writeFile(
)
await rename(registration + ".tmp", registration)
const shutdown = () => {
function shutdown() {
server.stop(true)
process.exit()
}

20
packages/client/test/promise.test.ts
View file

@ -16,7 +16,7 @@ test("exposes every standard HTTP API group", () => {
"generate",
"provider",
"integration",
"server.mcp",
"mcp",
"credential",
"project",
"form",
@ -61,6 +61,22 @@ test("server.get uses the public HTTP contract", async () => {
expect(request?.url).toBe("http://localhost:3000/api/server")
})
test("health.stop sends exact replacement identity", async () => {
let request: Request | undefined
const client = OpenCode.make({
baseUrl: "http://localhost:3000",
fetch: async (input, init) => {
request = input instanceof Request ? input : new Request(input, init)
return Response.json({ accepted: true })
},
})
expect(await client.health.stop({ instanceID: "instance", targetVersion: "next" })).toEqual({ accepted: true })
expect(request?.method).toBe("POST")
expect(request?.url).toBe("http://localhost:3000/api/service/stop")
expect(await request?.json()).toEqual({ instanceID: "instance", targetVersion: "next" })
})
test("MCP resource catalog uses the public HTTP contract", async () => {
let request: Request | undefined
const client = OpenCode.make({
@ -77,7 +93,7 @@ test("MCP resource catalog uses the public HTTP contract", async () => {
},
})
const result = await client["server.mcp"].resource.catalog({ location: { directory: "/tmp/project" } })
const result = await client.mcp.resource.catalog({ location: { directory: "/tmp/project" } })
expect(result.data.resources[0]?.uri).toBe("docs://readme")
expect(request?.method).toBe("GET")

159
packages/client/test/service.test.ts
View file

@ -43,6 +43,75 @@ test("a concurrent same-version start cannot invalidate a resolved endpoint", as
expect(starts).toEqual([])
expect(await Bun.file(registration).json()).toEqual(original)
expect(await health(resolved.url)).toEqual({ healthy: true, version: "test", pid: original.pid })
expect(await run(Service.status({ file: registration }))).toEqual({ type: "ready", version: "test" })
})
test("waits for a registered service to finish starting", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const process = spawn(registration, "starting")
await waitForFile(registration)
const statuses: Service.Status[] = []
const result = run(
Service.start({ file: registration, version: "test", command: [], onStatus: (status) => statuses.push(status) }),
)
await Bun.sleep(500)
expect(process.exitCode).toBe(null)
expect(statuses).toContainEqual({ type: "starting", version: "test" })
expect(statuses.filter((status) => status.type === "starting")).toHaveLength(1)
await writeFile(registration + ".release", "")
expect((await result).url).toBe((await Bun.file(registration).json()).url)
})
test("reports a failed registered service without spawning", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const process = spawn(registration, "failed-owner")
await waitForFile(registration)
await expect(run(Service.start({ file: registration, version: "test", command: [] }))).rejects.toMatchObject({
message: "Could not open the database.",
action: "Check the service logs.",
})
expect(process.exitCode).toBe(null)
})
test("requests graceful replacement of the exact service instance", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const process = spawn(registration, "graceful")
await waitForFile(registration)
const info = await Bun.file(registration).json()
await run(Service.stop({ file: registration }, { targetVersion: "next" }))
await process.exited
expect(await Bun.file(registration + ".stop").json()).toEqual({ instanceID: info.id, targetVersion: "next" })
})
test("does not spawn contenders while an incompatible service rejects replacement", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const contender = join(directory, "contender.json")
const existing = spawn(registration, "reject-stop")
await waitForFile(registration)
const controller = new AbortController()
const starting = Effect.runPromise(
Service.start({
file: registration,
version: "test",
command: [process.execPath, fixture, contender, "record-start"],
}).pipe(Effect.provide(NodeFileSystem.layer)),
{ signal: controller.signal },
)
await waitForFile(registration + ".stop-attempt")
await Bun.sleep(500)
controller.abort()
await starting.catch(() => undefined)
expect(await Bun.file(contender + ".started").exists()).toBe(false)
expect(existing.exitCode).toBe(null)
})
test("a legacy health response is still replaced", async () => {
@ -57,9 +126,95 @@ test("a legacy health response is still replaced", async () => {
await expect(result).rejects.toThrow("Missing service command")
expect(starts).toEqual(["version-mismatch"])
await existing.exited
})
}, 10_000)
function run<A, E>(effect: Effect.Effect<A, E, never>) {
test("waits for a slow winner while bounding lock probes", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const endpoint = await run(
Service.start({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "coordinated"],
}),
)
const info = await Bun.file(registration).json()
try {
expect(endpoint.url).toBe(info.url)
expect(await health(endpoint.url)).toEqual({ healthy: true, version: "test", pid: info.pid })
expect((await Bun.file(registration + ".starts").text()).trim().split("\n")).toHaveLength(2)
} finally {
process.kill(info.pid, "SIGTERM")
}
}, 15_000)
test("reports a contender that fails to start", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
await expect(
run(
Service.start({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "failed"],
}),
),
).rejects.toThrow("Server process exited with code 1")
}, 10_000)
test("reports a contender terminated by a signal", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
await expect(
run(
Service.start({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "signal"],
}),
),
).rejects.toThrow(/Server process (terminated by|exited with code)/)
}, 10_000)
test("reports a slow contender that eventually fails", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
await expect(
run(
Service.start({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "delayed-failed", "8000"],
}),
),
).rejects.toThrow("Server process exited with code 1")
}, 15_000)
test("replaces an incompatible owner that appears during startup", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const starting = run(
Service.start({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "delayed", "8000"],
}),
)
await Bun.sleep(1_000)
const old = spawn(registration, "old")
await waitForFile(registration)
const endpoint = await starting
const info = await Bun.file(registration).json()
try {
expect(endpoint.url).toBe(info.url)
expect(info.version).toBe("test")
await old.exited
} finally {
process.kill(info.pid, "SIGTERM")
}
}, 20_000)
function run<A, E>(effect: Effect.Effect<A, E>) {
return Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer)))
}

181
packages/core/src/util/process-lock.ts Normal file
View file

@ -0,0 +1,181 @@
import { dlopen, read, type Pointer } from "bun:ffi"
import { closeSync, existsSync, mkdirSync, openSync } from "node:fs"
import { connect, createServer, type Server, type Socket } from "node:net"
import path from "node:path"
import { Effect, Schema } from "effect"
import { Hash } from "./hash"
export namespace ProcessLock {
export class HeldError extends Schema.TaggedErrorClass<HeldError>()("ProcessLockHeldError", {
file: Schema.String,
}) {
override get message() {
return `Process lock is already held: ${this.file}`
}
}
export class SystemError extends Schema.TaggedErrorClass<SystemError>()("ProcessLockSystemError", {
file: Schema.String,
operation: Schema.Literals(["open", "acquire"]),
code: Schema.String,
}) {
override get message() {
return `Process lock ${this.operation} failed for ${this.file}: ${this.code}`
}
}
export type LockError = HeldError | SystemError
const acquirePosix = Effect.fnUntraced(function* (file: string) {
const fd = yield* Effect.try({
try: () => {
mkdirSync(path.dirname(file), { recursive: true })
return openSync(file, "a+", 0o600)
},
catch: (cause) =>
new SystemError({
file,
operation: "open",
code: cause instanceof Error ? cause.message : String(cause),
}),
})
const result = yield* Effect.try({
try: () => lock(fd),
catch: (cause) =>
new SystemError({
file,
operation: "acquire",
code: cause instanceof Error ? cause.message : String(cause),
}),
}).pipe(
Effect.tapError(() =>
Effect.sync(() => {
closeSync(fd)
}),
),
)
if (result.acquired) {
return fd
}
closeSync(fd)
return yield* result.held
? new HeldError({ file })
: new SystemError({ file, operation: "acquire", code: String(result.code) })
})
export const acquire = Effect.fn("ProcessLock.acquire")(function* (file: string) {
if (process.platform === "win32") {
yield* Effect.acquireRelease(acquireWindows(file), closeWindows)
return
}
yield* Effect.acquireRelease(acquirePosix(file), (fd) =>
Effect.sync(() => {
closeSync(fd)
}),
)
})
}
type Result =
| { readonly acquired: true }
| { readonly acquired: false; readonly held: true }
| { readonly acquired: false; readonly held: false; readonly code: number }
const LOCK_EX = 2
const LOCK_NB = 4
const DARWIN_EWOULDBLOCK = 35
const LINUX_EWOULDBLOCK = 11
function lock(fd: number): Result {
if (process.platform === "darwin") return lockDarwin(fd)
if (process.platform === "linux") return lockLinux(fd)
throw new Error(`Unsupported process lock platform: ${process.platform}`)
}
function lockDarwin(fd: number): Result {
const library = dlopen("/usr/lib/libSystem.B.dylib", {
flock: { args: ["i32", "i32"], returns: "i32" },
__error: { args: [], returns: "ptr" },
})
try {
const result = library.symbols.flock(fd, LOCK_EX | LOCK_NB)
const code = result === 0 ? 0 : errorCode(library.symbols.__error())
if (result === 0) return { acquired: true }
if (code === DARWIN_EWOULDBLOCK) return { acquired: false, held: true }
return { acquired: false, held: false, code }
} finally {
library.close()
}
}
function lockLinux(fd: number): Result {
const musl = `/lib/libc.musl-${process.arch === "arm64" ? "aarch64" : "x86_64"}.so.1`
const library = dlopen(existsSync(musl) ? musl : "libc.so.6", {
flock: { args: ["i32", "i32"], returns: "i32" },
__errno_location: { args: [], returns: "ptr" },
})
try {
const result = library.symbols.flock(fd, LOCK_EX | LOCK_NB)
const code = result === 0 ? 0 : errorCode(library.symbols.__errno_location())
if (result === 0) return { acquired: true }
if (code === LINUX_EWOULDBLOCK) return { acquired: false, held: true }
return { acquired: false, held: false, code }
} finally {
library.close()
}
}
function errorCode(pointer: Pointer | null) {
if (pointer === null) throw new Error("Failed to read process lock error code")
return read.i32(pointer, 0)
}
function acquireWindows(file: string) {
return Effect.callback<Server, ProcessLock.LockError>((resume) => {
const server = createServer()
let probe: Socket | undefined
const pipe = `\\\\.\\pipe\\opencode-process-lock-${Hash.sha256(path.resolve(file).toLowerCase())}`
const onError = (cause: NodeJS.ErrnoException) => {
server.off("listening", onListening)
probe = connect(pipe)
const onProbeError = () => {
probe?.off("connect", onConnect)
resume(
Effect.fail(
new ProcessLock.SystemError({
file,
operation: "acquire",
code: cause.code ?? cause.message,
}),
),
)
}
const onConnect = () => {
probe?.off("error", onProbeError)
probe?.destroy()
resume(Effect.fail(new ProcessLock.HeldError({ file })))
}
probe.once("connect", onConnect)
probe.once("error", onProbeError)
}
const onListening = () => {
server.off("error", onError)
resume(Effect.succeed(server))
}
server.once("error", onError)
server.once("listening", onListening)
server.on("connection", (socket) => socket.destroy())
server.listen(pipe)
return Effect.sync(() => {
probe?.destroy()
server.close()
})
})
}
function closeWindows(server: Server) {
return Effect.callback<void>((resume) => {
if (!server.listening) return resume(Effect.void)
server.close((error) => resume(error ? Effect.die(error) : Effect.void))
})
}

17
packages/core/test/fixture/process-lock-worker.ts Normal file
View file

@ -0,0 +1,17 @@
import { ProcessLock } from "@opencode-ai/core/util/process-lock"
import { Effect, Schema } from "effect"
import fs from "node:fs/promises"
const input = Schema.decodeUnknownSync(
Schema.fromJsonString(Schema.Struct({ file: Schema.String, ready: Schema.String })),
)(process.argv[2])
await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
yield* ProcessLock.acquire(input.file)
yield* Effect.promise(() => fs.writeFile(input.ready, String(process.pid)))
return yield* Effect.never
}),
),
)

71
packages/core/test/util/process-lock.test.ts Normal file
View file

@ -0,0 +1,71 @@
import { expect } from "bun:test"
import { ProcessLock } from "@opencode-ai/core/util/process-lock"
import { Effect } from "effect"
import fs from "node:fs/promises"
import os from "node:os"
import path from "node:path"
import { it } from "../lib/effect"
const worker = path.join(import.meta.dir, "../fixture/process-lock-worker.ts")
it.live(
"releases ownership when the scope closes",
Effect.gen(function* () {
const root = yield* temp("opencode-process-lock-")
const file = path.join(root, "service.lock")
yield* Effect.scoped(ProcessLock.acquire(file))
yield* Effect.scoped(ProcessLock.acquire(file))
}),
)
it.live(
"releases ownership when the process dies",
Effect.gen(function* () {
const root = yield* temp("opencode-process-lock-death-")
const file = path.join(root, "service.lock")
const ready = path.join(root, "ready")
const child = yield* Effect.acquireRelease(
Effect.sync(() =>
Bun.spawn([process.execPath, worker, JSON.stringify({ file, ready })], {
stdout: "ignore",
stderr: "pipe",
}),
),
(child) =>
Effect.promise(async () => {
kill(child)
await child.exited
}),
)
yield* Effect.promise(async () => {
for (let attempt = 0; attempt < 100 && !(await Bun.file(ready).exists()); attempt++) await Bun.sleep(20)
})
expect(yield* Effect.promise(() => Bun.file(ready).exists())).toBe(true)
const error = yield* Effect.scoped(ProcessLock.acquire(file)).pipe(Effect.flip)
expect(error._tag).toBe("ProcessLockHeldError")
if (process.platform !== "win32") {
process.kill(child.pid, "SIGSTOP")
const paused = yield* Effect.scoped(ProcessLock.acquire(file)).pipe(Effect.flip)
expect(paused._tag).toBe("ProcessLockHeldError")
process.kill(child.pid, "SIGCONT")
}
kill(child)
yield* Effect.promise(() => child.exited)
yield* Effect.scoped(ProcessLock.acquire(file))
}),
)
function temp(prefix: string) {
return Effect.acquireRelease(
Effect.promise(() => fs.mkdtemp(path.join(os.tmpdir(), prefix))),
(root) => Effect.promise(() => fs.rm(root, { recursive: true, force: true })),
)
}
function kill(child: Bun.Subprocess) {
if (process.platform === "win32") return child.kill()
return child.kill("SIGKILL")
}

59
packages/protocol/src/groups/health.ts
View file

@ -1,19 +1,64 @@
import { Schema } from "effect"
import { Effect, Schema } from "effect"
import { HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi"
export namespace ServiceStatus {
export const State = Schema.Union([
Schema.Struct({ type: Schema.Literal("starting") }),
Schema.Struct({ type: Schema.Literal("ready") }),
Schema.Struct({
type: Schema.Literal("stopping"),
targetVersion: Schema.String.pipe(Schema.optional),
}),
Schema.Struct({
type: Schema.Literal("failed"),
message: Schema.String,
action: Schema.String,
}),
]).annotate({ identifier: "ServiceStatus" })
export type State = typeof State.Type
export const Health = Schema.Struct({
healthy: Schema.Literal(true),
version: Schema.String,
pid: Schema.Int.check(Schema.isGreaterThan(0)),
instanceID: Schema.String.pipe(Schema.optional),
status: State.pipe(Schema.withDecodingDefaultKey(Effect.succeed({ type: "ready" as const }))),
}).annotate({ identifier: "ServiceHealth" })
export type Health = typeof Health.Type
export const StopRequest = Schema.Struct({
instanceID: Schema.String,
targetVersion: Schema.String.pipe(Schema.optional),
}).annotate({ identifier: "ServiceStopRequest" })
export type StopRequest = typeof StopRequest.Type
export const StopResponse = Schema.Struct({
accepted: Schema.Boolean,
}).annotate({ identifier: "ServiceStopResponse" })
export type StopResponse = typeof StopResponse.Type
}
export const HealthGroup = HttpApiGroup.make("server.health")
.add(
HttpApiEndpoint.get("health.get", "/api/health", {
success: Schema.Struct({
healthy: Schema.Literal(true),
version: Schema.String,
pid: Schema.Int.check(Schema.isGreaterThan(0)),
}),
success: ServiceStatus.Health,
}).annotateMerge(
OpenApi.annotations({
identifier: "v2.health.get",
summary: "Check server health",
description: "Check whether the API server is ready to accept requests.",
description: "Report the owning server process and its application status.",
}),
),
)
.add(
HttpApiEndpoint.post("health.stop", "/api/service/stop", {
payload: ServiceStatus.StopRequest,
success: ServiceStatus.StopResponse,
}).annotateMerge(
OpenApi.annotations({
identifier: "v2.health.stop",
summary: "Stop the managed server",
description: "Request graceful shutdown of one exact managed server instance.",
}),
),
)

13
packages/server/src/handlers/health.ts
View file

@ -4,7 +4,14 @@ import { HttpApiBuilder } from "effect/unstable/httpapi"
import { Api } from "../api"
export const HealthHandler = HttpApiBuilder.group(Api, "server.health", (handlers) =>
handlers.handle("health.get", () =>
Effect.succeed({ healthy: true as const, version: InstallationVersion, pid: process.pid }),
),
handlers
.handle("health.get", () =>
Effect.succeed({
healthy: true as const,
version: InstallationVersion,
pid: process.pid,
status: { type: "ready" },
}),
)
.handle("health.stop", () => Effect.succeed({ accepted: false })),
)

7
packages/server/src/middleware/authorization.ts
View file

@ -35,6 +35,10 @@ function credentialFromRequest(request: HttpServerRequest.HttpServerRequest) {
return Effect.succeed(emptyCredential())
}
export function authorizedRequest(request: HttpServerRequest.HttpServerRequest, config: ServerAuth.Info) {
return credentialFromRequest(request).pipe(Effect.map((credential) => ServerAuth.authorized(credential, config)))
}
export const authorizationLayer = Layer.effect(
Authorization,
Effect.gen(function* () {
@ -46,8 +50,7 @@ export const authorizationLayer = Layer.effect(
// Browsers cannot set headers on WebSocket upgrades, so a ticketed PTY connect skips
// credential checks here; the connect handler consumes and validates the ticket.
if (hasPtyConnectTicketURL(new URL(request.url, "http://localhost"))) return yield* effect
const credential = yield* credentialFromRequest(request)
if (ServerAuth.authorized(credential, config)) return yield* effect
if (yield* authorizedRequest(request, config)) return yield* effect
yield* HttpEffect.appendPreResponseHandler((_request, response) =>
Effect.succeed(HttpServerResponse.setHeader(response, "www-authenticate", WWW_AUTHENTICATE)),
)

249
packages/server/src/process.ts
View file

@ -1,74 +1,211 @@
export * as ServerProcess from "./process"
import { NodeHttpClient, NodeHttpServer } from "@effect/platform-node"
import { HealthGroup } from "@opencode-ai/protocol/groups/health"
import { Context, Effect, Layer, Option } from "effect"
import { HttpClient, HttpClientRequest, HttpMiddleware, HttpRouter, HttpServer } from "effect/unstable/http"
import { HttpApi, HttpApiClient } from "effect/unstable/httpapi"
import { createServer } from "node:http"
import { NodeHttpServer, NodeHttpServerRequest } from "@effect/platform-node"
import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
import { hasPtyConnectTicketURL } from "@opencode-ai/protocol/groups/pty"
import { Cause, Context, Deferred, Effect, Exit, Layer, Option, Ref, Schema, Scope } from "effect"
import { HttpMiddleware, HttpRouter, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
import { createServer } from "node:http"
import { ServerAuth } from "./auth"
import { authorizedRequest } from "./middleware/authorization"
import { createRoutes } from "./routes"
import { ServerInfo } from "./server-info"
import { Status } from "./service-status"
export type Options = {
export type Options<E = never, R = never> = {
readonly hostname: string
readonly port: Option.Option<number>
readonly password: string
readonly restartContinuity?: boolean
readonly instanceID: string
readonly service?: {
readonly onListen: (address: HttpServer.Address) => Effect.Effect<void, E, R>
}
}
const ReadinessApi = HttpApi.make("readiness").add(HealthGroup)
type App = Effect.Effect<
HttpServerResponse.HttpServerResponse,
unknown,
HttpServerRequest.HttpServerRequest | Scope.Scope
>
export const start = Effect.fn("ServerProcess.start")(function* (options: Options) {
export const start = Effect.fn("ServerProcess.start")(function* <E, R>(options: Options<E, R>) {
if (!options.password) return yield* Effect.fail(new Error("Missing server password"))
const address = yield* listen(options)
yield* Effect.gen(function* () {
const client = yield* HttpApiClient.make(ReadinessApi, {
baseUrl: HttpServer.formatAddress(address),
transformClient: HttpClient.mapRequest((request) =>
HttpClientRequest.setHeader(
request,
"authorization",
ServerAuth.header({ username: "opencode", password: options.password }) ?? "",
),
const shutdown = yield* Deferred.make<void>()
const status = yield* Status.make({
instanceID: options.instanceID,
managed: options.service !== undefined,
})
const bound = yield* listen(options)
const application = yield* Ref.make(Option.none<App>())
yield* bound.http.serve(dispatch(options.password, status, application, shutdown), HttpMiddleware.logger)
if (options.service) yield* options.service.onListen(bound.http.address)
const parentScope = yield* Scope.Scope
const applicationScope = yield* Scope.fork(parentScope)
yield* Effect.addFinalizer(() =>
status
.beginStopping()
.pipe(
Effect.andThen(Ref.set(application, Option.none())),
Effect.andThen(Effect.sync(() => bound.server.closeAllConnections())),
),
})
yield* client["server.health"]["health.get"]({})
}).pipe(Effect.provide(NodeHttpClient.layerNodeHttp))
return address
)
const boot = Effect.gen(function* () {
const context = yield* Layer.buildWithScope(
createRoutes(options.password, () => {
const address = bound.server.address()
if (address === null || typeof address === "string") return []
const host = address.family === "IPv6" ? `[${address.address}]` : address.address
return ServerInfo.connectionURLs(`http://${host}:${address.port}`, options.hostname)
}).pipe(Layer.provide(NodeHttpServer.layerHttpServices)),
applicationScope,
)
if (options.service) {
yield* installRestartContinuity(Context.get(context, SessionRestart.Service)).pipe(
Effect.provideService(Scope.Scope, applicationScope),
)
}
yield* Ref.set(application, Option.some(Context.get(context, HttpRouter.HttpRouter).asHttpEffect()))
yield* status.ready
return { address: bound.http.address, shutdown: Deferred.await(shutdown) }
}).pipe(
Effect.catchCause((cause) => {
if (!options.service || Cause.hasInterruptsOnly(cause)) return Effect.failCause(cause)
return status
.fail({
message: "The background service could not start.",
action: "Run `opencode service restart` after checking the service logs.",
})
.pipe(
Effect.andThen(
Scope.close(applicationScope, Exit.failCause(cause)).pipe(
Effect.catchCause((cleanupCause) =>
Effect.logError("failed to clean up background service boot", { cause: cleanupCause }),
),
),
),
Effect.andThen(Effect.logError("background service boot failed", { cause })),
Effect.andThen(Effect.never),
)
}),
)
if (!options.service) return yield* boot
return yield* Effect.raceFirst(boot, Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt)))
})
function listen(options: Options) {
if (Option.isSome(options.port)) return bind(options, options.port.value)
function listen(options: { readonly hostname: string; readonly port: Option.Option<number> }) {
if (Option.isSome(options.port)) return bind(options.hostname, options.port.value)
const next = (port: number): ReturnType<typeof bind> =>
bind(options, port).pipe(Effect.catch((error) => (port === 65_535 ? Effect.fail(error) : next(port + 1))))
bind(options.hostname, port).pipe(
Effect.catch((error) => (port < 65_535 && addressInUse(error) ? next(port + 1) : Effect.fail(error))),
)
return next(4096)
}
function bind(options: Options, port: number) {
const server = createServer()
return Layer.build(
createRoutes(options.password, () => {
const address = server.address()
if (address === null || typeof address === "string") return []
const host = address.family === "IPv6" ? `[${address.address}]` : address.address
return ServerInfo.connectionURLs(`http://${host}:${address.port}`, options.hostname)
function bind(hostname: string, port: number) {
return Effect.gen(function* () {
const parentScope = yield* Scope.Scope
const serverScope = yield* Scope.fork(parentScope)
const server = createServer()
return yield* Effect.gen(function* () {
const http = yield* NodeHttpServer.make(() => server, { port, host: hostname })
yield* Effect.addFinalizer(() => Effect.sync(() => server.closeAllConnections()))
return { http, server }
}).pipe(
Layer.flatMap((context) => {
const serve = HttpServer.serve(
Context.get(context, HttpRouter.HttpRouter).asHttpEffect(),
HttpMiddleware.logger,
)
if (!options.restartContinuity) return serve
const restart = Context.get(context, SessionRestart.Service)
return Layer.merge(serve, restartContinuity(restart))
}),
Layer.provideMerge(NodeHttpServer.layer(() => server, { port, host: options.hostname })),
),
).pipe(
Effect.tap(() => Effect.addFinalizer(() => Effect.sync(() => server.closeAllConnections()))),
Effect.map((context) => Context.get(context, HttpServer.HttpServer).address),
Effect.provideService(Scope.Scope, serverScope),
Effect.onError((cause) => Scope.close(serverScope, Exit.failCause(cause))),
)
})
}
function addressInUse(error: unknown) {
if (typeof error !== "object" || error === null || !("cause" in error)) return false
const cause = error.cause
return typeof cause === "object" && cause !== null && "code" in cause && cause.code === "EADDRINUSE"
}
function dispatch(
password: string,
status: Status.Interface,
application: Ref.Ref<Option.Option<App>>,
shutdown: Deferred.Deferred<void>,
): App {
const auth = ServerAuth.Config.of({ username: "opencode", password: Option.some(password) })
return Effect.gen(function* () {
const request = yield* HttpServerRequest.HttpServerRequest
const url = new URL(request.url, "http://localhost")
const lifecycle =
request.method === "GET" && url.pathname === "/api/health"
? "health"
: request.method === "POST" && url.pathname === "/api/service/stop"
? "stop"
: undefined
if (lifecycle !== undefined) {
if (!(yield* authorizedRequest(request, auth))) return unauthorized()
return yield* control(request, lifecycle, status, () => Deferred.doneUnsafe(shutdown, Effect.void))
}
const state = yield* status.current
const app = yield* Ref.get(application)
const ready = state.type === "ready" && Option.isSome(app)
if ((!ready || !hasPtyConnectTicketURL(url)) && !(yield* authorizedRequest(request, auth))) return unauthorized()
if (ready) return yield* app.value
return unavailable(state)
})
}
function unauthorized() {
return HttpServerResponse.empty({
status: 401,
headers: { "www-authenticate": 'Basic realm="Secure Area"' },
})
}
const control = Effect.fnUntraced(function* (
request: HttpServerRequest.HttpServerRequest,
route: "health" | "stop",
status: Status.Interface,
stop: () => void,
) {
if (route === "health") return yield* healthResponse(status)
const body = yield* request.json.pipe(Effect.option)
const input = Option.isSome(body) ? Schema.decodeUnknownOption(ServiceStatus.StopRequest)(body.value) : Option.none()
if (Option.isNone(input)) return HttpServerResponse.jsonUnsafe({ code: "invalid_request" }, { status: 400 })
const accepted = yield* status.requestStop(input.value)
if (accepted) {
const response = NodeHttpServerRequest.toServerResponse(request)
yield* Effect.sync(() => {
const complete = () => {
response.off("finish", complete)
response.off("close", complete)
stop()
}
response.once("finish", complete)
response.once("close", complete)
})
}
return HttpServerResponse.jsonUnsafe({ accepted })
})
const healthResponse = Effect.fnUntraced(function* (status: Status.Interface) {
const health = yield* status.health
return HttpServerResponse.jsonUnsafe(health, {
status: health.status.type === "ready" ? 200 : 503,
headers:
health.status.type === "starting" || health.status.type === "stopping" ? { "retry-after": "1" } : undefined,
})
})
function unavailable(status: ServiceStatus.State) {
if (status.type === "failed")
return HttpServerResponse.jsonUnsafe(
{ code: "service_failed", message: status.message, action: status.action },
{ status: 503 },
)
return HttpServerResponse.jsonUnsafe(
{ code: status.type === "stopping" ? "service_stopping" : "service_starting" },
{ status: 503, headers: { "retry-after": "1" } },
)
}
@ -77,12 +214,8 @@ function bind(options: Options, port: number) {
* suspends its own active Sessions on graceful shutdown. Suspension runs while the drains are still
* alive: connections close first, this finalizer runs next, and Session execution teardown follows.
*/
function restartContinuity(restart: SessionRestart.Interface) {
return Layer.effectDiscard(
Effect.gen(function* () {
yield* Effect.forkScoped(restart.resumeSuspendedSessions)
// Registered after the fork so suspension observes still-running resumed drains during teardown.
yield* Effect.addFinalizer(() => restart.suspendActiveSessions)
}),
)
}
const installRestartContinuity = Effect.fnUntraced(function* (restart: SessionRestart.Interface) {
yield* Effect.forkScoped(restart.resumeSuspendedSessions)
// Registered after the fork so suspension observes still-running resumed drains during teardown.
yield* Effect.addFinalizer(() => restart.suspendActiveSessions)
})

5
packages/server/src/routes.ts
View file

@ -120,6 +120,5 @@ function driveEnabled() {
return !!process.env.OPENCODE_DRIVE
}
export const routes = createRoutes()
export const webHandler = () => HttpRouter.toWebHandler(routes.pipe(Layer.provide(HttpServer.layerServices)))
export const webHandler = () =>
HttpRouter.toWebHandler(createRoutes().pipe(Layer.provide(HttpServer.layerServices)))

57
packages/server/src/service-status.ts Normal file
View file

@ -0,0 +1,57 @@
export * as Status from "./service-status"
import { InstallationVersion } from "@opencode-ai/core/installation/version"
import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
import { Effect, Ref } from "effect"
export interface Interface {
readonly health: Effect.Effect<ServiceStatus.Health>
readonly current: Effect.Effect<ServiceStatus.State>
readonly ready: Effect.Effect<void>
readonly fail: (failure: { readonly message: string; readonly action: string }) => Effect.Effect<void>
readonly beginStopping: (targetVersion?: string) => Effect.Effect<void>
readonly requestStop: (request: ServiceStatus.StopRequest) => Effect.Effect<boolean>
}
export const make = Effect.fnUntraced(function* (options: {
readonly instanceID: string
readonly managed: boolean
readonly initial?: ServiceStatus.State
}) {
const current = yield* Ref.make(options.initial ?? ({ type: "starting" } satisfies ServiceStatus.State))
const transitionToStopping = (targetVersion?: string) =>
Ref.update(current, (status) => {
if (status.type === "stopping") return status
return (
targetVersion === undefined || targetVersion === InstallationVersion
? { type: "stopping" }
: { type: "stopping", targetVersion }
) satisfies ServiceStatus.State
})
return {
current: Ref.get(current),
health: Effect.gen(function* () {
return {
healthy: true as const,
version: InstallationVersion,
pid: process.pid,
instanceID: options.instanceID,
status: yield* Ref.get(current),
}
}),
ready: Ref.update(current, (status) =>
status.type === "starting" ? ({ type: "ready" } satisfies ServiceStatus.State) : status,
),
fail: (failure) =>
Ref.update(current, (status) =>
status.type === "starting" ? ({ type: "failed", ...failure } satisfies ServiceStatus.State) : status,
),
beginStopping: transitionToStopping,
requestStop: (request) => {
if (!options.managed || request.instanceID !== options.instanceID)
return Effect.succeed(false)
return transitionToStopping(request.targetVersion).pipe(Effect.as(true))
},
} satisfies Interface
})

51
packages/server/test/service-status.test.ts Normal file
View file

@ -0,0 +1,51 @@
import { InstallationVersion } from "@opencode-ai/core/installation/version"
import { expect } from "bun:test"
import { Effect } from "effect"
import { it } from "../../core/test/lib/effect"
import { Status } from "../src/service-status"
it.effect("moves from starting to ready", () =>
Effect.gen(function* () {
const status = yield* Status.make({ instanceID: "one", managed: false })
expect(yield* status.current).toEqual({ type: "starting" })
yield* status.ready
expect(yield* status.current).toEqual({ type: "ready" })
}),
)
it.effect("keeps a startup failure until shutdown", () =>
Effect.gen(function* () {
const status = yield* Status.make({ instanceID: "one", managed: true })
yield* status.fail({ message: "Could not open the database.", action: "Check the database path." })
yield* status.ready
yield* status.fail({ message: "Different failure.", action: "Different action." })
expect(yield* status.current).toEqual({
type: "failed",
message: "Could not open the database.",
action: "Check the database path.",
})
}),
)
it.effect("stops only the addressed managed instance", () =>
Effect.gen(function* () {
const status = yield* Status.make({ instanceID: "one", managed: true })
expect(yield* status.requestStop({ instanceID: "other", targetVersion: "next" })).toBe(false)
expect(yield* status.current).toEqual({ type: "starting" })
expect(yield* status.requestStop({ instanceID: "one", targetVersion: "next" })).toBe(true)
expect(yield* status.requestStop({ instanceID: "one", targetVersion: InstallationVersion })).toBe(true)
expect(yield* status.current).toEqual({ type: "stopping", targetVersion: "next" })
}),
)
it.effect("preserves the original stopping target after shutdown begins", () =>
Effect.gen(function* () {
const status = yield* Status.make({ instanceID: "one", managed: true })
yield* status.beginStopping("next")
expect(yield* status.current).toEqual({ type: "stopping", targetVersion: "next" })
expect(yield* status.requestStop({ instanceID: "one" })).toBe(true)
expect(yield* status.current).toEqual({ type: "stopping", targetVersion: "next" })
}),
)

8
packages/tui/src/app.tsx
View file

@ -139,7 +139,7 @@ const appBindingCommands = [
export type TuiInput = {
server: {
endpoint: Service.Endpoint
reconnect?: (attempt: number) => Promise<Service.Endpoint>
reconnect?: (onStatus: (status: Service.Status) => void, signal: AbortSignal) => Promise<Service.Endpoint>
reload?: () => Promise<void>
}
args: Args
@ -186,8 +186,8 @@ export const run = Effect.fn("Tui.run")(function* (input: TuiInput) {
const handoff = input.terminalHandoff ? yield* Effect.promise(input.terminalHandoff) : undefined
const reconnectEndpoint = input.server.reconnect
const reconnect = reconnectEndpoint
? async (attempt: number) => {
const endpoint = await reconnectEndpoint(attempt)
? async (onStatus: (status: Service.Status) => void, signal: AbortSignal) => {
const endpoint = await reconnectEndpoint(onStatus, signal)
const next = { baseUrl: endpoint.url, headers: Service.headers(endpoint) }
return {
api: OpenCode.make(next),
@ -1121,7 +1121,7 @@ function App(props: { pair?: DialogPairCredentials }) {
<StartupLoading ready={plugins.ready} />
</Show>
<Show when={showReconnecting()}>
<Reconnecting attempt={client.connection.attempt()} error={client.connection.error()} />
<Reconnecting status={client.connection.service()} />
</Show>
</box>
)

51
packages/tui/src/component/reconnecting.tsx
View file

@ -1,9 +1,11 @@
import type { Service } from "@opencode-ai/client/effect"
import { Show } from "solid-js"
import { useTheme } from "../context/theme"
import { Spinner } from "./spinner"
export function Reconnecting(props: { attempt: number; error?: string }) {
export function Reconnecting(props: { status?: Service.Status }) {
const theme = useTheme().theme
const copy = () => reconnectingCopy(props.status)
return (
<box
@ -17,16 +19,47 @@ export function Reconnecting(props: { attempt: number; error?: string }) {
alignItems="center"
justifyContent="center"
>
<box width={54} maxWidth="90%" flexDirection="column" alignItems="center" gap={1}>
<text fg={theme.text}>Connection lost</text>
<Spinner color={theme.textMuted}>Reconnecting to server...</Spinner>
<text fg={theme.textMuted}>Attempt {props.attempt}</text>
<Show when={props.error}>
<text fg={theme.error} wrapMode="word">
{props.error}
</text>
<box width={62} maxWidth="90%" flexDirection="column" alignItems="center" gap={1}>
<Show when={!copy().loading} fallback={<Spinner color={theme.textMuted}>{copy().message}</Spinner>}>
<text fg={theme.error}>{copy().message}</text>
<Show when={copy().detail}>
{(detail) => (
<text fg={theme.textMuted} wrapMode="word">
{detail()}
</text>
)}
</Show>
<Show when={copy().action}>
{(action) => (
<text fg={theme.text} wrapMode="word">
{action()}
</text>
)}
</Show>
</Show>
</box>
</box>
)
}
export function reconnectingCopy(status?: Service.Status) {
if (status?.type === "starting")
return {
loading: true,
message: status.version ? `Starting OpenCode ${status.version}...` : "Starting background service...",
}
if (status?.type === "stopping")
return {
loading: true,
message: status.targetVersion ? `Updating to ${status.targetVersion}...` : "Restarting background service...",
}
if (status?.type === "failed")
return { loading: false, message: "Background service failed", detail: status.message, action: status.action }
if (status?.type === "unresponsive")
return {
loading: false,
message: "Background service is not responding",
action: "Run `opencode service restart` to recover it.",
}
return { loading: true, message: "Waiting for background service..." }
}

61
packages/tui/src/context/client.tsx
View file

@ -1,7 +1,9 @@
import type { OpenCodeClient, OpenCodeEvent } from "@opencode-ai/client"
import type { Service } from "@opencode-ai/client/effect"
import { createGlobalEmitter } from "@solid-primitives/event-bus"
import { onCleanup, onMount } from "solid-js"
import { createSignal, onCleanup, onMount } from "solid-js"
import { createStore } from "solid-js/store"
import { errorMessage } from "../util/error"
import { createSimpleContext } from "./helper"
import { useLog } from "./log"
@ -24,7 +26,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
name: "Client",
init: (props: {
api: OpenCodeClient
reconnect?: (attempt: number) => Promise<{ api: OpenCodeClient }>
reconnect?: (onStatus: (status: Service.Status) => void, signal: AbortSignal) => Promise<{ api: OpenCodeClient }>
// Stops and starts the managed service; present only in service mode.
reload?: () => Promise<void>
}) => {
@ -41,6 +43,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
status: "connecting",
attempt: 0,
})
const [service, setService] = createSignal<Service.Status>()
let stream: AbortController | undefined
function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) {
@ -51,43 +54,37 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
function start() {
stream?.abort()
const controller = new AbortController()
let connected!: () => void
const ready = new Promise<void>((resolve) => {
connected = resolve
})
stream = controller
void (async () => {
let attempt = 0
while (!abort.signal.aborted && !controller.signal.aborted) {
const connection = new AbortController()
const cancel = () => connection.abort(controller.signal.reason)
const timeout = setTimeout(
() => connection.abort(new Error("Timed out connecting to server")),
connectTimeout,
)
let connectedAt: number | undefined
const request = new AbortController()
const cancel = () => request.abort(controller.signal.reason)
const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout)
controller.signal.addEventListener("abort", cancel, { once: true })
const error = await (async () => {
record(attempt === 0 ? "connecting" : "reconnecting", attempt)
log.info("event stream connecting", { attempt })
const iterator = api.event.subscribe({ signal: connection.signal })[Symbol.asyncIterator]()
const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]()
const first = await iterator.next()
if (abort.signal.aborted || controller.signal.aborted) return
if (abort.signal.aborted || controller.signal.aborted) return undefined
if (first.done)
return connection.signal.reason instanceof Error
? connection.signal.reason
return request.signal.reason instanceof Error
? request.signal.reason
: new Error("Event stream disconnected")
if (first.value.type !== "server.connected")
return new Error("Event stream did not start with server.connected")
clearTimeout(timeout)
record("connected", attempt)
attempt = 0
connectedAt = Date.now()
log.info("event stream connected")
events.emit(first.value.type, first.value)
setConnection({ status: "connected", attempt: 0, error: undefined })
connected()
setService(undefined)
while (!abort.signal.aborted && !controller.signal.aborted) {
const event = await iterator.next()
if (abort.signal.aborted || controller.signal.aborted) return
if (abort.signal.aborted || controller.signal.aborted) return undefined
if (event.done) return new Error("Event stream disconnected")
if ("durable" in event.value)
log.debug("event", {
@ -97,42 +94,47 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
})
events.emit(event.value.type, event.value)
}
return undefined
})()
.catch((error) => error)
.finally(() => {
request.abort()
clearTimeout(timeout)
controller.signal.removeEventListener("abort", cancel)
})
if (abort.signal.aborted || controller.signal.aborted) return
if (connectedAt !== undefined && Date.now() - connectedAt >= 1_000) attempt = 0
attempt += 1
const message = error instanceof Error ? error.message : String(error)
const message = errorMessage(error)
record("disconnected", attempt, message)
log.info("event stream disconnected", {
attempt,
error: message,
})
setConnection({ status: "reconnecting", attempt, error: message })
// Re-resolve the transport before retrying: the server may have
// moved (service restarted on a new port) or need starting. Static
// transports (--server, standalone) resolve to the same address.
if (props.reconnect) {
const next = await props.reconnect(attempt).catch(() => undefined)
const next = await props.reconnect(setService, controller.signal).catch((error) => {
if (!controller.signal.aborted)
log.info("server resolution failed", {
attempt,
error: errorMessage(error),
})
})
if (abort.signal.aborted || controller.signal.aborted) return
if (next) {
api = next.api
if (attempt === 1) continue
}
}
setConnection({
status: "reconnecting",
attempt,
error: message,
})
await wait(1_000, controller.signal)
}
})()
return ready
}
onMount(() => void start())
onMount(start)
onCleanup(() => {
abort.abort()
stream?.abort()
@ -157,6 +159,9 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
error() {
return connection.error
},
service() {
return service()
},
internal: {
history() {
return history.slice()

31
packages/tui/test/cli/tui/reconnecting.test.ts Normal file
View file

@ -0,0 +1,31 @@
import { expect, test } from "bun:test"
import { reconnectingCopy } from "../../../src/component/reconnecting"
test("describes service status without transport diagnostics", () => {
expect(reconnectingCopy({ type: "starting", version: "2.0.0" })).toEqual({
loading: true,
message: "Starting OpenCode 2.0.0...",
})
expect(reconnectingCopy({ type: "stopping", targetVersion: "2.0.0" })).toEqual({
loading: true,
message: "Updating to 2.0.0...",
})
expect(
reconnectingCopy({
type: "failed",
message: "Could not open the database.",
action: "Check the service logs.",
}),
).toEqual({
loading: false,
message: "Background service failed",
detail: "Could not open the database.",
action: "Check the service logs.",
})
expect(reconnectingCopy({ type: "unresponsive" })).toEqual({
loading: false,
message: "Background service is not responding",
action: "Run `opencode service restart` to recover it.",
})
expect(JSON.stringify(reconnectingCopy())).not.toMatch(/Attempt|ECONNREFUSED|Event stream disconnected/)
})

109
packages/tui/test/cli/tui/use-event.test.tsx
View file

@ -1,6 +1,7 @@
/** @jsxImportSource @opentui/solid */
import { describe, expect, test } from "bun:test"
import type { OpenCodeClient, OpenCodeEvent } from "@opencode-ai/client"
import type { Service } from "@opencode-ai/client/effect"
import { testRender } from "@opentui/solid"
import { onMount } from "solid-js"
import { ProjectProvider, useProject } from "../../../src/context/project"
@ -53,7 +54,7 @@ function update(version: string): OpenCodeEvent {
}
async function mount(
reconnect?: (attempt: number) => Promise<{ api: OpenCodeClient }>,
reconnect?: (onStatus: (status: Service.Status) => void, signal: AbortSignal) => Promise<{ api: OpenCodeClient }>,
log?: LogSink,
) {
const events = createEventStream()
@ -194,8 +195,8 @@ describe("useEvent", () => {
const replacementEvents = createEventStream()
const replacementCalls = createFetch(undefined, replacementEvents)
const replacement = { api: createApi(replacementCalls.fetch) }
const { app, events, client, seen } = await mount(async (attempt) => {
attempts.push(attempt)
const { app, events, client, seen } = await mount(async () => {
attempts.push(attempts.length + 1)
return replacement
})
@ -246,4 +247,106 @@ describe("useEvent", () => {
app.renderer.destroy()
}
})
test("backs off when a resolved event stream keeps failing", async () => {
let calls = 0
const encoder = new TextEncoder()
const replacementCalls = createFetch((url) => {
if (url.pathname !== "/api/event") return undefined
return new Response(
new ReadableStream({
start(controller) {
controller.enqueue(
encoder.encode('data: {"id":"evt_connected","type":"server.connected","data":{}}\n\n'),
)
controller.close()
},
}),
{ headers: { "content-type": "text/event-stream" } },
)
})
const replacement = {
api: createApi(replacementCalls.fetch),
}
const { app, events, client } = await mount(async () => {
calls += 1
return replacement
})
try {
await wait(() => client.connection.status() === "connected")
events.disconnect()
await Promise.race([
wait(() => calls === 2),
Bun.sleep(500).then(() => {
throw new Error("resolved event stream did not retry immediately")
}),
])
await Bun.sleep(200)
expect(calls).toBe(2)
} finally {
app.renderer.destroy()
}
})
test("reports service status while endpoint resolution is pending", async () => {
const replacementEvents = createEventStream()
const replacement = { api: createApi(createFetch(undefined, replacementEvents).fetch) }
let report!: (status: Service.Status) => void
let resolve!: (value: typeof replacement) => void
const endpoint = new Promise<typeof replacement>((done) => {
resolve = done
})
const { app, events, client } = await mount(async (onStatus) => {
report = onStatus
onStatus({ type: "starting", version: "2.0.0" })
return endpoint
})
try {
await wait(() => client.connection.status() === "connected")
events.disconnect()
await wait(
() => client.connection.status() === "reconnecting" && client.connection.service()?.type === "starting",
)
expect(client.connection.service()).toEqual({ type: "starting", version: "2.0.0" })
report({ type: "failed", message: "Could not open the database.", action: "Check the service logs." })
await wait(() => client.connection.service()?.type === "failed")
expect(client.connection.service()).toEqual({
type: "failed",
message: "Could not open the database.",
action: "Check the service logs.",
})
resolve(replacement)
await wait(() => client.connection.status() === "connected")
expect(client.connection.service()).toBeUndefined()
} finally {
app.renderer.destroy()
}
})
test("cancels pending endpoint resolution on cleanup", async () => {
let aborted = false
const { app, events, client } = await mount(
(_onStatus, signal) =>
new Promise((_, reject) => {
signal.addEventListener(
"abort",
() => {
aborted = true
reject(signal.reason)
},
{ once: true },
)
}),
)
await wait(() => client.connection.status() === "connected")
events.disconnect()
await wait(() => client.connection.status() === "reconnecting")
app.renderer.destroy()
await wait(() => aborted)
})
})