From 2b8a3e7ee079115709ab9d4fcf94e38692acfc77 Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 27 Aug 2026 11:52:48 -0400 Subject: [PATCH] fix(pty): own daemon lifetime and hand off restarts (#45551) --- bun.lock | 18 +- packages/cli/package.json | 2 +- packages/cli/script/opencode-pty.ts | 9 +- packages/cli/src/server-process.ts | 12 +- packages/client/src/effect/api/api.ts | 14 + .../client/src/effect/generated/client.ts | 7 + packages/client/src/effect/service.ts | 21 +- .../client/src/promise/generated/client.ts | 12 + .../client/src/promise/generated/types.ts | 4 + packages/client/src/promise/service.ts | 22 +- packages/client/src/pty-handoff.ts | 139 +++++ packages/client/src/service-contender.ts | 2 +- packages/client/test/fixture/service.ts | 22 +- packages/client/test/promise-service.test.ts | 18 + packages/core/package.json | 2 +- packages/core/src/persistent-pty/daemon.ts | 195 +++++-- packages/core/src/persistent-pty/index.ts | 447 ++++++++-------- .../core/test/persistent-pty-daemon.test.ts | 308 ++++++------ .../protocol/src/groups/persistent-pty.ts | 6 + packages/schema/src/persistent-pty.ts | 8 + .../server/src/handlers/persistent-pty.ts | 6 + packages/server/src/options.ts | 2 + packages/server/src/routes.ts | 1 + packages/server/test/persistent-pty.test.ts | 475 ++++++++++-------- 24 files changed, 1107 insertions(+), 645 deletions(-) create mode 100644 packages/client/src/pty-handoff.ts diff --git a/bun.lock b/bun.lock index 44d2ed1b4fe..491f9331890 100644 --- a/bun.lock +++ b/bun.lock @@ -125,7 +125,7 @@ "@effect/platform-node": "catalog:", "@opencode-ai/client": "workspace:*", "@opencode-ai/plugin": "workspace:*", - "@opencode-ai/pty": "0.1.12", + "@opencode-ai/pty": "0.1.13", "@opencode-ai/schema": "workspace:*", "@opencode-ai/server": "workspace:*", "@opencode-ai/tui": "workspace:*", @@ -364,7 +364,7 @@ "@opencode-ai/ai": "workspace:*", "@opencode-ai/codemode": "workspace:*", "@opencode-ai/plugin": "workspace:*", - "@opencode-ai/pty": "0.1.12", + "@opencode-ai/pty": "0.1.13", "@opencode-ai/schema": "workspace:*", "@opencode-ai/util": "workspace:*", "@parcel/watcher": "2.5.1", @@ -2175,19 +2175,19 @@ "@opencode-ai/protocol": ["@opencode-ai/protocol@workspace:packages/protocol"], - "@opencode-ai/pty": ["@opencode-ai/pty@0.1.12", "", { "optionalDependencies": { "@opencode-ai/pty-darwin-arm64": "0.1.12", "@opencode-ai/pty-darwin-x64": "0.1.12", "@opencode-ai/pty-linux-arm64-gnu": "0.1.12", "@opencode-ai/pty-linux-arm64-musl": "0.1.12", "@opencode-ai/pty-linux-x64-gnu": "0.1.12", "@opencode-ai/pty-linux-x64-musl": "0.1.12" }, "bin": { "opencode-pty": "bin/opencode-pty.js" } }, "sha512-dl4FyJUhTXThsWYY8txG/8/nwN7dE0M5Sic9r4L9f2pvtJnbR5zrCrPoiPIBIxZle1wVks1dhz4z/CfqLf5sCg=="], + "@opencode-ai/pty": ["@opencode-ai/pty@0.1.13", "", { "optionalDependencies": { "@opencode-ai/pty-darwin-arm64": "0.1.13", "@opencode-ai/pty-darwin-x64": "0.1.13", "@opencode-ai/pty-linux-arm64-gnu": "0.1.13", "@opencode-ai/pty-linux-arm64-musl": "0.1.13", "@opencode-ai/pty-linux-x64-gnu": "0.1.13", "@opencode-ai/pty-linux-x64-musl": "0.1.13" }, "bin": { "opencode-pty": "bin/opencode-pty.js" } }, "sha512-WPCN8h8HaZhhUcrMG0zu+4D9vco0EZiEg/gCF1K3JPRN6UsHMiXq1HVIy5IlyfcoyjfViRmQmXYE4AuU3laBjA=="], - "@opencode-ai/pty-darwin-arm64": ["@opencode-ai/pty-darwin-arm64@0.1.12", "", { "os": "darwin", "cpu": "arm64" }, "sha512-tMvoriq3VegVlj1uEglc6qE0M7VXy61nyf9Si7tTO7xa8JiyxuFJSXOZ1pGeErDu+pe24hvTyVOR+gkdew8w9g=="], + "@opencode-ai/pty-darwin-arm64": ["@opencode-ai/pty-darwin-arm64@0.1.13", "", { "os": "darwin", "cpu": "arm64" }, "sha512-fVtQZqVLBuJx/aB+5ojfmQifS1KMc9gxlxpFQ6bxEFU8tn8xHQTiFPaNroZgOtaw7I4ceGyx/eXieK1wp68yAA=="], - "@opencode-ai/pty-darwin-x64": ["@opencode-ai/pty-darwin-x64@0.1.12", "", { "os": "darwin", "cpu": "x64" }, "sha512-Sn5vMLL5giHOhx7J5H6zwDp4YjjXorY+QV0IEYY+SCT4wQfRBliokIyj23pRl6P2RK3u9bDLXJHDNMfDVZ2Rxg=="], + "@opencode-ai/pty-darwin-x64": ["@opencode-ai/pty-darwin-x64@0.1.13", "", { "os": "darwin", "cpu": "x64" }, "sha512-b/tAEm0hCMXraPM9cxR8Rg7X1UBZInRTaxWAS4Ht9eH1nWj1rANOLvHWiWX/vVh5TB0Ubg8bWPu4B0nZkEHROQ=="], - "@opencode-ai/pty-linux-arm64-gnu": ["@opencode-ai/pty-linux-arm64-gnu@0.1.12", "", { "os": "linux", "cpu": "arm64" }, "sha512-HbnlKZy052l7G527wK0+05EXaUpZ4ykVAmNBEzqWCoi4TeQj2+Nr9kJ9trx9o1KrVcT4Ki58CCvN5QOls6Z0yQ=="], + "@opencode-ai/pty-linux-arm64-gnu": ["@opencode-ai/pty-linux-arm64-gnu@0.1.13", "", { "os": "linux", "cpu": "arm64" }, "sha512-I124aSYBBjpGZnYExHfIajkvVK1FiK+//OJBGdqqFp5pas2Oruq4O8tv+pMoxomZIYh2ce/QhOOYLHRwXsthTg=="], - "@opencode-ai/pty-linux-arm64-musl": ["@opencode-ai/pty-linux-arm64-musl@0.1.12", "", { "os": "linux", "cpu": "arm64" }, "sha512-2nTN7ggu1h9XgjNcoQMYjP5sirfYnAskpdFCOqjokLqhytX/IMMmkRTQs+foaEaPz0dAIQD3DQplR2jZIgxp1w=="], + "@opencode-ai/pty-linux-arm64-musl": ["@opencode-ai/pty-linux-arm64-musl@0.1.13", "", { "os": "linux", "cpu": "arm64" }, "sha512-feWsfKpaDytGJzutoK43GqQwVghG2vHZt6BE/ydPZNuqIrySQ/6JfliUAMwn5BWs/Ky7ouSwKHCyAVeukusSvg=="], - "@opencode-ai/pty-linux-x64-gnu": ["@opencode-ai/pty-linux-x64-gnu@0.1.12", "", { "os": "linux", "cpu": "x64" }, "sha512-FnD5ndnObTQKAoaVvxLKi5W+r3/+dsaMsobz6uK0B9hlmffXxY5CQ6HQyUU/h3aIKLWXxho5XYkA2b9yrp8/gA=="], + "@opencode-ai/pty-linux-x64-gnu": ["@opencode-ai/pty-linux-x64-gnu@0.1.13", "", { "os": "linux", "cpu": "x64" }, "sha512-jliNgsevGuxfIeX7eyzjHhrJkF8uEUPnDLbF2v16uv69FhEHrraf7jyWkxazMP6rNvn2CGtwMMc4BXPS5pzjhg=="], - "@opencode-ai/pty-linux-x64-musl": ["@opencode-ai/pty-linux-x64-musl@0.1.12", "", { "os": "linux", "cpu": "x64" }, "sha512-prkrNu6uvjqoffxdGiDHSU5C0Y+kCSfv+lslu7dfRPgPKenVELNpRTAbOduyrWPac2vGt8j5NM61icJyodbJmA=="], + "@opencode-ai/pty-linux-x64-musl": ["@opencode-ai/pty-linux-x64-musl@0.1.13", "", { "os": "linux", "cpu": "x64" }, "sha512-rXDpidW66gz2b2M/NbUN8ZKmAxaJcASnuHATeXevlrFdiPUv8uJwvkRd6Pla1fp01Q65MkBmgRa7Q9c+H1PlzA=="], "@opencode-ai/schema": ["@opencode-ai/schema@workspace:packages/schema"], diff --git a/packages/cli/package.json b/packages/cli/package.json index 29e66c97763..deac1fa518d 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -27,7 +27,7 @@ "@effect/platform-node": "catalog:", "@opencode-ai/client": "workspace:*", "@opencode-ai/plugin": "workspace:*", - "@opencode-ai/pty": "0.1.12", + "@opencode-ai/pty": "0.1.13", "@opencode-ai/schema": "workspace:*", "@opencode-ai/server": "workspace:*", "@opencode-ai/tui": "workspace:*", diff --git a/packages/cli/script/opencode-pty.ts b/packages/cli/script/opencode-pty.ts index feda2e798e9..ae3ec306057 100644 --- a/packages/cli/script/opencode-pty.ts +++ b/packages/cli/script/opencode-pty.ts @@ -27,8 +27,13 @@ export async function resolveOpencodePty(target: Target): Promise (value === "glibc" ? "gnu" : value)) .join("-") const name = `@opencode-ai/pty-${suffix}` - const source = pty.resolve(`${name}/bin/opencode-pty`) - const manifest: unknown = JSON.parse(await readFile(pty.resolve(`${name}/package.json`), "utf8")) + const local = process.env.OPENCODE_PTY_BIN + if (local && (target.platform !== process.platform || target.arch !== process.arch)) + throw new Error("OPENCODE_PTY_BIN can only be embedded in a build for the current platform and architecture") + const source = local ? path.resolve(local) : pty.resolve(`${name}/bin/opencode-pty`) + const manifest: unknown = local + ? { version: "local" } + : JSON.parse(await readFile(pty.resolve(`${name}/package.json`), "utf8")) if (!manifest || typeof manifest !== "object" || !("version" in manifest) || typeof manifest.version !== "string") throw new Error(`Invalid package metadata for ${name}`) diff --git a/packages/cli/src/server-process.ts b/packages/cli/src/server-process.ts index 66a37af07c1..8a1dcda2f6b 100644 --- a/packages/cli/src/server-process.ts +++ b/packages/cli/src/server-process.ts @@ -7,7 +7,8 @@ import { Global } from "@opencode-ai/util/global" import { OPENCODE_CHANNEL, OPENCODE_VERSION } from "./version" import { AppProcess } from "@opencode-ai/util/process" import { randomBytes, randomUUID } from "node:crypto" -import { Effect, Option, Redacted, Schedule } from "effect" +import { Effect, Option, Redacted, Schedule, Schema } from "effect" +import { PersistentPty } from "@opencode-ai/schema/persistent-pty" import { HttpServer } from "effect/unstable/http" import { Env } from "./env" import { ServiceConfig } from "./services/service-config" @@ -40,6 +41,14 @@ export const run = Effect.fnUntraced(function* (options: Options) { }) const processEffect = Effect.fnUntraced(function* (options: Options) { + const inherited = process.env.OPENCODE_PTY_HANDOFF + delete process.env.OPENCODE_PTY_HANDOFF + const handoff = + inherited === undefined + ? undefined + : yield* Schema.decodeUnknownEffect(Schema.fromJsonString(PersistentPty.Handoff))(inherited).pipe( + Effect.mapError(() => new Error("Invalid PTY restart handoff")), + ) const global = yield* Global.Service if (options.mode === "service") yield* Effect.sync(() => process.chdir(global.home)) return yield* Effect.scoped( @@ -80,6 +89,7 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { hostname, port, password, + pty: { handoff }, simulation: truthy(process.env.OPENCODE_SIMULATE), database: { path: diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 5a6b3f681fc..2b42f6f6f35 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -1689,6 +1689,19 @@ export type ExperimentalPersistentPtyShutdownOperation = () => Effect E > +export type ExperimentalPersistentPtyHandoffOutput = { + readonly handoff: { + readonly directory: string + readonly instanceID: string + readonly ticket: string + readonly expiresAt: number + } | null +} +export type ExperimentalPersistentPtyHandoffOperation = () => Effect.Effect< + ExperimentalPersistentPtyHandoffOutput, + E +> + export type ExperimentalPersistentPtyGetInput = { readonly ptyID: Pty.ID } export type ExperimentalPersistentPtyGetOutput = { readonly id: Pty.ID @@ -1775,6 +1788,7 @@ export interface ExperimentalApi { readonly list: ExperimentalPersistentPtyListOperation readonly create: ExperimentalPersistentPtyCreateOperation readonly shutdown: ExperimentalPersistentPtyShutdownOperation + readonly handoff: ExperimentalPersistentPtyHandoffOperation readonly get: ExperimentalPersistentPtyGetOperation readonly update: ExperimentalPersistentPtyUpdateOperation readonly snapshot: ExperimentalPersistentPtySnapshotOperation diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index 697d771e6f4..92aae3154d4 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -203,6 +203,7 @@ import type { ExperimentalPersistentPtyCreateInput, ExperimentalPersistentPtyCreateOutput, ExperimentalPersistentPtyShutdownOutput, + ExperimentalPersistentPtyHandoffOutput, ExperimentalPersistentPtyGetInput, ExperimentalPersistentPtyGetOutput, ExperimentalPersistentPtyUpdateInput, @@ -1265,6 +1266,11 @@ const EndpointExperimentalPersistentPtyShutdown = (raw: RawClient["server.experi raw["persistentPty.shutdown"]({}).pipe(Effect.mapError(mapClientError)), ) +const EndpointExperimentalPersistentPtyHandoff = (raw: RawClient["server.experimental"]) => () => + preserveEffect()( + raw["persistentPty.handoff"]({}).pipe(Effect.mapError(mapClientError)), + ) + const EndpointExperimentalPersistentPtyGet = (raw: RawClient["server.experimental"]) => (input: ExperimentalPersistentPtyGetInput) => preserveEffect()( @@ -1318,6 +1324,7 @@ const adaptGroupExperimental = (raw: RawClient["server.experimental"]) => ({ list: EndpointExperimentalPersistentPtyList(raw), create: EndpointExperimentalPersistentPtyCreate(raw), shutdown: EndpointExperimentalPersistentPtyShutdown(raw), + handoff: EndpointExperimentalPersistentPtyHandoff(raw), get: EndpointExperimentalPersistentPtyGet(raw), update: EndpointExperimentalPersistentPtyUpdate(raw), snapshot: EndpointExperimentalPersistentPtySnapshot(raw), diff --git a/packages/client/src/effect/service.ts b/packages/client/src/effect/service.ts index 33508c87b03..0a72124a6ed 100644 --- a/packages/client/src/effect/service.ts +++ b/packages/client/src/effect/service.ts @@ -11,6 +11,7 @@ import { } from "../service-contender.js" import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js" import { matchesVersion } from "../service-version.js" +import { PtyHandoff } from "../pty-handoff.js" export * from "../service.js" /** Contents of the local service registration file. */ @@ -65,9 +66,10 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti 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")) + const env = yield* Effect.tryPromise(() => PtyHandoff.environment(options.file ?? fallback(), options.env)) return yield* Effect.try({ try: () => { - return spawnServiceContender(command, args, options.env) + return spawnServiceContender(command, args, env) }, catch: (cause) => new Error("Failed to start server", { cause }), }) @@ -83,6 +85,8 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti } if (timeouts.count >= 3) { yield* announce("missing") + yield* Effect.logWarning("Background service is unresponsive; recovery cannot preserve persistent terminals") + yield* Effect.tryPromise(() => PtyHandoff.clear(options.file ?? fallback())) yield* terminate(info, options, timing) timeouts = undefined lastSpawn = Date.now() - spawnDelay @@ -91,11 +95,23 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti if (service !== undefined) { spawnDelay = timing.spawnDelay const compatible = !service.legacy && matchesVersion(service.version, options) - if (compatible && service.state === "ready") return Option.some(service) + if (compatible && service.state === "ready") { + yield* Effect.tryPromise(() => PtyHandoff.complete(options.file ?? fallback(), service.info)) + 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() yield* announce("version-mismatch", service.version) + if (!service.legacy && service.state === "ready") + yield* Effect.tryPromise(() => + PtyHandoff.prepare(options.file ?? fallback(), service.info, timing.requestTimeout), + ) + else { + if (!service.legacy) + yield* Effect.logWarning("Background service is not ready; replacement cannot preserve persistent terminals") + yield* Effect.tryPromise(() => PtyHandoff.clear(options.file ?? fallback())) + } yield* terminate(service.info, options, timing).pipe(Effect.ignore) lastSpawn = 0 return Option.none() @@ -129,6 +145,7 @@ 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 = {}) { + yield* Effect.tryPromise(() => PtyHandoff.clear(options.file ?? fallback())) const info = yield* read(options.file) if (info !== undefined) yield* terminate(info, options, defaultEnsureTiming) }) diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 3cea8f8e48b..35753a96bef 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -199,6 +199,7 @@ import type { ExperimentalPersistentPtyCreateInput, ExperimentalPersistentPtyCreateOutput, ExperimentalPersistentPtyShutdownOutput, + ExperimentalPersistentPtyHandoffOutput, ExperimentalPersistentPtyGetInput, ExperimentalPersistentPtyGetOutput, ExperimentalPersistentPtyUpdateInput, @@ -1723,6 +1724,17 @@ export function make(options: ClientOptions) { }, requestOptions, ), + handoff: (requestOptions?: RequestOptions) => + request( + { + method: "POST", + path: `/api/experimental/persistent-pty/handoff`, + successStatus: 200, + declaredStatuses: [503, 401, 400], + empty: false, + }, + requestOptions, + ), get: (input: ExperimentalPersistentPtyGetInput, requestOptions?: RequestOptions) => request<{ readonly data: ExperimentalPersistentPtyGetOutput }>( { diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 0af1a23ca47..4c8b77379d2 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -376,6 +376,8 @@ export type SessionStatus = export type PtyTicketConnectToken = { ticket: string; expires_in: number } +export type PersistentPtyHandoff = { directory: string; instanceID: string; ticket: string; expiresAt: number } + export type ShellInfo1 = { id: string status: "running" | "exited" | "timeout" | "killed" @@ -5807,6 +5809,8 @@ export type ExperimentalPersistentPtyCreateOutput = { data: PersistentPtyInfo }[ export type ExperimentalPersistentPtyShutdownOutput = void +export type ExperimentalPersistentPtyHandoffOutput = { handoff: PersistentPtyHandoff | null } + export type ExperimentalPersistentPtyGetInput = { readonly ptyID: { readonly ptyID: string }["ptyID"] } export type ExperimentalPersistentPtyGetOutput = { data: PersistentPtyInfo }["data"] diff --git a/packages/client/src/promise/service.ts b/packages/client/src/promise/service.ts index 1441cdd88ba..680b08ca827 100644 --- a/packages/client/src/promise/service.ts +++ b/packages/client/src/promise/service.ts @@ -10,6 +10,7 @@ import { } from "../service-contender.js" import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js" import { matchesVersion } from "../service-version.js" +import { PtyHandoff } from "../pty-handoff.js" import type { ServiceHealth } from "./generated/types.js" export * from "../service.js" @@ -43,11 +44,11 @@ export async function ensure(options: EnsureOptions = {}): Promise { announced = true options.onStart?.(reason, previousVersion) } - const spawnContender = () => { + const spawnContender = async () => { const [command, ...args] = options.command ?? ["opencode", "serve", "--service"] if (command === undefined) throw new Error("Missing service command") try { - return spawnServiceContender(command, args, options.env) + return spawnServiceContender(command, args, await PtyHandoff.environment(options.file ?? fallback(), options.env)) } catch (cause) { throw new Error("Failed to start server", { cause }) } @@ -64,6 +65,8 @@ export async function ensure(options: EnsureOptions = {}): Promise { } if (timeouts.count >= 3) { announce("missing") + console.warn("Background service is unresponsive; recovery cannot preserve persistent terminals") + await PtyHandoff.clear(options.file ?? fallback()) await terminate(registration.info, options, timing) timeouts = undefined lastSpawn = Date.now() - spawnDelay @@ -74,10 +77,20 @@ export async function ensure(options: EnsureOptions = {}): Promise { spawnDelay = timing.spawnDelay const service = registration.service const compatible = !service.legacy && matchesVersion(service.version, options) - if (compatible && service.state === "ready") return service.endpoint + if (compatible && service.state === "ready") { + await PtyHandoff.complete(options.file ?? fallback(), service.info) + return service.endpoint + } if (compatible && service.state === "failed") throw new Error("Background service failed to start") if (!compatible) { announce("version-mismatch", service.version) + if (!service.legacy && service.state === "ready") + await PtyHandoff.prepare(options.file ?? fallback(), service.info, timing.requestTimeout) + else { + if (!service.legacy) + console.warn("Background service is not ready; replacement cannot preserve persistent terminals") + await PtyHandoff.clear(options.file ?? fallback()) + } await terminate(service.info, options, timing).catch(() => undefined) lastSpawn = 0 } @@ -93,7 +106,7 @@ export async function ensure(options: EnsureOptions = {}): Promise { // 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") - contenders.add(spawnContender()) + contenders.add(await spawnContender()) lastSpawn = Date.now() } } @@ -106,6 +119,7 @@ export async function ensure(options: EnsureOptions = {}): Promise { /** Stop the registered local service. */ export async function stop(options: StopOptions = {}) { + await PtyHandoff.clear(options.file ?? fallback()) const info = await read(options.file) if (info !== undefined) await terminate(info, options, defaultEnsureTiming) } diff --git a/packages/client/src/pty-handoff.ts b/packages/client/src/pty-handoff.ts new file mode 100644 index 00000000000..61e9568c449 --- /dev/null +++ b/packages/client/src/pty-handoff.ts @@ -0,0 +1,139 @@ +export * as PtyHandoff from "./pty-handoff.js" + +import type { PersistentPty } from "@opencode-ai/schema/persistent-pty" +import { readFile, rename, rm, writeFile } from "node:fs/promises" +import type { Info } from "./service.js" + +type Sidecar = { + readonly source: Pick + readonly handoff: PersistentPty.Handoff | null + readonly expiresAt: number +} + +/** Publish the ticket before stopping its owner so every replacement contender can adopt it. */ +export async function prepare(file: string, info: Info, timeout: number) { + const existing = await read(file) + if (existing !== undefined && existing.expiresAt > Date.now() && same(existing.source, info)) return + const { ClientError, OpenCode } = await import("./promise/index.js") + const client = OpenCode.make({ + baseUrl: info.url, + headers: + info.password === undefined + ? undefined + : { authorization: "Basic " + Buffer.from(`opencode:${info.password}`).toString("base64") }, + }) + const missing = (error: unknown) => + error instanceof ClientError && + error.reason === "UnexpectedStatus" && + typeof error.cause === "object" && + error.cause !== null && + "status" in error.cause && + error.cause.status === 404 + const result = await client.experimental.persistentPty.handoff({ signal: AbortSignal.timeout(timeout) }).then( + (value) => ({ value }), + (cause: unknown) => ({ cause }), + ) + if ("cause" in result) { + // Another caller may already have prepared and stopped this server. + const concurrent = await read(file) + if (concurrent !== undefined && concurrent.expiresAt > Date.now() && same(concurrent.source, info)) return + if (!missing(result.cause)) + throw new Error("Failed to prepare persistent terminals for service replacement", { cause: result.cause }) + console.warn("Background service cannot hand off persistent terminals; shutting them down before replacement") + await client.experimental.persistentPty + .shutdown({ signal: AbortSignal.timeout(timeout) }) + .catch((cause: unknown) => { + if (missing(cause)) return + throw new Error("Failed to shut down persistent terminals before service replacement", { cause }) + }) + await publish(file, info, null) + return + } + const body: unknown = result.value + if (typeof body !== "object" || body === null || !("handoff" in body)) + throw new Error("Invalid persistent terminal handoff response") + if (body.handoff === null) { + await publish(file, info, null) + return + } + if (!isHandoff(body.handoff) || body.handoff.expiresAt <= Date.now()) + throw new Error("Invalid or expired persistent terminal handoff") + await publish(file, info, body.handoff) +} + +async function publish(file: string, info: Info, handoff: PersistentPty.Handoff | null) { + const temporary = `${file}.pty-handoff.${crypto.randomUUID()}.tmp` + await writeFile( + temporary, + JSON.stringify({ + source: { id: info.id, pid: info.pid, url: info.url }, + handoff, + expiresAt: handoff?.expiresAt ?? Date.now() + 30_000, + } satisfies Sidecar), + { mode: 0o600, flag: "wx" }, + ) + await rename(temporary, file + ".pty-handoff").finally(() => rm(temporary, { force: true })) +} + +export async function environment(file: string, env?: Readonly>) { + const record = await read(file) + const current: Info | undefined = await readFile(file, "utf8") + .then((text) => JSON.parse(text)) + .catch(() => undefined) + const handoff = + record !== undefined && record.expiresAt > Date.now() && (current === undefined || same(record.source, current)) + ? record.handoff + : undefined + return { ...env, OPENCODE_PTY_HANDOFF: handoff == null ? undefined : JSON.stringify(handoff) } +} + +export async function complete(file: string, info: Info) { + const record = await read(file) + if (record !== undefined && !same(record.source, info)) await clear(file) +} + +export async function clear(file: string) { + await rm(file + ".pty-handoff", { force: true }) +} + +async function read(file: string): Promise { + const value: unknown = await readFile(file + ".pty-handoff", "utf8") + .then((text) => JSON.parse(text)) + .catch(() => undefined) + if (typeof value !== "object" || value === null || !("source" in value) || !("handoff" in value)) return + if (typeof value.source !== "object" || value.source === null) return + if (!("pid" in value.source) || typeof value.source.pid !== "number") return + if (!("url" in value.source) || typeof value.source.url !== "string") return + if ("id" in value.source && typeof value.source.id !== "string") return + if (value.handoff !== null && !isHandoff(value.handoff)) return + if (!("expiresAt" in value) || typeof value.expiresAt !== "number" || !Number.isFinite(value.expiresAt)) return + return { + source: { + id: "id" in value.source && typeof value.source.id === "string" ? value.source.id : undefined, + pid: value.source.pid, + url: value.source.url, + }, + handoff: value.handoff, + expiresAt: value.expiresAt, + } +} + +function same(left: Sidecar["source"], right: Info) { + return left.id === right.id && left.pid === right.pid && left.url === right.url +} + +function isHandoff(value: unknown): value is PersistentPty.Handoff { + return ( + typeof value === "object" && + value !== null && + "directory" in value && + typeof value.directory === "string" && + "instanceID" in value && + typeof value.instanceID === "string" && + "ticket" in value && + typeof value.ticket === "string" && + "expiresAt" in value && + typeof value.expiresAt === "number" && + Number.isFinite(value.expiresAt) + ) +} diff --git a/packages/client/src/service-contender.ts b/packages/client/src/service-contender.ts index 5d60bf4d600..b6c710992ce 100644 --- a/packages/client/src/service-contender.ts +++ b/packages/client/src/service-contender.ts @@ -13,7 +13,7 @@ const stderrLimit = 8 * 1024 export function spawnServiceContender( command: string, args: ReadonlyArray, - env?: Readonly>, + env?: Readonly>, ): ServiceContender { const child = spawn(command, args, { detached: true, diff --git a/packages/client/test/fixture/service.ts b/packages/client/test/fixture/service.ts index 3128f9f8a69..cec27b2d0d2 100644 --- a/packages/client/test/fixture/service.ts +++ b/packages/client/test/fixture/service.ts @@ -11,8 +11,10 @@ if (mode === "record-start") { await writeFile(registration + ".started", "") process.exit(1) } -if (mode === "environment") +if (mode === "environment") { await writeFile(registration + ".environment", process.env.OPENCODE_SERVICE_ENV_TEST ?? "") + await writeFile(registration + ".handoff", process.env.OPENCODE_PTY_HANDOFF ?? "null") +} if (mode === "signal") process.kill(process.pid, process.platform === "win32" ? "SIGTERM" : "SIGKILL") if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated" || mode === "coordinated-failed-loser") { @@ -30,14 +32,26 @@ if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated" || let requests = 0 let version = "test" -if (mode === "old") version = "old" +if (mode === "old" || mode === "handoff") 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() +const handoff = { + directory: registration + ".daemon", + instanceID: crypto.randomUUID(), + ticket: crypto.randomUUID(), + expiresAt: Date.now() + 30_000, +} const server = Bun.serve({ port: 0, async fetch(request) { const pathname = new URL(request.url).pathname + if (pathname === "/api/experimental/persistent-pty/handoff" && mode === "handoff") { + if (request.method !== "POST" || request.headers.get("authorization") !== "Basic " + btoa("opencode:private")) + return new Response(null, { status: 401 }) + await writeFile(registration + ".prepared", JSON.stringify(handoff)) + return Response.json({ handoff }) + } if (pathname !== "/api/health") return new Response(null, { status: 404 }) requests += 1 if (mode === "starting") await writeFile(registration + ".health-request", "") @@ -54,8 +68,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") - return Response.json({ healthy: true, version, pid: process.pid }) + if (mode === "starting" || mode === "graceful") return Response.json({ healthy: true, version, pid: process.pid }) return Response.json({ healthy: true, version, pid: process.pid }) }, }) @@ -67,6 +80,7 @@ await writeFile( version: mode === "legacy" ? undefined : version, url: server.url.toString(), pid: process.pid, + password: "private", }), { mode: 0o600 }, ) diff --git a/packages/client/test/promise-service.test.ts b/packages/client/test/promise-service.test.ts index 740dd1b1172..dca0aac6d27 100644 --- a/packages/client/test/promise-service.test.ts +++ b/packages/client/test/promise-service.test.ts @@ -67,6 +67,24 @@ test("adds configured environment variables with native promises", async () => { expect(await Bun.file(registration + ".environment").text()).toBe("configured") }) +test("passes the prepared handoff to the replacement server", async () => { + await using fixture = await serviceFixture() + const registration = fixture.registration + fixture.spawn("handoff") + await fixture.waitForFile() + await ensure({ + file: registration, + version: "test", + command: fixture.command("environment"), + env: { OPENCODE_PTY_HANDOFF: "must-not-inherit" }, + }) + const replacement = await Bun.file(registration).json() + fixture.track(replacement.pid) + + expect(await Bun.file(registration + ".handoff").json()).toEqual(await Bun.file(registration + ".prepared").json()) + expect(await Bun.file(registration + ".pty-handoff").exists()).toBe(false) +}) + test("waits for a live contender when another native contender fails", async () => { await using fixture = await serviceFixture() const registration = fixture.registration diff --git a/packages/core/package.json b/packages/core/package.json index 545ab0e210b..3e1f656c553 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -118,7 +118,7 @@ "@ff-labs/fff-node": "0.10.5", "@opencode-ai/codemode": "workspace:*", "@opencode-ai/ai": "workspace:*", - "@opencode-ai/pty": "0.1.12", + "@opencode-ai/pty": "0.1.13", "@opencode-ai/schema": "workspace:*", "@opencode-ai/plugin": "workspace:*", "@opencode-ai/util": "workspace:*", diff --git a/packages/core/src/persistent-pty/daemon.ts b/packages/core/src/persistent-pty/daemon.ts index 46d45155604..dc351fedc5c 100644 --- a/packages/core/src/persistent-pty/daemon.ts +++ b/packages/core/src/persistent-pty/daemon.ts @@ -3,8 +3,9 @@ import { readFile } from "node:fs/promises" import net from "node:net" import path from "node:path" import { Data, Duration, Effect, Schema, Semaphore } from "effect" +import type { Handoff } from "@opencode-ai/schema/persistent-pty" -const ProtocolVersion = 6 +const ProtocolVersion = 7 const MaxFrameBytes = 8 * 1024 * 1024 const Lifecycle = Schema.Union([ @@ -48,6 +49,8 @@ export const WireResponse = Schema.Union([ Schema.Struct({ type: Schema.Literal("created"), terminal: WireTerminal }), Schema.Struct({ type: Schema.Literal("terminals"), terminals: Schema.Array(WireTerminal) }), Schema.Struct({ type: Schema.Literal("ok") }), + Schema.Struct({ type: Schema.Literal("owned") }), + Schema.Struct({ type: Schema.Literal("handoff"), ticket: Schema.String, expires_at: Schema.Number }), Schema.Struct({ type: Schema.Literal("snapshot"), terminal: WireTerminal, @@ -131,6 +134,7 @@ export interface DaemonTransport { readonly request: (value: object, start?: boolean) => Effect.Effect readonly requestIfRunning: (value: object) => Effect.Effect readonly shutdown: Effect.Effect + readonly handoff: Effect.Effect readonly subscribe: ( id: number, input: { @@ -147,19 +151,51 @@ export interface DaemonTransport { export const makeDaemonTransport = Effect.fn("PersistentPty.makeDaemonTransport")(function* ( directory: string, binary: () => Promise = () => Promise.resolve(process.env.OPENCODE_PTY_BIN || "opencode-pty"), + inherited?: Handoff, ) { const startup = Semaphore.makeUnsafe(1) let registration: Registration | undefined + let owner: Awaited> | undefined + let closed = false + yield* Effect.addFinalizer(() => + Effect.sync(() => { + closed = true + owner?.socket.destroy() + }), + ) + + const claim = (current: Registration, ticket?: string) => + Effect.tryPromise({ + try: async (signal) => { + const connection = await openOwner(current, ticket, signal) + if (closed || signal.aborted) { + connection.socket.destroy() + throw new Error("PTY owner scope is closed") + } + owner = connection + registration = current + connection.socket.once("close", () => { + if (owner !== connection) return + owner = undefined + registration = undefined + }) + }, + catch: (cause) => failure("connect", cause), + }).pipe( + Effect.timeoutOrElse({ + duration: Duration.seconds(5), + orElse: () => Effect.fail(new DaemonError({ kind: "connect", message: "PTY ownership claim timed out" })), + }), + ) const discover = Effect.fn("PersistentPty.daemon.discover")(function* () { const value = yield* Effect.tryPromise({ try: () => readFile(path.join(directory, "service.json"), "utf8"), catch: (cause) => failure("connect", cause), }) - const decoded = yield* Effect.try({ - try: () => Schema.decodeUnknownSync(Registration)(JSON.parse(value)), - catch: (cause) => failure("protocol", cause), - }) + const decoded = yield* Schema.decodeUnknownEffect(Schema.fromJsonString(Registration))(value).pipe( + Effect.mapError((cause) => failure("protocol", cause)), + ) if (decoded.protocol !== ProtocolVersion) return yield* Effect.fail( new DaemonError({ @@ -181,9 +217,9 @@ export const makeDaemonTransport = Effect.fn("PersistentPty.makeDaemonTransport" const start = Effect.fn("PersistentPty.daemon.start")(function* () { const executable = yield* Effect.tryPromise({ try: binary, catch: (cause) => failure("spawn", cause) }) - yield* Effect.tryPromise({ + const child = yield* Effect.tryPromise({ try: () => - new Promise((resolve, reject) => { + new Promise>((resolve, reject) => { const child = spawn(executable, ["daemon"], { detached: true, stdio: "ignore", @@ -191,43 +227,54 @@ export const makeDaemonTransport = Effect.fn("PersistentPty.makeDaemonTransport" }) child.once("spawn", () => { child.unref() - resolve() + resolve(child) }) child.once("error", reject) }), catch: (cause) => failure("spawn", cause), }) - const deadline = Date.now() + 5_000 - let last: DaemonError | undefined - while (Date.now() < deadline) { - const found = yield* discover().pipe( - Effect.map((value) => ({ value })), - Effect.catch((error) => { - last = error - return Effect.succeed(undefined) - }), + return yield* Effect.gen(function* () { + const deadline = Date.now() + 5_000 + let last: DaemonError | undefined + while (Date.now() < deadline) { + const found = yield* discover().pipe( + Effect.map((value) => ({ value })), + Effect.catch((error) => { + last = error + return Effect.succeed(undefined) + }), + ) + if (found) { + yield* claim(found.value) + return found.value + } + yield* Effect.sleep(50) + } + return yield* Effect.fail( + last ?? new DaemonError({ kind: "connect", message: "opencode-pty did not become ready" }), ) - if (found) return found.value - yield* Effect.sleep(50) - } - return yield* Effect.fail( - last ?? new DaemonError({ kind: "connect", message: "opencode-pty did not become ready" }), + }).pipe( + Effect.onError(() => + Effect.sync(() => { + child.kill("SIGTERM") + }), + ), ) }) const connect = Effect.fn("PersistentPty.daemon.connect")(function* (shouldStart: boolean) { + if (closed) return yield* Effect.fail(new DaemonError({ kind: "connect", message: "PTY owner scope is closed" })) if (registration) return registration return yield* startup.withPermit( Effect.gen(function* () { if (registration) return registration const found = yield* discover().pipe( Effect.catch((error) => { - if (!shouldStart) return Effect.fail(error) - if (error.kind === "connect") return start() - if (error.kind !== "protocol" || error.pid === undefined) return Effect.fail(error) - return terminate(error.pid).pipe(Effect.andThen(start())) + if (shouldStart && error.kind === "connect") return start() + return Effect.fail(error) }), ) + if (!owner) yield* claim(found) registration = found return found }), @@ -277,6 +324,27 @@ export const makeDaemonTransport = Effect.fn("PersistentPty.makeDaemonTransport" return yield* Effect.fail(new DaemonError({ kind: "connect", message: "opencode-pty did not stop" })) }) + const handoff = startup.withPermit( + Effect.gen(function* () { + const current = owner + const registered = registration + if (!current || !registered) return null + const response = yield* Effect.tryPromise({ + try: (signal) => current.exchange({ op: "prepare_handoff" }, signal), + catch: (cause) => failure("response", cause), + }).pipe( + Effect.timeoutOrElse({ + duration: Duration.seconds(5), + orElse: () => + Effect.fail(new DaemonError({ kind: "response", message: "PTY handoff preparation timed out" })), + }), + ) + if (response.type !== "handoff") + return yield* Effect.fail(new DaemonError({ kind: "protocol", message: "Expected PTY handoff ticket" })) + return { directory, instanceID: registered.instance_id, ticket: response.ticket, expiresAt: response.expires_at } + }), + ) + const subscribe = Effect.fn("PersistentPty.daemon.subscribe")(function* ( id: number, input: Parameters[1], @@ -296,9 +364,56 @@ export const makeDaemonTransport = Effect.fn("PersistentPty.makeDaemonTransport" return yield* attempt.pipe(Effect.catch((error) => (error.kind === "registration" ? attempt : Effect.fail(error)))) }) - return { request, requestIfRunning, shutdown, subscribe } satisfies DaemonTransport + // A replacement must own its inherited daemon before the server becomes ready. + if (inherited) { + if (inherited.expiresAt <= Date.now()) + return yield* Effect.fail(new DaemonError({ kind: "registration", message: "PTY restart handoff expired" })) + const current = yield* discover() + if (current.instance_id !== inherited.instanceID) + return yield* Effect.fail(new DaemonError({ kind: "registration", message: "PTY restart daemon changed" })) + yield* claim(current, inherited.ticket) + } + + return { request, requestIfRunning, shutdown, handoff, subscribe } satisfies DaemonTransport }) +async function openOwner(registration: Registration, ticket: string | undefined, signal: AbortSignal) { + const socket = net.createConnection({ path: registration.socket }) + const frames = decoder(socket) + const abort = () => socket.destroy() + signal.addEventListener("abort", abort, { once: true }) + const exchange = async (request: object, signal: AbortSignal) => { + if (signal.aborted) throw new Error("PTY ownership request interrupted") + signal.addEventListener("abort", abort, { once: true }) + try { + socket.write(encode({ token: registration.token, request })) + const frame = await frames.next() + if (frame.done) throw new Error("PTY daemon closed its ownership connection") + const response = decode(frame.value) + if (response.type === "error") throw new Error(response.message) + return response + } finally { + signal.removeEventListener("abort", abort) + } + } + try { + await new Promise((resolve, reject) => { + socket.once("connect", resolve) + socket.once("error", reject) + socket.once("close", () => reject(new Error("PTY ownership connection closed"))) + }) + const response = await exchange({ op: "own", instance_id: registration.instance_id, ticket }, signal) + if (response.type !== "owned") throw new Error("PTY daemon does not support server ownership") + socket.unref() + return { socket, exchange } + } catch (error) { + socket.destroy() + throw error + } finally { + signal.removeEventListener("abort", abort) + } +} + const oneShot = Effect.fn("PersistentPty.daemon.oneShot")(function* (registration: Registration, request: object) { const payload = yield* Effect.try({ try: () => encode({ token: registration.token, request }), @@ -507,29 +622,3 @@ function decode(payload: Uint8Array) { function failure(kind: DaemonError["kind"], cause: unknown) { return new DaemonError({ kind, message: cause instanceof Error ? cause.message : String(cause) }) } - -const terminate = Effect.fn("PersistentPty.daemon.terminate-incompatible")(function* (pid: number) { - yield* Effect.logWarning("replacing incompatible opencode-pty daemon", { pid }) - yield* Effect.try({ try: () => process.kill(pid, "SIGTERM"), catch: (cause) => failure("spawn", cause) }).pipe( - Effect.catch((error) => (isMissingProcess(error) ? Effect.void : Effect.fail(error))), - ) - const deadline = Date.now() + 2_000 - while (Date.now() < deadline && processRunning(pid)) yield* Effect.sleep(25) - if (!processRunning(pid)) return - yield* Effect.try({ try: () => process.kill(pid, "SIGKILL"), catch: (cause) => failure("spawn", cause) }).pipe( - Effect.catch((error) => (isMissingProcess(error) ? Effect.void : Effect.fail(error))), - ) -}) - -function processRunning(pid: number) { - try { - process.kill(pid, 0) - return true - } catch { - return false - } -} - -function isMissingProcess(error: DaemonError) { - return error.message.includes("ESRCH") || error.message.includes("no such process") -} diff --git a/packages/core/src/persistent-pty/index.ts b/packages/core/src/persistent-pty/index.ts index 440e14b81ea..6e6e028f304 100644 --- a/packages/core/src/persistent-pty/index.ts +++ b/packages/core/src/persistent-pty/index.ts @@ -1,14 +1,12 @@ export * as PersistentPty from "./index.js" -import { createHash } from "node:crypto" import os from "node:os" import path from "node:path" import { Context, Effect, Layer, Schema } from "effect" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" -import { Added, Removed } from "@opencode-ai/schema/persistent-pty" +import { Added, Handoff, Removed } from "@opencode-ai/schema/persistent-pty" import { Session } from "@opencode-ai/schema/session" import { Bus } from "../bus.js" -import { Database } from "../database/database.js" import { Pty } from "@opencode-ai/schema/pty" import { Global } from "@opencode-ai/util/global" import { ShellSelect } from "../shell/select.js" @@ -23,6 +21,10 @@ import { import { resolveBinary } from "#persistent-pty-binary" export type { Role, StreamEvent } from "./daemon.js" +export { Handoff } from "@opencode-ai/schema/persistent-pty" + +export const Options = Schema.Struct({ handoff: Schema.optional(Handoff) }) +export type Options = typeof Options.Type export type Info = Pty.Info & { readonly sessionID: Session.ID @@ -103,6 +105,7 @@ export interface Interface { readonly snapshot: (id: Pty.ID) => Effect.Effect readonly remove: (id: Pty.ID) => Effect.Effect readonly shutdown: () => Effect.Effect + readonly handoff: () => Effect.Effect readonly attach: ( id: Pty.ID, input: { @@ -118,218 +121,235 @@ export interface Interface { export class Service extends Context.Service()("@opencode/PersistentPty") {} -export const layer = Layer.effect( - Service, - Effect.gen(function* () { - const bus = yield* Bus.Service - const database = yield* Database.Service - const global = yield* Global.Service - const context = yield* Effect.context() - const runFork = Effect.runForkWith(context) - let binary: Promise | undefined - const daemon = yield* makeDaemonTransport( - runtimeDirectory(databasePath(database.db)), - () => - (binary ??= resolveBinary(global.bin).catch((error) => { - binary = undefined - throw error - })), - ) - const removing = new Set() +export const configured = (options: Options = {}) => + Layer.effect( + Service, + Effect.gen(function* () { + const bus = yield* Bus.Service + const global = yield* Global.Service + const context = yield* Effect.context() + const runFork = Effect.runForkWith(context) + let binary: Promise | undefined + const daemon = yield* makeDaemonTransport( + options.handoff?.directory ?? runtimeDirectory(), + () => + (binary ??= resolveBinary(global.bin).catch((error) => { + binary = undefined + throw error + })), + options.handoff, + ).pipe(Effect.mapError(unavailable)) + const removing = new Set() - const list = Effect.fn("PersistentPty.list")(function* (sessionID?: Session.ID) { - const response = yield* optionalRequest(daemon, { op: "list" }) - if (!response) return [] - if (response.type !== "terminals") return yield* unexpected(response) - return response.terminals - .map(toInfo) - .filter((terminal) => sessionID === undefined || terminal.sessionID === sessionID) - }) + const list = Effect.fn("PersistentPty.list")(function* (sessionID?: Session.ID) { + const response = yield* optionalRequest(daemon, { op: "list" }) + if (!response) return [] + if (response.type !== "terminals") return yield* unexpected(response) + return response.terminals + .map(toInfo) + .filter((terminal) => sessionID === undefined || terminal.sessionID === sessionID) + }) - const get = Effect.fn("PersistentPty.get")(function* (id: Pty.ID) { - const found = (yield* list()).find((terminal) => terminal.id === id) - if (!found) return yield* new NotFoundError({ ptyID: id }) - return found - }) + const get = Effect.fn("PersistentPty.get")(function* (id: Pty.ID) { + const found = (yield* list()).find((terminal) => terminal.id === id) + if (!found) return yield* new NotFoundError({ ptyID: id }) + return found + }) - const create = Effect.fn("PersistentPty.create")(function* ( - sessionID: Session.ID, - input: { - readonly command?: string - readonly args: readonly string[] - readonly cwd?: string - readonly title: string - readonly env: Readonly> - readonly cols?: number - readonly rows?: number - }, - ) { - const response = yield* request( - daemon, - { - op: "create", - program: input.command ?? ShellSelect.environment(global.bin), - args: input.args, - cwd: input.cwd ?? path.resolve("/"), - title: input.title, - group_id: sessionID, - env: input.env, - cols: input.cols ?? 80, - rows: input.rows ?? 24, + const create = Effect.fn("PersistentPty.create")(function* ( + sessionID: Session.ID, + input: { + readonly command?: string + readonly args: readonly string[] + readonly cwd?: string + readonly title: string + readonly env: Readonly> + readonly cols?: number + readonly rows?: number }, - true, - ) - if (response.type !== "created") return yield* unexpected(response) - const terminal = toInfo(response.terminal) - yield* bus.publish(Added, { sessionID, terminal }) - return terminal - }) - - const write = Effect.fn("PersistentPty.write")(function* (id: Pty.ID, data: string, attachmentID?: string) { - yield* get(id) - const response = yield* request(daemon, { - op: "write", - id: fromID(id), - attachment_id: attachmentID ?? null, - data_base64: Buffer.from(data).toString("base64"), - }) - if (response.type !== "ok") return yield* unexpected(response) - return undefined - }) - - const resize = Effect.fn("PersistentPty.resize")(function* ( - id: Pty.ID, - cols: number, - rows: number, - attachmentID?: string, - ) { - yield* get(id) - const response = yield* request(daemon, { - op: "resize", - id: fromID(id), - attachment_id: attachmentID ?? null, - cols, - rows, - }) - if (response.type !== "ok") return yield* unexpected(response) - return undefined - }) - - const control = Effect.fn("PersistentPty.control")(function* ( - id: Pty.ID, - attachmentID: string, - cols: number, - rows: number, - ) { - yield* get(id) - const response = yield* request(daemon, { - op: "control", - id: fromID(id), - attachment_id: attachmentID, - cols, - rows, - }) - if (response.type !== "ok") return yield* unexpected(response) - return undefined - }) - - const input = Effect.fn("PersistentPty.input")(function* ( - id: Pty.ID, - attachmentID: string, - cols: number, - rows: number, - data: Uint8Array, - ) { - yield* get(id) - const response = yield* request(daemon, { - op: "input", - id: fromID(id), - attachment_id: attachmentID, - cols, - rows, - data_base64: Buffer.from(data).toString("base64"), - }) - if (response.type !== "ok") return yield* unexpected(response) - return undefined - }) - - const snapshot = Effect.fn("PersistentPty.snapshot")(function* (id: Pty.ID) { - yield* get(id) - const response = yield* request(daemon, { op: "snapshot", id: fromID(id) }) - if (response.type !== "snapshot") return yield* unexpected(response) - return { - info: toInfo(response.terminal), - text: response.text, - checkpoint: Buffer.from(response.checkpoint_base64, "base64"), - cursor: { x: response.cursor_x, y: response.cursor_y }, - } - }) - - const remove = Effect.fn("PersistentPty.remove")(function* (id: Pty.ID) { - const terminal = yield* get(id) - const response = yield* request(daemon, { op: "terminate", id: fromID(id) }) - if (response.type !== "ok") return yield* unexpected(response) - yield* bus.publish(Removed, { sessionID: terminal.sessionID, ptyID: id }) - return undefined - }) - - const shutdown = Effect.fn("PersistentPty.shutdown")(function* () { - const response = yield* daemon.shutdown.pipe(Effect.mapError(unavailable)) - if (!response) return - if (response.type !== "ok") return yield* unexpected(response) - }) - - const removeVisibleExit = (id: Pty.ID) => { - if (removing.has(id)) return - removing.add(id) - runFork( - remove(id).pipe( - Effect.catchTags({ - "PersistentPty.NotFoundError": () => Effect.void, - "PersistentPty.UnavailableError": (error) => - Effect.logWarning("failed to remove visible exited terminal", { id, error: error.message }), - }), - Effect.ensuring(Effect.sync(() => removing.delete(id))), - ), - ) - } - - const attach = Effect.fn("PersistentPty.attach")(function* ( - id: Pty.ID, - input: { - readonly cursor: number - readonly attachmentID: string - readonly role: Role - readonly takeover?: boolean - readonly onEvent: (event: StreamEvent) => void - readonly onEnd: () => void - }, - ) { - yield* get(id) - const attachment = yield* daemon - .subscribe(fromID(id), { - ...input, - onEvent: (event) => { - if (event.type === "exited") removeVisibleExit(id) - input.onEvent(event) + ) { + const response = yield* request( + daemon, + { + op: "create", + program: input.command ?? ShellSelect.environment(global.bin), + args: input.args, + cwd: input.cwd ?? path.resolve("/"), + title: input.title, + group_id: sessionID, + env: input.env, + cols: input.cols ?? 80, + rows: input.rows ?? 24, }, + true, + ) + if (response.type !== "created") return yield* unexpected(response) + const terminal = toInfo(response.terminal) + yield* bus.publish(Added, { sessionID, terminal }) + return terminal + }) + + const write = Effect.fn("PersistentPty.write")(function* (id: Pty.ID, data: string, attachmentID?: string) { + yield* get(id) + const response = yield* request(daemon, { + op: "write", + id: fromID(id), + attachment_id: attachmentID ?? null, + data_base64: Buffer.from(data).toString("base64"), }) - .pipe(Effect.mapError(unavailable)) - return { - info: toInfo(attachment.terminal), - role: attachment.role, - generation: attachment.generation, - replay: attachment.replay, - activate: attachment.activate, - detach: attachment.detach, + if (response.type !== "ok") return yield* unexpected(response) + return undefined + }) + + const resize = Effect.fn("PersistentPty.resize")(function* ( + id: Pty.ID, + cols: number, + rows: number, + attachmentID?: string, + ) { + yield* get(id) + const response = yield* request(daemon, { + op: "resize", + id: fromID(id), + attachment_id: attachmentID ?? null, + cols, + rows, + }) + if (response.type !== "ok") return yield* unexpected(response) + return undefined + }) + + const control = Effect.fn("PersistentPty.control")(function* ( + id: Pty.ID, + attachmentID: string, + cols: number, + rows: number, + ) { + yield* get(id) + const response = yield* request(daemon, { + op: "control", + id: fromID(id), + attachment_id: attachmentID, + cols, + rows, + }) + if (response.type !== "ok") return yield* unexpected(response) + return undefined + }) + + const input = Effect.fn("PersistentPty.input")(function* ( + id: Pty.ID, + attachmentID: string, + cols: number, + rows: number, + data: Uint8Array, + ) { + yield* get(id) + const response = yield* request(daemon, { + op: "input", + id: fromID(id), + attachment_id: attachmentID, + cols, + rows, + data_base64: Buffer.from(data).toString("base64"), + }) + if (response.type !== "ok") return yield* unexpected(response) + return undefined + }) + + const snapshot = Effect.fn("PersistentPty.snapshot")(function* (id: Pty.ID) { + yield* get(id) + const response = yield* request(daemon, { op: "snapshot", id: fromID(id) }) + if (response.type !== "snapshot") return yield* unexpected(response) + return { + info: toInfo(response.terminal), + text: response.text, + checkpoint: Buffer.from(response.checkpoint_base64, "base64"), + cursor: { x: response.cursor_x, y: response.cursor_y }, + } + }) + + const remove = Effect.fn("PersistentPty.remove")(function* (id: Pty.ID) { + const terminal = yield* get(id) + const response = yield* request(daemon, { op: "terminate", id: fromID(id) }) + if (response.type !== "ok") return yield* unexpected(response) + yield* bus.publish(Removed, { sessionID: terminal.sessionID, ptyID: id }) + return undefined + }) + + const shutdown = Effect.fn("PersistentPty.shutdown")(function* () { + const response = yield* daemon.shutdown.pipe(Effect.mapError(unavailable)) + if (!response) return + if (response.type !== "ok") return yield* unexpected(response) + }) + + const handoff = Effect.fn("PersistentPty.handoff")(() => daemon.handoff.pipe(Effect.mapError(unavailable))) + + const removeVisibleExit = (id: Pty.ID) => { + if (removing.has(id)) return + removing.add(id) + runFork( + remove(id).pipe( + Effect.catchTags({ + "PersistentPty.NotFoundError": () => Effect.void, + "PersistentPty.UnavailableError": (error) => + Effect.logWarning("failed to remove visible exited terminal", { id, error: error.message }), + }), + Effect.ensuring(Effect.sync(() => removing.delete(id))), + ), + ) } - }) - return Service.of({ list, get, create, write, resize, control, input, snapshot, remove, shutdown, attach }) - }), -) + const attach = Effect.fn("PersistentPty.attach")(function* ( + id: Pty.ID, + input: { + readonly cursor: number + readonly attachmentID: string + readonly role: Role + readonly takeover?: boolean + readonly onEvent: (event: StreamEvent) => void + readonly onEnd: () => void + }, + ) { + yield* get(id) + const attachment = yield* daemon + .subscribe(fromID(id), { + ...input, + onEvent: (event) => { + if (event.type === "exited") removeVisibleExit(id) + input.onEvent(event) + }, + }) + .pipe(Effect.mapError(unavailable)) + return { + info: toInfo(attachment.terminal), + role: attachment.role, + generation: attachment.generation, + replay: attachment.replay, + activate: attachment.activate, + detach: attachment.detach, + } + }) -export const node = makeGlobalNode({ service: Service, layer, deps: [Bus.node, Database.node, Global.node] }) + return Service.of({ + list, + get, + create, + write, + resize, + control, + input, + snapshot, + remove, + shutdown, + handoff, + attach, + }) + }), + ) + +export const layer = configured() +export const node = makeGlobalNode({ service: Service, layer, deps: [Bus.node, Global.node] }) const request = (daemon: DaemonTransport, value: object, start = false) => daemon.request(value, start).pipe(Effect.mapError(unavailable)) @@ -343,17 +363,7 @@ const unexpected = (response: WireResponse) => const unavailable = (error: unknown) => new UnavailableError({ message: error instanceof Error ? error.message : String(error) }) -function databasePath(db: Database.Interface["db"]) { - const client: unknown = db.$client - if ((typeof client !== "object" && typeof client !== "function") || client === null || !("config" in client)) - return undefined - const config = client.config - if (typeof config !== "object" || config === null || !("filename" in config)) return undefined - if (typeof config.filename !== "string" || config.filename === ":memory:") return undefined - return path.resolve(config.filename) -} - -const runtimeDirectory = (databasePath?: string) => { +const runtimeDirectory = () => { const root = process.env.OPENCODE_PTY_RUNTIME_DIR ?? (process.env.XDG_RUNTIME_DIR @@ -362,8 +372,7 @@ const runtimeDirectory = (databasePath?: string) => { os.tmpdir(), `opencode-pty-${typeof process.getuid === "function" ? process.getuid() : process.env.USER || "unknown"}`, )) - const identity = databasePath ?? `memory:${crypto.randomUUID()}` - return path.join(root, createHash("sha256").update(identity).digest("hex").slice(0, 16)) + return path.join(root, crypto.randomUUID()) } function toInfo(value: WireTerminal): Info { diff --git a/packages/core/test/persistent-pty-daemon.test.ts b/packages/core/test/persistent-pty-daemon.test.ts index 2280019fd55..3df36331dc8 100644 --- a/packages/core/test/persistent-pty-daemon.test.ts +++ b/packages/core/test/persistent-pty-daemon.test.ts @@ -1,4 +1,4 @@ -import { expect, test } from "bun:test" +import { expect } from "bun:test" import { spawn } from "node:child_process" import { mkdtemp, rm, writeFile } from "node:fs/promises" import net from "node:net" @@ -6,198 +6,208 @@ import os from "node:os" import path from "node:path" import { Effect } from "effect" import { makeDaemonTransport } from "../src/persistent-pty/daemon" +import { it } from "./lib/effect" -const pong = { type: "pong", instance_id: "test", pid: process.pid, protocol: 6 } +const pong = { type: "pong", instance_id: "test", pid: process.pid, protocol: 7 } -test("rediscovers a same-protocol daemon after its registration rotates", async () => { - const directory = await mkdtemp(path.join(os.tmpdir(), "opencode-pty-registration-")) - const socketPath = path.join(directory, "daemon.sock") - let token = "old-token" - let instance = "old-instance" - let creates = 0 - const server = await listen(socketPath, (_socket, request, receivedToken) => { - if (receivedToken !== token) return { type: "error", message: "authentication failed" } - if (request.op === "ping") return { ...pong, instance_id: instance } - if (request.op === "create") creates++ - return request.op === "list" ? { type: "terminals", terminals: [] } : { type: "ok" } - }) - try { - await writeRegistration(directory, socketPath, instance, token) - const daemon = await Effect.runPromise(makeDaemonTransport(directory)) - await Effect.runPromise(daemon.request({ op: "list" })) +it.live("rediscovers a same-protocol daemon after its registration rotates", () => + Effect.gen(function* () { + const directory = yield* temporaryDirectory() + const socketPath = path.join(directory, "daemon.sock") + let token = "old-token" + let instance = "old-instance" + let creates = 0 + yield* listen(socketPath, (_socket, request, receivedToken) => { + if (receivedToken !== token) return { type: "error", message: "authentication failed" } + if (request.op === "ping") return { ...pong, instance_id: instance } + if (request.op === "create") creates++ + return request.op === "list" ? { type: "terminals", terminals: [] } : { type: "ok" } + }) + yield* Effect.promise(() => writeRegistration(directory, socketPath, instance, token)) + const daemon = yield* makeDaemonTransport(directory) + yield* daemon.request({ op: "list" }) token = "new-token" instance = "new-instance" - await writeRegistration(directory, socketPath, instance, token) + yield* Effect.promise(() => writeRegistration(directory, socketPath, instance, token)) - const running = await Effect.runPromise(daemon.requestIfRunning({ op: "list" })) + const running = yield* daemon.requestIfRunning({ op: "list" }) expect(running).toEqual({ type: "terminals", terminals: [] }) token = "newest-token" instance = "newest-instance" - await writeRegistration(directory, socketPath, instance, token) - await Effect.runPromise(daemon.request({ op: "create" }, true)) + yield* Effect.promise(() => writeRegistration(directory, socketPath, instance, token)) + yield* daemon.request({ op: "create" }, true) expect(creates).toBe(1) - } finally { - await close(server) - await rm(directory, { recursive: true, force: true }) - } -}) + }), +) -test("rediscovers a rotated registration when acquiring a subscription", async () => { - const directory = await mkdtemp(path.join(os.tmpdir(), "opencode-pty-subscription-")) - const socketPath = path.join(directory, "daemon.sock") - let token = "old-token" - let instance = "old-instance" - let subscriptions = 0 - const server = await listen(socketPath, (_socket, request, receivedToken) => { - if (receivedToken !== token) return { type: "error", message: "authentication failed" } - if (request.op === "ping") return { ...pong, instance_id: instance } - if (request.op !== "subscribe") return { type: "terminals", terminals: [] } - subscriptions++ - return { - type: "attached", - terminal: terminal(1), - role: "observer", - generation: 1, - requested_offset: 0, - available_offset: 0, - end_offset: 0, - truncated: false, - replay_base64: "", - } - }) - try { - await writeRegistration(directory, socketPath, instance, token) - const daemon = await Effect.runPromise(makeDaemonTransport(directory)) - await Effect.runPromise(daemon.request({ op: "list" })) +it.live("rediscovers a rotated registration when acquiring a subscription", () => + Effect.gen(function* () { + const directory = yield* temporaryDirectory() + const socketPath = path.join(directory, "daemon.sock") + let token = "old-token" + let instance = "old-instance" + let subscriptions = 0 + yield* listen(socketPath, (_socket, request, receivedToken) => { + if (receivedToken !== token) return { type: "error", message: "authentication failed" } + if (request.op === "ping") return { ...pong, instance_id: instance } + if (request.op !== "subscribe") return { type: "terminals", terminals: [] } + subscriptions++ + return { + type: "attached", + terminal: terminal(1), + role: "observer", + generation: 1, + requested_offset: 0, + available_offset: 0, + end_offset: 0, + truncated: false, + replay_base64: "", + } + }) + yield* Effect.promise(() => writeRegistration(directory, socketPath, instance, token)) + const daemon = yield* makeDaemonTransport(directory) + yield* daemon.request({ op: "list" }) token = "new-token" instance = "new-instance" - await writeRegistration(directory, socketPath, instance, token) + yield* Effect.promise(() => writeRegistration(directory, socketPath, instance, token)) - const attachment = await Effect.runPromise( - daemon.subscribe(1, { - cursor: 0, - attachmentID: "attachment", - role: "observer", - onEvent: () => {}, - onEnd: () => {}, - }), - ) + const attachment = yield* daemon.subscribe(1, { + cursor: 0, + attachmentID: "attachment", + role: "observer", + onEvent: () => {}, + onEnd: () => {}, + }) expect(attachment.terminal.id).toBe(1) expect(subscriptions).toBe(1) attachment.detach() - } finally { - await close(server) - await rm(directory, { recursive: true, force: true }) - } -}) + }), +) -test("retries a start-required request when connection fails before dispatch", async () => { - const directory = await mkdtemp(path.join(os.tmpdir(), "opencode-pty-retry-")) - const firstSocket = path.join(directory, "first.sock") - const secondSocket = path.join(directory, "second.sock") - const first = await listen(firstSocket, (_socket, request) => { - if (request.op === "ping") return pong - return { type: "terminals", terminals: [] } - }) - try { - await writeRegistration(directory, firstSocket) - const daemon = await Effect.runPromise(makeDaemonTransport(directory)) - await Effect.runPromise(daemon.request({ op: "list" })) - await close(first) +it.live("retries a start-required request when connection fails before dispatch", () => + Effect.gen(function* () { + const directory = yield* temporaryDirectory() + const firstSocket = path.join(directory, "first.sock") + const secondSocket = path.join(directory, "second.sock") + const first = yield* listen(firstSocket, (_socket, request) => { + if (request.op === "ping") return pong + return { type: "terminals", terminals: [] } + }) + yield* Effect.promise(() => writeRegistration(directory, firstSocket)) + const daemon = yield* makeDaemonTransport(directory) + yield* daemon.request({ op: "list" }) + yield* Effect.promise(first.close) let creates = 0 - const second = await listen(secondSocket, (_socket, request) => { + yield* listen(secondSocket, (_socket, request) => { if (request.op === "ping") return pong creates++ return { type: "ok" } }) - try { - await writeRegistration(directory, secondSocket) - await Effect.runPromise(daemon.request({ op: "create" }, true)) - expect(creates).toBe(1) - } finally { - await close(second) - } - } finally { - await close(first) - await rm(directory, { recursive: true, force: true }) - } -}) + yield* Effect.promise(() => writeRegistration(directory, secondSocket)) + yield* daemon.request({ op: "create" }, true) + expect(creates).toBe(1) + }), +) -test("does not replay a dispatched mutating request when its response is lost", async () => { - const directory = await mkdtemp(path.join(os.tmpdir(), "opencode-pty-response-")) - const socketPath = path.join(directory, "daemon.sock") - let creates = 0 - const server = await listen(socketPath, (socket, request) => { - if (request.op === "ping") return pong - creates++ - socket.destroy() - return undefined - }) - try { - await writeRegistration(directory, socketPath) - const daemon = await Effect.runPromise(makeDaemonTransport(directory)) - const error = await Effect.runPromise(Effect.flip(daemon.request({ op: "create" }, true))) +it.live("does not replay a dispatched mutating request when its response is lost", () => + Effect.gen(function* () { + const directory = yield* temporaryDirectory() + const socketPath = path.join(directory, "daemon.sock") + let creates = 0 + yield* listen(socketPath, (socket, request) => { + if (request.op === "ping") return pong + creates++ + socket.destroy() + return undefined + }) + yield* Effect.promise(() => writeRegistration(directory, socketPath)) + const daemon = yield* makeDaemonTransport(directory) + const error = yield* Effect.flip(daemon.request({ op: "create" }, true)) expect(error.kind).toBe("response") expect(creates).toBe(1) - } finally { - await close(server) - await rm(directory, { recursive: true, force: true }) - } -}) + }), +) -test("reports protocol mismatches until a start-required request replaces the daemon", async () => { - const directory = await mkdtemp(path.join(os.tmpdir(), "opencode-pty-mismatch-")) - const existing = spawn("sleep", ["30"]) - const exited = new Promise((resolve) => existing.once("exit", () => resolve())) - try { +it.live("rejects incompatible daemons without replacing or killing them", () => + Effect.gen(function* () { + const directory = yield* temporaryDirectory() + const existing = yield* Effect.acquireRelease( + Effect.sync(() => spawn("sleep", ["30"])), + (child) => + Effect.promise(async () => { + if (child.exitCode !== null || child.signalCode !== null) return + const exited = new Promise((resolve) => child.once("exit", () => resolve())) + child.kill("SIGKILL") + await exited + }), + ) if (existing.pid === undefined) throw new Error("Expected fixture process PID") - await writeFile( - path.join(directory, "service.json"), - JSON.stringify({ instance_id: "old", pid: existing.pid, protocol: 5, socket: "/unused", token: "old" }), - ) - const daemon = await Effect.runPromise( - makeDaemonTransport(directory, () => Promise.resolve("/missing/opencode-pty")), + yield* Effect.promise(() => + writeFile( + path.join(directory, "service.json"), + JSON.stringify({ instance_id: "old", pid: existing.pid, protocol: 6, socket: "/unused", token: "old" }), + ), ) + const daemon = yield* makeDaemonTransport(directory, () => Promise.resolve("/missing/opencode-pty")) - const optional = await Effect.runPromise(Effect.flip(daemon.requestIfRunning({ op: "list" }))) + const optional = yield* Effect.flip(daemon.requestIfRunning({ op: "list" })) expect(optional).toMatchObject({ kind: "protocol", - message: "opencode-pty protocol mismatch: daemon=5, client=6", + message: "opencode-pty protocol mismatch: daemon=6, client=7", pid: existing.pid, }) - expect(existing.exitCode).toBeNull() - const starting = await Effect.runPromise(Effect.flip(daemon.request({ op: "create" }, true))) - await exited - expect(starting).toMatchObject({ kind: "spawn" }) - expect(existing.signalCode).toBe(process.platform === "win32" ? null : "SIGTERM") - } finally { - existing.kill("SIGKILL") - await rm(directory, { recursive: true, force: true }) - } -}) + const starting = yield* Effect.flip(daemon.request({ op: "create" }, true)) + expect(starting).toEqual(optional) + expect(existing.exitCode).toBeNull() + expect(existing.signalCode).toBeNull() + expect(process.kill(existing.pid, 0)).toBeTrue() + }), +) + +function temporaryDirectory() { + return Effect.acquireRelease( + Effect.promise(() => mkdtemp(path.join(os.tmpdir(), "opencode-pty-test-"))), + (directory) => Effect.promise(() => rm(directory, { recursive: true, force: true })), + ) +} function listen( socketPath: string, handle: (socket: net.Socket, request: Record, token: string) => object | undefined, ) { - const server = net.createServer((socket) => { - void readRequest(socket) - .then((envelope) => { - const response = handle(socket, envelope.request, envelope.token) - if (response) socket.write(frame(response)) + return Effect.acquireRelease( + Effect.promise(async () => { + const sockets = new Set() + const server = net.createServer((socket) => { + sockets.add(socket) + socket.once("close", () => sockets.delete(socket)) + void readRequest(socket) + .then((envelope) => { + const response = + envelope.request.op === "own" ? { type: "owned" } : handle(socket, envelope.request, envelope.token) + if (response) socket.write(frame(response)) + }) + .catch(() => socket.destroy()) }) - .catch(() => socket.destroy()) - }) - return new Promise((resolve, reject) => { - server.once("error", reject) - server.listen(socketPath, () => resolve(server)) - }) + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(socketPath, resolve) + }) + return { + close: () => { + sockets.forEach((socket) => socket.destroy()) + return close(server) + }, + } + }), + (server) => Effect.promise(server.close), + ) } function readRequest(socket: net.Socket) { @@ -246,7 +256,7 @@ function frame(value: object) { function writeRegistration(directory: string, socket: string, instance = "test", token = "test") { return writeFile( path.join(directory, "service.json"), - JSON.stringify({ instance_id: instance, pid: process.pid, protocol: 6, socket, token }), + JSON.stringify({ instance_id: instance, pid: process.pid, protocol: 7, socket, token }), ) } diff --git a/packages/protocol/src/groups/persistent-pty.ts b/packages/protocol/src/groups/persistent-pty.ts index 493564b66f6..1079f77267b 100644 --- a/packages/protocol/src/groups/persistent-pty.ts +++ b/packages/protocol/src/groups/persistent-pty.ts @@ -40,6 +40,12 @@ export const PersistentPtyGroup = HttpApiGroup.make("server.experimental") error: [ServiceUnavailableError], }), ) + .add( + HttpApiEndpoint.post("persistentPty.handoff", "/api/experimental/persistent-pty/handoff", { + success: Schema.Struct({ handoff: Schema.NullOr(PersistentPty.Handoff) }), + error: [ServiceUnavailableError], + }), + ) .add( HttpApiEndpoint.get("persistentPty.get", "/api/experimental/persistent-pty/:ptyID", { params: { ptyID: Pty.ID }, diff --git a/packages/schema/src/persistent-pty.ts b/packages/schema/src/persistent-pty.ts index ad5e4a364dc..82d7a02aeae 100644 --- a/packages/schema/src/persistent-pty.ts +++ b/packages/schema/src/persistent-pty.ts @@ -15,6 +15,14 @@ export const Info = Schema.Struct({ }).annotate({ identifier: "PersistentPty.Info" }) export interface Info extends Schema.Schema.Type {} +export const Handoff = Schema.Struct({ + directory: Schema.String, + instanceID: Schema.String, + ticket: Schema.String, + expiresAt: Schema.Number, +}).annotate({ identifier: "PersistentPty.Handoff" }) +export interface Handoff extends Schema.Schema.Type {} + export const CreateInput = Schema.Struct({ command: optional(Schema.String), args: Schema.Array(Schema.String), diff --git a/packages/server/src/handlers/persistent-pty.ts b/packages/server/src/handlers/persistent-pty.ts index fee71afed32..ab1476a9bf4 100644 --- a/packages/server/src/handlers/persistent-pty.ts +++ b/packages/server/src/handlers/persistent-pty.ts @@ -51,6 +51,12 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experiment return HttpApiSchema.NoContent.make() }), ) + .handle( + "persistentPty.handoff", + Effect.fn(function* () { + return { handoff: yield* pty.handoff().pipe(mapUnavailable) } + }), + ) .handle( "persistentPty.get", Effect.fn(function* (ctx) { diff --git a/packages/server/src/options.ts b/packages/server/src/options.ts index 96a70bd1800..b45b52ed590 100644 --- a/packages/server/src/options.ts +++ b/packages/server/src/options.ts @@ -1,5 +1,6 @@ import { Database } from "@opencode-ai/core/database/database" import { ModelsDev } from "@opencode-ai/core/models-dev" +import { PersistentPty } from "@opencode-ai/core/persistent-pty" import { Schema } from "effect" export const ServerOptions = Schema.Struct({ @@ -15,6 +16,7 @@ export const ServerOptions = Schema.Struct({ password: Schema.optional(Schema.String), simulation: Schema.optional(Schema.Boolean), database: Schema.optional(Database.Options), + pty: Schema.optional(PersistentPty.Options), events: Schema.optional( Schema.Struct({ persist: Schema.optional(Schema.Boolean), diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index 92e59199127..d02df53dc7d 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -100,6 +100,7 @@ function makeRoutes( const pluginRuntimeCell = PluginRuntime.makeCell() const standard: LayerNode.Replacements = [ [Database.node, Database.configured(options.database)], + [PersistentPty.node, PersistentPty.configured(options.pty)], [Bus.node, Bus.configured({ persist: options.events?.persist })], [App.node, App.configured(options.app)], [ModelsDev.node, ModelsDev.configured(options.models)], diff --git a/packages/server/test/persistent-pty.test.ts b/packages/server/test/persistent-pty.test.ts index 8046f0cc669..6ab8e358422 100644 --- a/packages/server/test/persistent-pty.test.ts +++ b/packages/server/test/persistent-pty.test.ts @@ -1,12 +1,11 @@ import { existsSync } from "node:fs" import fs from "node:fs/promises" -import { createHash } from "node:crypto" import os from "node:os" import path from "node:path" import { expect } from "bun:test" import { PersistentPty } from "@opencode-ai/schema/persistent-pty" import { Session } from "@opencode-ai/schema/session" -import { Effect, Schema } from "effect" +import { Effect, Exit, Schema, Scope } from "effect" import { HttpServer } from "effect/unstable/http" import { it } from "../../core/test/lib/effect" import { ServerProcess } from "../src/process" @@ -17,210 +16,288 @@ const smoke = existsSync(binary) ? it.live : it.live.skip smoke( "creates two persistent terminals for one session through the client API", () => - Effect.acquireUseRelease( - Effect.promise(async () => { - const environment = { - binary: process.env.OPENCODE_PTY_BIN, - runtime: process.env.OPENCODE_PTY_RUNTIME_DIR, - xdg: process.env.XDG_RUNTIME_DIR, - shell: process.env.SHELL, + Effect.gen(function* () { + const fixture = yield* testDirectory("xdg") + const server = yield* ServerProcess.start({ + hostname: "127.0.0.1", + port: 0, + password: "secret", + app: { version: "test-version" }, + database: { path: fixture.database }, + fs: { filewatcher: false }, + }) + const base = HttpServer.formatAddress(server.address) + const sessionID = Session.ID.make("ses_persistent_pty_test") + expect(existsSync(fixture.directory)).toBeFalse() + expect(yield* request(base, "POST", "/api/experimental/persistent-pty/handoff")).toEqual({ + handoff: null, + }) + expect((yield* request(base, "GET", `/api/experimental/session/${sessionID}/terminal`)).data).toEqual([]) + expect(existsSync(fixture.directory)).toBeFalse() + const defaults = { + args: [], + cwd: fixture.root, + title: "default shell", + env: { SHELL: "/missing/client/zsh" }, + } + const createPath = `/api/experimental/session/${sessionID}/terminal` + const terminal = Schema.decodeUnknownSync(PersistentPty.Info)( + (yield* request(base, "POST", createPath, defaults)).data, + ) + expect(terminal.command).toBe("/bin/sh") + expect(terminal.cwd).toBe(fixture.root) + expect(terminal.cwd).not.toBe(process.cwd()) + yield* request(base, "DELETE", `/api/experimental/persistent-pty/${terminal.id}`) + const root = Schema.decodeUnknownSync(PersistentPty.Info)( + (yield* request(base, "POST", createPath, { + args: ["-c", "printf 'root-cwd:%s\\n' \"$PWD\"; cat"], + title: "root directory", + env: {}, + })).data, + ) + expect(root.cwd).toBe(path.parse(fixture.root).root) + expect(root.cwd).not.toBe(process.cwd()) + expect(yield* waitForText(base, root.id, `root-cwd:${root.cwd}`)).toContain(`root-cwd:${root.cwd}`) + yield* request(base, "DELETE", `/api/experimental/persistent-pty/${root.id}`) + const events = yield* Effect.promise(() => openEventStream(base)) + const first = Schema.decodeUnknownSync(PersistentPty.Info)( + (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, { + command: "/usr/bin/env", + args: ["/bin/sh", "-c", "stty -echo; printf terminal-one; cat"], + cwd: process.cwd(), + title: "first", + env: {}, + })).data, + ) + expect(first.command).toBe("/usr/bin/env") + expect(first.args).toEqual(["/bin/sh", "-c", "stty -echo; printf terminal-one; cat"]) + expect(first.cwd).toBe(process.cwd()) + expect(yield* Effect.promise(() => events.next("persistent-pty.added"))).toMatchObject({ + data: { sessionID, terminal: { id: first.id } }, + }) + expect(first.size).toEqual({ cols: 80, rows: 24 }) + const directories = yield* Effect.promise(() => fs.readdir(fixture.directory)) + expect(directories).toHaveLength(1) + expect(directories[0]).toMatch(/^[0-9a-f-]{36}$/) + if (!directories[0]) throw new Error("Missing daemon directory") + expect(existsSync(path.join(fixture.directory, directories[0], "service.json"))).toBeTrue() + const second = Schema.decodeUnknownSync(PersistentPty.Info)( + (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, { + command: "/bin/sh", + args: ["-c", "printf terminal-two; sleep 30"], + cwd: process.cwd(), + title: "second", + env: {}, + })).data, + ) + + const terminals = Schema.decodeUnknownSync(Schema.Array(PersistentPty.Info))( + (yield* request(base, "GET", `/api/experimental/session/${sessionID}/terminal`)).data, + ) + expect(terminals.map((terminal) => terminal.id).sort()).toEqual([first.id, second.id].sort()) + expect(yield* waitForText(base, first.id, "terminal-one")).toContain("terminal-one") + expect(yield* waitForText(base, second.id, "terminal-two")).toContain("terminal-two") + yield* Effect.promise(() => verifySharedControl(base, first.id)) + const snapshot = yield* request(base, "GET", `/api/experimental/persistent-pty/${first.id}/snapshot`) + if ( + !isRecord(snapshot.data) || + typeof snapshot.data.checkpoint !== "string" || + !isRecord(snapshot.data.info) || + !isRecord(snapshot.data.info.output) || + typeof snapshot.data.info.output.tail !== "number" + ) + throw new Error("Persistent PTY snapshot response was invalid") + expect(Buffer.from(snapshot.data.checkpoint, "base64").byteLength).toBeGreaterThan(0) + expect(snapshot.data.info.output.tail).toBeGreaterThan(0) + + const ticket = yield* request( + base, + "POST", + `/api/experimental/persistent-pty/${first.id}/connect-token`, + undefined, + { + "x-opencode-ticket": "1", + }, + ) + if (!isRecord(ticket.data) || typeof ticket.data.ticket !== "string") throw new Error("Invalid connect ticket") + const connectTicket = ticket.data.ticket + yield* request(base, "DELETE", `/api/experimental/persistent-pty/${first.id}`) + yield* Effect.promise(async () => { + const url = new URL(`/api/experimental/persistent-pty/${first.id}/connect`, base) + url.searchParams.set("ticket", "invalid") + expect((await fetch(url)).status).toBe(403) + url.protocol = "ws:" + url.searchParams.set("ticket", connectTicket) + const socket = new WebSocket(url) + try { + const closed = await new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error("Removed terminal socket did not close")), 5_000) + socket.addEventListener("close", (event) => { + clearTimeout(timeout) + resolve(event) + }) + socket.addEventListener("error", () => { + clearTimeout(timeout) + reject(new Error("Valid ticket should upgrade before the missing terminal is reported")) + }) + }) + expect(closed.code).toBe(4404) + expect(closed.reason).toBe("terminal unavailable") + } finally { + socket.close() } - const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-pty-server-test-")) - const database = path.join(root, "opencode.db") - const runtime = path.join(root, "runtime") - process.env.OPENCODE_PTY_BIN = binary - delete process.env.OPENCODE_PTY_RUNTIME_DIR - process.env.XDG_RUNTIME_DIR = runtime - process.env.SHELL = "/bin/sh" - return { - database, - directory: path.join( - runtime, - "opencode-pty", - createHash("sha256").update(database).digest("hex").slice(0, 16), - ), - environment, - root, - } - }), - (fixture) => - Effect.gen(function* () { - const server = yield* ServerProcess.start({ - hostname: "127.0.0.1", - port: 0, - password: "secret", - app: { version: "test-version" }, - database: { path: fixture.database }, - fs: { filewatcher: false }, - }) - const base = HttpServer.formatAddress(server.address) - const sessionID = Session.ID.make("ses_persistent_pty_test") - expect(existsSync(path.join(fixture.directory, "service.json"))).toBeFalse() - expect((yield* request(base, "GET", `/api/experimental/session/${sessionID}/terminal`)).data).toEqual([]) - expect(existsSync(path.join(fixture.directory, "service.json"))).toBeFalse() - const defaults = { - args: [], - cwd: fixture.root, - title: "default shell", - env: { SHELL: "/missing/client/zsh" }, - } - const createPath = `/api/experimental/session/${sessionID}/terminal` - const terminal = Schema.decodeUnknownSync(PersistentPty.Info)( - (yield* request(base, "POST", createPath, defaults)).data, - ) - expect(terminal.command).toBe("/bin/sh") - expect(terminal.cwd).toBe(fixture.root) - expect(terminal.cwd).not.toBe(process.cwd()) - yield* request(base, "DELETE", `/api/experimental/persistent-pty/${terminal.id}`) - const root = Schema.decodeUnknownSync(PersistentPty.Info)( - (yield* request(base, "POST", createPath, { - args: ["-c", "printf 'root-cwd:%s\\n' \"$PWD\"; cat"], - title: "root directory", - env: {}, - })).data, - ) - expect(root.cwd).toBe(path.parse(fixture.root).root) - expect(root.cwd).not.toBe(process.cwd()) - expect(yield* waitForText(base, root.id, `root-cwd:${root.cwd}`)).toContain(`root-cwd:${root.cwd}`) - yield* request(base, "DELETE", `/api/experimental/persistent-pty/${root.id}`) - const events = yield* Effect.promise(() => openEventStream(base)) - const first = Schema.decodeUnknownSync(PersistentPty.Info)( - (yield* request(base, "POST", createPath, { - command: "/usr/bin/env", - args: ["/bin/sh", "-c", "stty -echo; printf terminal-one; cat"], - cwd: process.cwd(), - title: "first", - env: {}, - })).data, - ) - expect(first.command).toBe("/usr/bin/env") - expect(first.args).toEqual(["/bin/sh", "-c", "stty -echo; printf terminal-one; cat"]) - expect(first.cwd).toBe(process.cwd()) - expect(yield* Effect.promise(() => events.next("persistent-pty.added"))).toMatchObject({ - data: { sessionID, terminal: { id: first.id } }, - }) - expect(first.size).toEqual({ cols: 80, rows: 24 }) - expect(existsSync(path.join(fixture.directory, "service.json"))).toBeTrue() - const second = Schema.decodeUnknownSync(PersistentPty.Info)( - (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, { - command: "/bin/sh", - args: ["-c", "printf terminal-two; sleep 30"], - cwd: process.cwd(), - title: "second", - env: {}, - })).data, - ) + }) + expect(yield* Effect.promise(() => events.next("persistent-pty.removed"))).toMatchObject({ + data: { sessionID, ptyID: first.id }, + }) + yield* request(base, "DELETE", `/api/experimental/persistent-pty/${second.id}`) + expect((yield* request(base, "GET", `/api/experimental/session/${sessionID}/terminal`)).data).toEqual([]) - const terminals = Schema.decodeUnknownSync(Schema.Array(PersistentPty.Info))( - (yield* request(base, "GET", `/api/experimental/session/${sessionID}/terminal`)).data, - ) - expect(terminals.map((terminal) => terminal.id).sort()).toEqual([first.id, second.id].sort()) - expect(yield* waitForText(base, first.id, "terminal-one")).toContain("terminal-one") - expect(yield* waitForText(base, second.id, "terminal-two")).toContain("terminal-two") - yield* Effect.promise(() => verifySharedControl(base, first.id)) - const snapshot = yield* request(base, "GET", `/api/experimental/persistent-pty/${first.id}/snapshot`) - if ( - !isRecord(snapshot.data) || - typeof snapshot.data.checkpoint !== "string" || - !isRecord(snapshot.data.info) || - !isRecord(snapshot.data.info.output) || - typeof snapshot.data.info.output.tail !== "number" - ) - throw new Error("Persistent PTY snapshot response was invalid") - expect(Buffer.from(snapshot.data.checkpoint, "base64").byteLength).toBeGreaterThan(0) - expect(snapshot.data.info.output.tail).toBeGreaterThan(0) + yield* request(base, "POST", "/api/experimental/persistent-pty/shutdown") - const ticket = yield* request( - base, - "POST", - `/api/experimental/persistent-pty/${first.id}/connect-token`, - undefined, - { - "x-opencode-ticket": "1", - }, - ) - if (!isRecord(ticket.data) || typeof ticket.data.ticket !== "string") - throw new Error("Invalid connect ticket") - const connectTicket = ticket.data.ticket - yield* request(base, "DELETE", `/api/experimental/persistent-pty/${first.id}`) - yield* Effect.promise(async () => { - const url = new URL(`/api/experimental/persistent-pty/${first.id}/connect`, base) - url.searchParams.set("ticket", "invalid") - expect((await fetch(url)).status).toBe(403) - url.protocol = "ws:" - url.searchParams.set("ticket", connectTicket) - const socket = new WebSocket(url) - try { - const closed = await new Promise((resolve, reject) => { - const timeout = setTimeout(() => reject(new Error("Removed terminal socket did not close")), 5_000) - socket.addEventListener("close", (event) => { - clearTimeout(timeout) - resolve(event) - }) - socket.addEventListener("error", () => { - clearTimeout(timeout) - reject(new Error("Valid ticket should upgrade before the missing terminal is reported")) - }) - }) - expect(closed.code).toBe(4404) - expect(closed.reason).toBe("terminal unavailable") - } finally { - socket.close() - } - }) - expect(yield* Effect.promise(() => events.next("persistent-pty.removed"))).toMatchObject({ - data: { sessionID, ptyID: first.id }, - }) - yield* request(base, "DELETE", `/api/experimental/persistent-pty/${second.id}`) - expect((yield* request(base, "GET", `/api/experimental/session/${sessionID}/terminal`)).data).toEqual([]) + const unattended = Schema.decodeUnknownSync(PersistentPty.Info)( + (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, { + command: "/bin/sh", + args: ["-c", "exit 7"], + cwd: process.cwd(), + title: "unattended", + env: {}, + })).data, + ) + yield* waitForStatus(base, unattended.id, "exited") + expect((yield* request(base, "GET", `/api/experimental/session/${sessionID}/terminal`)).data).toMatchObject([ + { id: unattended.id, status: "exited" }, + ]) + yield* request(base, "DELETE", `/api/experimental/persistent-pty/${unattended.id}`) - yield* request(base, "POST", "/api/experimental/persistent-pty/shutdown") - - const unattended = Schema.decodeUnknownSync(PersistentPty.Info)( - (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, { - command: "/bin/sh", - args: ["-c", "exit 7"], - cwd: process.cwd(), - title: "unattended", - env: {}, - })).data, - ) - yield* waitForStatus(base, unattended.id, "exited") - expect((yield* request(base, "GET", `/api/experimental/session/${sessionID}/terminal`)).data).toMatchObject([ - { id: unattended.id, status: "exited" }, - ]) - yield* request(base, "DELETE", `/api/experimental/persistent-pty/${unattended.id}`) - - const visible = Schema.decodeUnknownSync(PersistentPty.Info)( - (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, { - command: "/bin/sh", - args: ["-c", "read value"], - cwd: process.cwd(), - title: "visible", - env: {}, - })).data, - ) - yield* attachAndExit(base, visible.id) - yield* waitForTerminals(base, sessionID, []) - yield* Effect.promise(() => events.close()) - }), - (fixture) => - Effect.promise(async () => { - await Bun.spawn([binary, "stop"], { - env: { ...process.env, OPENCODE_PTY_RUNTIME_DIR: fixture.directory }, - stdout: "ignore", - stderr: "ignore", - }).exited - await fs.rm(fixture.root, { recursive: true, force: true }) - restore("OPENCODE_PTY_BIN", fixture.environment.binary) - restore("OPENCODE_PTY_RUNTIME_DIR", fixture.environment.runtime) - restore("XDG_RUNTIME_DIR", fixture.environment.xdg) - restore("SHELL", fixture.environment.shell) - }), - ), + const visible = Schema.decodeUnknownSync(PersistentPty.Info)( + (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, { + command: "/bin/sh", + args: ["-c", "read value"], + cwd: process.cwd(), + title: "visible", + env: {}, + })).data, + ) + yield* attachAndExit(base, visible.id) + yield* waitForTerminals(base, sessionID, []) + yield* Effect.promise(() => events.close()) + }), 20_000, ) +smoke( + "isolates servers sharing a database and preserves terminals only through explicit restart handoff", + () => + Effect.gen(function* () { + const fixture = yield* testDirectory("override") + const scope = yield* Scope.Scope + const originalScope = yield* Scope.fork(scope) + const options = { + hostname: "127.0.0.1", + port: 0, + password: "secret", + app: { version: "test-version" }, + database: { path: fixture.database }, + fs: { filewatcher: false }, + } + const original = yield* ServerProcess.start(options).pipe( + Effect.provideService(Scope.Scope, originalScope), + ) + const base = HttpServer.formatAddress(original.address) + const sessionID = Session.ID.make("ses_persistent_pty_restart") + const first = Schema.decodeUnknownSync(PersistentPty.Info)( + (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, { + command: "/bin/sh", + args: ["-c", "stty -echo; printf before-restart; exec cat"], + cwd: process.cwd(), + title: "survivor", + env: {}, + })).data, + ) + expect(yield* waitForText(base, first.id, "before-restart")).toContain("before-restart") + + const independent = yield* ServerProcess.start(options) + const otherBase = HttpServer.formatAddress(independent.address) + expect((yield* request(otherBase, "GET", `/api/experimental/session/${sessionID}/terminal`)).data).toEqual([]) + + const handoff = Schema.decodeUnknownSync(PersistentPty.Handoff)( + (yield* request(base, "POST", "/api/experimental/persistent-pty/handoff")).handoff, + ) + const registration = Schema.decodeUnknownSync(Schema.Struct({ pid: Schema.Number }))( + yield* Effect.promise(() => Bun.file(path.join(handoff.directory, "service.json")).json()), + ) + yield* Scope.close(originalScope, Exit.void) + expect(process.kill(registration.pid, 0)).toBeTrue() + expect(process.kill(first.pid, 0)).toBeTrue() + + const replacementScope = yield* Scope.fork(scope) + const replacement = yield* ServerProcess.start({ ...options, pty: { handoff } }).pipe( + Effect.provideService(Scope.Scope, replacementScope), + ) + const replacementBase = HttpServer.formatAddress(replacement.address) + expect( + (yield* request(replacementBase, "GET", `/api/experimental/session/${sessionID}/terminal`)).data, + ).toMatchObject([{ id: first.id, pid: first.pid }]) + expect(yield* waitForText(replacementBase, first.id, "before-restart")).toContain("before-restart") + + yield* Scope.close(replacementScope, Exit.void) + yield* waitForExit(registration.pid) + yield* waitForExit(first.pid) + expect(existsSync(path.join(handoff.directory, "service.json"))).toBeFalse() + }), + 30_000, +) + +function testDirectory(mode: "xdg" | "override") { + return Effect.acquireRelease( + Effect.promise(async () => { + const environment = { + binary: process.env.OPENCODE_PTY_BIN, + runtime: process.env.OPENCODE_PTY_RUNTIME_DIR, + xdg: process.env.XDG_RUNTIME_DIR, + shell: process.env.SHELL, + } + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-pty-server-test-")) + const runtime = path.join(root, "runtime") + process.env.OPENCODE_PTY_BIN = binary + delete process.env.OPENCODE_PTY_RUNTIME_DIR + process.env.XDG_RUNTIME_DIR = runtime + process.env.SHELL = "/bin/sh" + if (mode === "override") process.env.OPENCODE_PTY_RUNTIME_DIR = runtime + return { + database: path.join(root, "opencode.db"), + directory: mode === "override" ? runtime : path.join(runtime, "opencode-pty"), + environment, + root, + } + }), + (fixture) => + Effect.promise(async () => { + await fs.rm(fixture.root, { recursive: true, force: true }) + restore("OPENCODE_PTY_BIN", fixture.environment.binary) + restore("OPENCODE_PTY_RUNTIME_DIR", fixture.environment.runtime) + restore("XDG_RUNTIME_DIR", fixture.environment.xdg) + restore("SHELL", fixture.environment.shell) + }), + ) +} + +function waitForExit(pid: number) { + return Effect.promise(async () => { + for (let attempt = 0; attempt < 100; attempt++) { + try { + process.kill(pid, 0) + } catch (error) { + if (isRecord(error) && error.code === "ESRCH") return + throw error + } + await Bun.sleep(20) + } + throw new Error(`Process ${pid} survived its server scope`) + }) +} + function request(base: string, method: string, pathname: string, body?: unknown, headers?: Record) { return Effect.tryPromise({ try: async () => {