vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-08-21 11:03:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

refactor(client): move service shutdown client-side (#43252)

Browse source
This commit is contained in:
Dax 2026-08-18 11:02:37 -04:00 • committed by GitHub
parent 8df039d261
commit b38d9d812f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 80 additions and 536 deletions
Download patch file Download diff file Expand all files Collapse all files

19
packages/cli/script/service-smoke.ts
View file

@ -1,8 +1,9 @@
#!/usr/bin/env bun
import { NodeFileSystem } from "@effect/platform-node"
import { Service } from "@opencode-ai/client/effect/service"
import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
import { Schema } from "effect"
import { Effect, Schema } from "effect"
import fs from "node:fs/promises"
import os from "node:os"
import path from "node:path"
@ -63,28 +64,22 @@ try {
})
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), {
const stopRoute = await fetch(new URL("/api/service/stop", info.url), {
method: "POST",
headers: { "content-type": "application/json" },
headers: { ...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")
if (stopRoute.status !== 404) throw new Error("Compiled service exposed the removed HTTP stop route")
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 }),
signal: AbortSignal.timeout(5_000),
}).then((response) => response.json()),
await Effect.runPromise(
Service.stop({ file: registration }).pipe(Effect.provide(NodeFileSystem.layer)),
)
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")

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

@ -117,7 +117,6 @@ const processEffect = Effect.fnUntraced(function* (options: Options) {
serviceOptions === undefined
? undefined
: {
instanceID,
onListen: (address, shutdown) =>
Effect.gen(function* () {
if (!config.password) yield* ServiceConfig.password(password)

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

@ -41,13 +41,8 @@ import type { Config } from "@opencode-ai/schema/config"
export type Endpoint0_0Output = { readonly healthy: true; readonly version: string; readonly pid: number }
export type HealthGetOperation<E = never> = () => Effect.Effect<Endpoint0_0Output, E>
export type Endpoint0_1Input = { readonly instanceID: string }
export type Endpoint0_1Output = { readonly accepted: boolean }
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 = { readonly urls: ReadonlyArray<string> }

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

@ -6,8 +6,6 @@ import { HttpApiClient } from "effect/unstable/httpapi"
import { ClientApi } from "../../contract"
import type {
Endpoint0_0Output,
Endpoint0_1Input,
Endpoint0_1Output,
Endpoint1_0Output,
Endpoint2_0Input,
Endpoint2_0Output,
@ -248,12 +246,7 @@ const preserveStream =
const Endpoint0_0 = (raw: RawClient["server.health"]) => () =>
preserveEffect<Endpoint0_0Output>()(raw["health.get"]({}).pipe(Effect.mapError(mapClientError)))
const Endpoint0_1 = (raw: RawClient["server.health"]) => (input: Endpoint0_1Input) =>
preserveEffect<Endpoint0_1Output>()(
raw["health.stop"]({ payload: { instanceID: input["instanceID"] } }).pipe(Effect.mapError(mapClientError)),
)
const adaptGroup0 = (raw: RawClient["server.health"]) => ({ get: Endpoint0_0(raw), stop: Endpoint0_1(raw) })
const adaptGroup0 = (raw: RawClient["server.health"]) => ({ get: Endpoint0_0(raw) })
const Endpoint1_0 = (raw: RawClient["server.server"]) => () =>
preserveEffect<Endpoint1_0Output>()(raw["server.get"]({}).pipe(Effect.mapError(mapClientError)))

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

@ -87,7 +87,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
}
if (timeouts.count >= 3) {
yield* announce("missing")
yield* evict(info, options, timing)
yield* terminate(info, options, timing)
timeouts = undefined
lastSpawn = Date.now() - spawnDelay
}
@ -100,7 +100,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
return yield* Effect.fail(new Error("Background service failed to start"))
if (compatible) return Option.none<LocalService>()
yield* announce("version-mismatch", service.version)
yield* kill(service, options, timing).pipe(Effect.ignore)
yield* terminate(service.info, options, timing).pipe(Effect.ignore)
lastSpawn = 0
return Option.none<LocalService>()
} else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now()
@ -133,8 +133,8 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
/** Stop the registered local service. */
export const stop = Effect.fn("service.stop")(function* (options: StopOptions = {}) {
const existing = yield* find(options)
if (existing !== undefined) yield* kill(existing, options, defaultEnsureTiming)
const info = yield* read(options.file)
if (info !== undefined) yield* terminate(info, options, defaultEnsureTiming)
})
function fallback() {
@ -243,12 +243,6 @@ const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = fal
return { info, ...(yield* probeResult(info, allowLegacy, timeout)) }
})
// Health-checked lookup without the version gate: lifecycle operations must be
// able to see (and replace or stop) a server from a different version.
const find = Effect.fnUntraced(function* (options: { readonly file?: string }) {
return (yield* registered(options.file, true)).service
})
// 50ms cadence bounded at ~5s, shared by stop escalation and each ensure
// discovery window.
const poll = (timing: EnsureTiming) =>
@ -269,59 +263,21 @@ function same(left: Info, right: Info) {
return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
}
const evict = Effect.fnUntraced(function* (info: Info, options: { readonly file?: string }, timing: EnsureTiming) {
const terminate = Effect.fnUntraced(function* (info: Info, options: { readonly file?: string }, timing: EnsureTiming) {
const current = yield* read(options.file)
if (current === undefined || !same(current, info)) return
yield* signal(info.pid, "SIGTERM")
const done = yield* stopped(info.pid).pipe(Effect.retry(poll(timing)), Effect.option)
if (Option.isSome(done)) return
if (Option.isNone(done)) {
const latest = yield* read(options.file)
if (latest === undefined || !same(latest, info)) return
yield* signal(info.pid, "SIGKILL")
yield* stopped(info.pid).pipe(Effect.retry(poll(timing)))
}
const latest = yield* read(options.file)
if (latest === undefined || !same(latest, info)) return
yield* signal(info.pid, "SIGKILL")
yield* stopped(info.pid).pipe(Effect.retry(poll(timing)))
})
const kill = Effect.fnUntraced(function* (
service: LocalService,
options: { readonly file?: string },
timing: EnsureTiming,
) {
const requested = yield* requestStop(service, timing.requestTimeout)
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(timing)), Effect.option)
if (Option.isSome(done)) return
const latest = yield* find(options)
if (latest === undefined || !same(latest.info, service.info)) return
yield* signal(service.info.pid, "SIGKILL")
yield* stopped(service.info.pid).pipe(Effect.retry(poll(timing)))
})
const decodeStopResponse = Schema.decodeUnknownOption(ServiceStatus.StopResponse)
const requestStop = Effect.fnUntraced(function* (service: LocalService, timeout = defaultEnsureTiming.requestTimeout) {
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 }),
signal: AbortSignal.timeout(timeout),
}),
).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
const fs = yield* FileSystem.FileSystem
yield* fs.remove(options.file ?? fallback()).pipe(Effect.ignore)
})
/** Effect-based local service lifecycle operations. */

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

@ -1,7 +1,5 @@
import type {
HealthGetOutput,
HealthStopInput,
HealthStopOutput,
ServerGetOutput,
LocationGetInput,
LocationGetOutput,
@ -367,18 +365,6 @@ 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"] },
successStatus: 200,
declaredStatuses: [401, 400],
empty: false,
},
requestOptions,
),
},
server: {
get: (requestOptions?: RequestOptions) =>

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

@ -2,8 +2,6 @@ export type JsonValue = null | boolean | number | string | Array<JsonValue> | {
export type ServiceHealth = { healthy: true; version: string; pid: number }
export type ServiceStopResponse = { accepted: boolean }
export type ModelRef = { id: string; providerID: string; variant?: string }
export type ProviderSettings = { [x: string]: any }
@ -2273,10 +2271,6 @@ export const isWorktreeError = (value: unknown): value is WorktreeError =>
export type HealthGetOutput = ServiceHealth
export type HealthStopInput = { readonly instanceID: { readonly instanceID: string }["instanceID"] }
export type HealthStopOutput = ServiceStopResponse
export type ServerGetOutput = { urls: Array<string> }
export type LocationGetInput = {

60
packages/client/src/promise/service.ts
View file

@ -1,4 +1,4 @@
import { readFile } from "node:fs/promises"
import { readFile, rm } from "node:fs/promises"
import { homedir } from "node:os"
import { join } from "node:path"
import type { DiscoverOptions, Endpoint, Info, EnsureOptions, StopOptions } from "../service.js"
@ -10,7 +10,7 @@ import {
} from "../service-contender.js"
import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js"
import { matchesVersion } from "../service-version.js"
import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js"
import type { ServiceHealth } from "./generated/types.js"
export * from "../service.js"
@ -68,7 +68,7 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
}
if (timeouts.count >= 3) {
announce("missing")
await evict(registration.info, options, timing)
await terminate(registration.info, options, timing)
timeouts = undefined
lastSpawn = Date.now() - spawnDelay
}
@ -82,7 +82,7 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
if (compatible && service.state === "failed") throw new Error("Background service failed to start")
if (!compatible) {
announce("version-mismatch", service.version)
await kill(service, options, timing).catch(() => undefined)
await terminate(service.info, options, timing).catch(() => undefined)
lastSpawn = 0
}
} else {
@ -110,8 +110,8 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
/** Stop the registered local service. */
export async function stop(options: StopOptions = {}) {
const existing = await find(options)
if (existing !== undefined) await kill(existing, options, defaultEnsureTiming)
const info = await read(options.file)
if (info !== undefined) await terminate(info, options, defaultEnsureTiming)
}
function fallback() {
@ -199,10 +199,6 @@ async function registered(file?: string, allowLegacy = false, timeout?: number)
return { info, ...(await probeResult(info, allowLegacy, timeout)) }
}
async function find(options: { readonly file?: string }) {
return (await registered(options.file, true)).service
}
function signal(pid: number, name: NodeJS.Signals) {
try {
process.kill(pid, name)
@ -230,47 +226,19 @@ function same(left: Info, right: Info) {
return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
}
async function evict(info: Info, options: { readonly file?: string }, timing: EnsureTiming) {
async function terminate(info: Info, options: { readonly file?: string }, timing: EnsureTiming) {
const current = await read(options.file)
if (current === undefined || !same(current, info)) return
signal(info.pid, "SIGTERM")
if (await waitUntilStopped(info.pid, timing)) return
if (!(await waitUntilStopped(info.pid, timing))) {
const latest = await read(options.file)
if (latest === undefined || !same(latest, info)) return
signal(info.pid, "SIGKILL")
if (!(await waitUntilStopped(info.pid, timing))) throw new Error(`Server process ${info.pid} is still running`)
}
const latest = await read(options.file)
if (latest === undefined || !same(latest, info)) return
signal(info.pid, "SIGKILL")
if (!(await waitUntilStopped(info.pid, timing))) throw new Error(`Server process ${info.pid} is still running`)
}
async function kill(service: LocalService, options: { readonly file?: string }, timing: EnsureTiming) {
const requested = await requestStop(service, timing.requestTimeout)
if (requested === "rejected") return
if (requested === "unsupported") {
const current = await find(options)
if (current === undefined || !same(current.info, service.info)) return
signal(service.info.pid, "SIGTERM")
}
if (await waitUntilStopped(service.info.pid, timing)) return
const latest = await find(options)
if (latest === undefined || !same(latest.info, service.info)) return
signal(service.info.pid, "SIGKILL")
if (!(await waitUntilStopped(service.info.pid, timing)))
throw new Error(`Server process ${service.info.pid} is still running`)
}
async function requestStop(service: LocalService, timeout = defaultEnsureTiming.requestTimeout) {
if (service.info.id === undefined || service.legacy) return "unsupported" as const
const response = await 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 }),
signal: AbortSignal.timeout(timeout),
}).catch(() => undefined)
if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const
const body = (await response.json().catch(() => undefined)) as ServiceStopResponse | undefined
if (!response.ok || body?.accepted !== true) return "rejected" as const
return "accepted" as const
await rm(options.file ?? fallback(), { force: true })
}
function delay(milliseconds: number) {

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

@ -28,7 +28,7 @@ if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated" ||
let requests = 0
let version = "test"
if (mode === "old" || mode === "reject-stop") version = "old"
if (mode === "old") version = "old"
if (mode === "incompatible") version = "1.9.0"
if (mode === "compatible" || mode === "delayed-compatible") version = "2.1.0-next.1"
const id = crypto.randomUUID()
@ -36,17 +36,6 @@ const server = Bun.serve({
port: 0,
async fetch(request) {
const pathname = new URL(request.url).pathname
if (pathname === "/api/service/stop" && mode === "reject-stop") {
await appendFile(registration + ".stop-attempts", process.pid + "\n")
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 === "starting") await writeFile(registration + ".health-request", "")
@ -63,7 +52,7 @@ const server = Bun.serve({
if (mode === "starting" && !(await Bun.file(registration + ".release").exists()))
return Response.json({ healthy: true, version, pid: process.pid }, { status: 503 })
if (mode === "failed-owner") return Response.json({ healthy: true, version, pid: process.pid }, { status: 500 })
if (mode === "starting" || mode === "graceful" || mode === "reject-stop")
if (mode === "starting" || mode === "graceful")
return Response.json({ healthy: true, version, pid: process.pid })
return Response.json({ healthy: true, version, pid: process.pid })
},
@ -81,9 +70,10 @@ await writeFile(
)
await rename(registration + ".tmp", registration)
function shutdown() {
async function shutdown(signal?: NodeJS.Signals) {
if (signal !== undefined) await writeFile(registration + ".signal", signal)
server.stop(true)
process.exit()
}
process.on("SIGTERM", shutdown)
process.on("SIGINT", shutdown)
process.on("SIGTERM", () => void shutdown("SIGTERM"))
process.on("SIGINT", () => void shutdown("SIGINT"))

6
packages/client/test/promise-service.test.ts
View file

@ -126,13 +126,13 @@ test("evicts an unresponsive registered service before starting its replacement"
await waitForExit(replacement.pid)
})
test("requests graceful stop of the exact service instance", async () => {
test("signals the registered service process", async () => {
const registration = await setup("graceful")
const info = await Bun.file(registration).json()
await Service.stop({ file: registration })
expect(await Bun.file(registration + ".stop").json()).toEqual({ instanceID: info.id })
expect(await Bun.file(registration + ".signal").text()).toBe("SIGTERM")
expect(await Bun.file(registration).exists()).toBe(false)
})
async function setup(mode: string) {

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

@ -191,22 +191,6 @@ test("integration connections optionally submit a form answer", async () => {
expect(await requests[3].json()).toEqual({ methodID: "device" })
})
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" })).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" })
})
test("MCP resource catalog uses the public HTTP contract", async () => {
let request: Request | undefined
const client = OpenCode.make({

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

@ -143,40 +143,36 @@ test("evicts an unresponsive registered service before starting its replacement"
await waitForExit(replacement.pid)
})
test("requests graceful stop of the exact service instance", async () => {
test("signals an unresponsive registered service process", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const process = spawn(registration, "graceful")
const process = spawn(registration, "hanging")
await waitForFile(registration)
const info = await Bun.file(registration).json()
await run(Service.stop({ file: registration }))
await process.exited
expect(await Bun.file(registration + ".stop").json()).toEqual({ instanceID: info.id })
expect(await Bun.file(registration + ".signal").text()).toBe("SIGTERM")
expect(await Bun.file(registration).exists()).toBe(false)
})
test("does not spawn contenders while an incompatible service rejects replacement", async () => {
test("signals an incompatible service before starting its replacement", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const contender = join(directory, "contender.json")
const existing = spawn(registration, "reject-stop")
const existing = spawn(registration, "old")
await waitForFile(registration)
const controller = new AbortController()
const starting = Effect.runPromise(
const endpoint = await run(
ensure({
file: registration,
version: "test",
command: [process.execPath, fixture, contender, "record-start"],
}).pipe(Effect.provide(NodeFileSystem.layer)),
{ signal: controller.signal },
command: [process.execPath, fixture, registration, "delayed", "10"],
}),
)
const replacement = await Bun.file(registration).json()
await waitForLines(registration + ".stop-attempts", 2)
controller.abort()
await starting.catch(() => undefined)
expect(await Bun.file(contender + ".started").exists()).toBe(false)
expect(existing.exitCode).toBe(null)
expect(await existing.exited).toBe(0)
expect(endpoint.url).toBe(replacement.url)
process.kill(replacement.pid, "SIGTERM")
await waitForExit(replacement.pid)
})
test("a legacy health response is still replaced", async () => {
@ -344,17 +340,6 @@ async function waitForFile(file: string) {
throw new Error(`Timed out waiting for ${file}`)
}
async function waitForLines(file: string, count: number) {
for (let attempt = 0; attempt < 600; attempt++) {
const text = await Bun.file(file)
.text()
.catch(() => "")
if (text.trim().split("\n").length >= count) return
await Bun.sleep(5)
}
throw new Error(`Timed out waiting for ${count} lines in ${file}`)
}
async function health(url: string) {
return fetch(new URL("/api/health", url), { signal: AbortSignal.timeout(1_000) }).then((response) => response.json())
}

72
packages/protocol/openapi.json
View file

@ -48,58 +48,6 @@
"summary": "Check server health"
}
},
"/api/service/stop": {
"post": {
"tags": ["health"],
"operationId": "v2.health.stop",
"parameters": [],
"security": [],
"responses": {
"200": {
"description": "ServiceStopResponse",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ServiceStopResponse"
}
}
}
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
}
},
"description": "Request graceful shutdown of one exact managed server instance.",
"summary": "Stop the managed server",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ServiceStopRequest"
}
}
},
"required": true
}
}
},
"/api/server": {
"get": {
"tags": ["server"],
@ -9783,26 +9731,6 @@
"required": ["_tag", "message"],
"additionalProperties": false
},
"ServiceStopRequest": {
"type": "object",
"properties": {
"instanceID": {
"type": "string"
}
},
"required": ["instanceID"],
"additionalProperties": false
},
"ServiceStopResponse": {
"type": "object",
"properties": {
"accepted": {
"type": "boolean"
}
},
"required": ["accepted"],
"additionalProperties": false
},
"Union_1": {
"anyOf": [
{

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

@ -9,16 +9,6 @@ export namespace ServiceStatus {
pid: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)),
}).annotate({ identifier: "ServiceHealth" })
export type Health = typeof Health.Type
export const StopRequest = Schema.Struct({
instanceID: Schema.String,
}).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")
@ -33,16 +23,4 @@ export const HealthGroup = HttpApiGroup.make("server.health")
}),
),
)
.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.",
}),
),
)
.annotateMerge(OpenApi.annotations({ title: "health" }))

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

@ -4,17 +4,15 @@ import { Api } from "../api"
import { ServerInfo } from "../server-info"
export const HealthHandler = HttpApiBuilder.group(Api, "server.health", (handlers) =>
handlers
.handle("health.get", () =>
Effect.gen(function* () {
const info = yield* ServerInfo.Service
return {
healthy: true as const,
version: info.app.version ?? "unknown",
// Runtimes without OS process identity (workerd) report 0.
pid: process.pid ?? 0,
}
}),
)
.handle("health.stop", () => Effect.succeed({ accepted: false })),
handlers.handle("health.get", () =>
Effect.gen(function* () {
const info = yield* ServerInfo.Service
return {
healthy: true as const,
version: info.app.version ?? "unknown",
// Runtimes without OS process identity (workerd) report 0.
pid: process.pid ?? 0,
}
}),
),
)

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

@ -1,12 +1,10 @@
export * as ServerProcess from "./process"
import { NodeHttpServer, NodeHttpServerRequest } from "@effect/platform-node"
import { NodeHttpServer } 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 { Cause, Context, Deferred, Effect, Exit, Layer, Option, Ref, Scope } from "effect"
import { HttpMiddleware, HttpRouter, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
import { randomUUID } from "node:crypto"
import { createServer } from "node:http"
import { ServerAuth } from "./auth"
import { isAllowedCorsOrigin } from "./cors"
@ -18,7 +16,6 @@ import { Status } from "./service-status"
import type { ServerOptions } from "./options"
export interface Lifecycle<E = never, R = never> {
readonly instanceID: string
readonly onListen: (
address: HttpServer.Address,
shutdown: Effect.Effect<void>,
@ -51,16 +48,13 @@ export const start = Effect.fn("ServerProcess.start")(function* <E, R>(
const hostname = options.hostname ?? "127.0.0.1"
const port = Option.fromNullishOr(options.port)
const shutdown = yield* Deferred.make<void>()
const status = yield* Status.make({
instanceID: lifecycle?.instanceID ?? randomUUID(),
managed: lifecycle !== undefined,
})
const status = yield* Status.make()
const bound = yield* listen({ hostname, port })
const application = yield* Ref.make(Option.none<App>())
// Request fibers may continue inbound trace context, but must not inherit the server startup parent.
yield* bound.http
.serve(
dispatch(password, status, application, shutdown, options.app?.version ?? "unknown").pipe(
dispatch(password, status, application, options.app?.version ?? "unknown").pipe(
HttpMiddleware.cors({ allowedOrigins: isAllowedCorsOrigin, maxAge: 86_400 }),
),
errorResponseLogger,
@ -163,22 +157,15 @@ function dispatch(
password: string,
status: Status.Interface,
application: Ref.Ref<Option.Option<App>>,
shutdown: Deferred.Deferred<void>,
version: string,
): App {
const auth = ServerAuth.Config.of({ password: Option.some(password), username: "opencode" })
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 (request.method === "GET" && url.pathname === "/api/health") {
if (!(yield* authorizedRequest(request, auth))) return unauthorized()
return yield* control(request, lifecycle, status, () => Deferred.doneUnsafe(shutdown, Effect.void), version)
return yield* healthResponse(status, version)
}
const state = yield* status.current
const app = yield* Ref.get(application)
@ -196,33 +183,6 @@ function unauthorized() {
})
}
const control = Effect.fnUntraced(function* (
request: HttpServerRequest.HttpServerRequest,
route: "health" | "stop",
status: Status.Interface,
stop: () => void,
version: string,
) {
if (route === "health") return yield* healthResponse(status, version)
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, version: string) {
const state = yield* status.current
return HttpServerResponse.jsonUnsafe(

12
packages/server/src/service-status.ts
View file

@ -1,6 +1,5 @@
export * as Status from "./service-status"
import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
import { Effect, Ref } from "effect"
export type State =
@ -14,14 +13,9 @@ export interface Interface {
readonly ready: Effect.Effect<void>
readonly fail: Effect.Effect<void>
readonly beginStopping: 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?: State
}) {
export const make = Effect.fnUntraced(function* (options: { readonly initial?: State } = {}) {
const current = yield* Ref.make(options.initial ?? ({ type: "starting" } satisfies State))
const beginStopping = Ref.update(current, (status) =>
status.type === "stopping" ? status : ({ type: "stopping" } satisfies State),
@ -32,9 +26,5 @@ export const make = Effect.fnUntraced(function* (options: {
ready: Ref.update(current, (status) => (status.type === "starting" ? ({ type: "ready" } satisfies State) : status)),
fail: Ref.update(current, (status) => (status.type === "starting" ? ({ type: "failed" } satisfies State) : status)),
beginStopping,
requestStop: (request) => {
if (!options.managed || request.instanceID !== options.instanceID) return Effect.succeed(false)
return beginStopping.pipe(Effect.as(true))
},
} satisfies Interface
})

19
packages/server/test/service-status.test.ts
View file

@ -5,7 +5,7 @@ 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 })
const status = yield* Status.make()
expect(yield* status.current).toEqual({ type: "starting" })
yield* status.ready
expect(yield* status.current).toEqual({ type: "ready" })
@ -14,7 +14,7 @@ it.effect("moves from starting to ready", () =>
it.effect("keeps a startup failure until shutdown", () =>
Effect.gen(function* () {
const status = yield* Status.make({ instanceID: "one", managed: true })
const status = yield* Status.make()
yield* status.fail
yield* status.ready
yield* status.fail
@ -22,24 +22,13 @@ it.effect("keeps a startup failure until shutdown", () =>
}),
)
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" })).toBe(false)
expect(yield* status.current).toEqual({ type: "starting" })
expect(yield* status.requestStop({ instanceID: "one" })).toBe(true)
expect(yield* status.current).toEqual({ type: "stopping" })
}),
)
it.effect("keeps stopping after shutdown begins", () =>
Effect.gen(function* () {
const status = yield* Status.make({ instanceID: "one", managed: true })
const status = yield* Status.make()
yield* status.beginStopping
expect(yield* status.current).toEqual({ type: "stopping" })
expect(yield* status.requestStop({ instanceID: "one" })).toBe(true)
yield* status.beginStopping
expect(yield* status.current).toEqual({ type: "stopping" })
}),
)

72
packages/www/openapi.json
View file

@ -48,58 +48,6 @@
"summary": "Check server health"
}
},
"/api/service/stop": {
"post": {
"tags": ["health"],
"operationId": "v2.health.stop",
"parameters": [],
"security": [],
"responses": {
"200": {
"description": "ServiceStopResponse",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ServiceStopResponse"
}
}
}
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
}
},
"description": "Request graceful shutdown of one exact managed server instance.",
"summary": "Stop the managed server",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ServiceStopRequest"
}
}
},
"required": true
}
}
},
"/api/server": {
"get": {
"tags": ["server"],
@ -9783,26 +9731,6 @@
"required": ["_tag", "message"],
"additionalProperties": false
},
"ServiceStopRequest": {
"type": "object",
"properties": {
"instanceID": {
"type": "string"
}
},
"required": ["instanceID"],
"additionalProperties": false
},
"ServiceStopResponse": {
"type": "object",
"properties": {
"accepted": {
"type": "boolean"
}
},
"required": ["accepted"],
"additionalProperties": false
},
"Union_1": {
"anyOf": [
{

72
packages/www/public/openapi.json
View file

@ -48,58 +48,6 @@
"summary": "Check server health"
}
},
"/api/service/stop": {
"post": {
"tags": ["health"],
"operationId": "v2.health.stop",
"parameters": [],
"security": [],
"responses": {
"200": {
"description": "ServiceStopResponse",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ServiceStopResponse"
}
}
}
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
}
},
"description": "Request graceful shutdown of one exact managed server instance.",
"summary": "Stop the managed server",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ServiceStopRequest"
}
}
},
"required": true
}
}
},
"/api/server": {
"get": {
"tags": ["server"],
@ -9783,26 +9731,6 @@
"required": ["_tag", "message"],
"additionalProperties": false
},
"ServiceStopRequest": {
"type": "object",
"properties": {
"instanceID": {
"type": "string"
}
},
"required": ["instanceID"],
"additionalProperties": false
},
"ServiceStopResponse": {
"type": "object",
"properties": {
"accepted": {
"type": "boolean"
}
},
"required": ["accepted"],
"additionalProperties": false
},
"Union_1": {
"anyOf": [
{