Compare commits

..

2 Commits

Author SHA1 Message Date
Kit Langton f0308ec44a fix(cli): prevent stale service replacement 2026-08-04 16:25:14 -04:00
Kit Langton 1fd817ffcf fix(client): wait for live service contenders (#40515) 2026-08-04 20:23:57 +00:00
12 changed files with 239 additions and 165 deletions
@@ -5,6 +5,7 @@ import { Service } from "@opencode-ai/client/effect/service"
import { Effect, FileSystem, Option, Schema } from "effect"
import { randomBytes } from "crypto"
import path from "path"
import semver from "semver"
import { selfCommand } from "../util/process"
// The CLI's service configuration file, plus the Service.EnsureOptions binding that
@@ -104,10 +105,21 @@ export const options = Effect.fnUntraced(function* () {
return {
file,
version: OPENCODE_VERSION,
canReplace: (version: string | undefined) => canReplaceVersion(version),
command: [...selfCommand(), "serve", "--service"],
}
})
export function canReplaceVersion(serverVersion: string | undefined, clientVersion = OPENCODE_VERSION) {
if (serverVersion === undefined) return true
// Preview versions end in `<channel>-<build>[.<attempt>]`. Convert the build
// to a numeric semver identifier so next-15000 sorts after next-9999.
const server = serverVersion.replace(/-(\d+)(?=(?:\.\d+)?$)/, ".$1")
const client = clientVersion.replace(/-(\d+)(?=(?:\.\d+)?$)/, ".$1")
if (!semver.valid(server) || !semver.valid(client)) return true
return semver.lt(server, client)
}
export const read = Effect.fn("cli.service-config.read")(function* () {
const { fs, configFile, legacyConfigFile } = yield* paths
if (legacyConfigFile) yield* migrateConfig(legacyConfigFile, configFile)
+9
View File
@@ -47,6 +47,15 @@ test("service filenames share release channels and identify preview channels", (
expect(ServiceConfig.versionBelongsToChannel("1.2.3", "preview-a")).toBe(false)
})
test("only newer clients replace managed service versions", () => {
expect(ServiceConfig.canReplaceVersion("1.2.3", "1.2.4")).toBe(true)
expect(ServiceConfig.canReplaceVersion("1.2.4", "1.2.3")).toBe(false)
expect(ServiceConfig.canReplaceVersion("1.2.3", "1.2.3")).toBe(false)
expect(ServiceConfig.canReplaceVersion("0.0.0-next-9999", "0.0.0-next-15000")).toBe(true)
expect(ServiceConfig.canReplaceVersion("0.0.0-next-15000", "0.0.0-next-9999")).toBe(false)
expect(ServiceConfig.canReplaceVersion(undefined, "1.2.3")).toBe(true)
})
test("service config migrates from the hashed channel filename", async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-config-migration-"))
const legacy = path.join(root, ServiceConfig.legacyFilename("preview-a")!)
+11 -6
View File
@@ -3,7 +3,13 @@ import { Effect, FileSystem, Option, Schedule, Schema } from "effect"
import { spawn, type ChildProcess } from "node:child_process"
import { homedir } from "node:os"
import { join } from "node:path"
import type { DiscoverOptions, Endpoint, EnsureOptions, StopOptions } from "../service.js"
import {
VersionMismatchError,
type DiscoverOptions,
type Endpoint,
type EnsureOptions,
type StopOptions,
} from "../service.js"
export * from "../service.js"
/** Contents of the local service registration file. */
@@ -56,7 +62,6 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
let announced = false
let lastSpawn = 0
let spawnDelay = 5_000
let ownerHeld = false
const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) =>
Effect.sync(() => {
if (announced) return
@@ -84,27 +89,27 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
const info = registration.info
const service = registration.service
if (service !== undefined) {
ownerHeld = false
spawnDelay = 5_000
const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
if (compatible && service.state === "ready") return Option.some(service)
if (compatible && service.state === "failed")
return yield* Effect.fail(new Error("Background service failed to start"))
if (compatible) return Option.none<LocalService>()
if (options.canReplace?.(service.version) === false)
return yield* Effect.fail(new VersionMismatchError(options.version, service.version))
yield* announce("version-mismatch", service.version)
yield* kill(service, options).pipe(Effect.ignore)
lastSpawn = 0
return Option.none<LocalService>()
} else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now()
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)
const failure = finished.map(contenderFailure).find((error): error is Error => error !== undefined)
if (finished.some((item) => item.child.exitCode === 0)) {
ownerHeld = true
spawnDelay = Math.min(spawnDelay * 2, 30_000)
}
finished.forEach((item) => contenders.delete(item))
if (failure !== undefined && contenders.size === 0) return yield* Effect.fail(failure)
// 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")
+12 -6
View File
@@ -2,7 +2,14 @@ import { readFile } from "node:fs/promises"
import { spawn, type ChildProcess } from "node:child_process"
import { homedir } from "node:os"
import { join } from "node:path"
import type { DiscoverOptions, Endpoint, Info, EnsureOptions, StopOptions } from "../service.js"
import {
VersionMismatchError,
type DiscoverOptions,
type Endpoint,
type Info,
type EnsureOptions,
type StopOptions,
} from "../service.js"
import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js"
export * from "../service.js"
@@ -37,7 +44,6 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
let announced = false
let lastSpawn = 0
let spawnDelay = 5_000
let ownerHeld = false
const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) => {
if (announced) return
@@ -65,27 +71,27 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
const registration = await registered(options.file, true)
if (registration.service !== undefined) {
ownerHeld = false
spawnDelay = 5_000
const service = registration.service
const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
if (compatible && service.state === "ready") return service.endpoint
if (compatible && service.state === "failed") throw new Error("Background service failed to start")
if (!compatible) {
if (options.canReplace?.(service.version) === false)
throw new VersionMismatchError(options.version, service.version)
announce("version-mismatch", service.version)
await kill(service, options).catch(() => undefined)
lastSpawn = 0
}
} else {
if (lastSpawn === 0 && registration.info !== undefined) lastSpawn = Date.now()
const failure = [...contenders].map(contenderFailure).find((error) => error !== undefined)
if (failure !== undefined) throw failure
const finished = [...contenders].filter(contenderFinished)
const failure = finished.map(contenderFailure).find((error) => error !== undefined)
if (finished.some((item) => item.child.exitCode === 0)) {
ownerHeld = true
spawnDelay = Math.min(spawnDelay * 2, 30_000)
}
finished.forEach((item) => contenders.delete(item))
if (failure !== undefined && contenders.size === 0) throw failure
// Keep one candidate plus one lock probe so a pre-lock stall cannot block recovery.
if (contenders.size < 2 && Date.now() - lastSpawn >= spawnDelay) {
announce("missing")
+14
View File
@@ -28,10 +28,24 @@ export type EnsureReason = "missing" | "version-mismatch"
export type EnsureOptions = DiscoverOptions & {
/** Service command and arguments. Defaults to `opencode serve --service`. */
readonly command?: ReadonlyArray<string>
/** Decide whether a version-mismatched service may be replaced. Defaults to true. */
readonly canReplace?: (version: string | undefined) => boolean
/** Called once before spawning a new service process. */
readonly onStart?: (reason: EnsureReason, previousVersion?: string) => void
}
/** A healthy service exists, but the caller's replacement policy protects it. */
export class VersionMismatchError extends Error {
override readonly name = "VersionMismatchError"
constructor(
readonly clientVersion: string | undefined,
readonly serverVersion: string | undefined,
) {
super(`Client version ${clientVersion ?? "unknown"} cannot replace server version ${serverVersion ?? "unknown"}`)
}
}
/** Options used to stop the local OpenCode service. */
export type StopOptions = {
/** Absolute registration file path. Defaults to the XDG state directory. */
+9 -3
View File
@@ -9,14 +9,20 @@ if (mode === "record-start") {
}
if (mode === "signal") process.kill(process.pid, process.platform === "win32" ? "SIGTERM" : "SIGKILL")
if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated") {
if (
mode === "delayed" ||
mode === "delayed-failed" ||
mode === "coordinated" ||
mode === "coordinated-failed-loser"
) {
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") {
if (!owner) process.exit(mode === "coordinated-failed-loser" ? 1 : 0)
if (mode === "coordinated" || mode === "coordinated-failed-loser") {
while ((await Bun.file(registration + ".starts").text()).trim().split("\n").length < 2) await Bun.sleep(10)
if (mode === "coordinated-failed-loser") await Bun.sleep(1_500)
} else await Bun.sleep(Number(delay))
if (mode === "delayed-failed") process.exit(1)
}
@@ -44,6 +44,24 @@ test("ensures a missing service with native promises", async () => {
}
}, 15_000)
test("waits for a live contender when another native contender fails", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const endpoint = await Service.ensure({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "coordinated-failed-loser"],
})
const info = await Bun.file(registration).json()
try {
expect(endpoint.url).toBe(info.url)
} finally {
process.kill(info.pid, "SIGTERM")
await waitForExit(info.pid)
}
}, 15_000)
test("reports a failed registered service", async () => {
const registration = await setup("failed-owner")
@@ -52,6 +70,25 @@ test("reports a failed registered service", async () => {
)
})
test("does not replace a version rejected by the caller", async () => {
const registration = await setup("graceful")
const directory = await temp()
const contender = join(directory, "contender.json")
const info = await Bun.file(registration).json()
await expect(
Service.ensure({
file: registration,
version: "old",
canReplace: () => false,
command: [process.execPath, fixture, contender, "record-start"],
}),
).rejects.toThrow("Client version old cannot replace server version test")
expect(await Bun.file(contender + ".started").exists()).toBe(false)
expect(process.kill(info.pid, 0)).toBe(true)
})
test("requests graceful stop of the exact service instance", async () => {
const registration = await setup("graceful")
const info = await Bun.file(registration).json()
+40
View File
@@ -107,6 +107,28 @@ test("does not spawn contenders while an incompatible service rejects replacemen
expect(existing.exitCode).toBe(null)
})
test("does not replace a version rejected by the caller", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const contender = join(directory, "contender.json")
const existing = spawn(registration, "graceful")
await waitForFile(registration)
await expect(
run(
Service.ensure({
file: registration,
version: "old",
canReplace: () => false,
command: [process.execPath, fixture, contender, "record-start"],
}),
),
).rejects.toThrow("Client version old cannot replace server version test")
expect(await Bun.file(contender + ".started").exists()).toBe(false)
expect(existing.exitCode).toBe(null)
})
test("a legacy health response is still replaced", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
@@ -141,6 +163,24 @@ test("waits for a slow winner while bounding lock probes", async () => {
}
}, 15_000)
test("waits for a live contender when another contender fails", async () => {
const directory = await temp()
const registration = join(directory, "service.json")
const endpoint = await run(
Service.ensure({
file: registration,
version: "test",
command: [process.execPath, fixture, registration, "coordinated-failed-loser"],
}),
)
const info = await Bun.file(registration).json()
try {
expect(endpoint.url).toBe(info.url)
} 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")
+4 -1
View File
@@ -327,7 +327,10 @@ export function Prompt(props: PromptProps) {
if (!session) return
const agent = session.agent && local.agent.list().find((agent) => agent.id === session.agent)
if (agent && !args.agent) local.agent.set(agent.id)
if (!local.model.hydrate(session.model)) return
if (session.model) {
local.model.set({ providerID: session.model.providerID, modelID: session.model.id })
local.model.variant.set(session.model.variant)
}
syncedSessionID = sessionID
})
+18 -45
View File
@@ -6,6 +6,7 @@ import { useEvent } from "./event"
import path from "path"
import { useTuiPaths } from "./runtime"
import { useArgs } from "./args"
import { useClient } from "./client"
import { RGBA } from "@opentui/core"
import { readJson, writeJsonAtomic } from "../util/persistence"
import {
@@ -47,6 +48,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
name: "Local",
init: () => {
const data = useData()
const client = useClient()
const toast = useToast()
const theme = useTheme()
const { mode } = useThemes()
@@ -208,47 +210,6 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
)
})
function select(model: ModelPreferenceModel, options?: { recent?: boolean }) {
batch(() => {
if (!isModelValid(model)) return
const a = agent.current()
if (!a) return
setModelStore("model", a.id, model)
if (!options?.recent) return
setModelStore("recent", recentModels(model, modelStore.recent))
save()
})
}
function selectVariant(value: string | undefined) {
const model = currentModel()
if (!model) return
const key = modelPreferenceKey(model)
const variant = normalizeModelVariant(value)
if (modelStore.variant[key] === variant) return
setModelStore("variant", key, variant)
save()
}
function matches(model?: { providerID: string; id: string }) {
if (!modelStore.ready) return false
const current = currentModel()
if (!current) return false
if (!model) return true
return current.providerID === model.providerID && current.modelID === model.id
}
function hydrate(model?: { providerID: string; id: string; variant?: string }) {
if (!modelStore.ready) return false
if (!model) return true
if (data.location.model.list() === undefined) return false
const selected = { providerID: model.providerID, modelID: model.id }
if (!isModelValid(selected)) return false
select(selected)
selectVariant(model.variant)
return matches(model)
}
return {
current: currentModel,
get ready() {
@@ -260,8 +221,6 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
favorite() {
return modelStore.favorite
},
hydrate,
matches,
parsed: createMemo(() => {
const value = currentModel()
if (!value) {
@@ -326,7 +285,18 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
setModelStore("recent", recentModels(next, modelStore.recent))
save()
},
set: select,
set(model: { providerID: string; modelID: string }, options?: { recent?: boolean }) {
batch(() => {
if (!isModelValid(model)) return
const a = agent.current()
if (!a) return
setModelStore("model", a.id, model)
if (options?.recent) {
setModelStore("recent", recentModels(model, modelStore.recent))
save()
}
})
},
toggleFavorite(model: { providerID: string; modelID: string }) {
batch(() => {
if (!isModelValid(model)) return
@@ -363,7 +333,10 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
return info?.variants?.map((variant) => variant.id) ?? []
},
set(value: string | undefined) {
selectVariant(value)
const m = currentModel()
if (!m) return
setModelStore("variant", modelPreferenceKey(m), normalizeModelVariant(value))
save()
},
cycle() {
const variants = this.list()
+1 -1
View File
@@ -360,7 +360,7 @@ export function Session() {
createEffect(() => {
const current = prompt()
if (sent || !current || !synced() || !local.model.ready) return
if (!local.agent.current() || !local.model.matches(session()?.model)) return
if (!local.agent.current() || !local.model.current()) return
if (!args.prompt || route.prompt?.text !== args.prompt || current.current.text !== args.prompt) return
sent = true
current.submit()
+72 -103
View File
@@ -226,110 +226,79 @@ test("session title generated while an untitled session is loading remains visib
}
})
for (const scenario of [
{ name: "session startup prompt is submitted exactly once", delayed: false },
{ name: "session model hydration retries after the model catalog loads", delayed: true },
]) {
test(scenario.name, async () => {
const setup = await createTestRenderer({ width: 80, height: 24, useThread: false })
const core = await import("@opentui/core")
mock.module("@opentui/core", () => ({ ...core, createCliRenderer: async () => setup.renderer }))
const events = createEventStream()
const cwd = process.cwd()
const location = { directory: cwd, project: { id: "project", directory: cwd } }
const selected = scenario.delayed ? "selected" : "model"
const session = {
id: "dummy",
title: "Demo session",
projectID: "project",
location: { directory: cwd },
agent: "build",
model: { providerID: "provider", id: selected },
cost: 0,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
time: { created: 0, updated: 0 },
test("session startup prompt is submitted exactly once", async () => {
const setup = await createTestRenderer({ width: 80, height: 24, useThread: false })
const core = await import("@opentui/core")
mock.module("@opentui/core", () => ({ ...core, createCliRenderer: async () => setup.renderer }))
const events = createEventStream()
const cwd = process.cwd()
const location = { directory: cwd, project: { id: "project", directory: cwd } }
const session = {
id: "dummy",
title: "Demo session",
projectID: "project",
location: { directory: cwd },
agent: "build",
model: { providerID: "provider", id: "model" },
cost: 0,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
time: { created: 0, updated: 0 },
}
const bodies: unknown[] = []
const promptSubmitted = Promise.withResolvers<void>()
const calls = createFetch(async (url, request) => {
if (url.pathname === "/api/location") return json(location)
if (url.pathname === "/api/session") return json({ data: [session], cursor: {} })
if (url.pathname === "/api/session/dummy") return json({ data: session })
if (url.pathname === "/api/session/dummy/message") return json({ data: [], cursor: {} })
if (url.pathname === "/api/session/dummy/pending") return json({ data: [] })
if (url.pathname === "/api/session/dummy/permission") return json({ data: [] })
if (url.pathname === "/api/agent")
return json({
location,
data: [{ id: "build", mode: "primary", hidden: false, permissions: [] }],
})
if (url.pathname === "/api/model")
return json({
location,
data: [{ id: "model", providerID: "provider", name: "Model", variants: [] }],
})
if (url.pathname === "/api/session/dummy/prompt") {
bodies.push(await request.json())
promptSubmitted.resolve()
return json({ data: {} })
}
const sessionRequested = Promise.withResolvers<void>()
const modelRequested = Promise.withResolvers<void>()
const releaseModels = Promise.withResolvers<void>()
const promptSubmitted = Promise.withResolvers<void>()
const bodies: unknown[] = []
const selections: unknown[] = []
const calls = createFetch(async (url, request) => {
if (url.pathname === "/api/location") return json(location)
if (url.pathname === "/api/session") return json({ data: [session], cursor: {} })
if (url.pathname === "/api/session/dummy") {
sessionRequested.resolve()
return json({ data: session })
}
if (url.pathname === "/api/session/dummy/message") return json({ data: [], cursor: {} })
if (url.pathname === "/api/session/dummy/pending") return json({ data: [] })
if (url.pathname === "/api/session/dummy/permission") return json({ data: [] })
if (url.pathname === "/api/agent")
return json({
location,
data: [{ id: "build", mode: "primary", hidden: false, permissions: [] }],
})
if (url.pathname === "/api/model") {
modelRequested.resolve()
if (scenario.delayed) await releaseModels.promise
return json({
location,
data: [
...(scenario.delayed ? [{ id: "fallback", providerID: "provider", name: "Fallback", variants: [] }] : []),
{ id: selected, providerID: "provider", name: "Selected", variants: [] },
],
})
}
if (url.pathname === "/api/session/dummy/model") {
selections.push(await request.json())
return new Response(null, { status: 204 })
}
if (url.pathname === "/api/session/dummy/prompt") {
bodies.push(await request.json())
promptSubmitted.resolve()
return json({ data: {} })
}
}, events)
const server = Bun.serve({ port: 0, fetch: (request) => calls.fetch(request) })
}, events)
const server = Bun.serve({ port: 0, fetch: (request) => calls.fetch(request) })
try {
const { run } = await import("../src/app")
const task = Effect.runPromise(
run({
app: { name: "test", version: "test", channel: "test" },
server: { endpoint: { url: server.url.toString() } },
config: { get: async () => ({}), update: async () => ({}) },
packages: { resolve: async () => undefined },
args: { sessionID: "dummy", prompt: "RESUME_READY" },
log: () => {},
}).pipe(Effect.provide(AppNodeBuilder.build(Global.node)), Effect.provide(FileSystem.layerNoop({}))),
)
try {
const { run } = await import("../src/app")
const task = Effect.runPromise(
run({
app: { name: "test", version: "test", channel: "test" },
server: { endpoint: { url: server.url.toString() } },
config: { get: async () => ({}), update: async () => ({}) },
packages: { resolve: async () => undefined },
args: { sessionID: "dummy", prompt: "RESUME_READY" },
log: () => {},
}).pipe(Effect.provide(AppNodeBuilder.build(Global.node)), Effect.provide(FileSystem.layerNoop({}))),
)
if (scenario.delayed) {
await Promise.all([sessionRequested.promise, modelRequested.promise])
await Bun.sleep(20)
expect(bodies).toEqual([])
releaseModels.resolve()
}
await Promise.race([
promptSubmitted.promise,
Bun.sleep(2000).then(() => {
throw new Error("startup prompt was not submitted")
}),
])
await Bun.sleep(20)
setup.renderer.destroy()
await task
await Promise.race([
promptSubmitted.promise,
Bun.sleep(2000).then(() => {
throw new Error("startup prompt was not submitted")
}),
])
await Bun.sleep(20)
setup.renderer.destroy()
await task
expect(bodies).toHaveLength(1)
expect(bodies[0]).toMatchObject({ text: "RESUME_READY" })
expect(selections).toEqual([])
} finally {
releaseModels.resolve()
if (!setup.renderer.isDestroyed) setup.renderer.destroy()
await server.stop()
mock.restore()
}
})
}
expect(bodies).toHaveLength(1)
expect(bodies[0]).toMatchObject({ text: "RESUME_READY" })
} finally {
if (!setup.renderer.isDestroyed) setup.renderer.destroy()
await server.stop()
mock.restore()
}
})