diff --git a/packages/core/src/persistent-pty.ts b/packages/core/src/persistent-pty.ts new file mode 100644 index 00000000000..c99bbad90c5 --- /dev/null +++ b/packages/core/src/persistent-pty.ts @@ -0,0 +1,2 @@ +export { PersistentPty } from "./persistent-pty/index.js" +export { Group } from "./persistent-pty/group.js" diff --git a/packages/core/src/persistent-pty/group.ts b/packages/core/src/persistent-pty/group.ts new file mode 100644 index 00000000000..b5f2ebccdfd --- /dev/null +++ b/packages/core/src/persistent-pty/group.ts @@ -0,0 +1,79 @@ +export * as Group from "./group.js" + +import { Group } from "@opencode-ai/schema/group" +import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" +import { Context, Effect, Layer, Schema, Semaphore } from "effect" +import { KV } from "../kv.js" + +export const ID = Group.ID +export type ID = Group.ID +export const Item = Group.Item +export type Item = Group.Item +export const Info = Group.Info +export type Info = Group.Info + +export interface Interface { + readonly list: () => Effect.Effect> + readonly get: (id: ID) => Effect.Effect + readonly create: (items?: ReadonlyArray) => Effect.Effect + readonly set: (group: Info) => Effect.Effect + readonly remove: (id: ID) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/Group") {} + +const key = "group:v1" +const Document = Schema.Array(Info) + +const layer = Layer.effect( + Service, + Effect.gen(function* () { + const kv = yield* KV.Service + const lock = Semaphore.makeUnsafe(1) + + const list = Effect.fn("Group.list")(function* () { + const value = yield* kv.get(key) + return Schema.is(Document)(value) ? value : [] + }) + + return Service.of({ + list, + get: Effect.fn("Group.get")(function* (id) { + return (yield* list()).find((group) => group.id === id) + }), + create: Effect.fn("Group.create")(function* (items = []) { + return yield* lock.withPermit( + Effect.gen(function* () { + const group = Info.make({ id: ID.create(), items: Array.from(items) }) + yield* kv.set(key, (yield* list()).concat(group)) + return group + }), + ) + }), + set: Effect.fn("Group.set")(function* (group) { + yield* lock.withPermit( + Effect.gen(function* () { + const groups = yield* list() + const index = groups.findIndex((item) => item.id === group.id) + yield* kv.set( + key, + index === -1 ? groups.concat(group) : groups.map((item) => (item.id === group.id ? group : item)), + ) + }), + ) + }), + remove: Effect.fn("Group.remove")(function* (id) { + yield* lock.withPermit( + Effect.gen(function* () { + yield* kv.set( + key, + (yield* list()).filter((group) => group.id !== id), + ) + }), + ) + }), + }) + }), +) + +export const node = makeGlobalNode({ service: Service, layer, deps: [KV.node] }) diff --git a/packages/core/src/persistent-pty/index.ts b/packages/core/src/persistent-pty/index.ts new file mode 100644 index 00000000000..89e5b7a072d --- /dev/null +++ b/packages/core/src/persistent-pty/index.ts @@ -0,0 +1,607 @@ +export * as PersistentPty from "./index.js" + +import { spawn } from "node:child_process" +import { createHash } from "node:crypto" +import { readFile } from "node:fs/promises" +import net from "node:net" +import os from "node:os" +import path from "node:path" +import { setTimeout } from "node:timers/promises" +import { Context, Effect, Layer, Schema } from "effect" +import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" +import { Group } from "./group.js" +import { Database } from "../database/database.js" +import { Pty } from "@opencode-ai/schema/pty" + +const ProtocolVersion = 1 +const MaxFrameBytes = 8 * 1024 * 1024 + +const Lifecycle = Schema.Union([ + Schema.Struct({ status: Schema.Literal("running") }), + Schema.Struct({ status: Schema.Literal("exited"), exit_code: Schema.NullOr(Schema.Number) }), + Schema.Struct({ status: Schema.Literal("failed"), message: Schema.String }), +]) + +const WireTerminal = Schema.Struct({ + id: Schema.Number, + pid: Schema.NullOr(Schema.Number), + title: Schema.String, + group_id: Schema.String, + command: Schema.Array(Schema.String), + cwd: Schema.String, + cols: Schema.Number, + rows: Schema.Number, + lifecycle: Lifecycle, + output_head: Schema.Number, + output_tail: Schema.Number, +}) + +const Registration = Schema.Struct({ + instance_id: Schema.String, + pid: Schema.Number, + protocol: Schema.Number, + socket: Schema.String, + token: Schema.String, +}) + +const Response = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("pong"), + instance_id: Schema.String, + pid: Schema.Number, + protocol: Schema.Number, + }), + 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("snapshot"), + terminal: WireTerminal, + text: Schema.String, + checkpoint_base64: Schema.String, + cursor_x: Schema.Number, + cursor_y: Schema.Number, + }), + Schema.Struct({ + type: Schema.Literal("attached"), + terminal: WireTerminal, + role: Schema.Literals(["controller", "observer"]), + generation: Schema.Number, + requested_offset: Schema.Number, + available_offset: Schema.Number, + end_offset: Schema.Number, + truncated: Schema.Boolean, + replay_base64: Schema.String, + }), + Schema.Struct({ + type: Schema.Literal("output"), + start: Schema.Number, + end: Schema.Number, + data_base64: Schema.String, + }), + Schema.Struct({ + type: Schema.Literal("resized"), + cols: Schema.Number, + rows: Schema.Number, + generation: Schema.Number, + }), + Schema.Struct({ + type: Schema.Literal("exited"), + exit_code: Schema.NullOr(Schema.Number), + final_offset: Schema.Number, + }), + Schema.Struct({ + type: Schema.Literal("controller_changed"), + attachment_id: Schema.NullOr(Schema.String), + generation: Schema.Number, + }), + Schema.Struct({ type: Schema.Literal("error"), message: Schema.String }), +]) + +type WireTerminal = typeof WireTerminal.Type +type WireResponse = typeof Response.Type +type Registration = typeof Registration.Type + +export type Role = "controller" | "observer" + +export type Info = Pty.Info & { + readonly groupID: Group.ID + readonly output: { readonly head: number; readonly tail: number } +} + +export type Snapshot = { + readonly info: Info + readonly text: string + readonly checkpoint: Uint8Array + readonly cursor: { readonly x: number; readonly y: number } +} + +export type StreamEvent = + | { readonly type: "output"; readonly start: number; readonly end: number; readonly data: Uint8Array } + | { readonly type: "resized"; readonly cols: number; readonly rows: number; readonly generation: number } + | { readonly type: "exited"; readonly exitCode?: number; readonly finalOffset: number } + | { readonly type: "controller_changed"; readonly attachmentID?: string; readonly generation: number } + +export type Attachment = { + readonly info: Info + readonly role: Role + readonly generation: number + readonly replay: { + readonly requestedOffset: number + readonly availableOffset: number + readonly endOffset: number + readonly truncated: boolean + readonly data: Uint8Array + } + readonly activate: () => void + readonly detach: () => void +} + +export class UnavailableError extends Schema.TaggedErrorClass()("PersistentPty.UnavailableError", { + message: Schema.String, +}) {} + +export class NotFoundError extends Schema.TaggedErrorClass()("PersistentPty.NotFoundError", { + ptyID: Pty.ID, +}) {} + +export class GroupNotFoundError extends Schema.TaggedErrorClass()( + "PersistentPty.GroupNotFoundError", + { groupID: Group.ID }, +) {} + +export interface Interface { + readonly list: (groupID?: Group.ID) => Effect.Effect + readonly get: (id: Pty.ID) => Effect.Effect + readonly create: ( + groupID: Group.ID, + input: { + readonly command: string + readonly args: readonly string[] + readonly cwd: string + readonly title: string + readonly env: Readonly> + readonly cols?: number + readonly rows?: number + }, + ) => Effect.Effect + readonly write: ( + id: Pty.ID, + data: string, + attachmentID?: string, + ) => Effect.Effect + readonly resize: ( + id: Pty.ID, + cols: number, + rows: number, + attachmentID?: string, + ) => Effect.Effect + readonly snapshot: (id: Pty.ID) => Effect.Effect + readonly remove: (id: Pty.ID) => Effect.Effect + readonly attach: ( + id: Pty.ID, + input: { + readonly cursor: number + readonly attachmentID: string + readonly role: Role + readonly takeover?: boolean + readonly onEvent: (event: StreamEvent) => void + readonly onEnd: () => void + }, + ) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/PersistentPty") {} + +export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const groups = yield* Group.Service + const database = yield* Database.Service + const client = new Client(runtimeDirectory(databasePath(database.db))) + + const list = Effect.fn("PersistentPty.list")(function* (groupID?: Group.ID) { + const response = yield* optionalRequest(client, { op: "list" }) + if (!response) return [] + if (response.type !== "terminals") return yield* unexpected(response) + return response.terminals + .map(toInfo) + .filter((terminal) => groupID === undefined || terminal.groupID === groupID) + }) + + 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* ( + groupID: Group.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 group = yield* groups.get(groupID) + if (!group) return yield* new GroupNotFoundError({ groupID }) + const response = yield* request(client, { + op: "create", + program: input.command, + args: input.args, + cwd: input.cwd, + title: input.title, + group_id: groupID, + 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* groups.set( + Group.Info.make({ + id: group.id, + items: group.items.concat({ type: "terminal", id: terminal.id }), + }), + ) + return terminal + }) + + const write = Effect.fn("PersistentPty.write")(function* ( + id: Pty.ID, + data: string, + attachmentID?: string, + ) { + yield* get(id) + const response = yield* request(client, { + 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(client, { + op: "resize", + id: fromID(id), + attachment_id: attachmentID ?? null, + cols, + rows, + }) + 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(client, { 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(client, { op: "terminate", id: fromID(id) }) + if (response.type !== "ok") return yield* unexpected(response) + const group = yield* groups.get(terminal.groupID) + if (!group) return undefined + yield* groups.set( + Group.Info.make({ + id: group.id, + items: group.items.filter((item) => item.type !== "terminal" || item.id !== id), + }), + ) + return undefined + }) + + 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) + return yield* Effect.tryPromise({ + try: () => client.subscribe(fromID(id), input), + catch: (error) => unavailable(error), + }) + }) + + return Service.of({ list, get, create, write, resize, snapshot, remove, attach }) + }), +) + +export const node = makeGlobalNode({ service: Service, layer, deps: [Group.node, Database.node] }) + +class Client { + private registration?: Promise + + constructor(private readonly directory: string) {} + + request(value: object, start = false) { + return this.connect(start).then((registration) => oneShot(registration, value)) + } + + requestIfRunning(value: object) { + return this.request(value).catch(() => undefined) + } + + async subscribe( + id: number, + input: { + readonly cursor: number + readonly attachmentID: string + readonly role: Role + readonly takeover?: boolean + readonly onEvent: (event: StreamEvent) => void + readonly onEnd: () => void + }, + ): Promise { + const registration = await this.connect(false) + const socket = net.createConnection(registration.socket) + const frames = decoder(socket) + await connected(socket) + socket.write( + encode({ + token: registration.token, + request: { + op: "subscribe", + id, + offset: input.cursor, + attachment_id: input.attachmentID, + role: input.role, + takeover: input.takeover ?? false, + }, + }), + ) + const initial = await frames.next() + if (initial.done) throw new Error("opencode-pty closed before attachment") + const response = decode(initial.value) + if (response.type === "error") throw new Error(response.message) + if (response.type !== "attached") throw new Error(`unexpected opencode-pty response: ${response.type}`) + let detached = false + const pump = async () => { + try { + for await (const frame of frames) { + const event = decode(frame) + if (event.type === "output") + input.onEvent({ + type: "output", + start: event.start, + end: event.end, + data: Buffer.from(event.data_base64, "base64"), + }) + if (event.type === "resized") + input.onEvent({ + type: "resized", + cols: event.cols, + rows: event.rows, + generation: event.generation, + }) + if (event.type === "controller_changed") + input.onEvent({ + type: "controller_changed", + attachmentID: event.attachment_id ?? undefined, + generation: event.generation, + }) + if (event.type === "exited") { + input.onEvent({ + type: "exited", + exitCode: event.exit_code ?? undefined, + finalOffset: event.final_offset, + }) + return + } + } + } finally { + if (!detached) input.onEnd() + } + } + let activated = false + return { + info: toInfo(response.terminal), + role: response.role, + generation: response.generation, + replay: { + requestedOffset: response.requested_offset, + availableOffset: response.available_offset, + endOffset: response.end_offset, + truncated: response.truncated, + data: Buffer.from(response.replay_base64, "base64"), + }, + activate() { + if (activated || detached) return + activated = true + void pump().catch(() => {}) + }, + detach() { + if (detached) return + detached = true + socket.destroy() + }, + } + } + + private connect(start: boolean) { + this.registration ??= start ? ensure(this.directory) : discover(this.directory) + return this.registration.catch((error) => { + this.registration = undefined + throw error + }) + } +} + +const request = (client: Client, value: object, start = false) => + Effect.tryPromise({ try: () => client.request(value, start), catch: (error) => unavailable(error) }) + +const optionalRequest = (client: Client, value: object) => + Effect.promise(() => client.requestIfRunning(value)) + +const unexpected = (response: WireResponse) => + Effect.fail(new UnavailableError({ message: `unexpected opencode-pty response: ${response.type}` })) + +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 root = + process.env.OPENCODE_PTY_RUNTIME_DIR ?? + (process.env.XDG_RUNTIME_DIR + ? path.join(process.env.XDG_RUNTIME_DIR, "opencode-pty") + : path.join( + 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)) +} + +const registrationPath = (directory: string) => path.join(directory, "service.json") + +async function ensure(directory: string) { + const found = await discover(directory).catch(() => undefined) + if (found) return found + spawn(process.env.OPENCODE_PTY_BIN || "opencode-pty", ["daemon"], { + detached: true, + stdio: "ignore", + env: { ...process.env, OPENCODE_PTY_RUNTIME_DIR: directory }, + }).unref() + const deadline = Date.now() + 5_000 + let last: unknown + while (Date.now() < deadline) { + try { + return await discover(directory) + } catch (error) { + last = error + await setTimeout(50) + } + } + throw last instanceof Error ? last : new Error("opencode-pty did not become ready") +} + +async function discover(directory: string) { + const registration = Schema.decodeUnknownSync(Registration)( + JSON.parse(await readFile(registrationPath(directory), "utf8")), + ) + if (registration.protocol !== ProtocolVersion) throw new Error("opencode-pty protocol mismatch") + const response = await oneShot(registration, { op: "ping" }) + if ( + response.type !== "pong" || + response.instance_id !== registration.instance_id || + response.pid !== registration.pid || + response.protocol !== ProtocolVersion + ) + throw new Error("opencode-pty registration mismatch") + return registration +} + +async function oneShot(registration: Registration, request: object) { + const socket = net.createConnection(registration.socket) + const frames = decoder(socket) + await connected(socket) + socket.write(encode({ token: registration.token, request })) + const first = await frames.next() + socket.end() + if (first.done) throw new Error("opencode-pty closed without response") + const response = decode(first.value) + if (response.type === "error") throw new Error(response.message) + return response +} + +function connected(socket: net.Socket) { + return new Promise((resolve, reject) => { + socket.once("connect", resolve) + socket.once("error", reject) + }) +} + +function encode(value: unknown) { + const payload = Buffer.from(JSON.stringify(value)) + if (payload.length > MaxFrameBytes) throw new Error("opencode-pty frame too large") + const output = Buffer.allocUnsafe(payload.length + 4) + output.writeUInt32BE(payload.length) + payload.copy(output, 4) + return output +} + +async function* decoder(socket: net.Socket) { + let pending = Buffer.alloc(0) + for await (const value of socket) { + const chunk = Buffer.isBuffer(value) ? value : Buffer.from(value) + pending = pending.length === 0 ? chunk : Buffer.concat([pending, chunk]) + while (pending.length >= 4) { + const length = pending.readUInt32BE(0) + if (length > MaxFrameBytes) throw new Error("opencode-pty frame too large") + if (pending.length < length + 4) break + yield pending.subarray(4, length + 4) + pending = pending.subarray(length + 4) + } + } + if (pending.length !== 0) throw new Error("opencode-pty truncated frame") +} + +function decode(payload: Uint8Array) { + return Schema.decodeUnknownSync(Response)(JSON.parse(Buffer.from(payload).toString("utf8"))) +} + +function toInfo(value: WireTerminal): Info { + const status = value.lifecycle.status + return { + ...Pty.Info.make({ + id: toID(value.id), + title: value.title, + command: value.command[0] || "", + args: value.command.slice(1), + cwd: value.cwd, + status: status === "running" ? "running" : "exited", + pid: value.pid ?? 0, + ...(status === "exited" ? { exitCode: value.lifecycle.exit_code ?? undefined } : {}), + }), + groupID: Group.ID.make(value.group_id), + output: { head: value.output_head, tail: value.output_tail }, + } +} + +function toID(value: number) { + return Pty.ID.make(`pty_persistent_${value}`) +} + +function fromID(value: Pty.ID) { + if (!value.startsWith("pty_persistent_")) throw new Error(`invalid persistent PTY ID: ${value}`) + const parsed = Number(value.slice("pty_persistent_".length)) + if (!Number.isSafeInteger(parsed) || parsed < 1) throw new Error(`invalid persistent PTY ID: ${value}`) + return parsed +} diff --git a/packages/core/test/group.test.ts b/packages/core/test/group.test.ts new file mode 100644 index 00000000000..a7a86c3769d --- /dev/null +++ b/packages/core/test/group.test.ts @@ -0,0 +1,52 @@ +import { describe, expect } from "bun:test" +import { Group } from "@opencode-ai/core/persistent-pty" +import { KV } from "@opencode-ai/core/kv" +import { Pty } from "@opencode-ai/schema/pty" +import { Session } from "@opencode-ai/schema/session" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { Effect } from "effect" +import { testEffect } from "./lib/effect" + +const it = testEffect(LayerNode.compile(LayerNode.group([Group.node, KV.node]))) + +describe("Group", () => { + it.effect("persists ordered groups in one versioned KV document", () => + Effect.gen(function* () { + const groups = yield* Group.Service + const kv = yield* KV.Service + const created = yield* groups.create([ + { type: "session", id: Session.ID.make("ses_one") }, + { type: "terminal", id: Pty.ID.make("pty_one") }, + ]) + + expect(yield* groups.get(created.id)).toEqual(created) + expect(yield* groups.list()).toEqual([created]) + expect(yield* kv.get("group:v1")).toEqual([created]) + + const updated = Group.Info.make({ + id: created.id, + items: [{ type: "terminal", id: Pty.ID.make("pty_two") }], + }) + yield* groups.set(updated) + expect(yield* groups.list()).toEqual([updated]) + + yield* groups.remove(created.id) + expect(yield* groups.get(created.id)).toBeUndefined() + expect(yield* kv.get("group:v1")).toEqual([]) + }), + ) + + it.effect("serializes concurrent document mutations", () => + Effect.gen(function* () { + const groups = yield* Group.Service + yield* Effect.all( + Array.from({ length: 20 }, (_, index) => + groups.create([{ type: "session", id: Session.ID.make(`ses_${index}`) }]), + ), + { concurrency: "unbounded" }, + ) + + expect(yield* groups.list()).toHaveLength(20) + }), + ) +}) diff --git a/packages/protocol/src/api.ts b/packages/protocol/src/api.ts index e6ca152bfda..fd329b67c8a 100644 --- a/packages/protocol/src/api.ts +++ b/packages/protocol/src/api.ts @@ -19,6 +19,7 @@ import { HealthGroup } from "./groups/health.js" import { ServerGroup } from "./groups/server.js" import { DebugGroup } from "./groups/debug.js" import { PtyGroup } from "./groups/pty.js" +import { PersistentPtyGroup } from "./groups/persistent-pty.js" import { ShellGroup } from "./groups/shell.js" import { ReferenceGroup } from "./groups/reference.js" import { Authorization } from "./middleware/authorization.js" @@ -86,6 +87,7 @@ type ApiGroups< | typeof DebugGroup | typeof MigrationGroup | typeof WorktreeGroup + | typeof PersistentPtyGroup | LocationGroups | FormGroups | SessionGroups @@ -166,6 +168,7 @@ const makeApiFromGroup = < .add(SkillGroup.middleware(locationMiddleware)) .add(eventGroup) .add(PtyGroup.middleware(locationMiddleware)) + .add(PersistentPtyGroup) .add(ShellGroup.middleware(locationMiddleware)) .add(ReferenceGroup.middleware(locationMiddleware)) .add(WorktreeGroup) diff --git a/packages/protocol/src/groups/persistent-pty.ts b/packages/protocol/src/groups/persistent-pty.ts new file mode 100644 index 00000000000..ddb85aecf39 --- /dev/null +++ b/packages/protocol/src/groups/persistent-pty.ts @@ -0,0 +1,131 @@ +import { Group } from "@opencode-ai/schema/group" +import { PersistentPty } from "@opencode-ai/schema/persistent-pty" +import { Pty } from "@opencode-ai/schema/pty" +import { PtyTicket } from "@opencode-ai/schema/pty-ticket" +import { Schema } from "effect" +import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/unstable/httpapi" +import { + ForbiddenError, + InvalidRequestError, + PtyNotFoundError, + ServiceUnavailableError, +} from "../errors.js" +import { + PTY_CONNECT_TICKET_QUERY, + PTY_CONNECT_TOKEN_HEADER, + PTY_CONNECT_TOKEN_HEADER_VALUE, +} from "./pty.js" + +export { PTY_CONNECT_TICKET_QUERY, PTY_CONNECT_TOKEN_HEADER, PTY_CONNECT_TOKEN_HEADER_VALUE } + +const CONNECT_PATH = /^\/api\/persistent-pty\/[^/]+\/connect$/ + +export function hasPersistentPtyConnectTicketURL(url: URL) { + return CONNECT_PATH.test(url.pathname) && !!url.searchParams.get(PTY_CONNECT_TICKET_QUERY) +} + +const errors = [InvalidRequestError, ServiceUnavailableError] as const +const terminalErrors = [PtyNotFoundError, ServiceUnavailableError] as const + +export const PersistentPtyGroup = HttpApiGroup.make("server.persistentPty") + .add( + HttpApiEndpoint.get("persistentPty.group.list", "/api/pty-group", { + success: Schema.Struct({ data: Schema.Array(Group.Info) }), + error: errors, + }), + ) + .add( + HttpApiEndpoint.post("persistentPty.group.create", "/api/pty-group", { + payload: Schema.Struct({ items: Schema.optional(Schema.Array(Group.Item)) }), + success: Schema.Struct({ data: Group.Info }), + error: errors, + }), + ) + .add( + HttpApiEndpoint.get("persistentPty.group.get", "/api/pty-group/:groupID", { + params: { groupID: Group.ID }, + success: Schema.Struct({ data: Group.Info }), + error: errors, + }), + ) + .add( + HttpApiEndpoint.put("persistentPty.group.set", "/api/pty-group/:groupID", { + params: { groupID: Group.ID }, + payload: Schema.Struct({ items: Schema.Array(Group.Item) }), + success: Schema.Struct({ data: Group.Info }), + error: errors, + }), + ) + .add( + HttpApiEndpoint.delete("persistentPty.group.remove", "/api/pty-group/:groupID", { + params: { groupID: Group.ID }, + success: HttpApiSchema.NoContent, + error: errors, + }), + ) + .add( + HttpApiEndpoint.get("persistentPty.list", "/api/pty-group/:groupID/terminal", { + params: { groupID: Group.ID }, + success: Schema.Struct({ data: Schema.Array(PersistentPty.Info) }), + error: errors, + }), + ) + .add( + HttpApiEndpoint.post("persistentPty.create", "/api/pty-group/:groupID/terminal", { + params: { groupID: Group.ID }, + payload: PersistentPty.CreateInput, + success: Schema.Struct({ data: PersistentPty.Info }), + error: errors, + }), + ) + .add( + HttpApiEndpoint.get("persistentPty.get", "/api/persistent-pty/:ptyID", { + params: { ptyID: Pty.ID }, + success: Schema.Struct({ data: PersistentPty.Info }), + error: terminalErrors, + }), + ) + .add( + HttpApiEndpoint.put("persistentPty.update", "/api/persistent-pty/:ptyID", { + params: { ptyID: Pty.ID }, + payload: PersistentPty.UpdateInput, + success: Schema.Struct({ data: PersistentPty.Info }), + error: terminalErrors, + }), + ) + .add( + HttpApiEndpoint.get("persistentPty.snapshot", "/api/persistent-pty/:ptyID/snapshot", { + params: { ptyID: Pty.ID }, + success: Schema.Struct({ data: PersistentPty.Snapshot }), + error: terminalErrors, + }), + ) + .add( + HttpApiEndpoint.delete("persistentPty.remove", "/api/persistent-pty/:ptyID", { + params: { ptyID: Pty.ID }, + success: HttpApiSchema.NoContent, + error: terminalErrors, + }), + ) + .add( + HttpApiEndpoint.post("persistentPty.connectToken", "/api/persistent-pty/:ptyID/connect-token", { + params: { ptyID: Pty.ID }, + success: Schema.Struct({ data: PtyTicket.ConnectToken }), + error: [ForbiddenError, PtyNotFoundError, ServiceUnavailableError], + }), + ) + .add( + HttpApiEndpoint.get("persistentPty.connect", "/api/persistent-pty/:ptyID/connect", { + params: { ptyID: Pty.ID }, + success: Schema.Boolean, + error: [ForbiddenError, PtyNotFoundError, ServiceUnavailableError], + }).annotateMerge( + OpenApi.annotations({ + identifier: "v2.persistentPty.connect", + summary: "Connect to a persistent PTY", + description: "Stream persistent PTY output through the OpenCode server.", + transform: (operation) => ({ ...operation, "x-websocket": true }), + }), + ), + ) + .annotateMerge(OpenApi.annotations({ title: "persistentPty", description: "Prototype persistent PTY routes." })) diff --git a/packages/schema/src/group.ts b/packages/schema/src/group.ts new file mode 100644 index 00000000000..b5cdf2ddb8d --- /dev/null +++ b/packages/schema/src/group.ts @@ -0,0 +1,38 @@ +export * as Group from "./group.js" + +import { Schema } from "effect" +import { ascending } from "./identifier.js" +import { Pty } from "./pty.js" +import { statics } from "./schema.js" +import { Session } from "./session.js" + +const IDSchema = Schema.String.check(Schema.isStartsWith("grp_")).pipe(Schema.brand("GroupID")) + +export const ID = IDSchema.pipe( + statics((schema: typeof IDSchema) => ({ create: () => schema.make("grp_" + ascending()) })), +) +export type ID = typeof ID.Type + +export const SessionItem = Schema.Struct({ + type: Schema.tag("session"), + id: Session.ID, +}) +export interface SessionItem extends Schema.Schema.Type {} + +export const TerminalItem = Schema.Struct({ + type: Schema.tag("terminal"), + id: Pty.ID, +}) +export interface TerminalItem extends Schema.Schema.Type {} + +export const Item = Schema.Union([SessionItem, TerminalItem]).pipe( + Schema.toTaggedUnion("type"), + Schema.annotate({ identifier: "Group.Item" }), +) +export type Item = typeof Item.Type + +export const Info = Schema.Struct({ + id: ID, + items: Schema.Array(Item), +}).annotate({ identifier: "Group.Info" }) +export interface Info extends Schema.Schema.Type {} diff --git a/packages/schema/src/index.ts b/packages/schema/src/index.ts index ec554688ac5..2e95b1cbbda 100644 --- a/packages/schema/src/index.ts +++ b/packages/schema/src/index.ts @@ -6,6 +6,7 @@ export { Credential } from "./credential.js" export { Event } from "./event.js" export { FileSystem } from "./filesystem.js" export { Form } from "./form.js" +export { Group } from "./group.js" export { Integration } from "./integration.js" export { LLM } from "./llm.js" export { Location } from "./location.js" @@ -30,6 +31,7 @@ export { Shell } from "./shell.js" export { Skill } from "./skill.js" export { TokenUsage } from "./token-usage.js" export { Pty } from "./pty.js" +export { PersistentPty } from "./persistent-pty.js" export { PtyTicket } from "./pty-ticket.js" export { Question } from "./question.js" export { Workspace } from "./workspace.js" diff --git a/packages/schema/src/persistent-pty.ts b/packages/schema/src/persistent-pty.ts new file mode 100644 index 00000000000..f5959f17722 --- /dev/null +++ b/packages/schema/src/persistent-pty.ts @@ -0,0 +1,37 @@ +export * as PersistentPty from "./persistent-pty.js" + +import { Schema } from "effect" +import { Group } from "./group.js" +import { Pty } from "./pty.js" +import { NonNegativeInt, PositiveInt, optional } from "./schema.js" + +export const Info = Schema.Struct({ + ...Pty.Info.fields, + groupID: Group.ID, + output: Schema.Struct({ head: NonNegativeInt, tail: NonNegativeInt }), +}).annotate({ identifier: "PersistentPty.Info" }) +export interface Info extends Schema.Schema.Type {} + +export const CreateInput = Schema.Struct({ + command: Schema.String, + args: Schema.Array(Schema.String), + cwd: Schema.String, + title: Schema.String, + env: Schema.Record(Schema.String, Schema.String), + size: optional(Schema.Struct({ cols: PositiveInt, rows: PositiveInt })), +}).annotate({ identifier: "PersistentPty.CreateInput" }) +export interface CreateInput extends Schema.Schema.Type {} + +export const UpdateInput = Schema.Struct({ + attachmentID: optional(Schema.String), + size: Schema.Struct({ cols: PositiveInt, rows: PositiveInt }), +}).annotate({ identifier: "PersistentPty.UpdateInput" }) +export interface UpdateInput extends Schema.Schema.Type {} + +export const Snapshot = Schema.Struct({ + info: Info, + text: Schema.String, + checkpoint: Schema.Uint8Array, + cursor: Schema.Struct({ x: NonNegativeInt, y: NonNegativeInt }), +}).annotate({ identifier: "PersistentPty.Snapshot" }) +export interface Snapshot extends Schema.Schema.Type {} diff --git a/packages/schema/test/group.test.ts b/packages/schema/test/group.test.ts new file mode 100644 index 00000000000..26b93a346f8 --- /dev/null +++ b/packages/schema/test/group.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, test } from "bun:test" +import { Schema } from "effect" +import { Group } from "../src/group.js" +import { Pty } from "../src/pty.js" +import { Session } from "../src/session.js" + +describe("Group", () => { + test("creates branded group IDs", () => { + expect(Group.ID.create()).toStartWith("grp_") + expect(() => Schema.decodeUnknownSync(Group.ID)("ses_invalid")).toThrow() + }) + + test("preserves one ordered session and terminal item list", () => { + const group = Schema.decodeUnknownSync(Group.Info)({ + id: Group.ID.create(), + items: [ + { type: "session", id: Session.ID.make("ses_one") }, + { type: "terminal", id: Pty.ID.make("pty_one") }, + { type: "session", id: Session.ID.make("ses_two") }, + ], + }) + + expect(group.items.map((item) => item.type)).toEqual(["session", "terminal", "session"]) + expect(() => + Schema.decodeUnknownSync(Group.Info)({ + id: group.id, + items: [{ type: "other", id: "other_one" }], + }), + ).toThrow() + }) +}) diff --git a/packages/server/src/handlers.ts b/packages/server/src/handlers.ts index e0233ba64a8..3f96802581b 100644 --- a/packages/server/src/handlers.ts +++ b/packages/server/src/handlers.ts @@ -16,6 +16,7 @@ import { HealthHandler } from "./handlers/health" import { ServerHandler } from "./handlers/server" import { DebugHandler } from "./handlers/debug" import { PtyHandler } from "./handlers/pty" +import { PersistentPtyHandler } from "./handlers/persistent-pty" import { ShellHandler } from "./handlers/shell" import { ReferenceHandler } from "./handlers/reference" import { LocationHandler } from "./handlers/location" @@ -55,6 +56,7 @@ export const handlers = Layer.mergeAll( SkillHandler, EventHandler.pipe(Layer.provide(EventFeed.layer)), PtyHandler, + PersistentPtyHandler, ShellHandler, ReferenceHandler, WorktreeHandler, diff --git a/packages/server/src/handlers/persistent-pty.ts b/packages/server/src/handlers/persistent-pty.ts new file mode 100644 index 00000000000..bfb48ff6996 --- /dev/null +++ b/packages/server/src/handlers/persistent-pty.ts @@ -0,0 +1,260 @@ +import { Group, PersistentPty } from "@opencode-ai/core/persistent-pty" +import { PtyTicket } from "@opencode-ai/core/pty/ticket" +import { + ForbiddenError, + InvalidRequestError, + PtyNotFoundError, + ServiceUnavailableError, +} from "@opencode-ai/protocol/errors" +import { + PTY_CONNECT_TICKET_QUERY, + PTY_CONNECT_TOKEN_HEADER, + PTY_CONNECT_TOKEN_HEADER_VALUE, +} from "@opencode-ai/protocol/groups/persistent-pty" +import { Effect, Queue } from "effect" +import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http" +import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi" +import { Socket } from "effect/unstable/socket" +import { Api } from "../api" +import { CorsConfig, isAllowedRequestOrigin } from "../cors" + +export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.persistentPty", (handlers) => + Effect.gen(function* () { + const tickets = yield* PtyTicket.Service + const cors = yield* CorsConfig + const groups = yield* Group.Service + const pty = yield* PersistentPty.Service + + return handlers + .handle( + "persistentPty.group.list", + Effect.fn(function* () { + return { data: yield* groups.list() } + }), + ) + .handle( + "persistentPty.group.create", + Effect.fn(function* (ctx) { + return { data: yield* groups.create(ctx.payload.items) } + }), + ) + .handle( + "persistentPty.group.get", + Effect.fn(function* (ctx) { + const group = yield* groups.get(ctx.params.groupID) + if (!group) + return yield* new InvalidRequestError({ + message: `Group not found: ${ctx.params.groupID}`, + field: "groupID", + }) + return { data: group } + }), + ) + .handle( + "persistentPty.group.set", + Effect.fn(function* (ctx) { + const group = Group.Info.make({ id: ctx.params.groupID, items: ctx.payload.items }) + yield* groups.set(group) + return { data: group } + }), + ) + .handle( + "persistentPty.group.remove", + Effect.fn(function* (ctx) { + yield* groups.remove(ctx.params.groupID) + return HttpApiSchema.NoContent.make() + }), + ) + .handle( + "persistentPty.list", + Effect.fn(function* (ctx) { + return { data: yield* pty.list(ctx.params.groupID).pipe(mapUnavailable) } + }), + ) + .handle( + "persistentPty.create", + Effect.fn(function* (ctx) { + return { + data: yield* pty + .create(ctx.params.groupID, { + command: ctx.payload.command, + args: ctx.payload.args, + cwd: ctx.payload.cwd, + title: ctx.payload.title, + env: ctx.payload.env, + cols: ctx.payload.size?.cols, + rows: ctx.payload.size?.rows, + }) + .pipe( + Effect.catchTags({ + "PersistentPty.GroupNotFoundError": () => + new InvalidRequestError({ + message: `Group not found: ${ctx.params.groupID}`, + field: "groupID", + }), + "PersistentPty.UnavailableError": unavailable, + }), + ), + } + }), + ) + .handle( + "persistentPty.get", + Effect.fn(function* (ctx) { + return { data: yield* pty.get(ctx.params.ptyID).pipe(mapTerminalError) } + }), + ) + .handle( + "persistentPty.update", + Effect.fn(function* (ctx) { + yield* pty + .resize( + ctx.params.ptyID, + ctx.payload.size.cols, + ctx.payload.size.rows, + ctx.payload.attachmentID, + ) + .pipe(mapTerminalError) + return { data: yield* pty.get(ctx.params.ptyID).pipe(mapTerminalError) } + }), + ) + .handle( + "persistentPty.snapshot", + Effect.fn(function* (ctx) { + return { data: yield* pty.snapshot(ctx.params.ptyID).pipe(mapTerminalError) } + }), + ) + .handle( + "persistentPty.remove", + Effect.fn(function* (ctx) { + yield* pty.remove(ctx.params.ptyID).pipe(mapTerminalError) + return HttpApiSchema.NoContent.make() + }), + ) + .handle( + "persistentPty.connectToken", + Effect.fn(function* (ctx) { + const request = yield* HttpServerRequest.HttpServerRequest + if ( + request.headers[PTY_CONNECT_TOKEN_HEADER] !== PTY_CONNECT_TOKEN_HEADER_VALUE || + !isAllowedRequestOrigin(request.headers.origin, request.headers.host, cors) + ) + return yield* new ForbiddenError({ message: "Invalid persistent PTY connect token request" }) + yield* pty.get(ctx.params.ptyID).pipe(mapTerminalError) + return { data: yield* tickets.issue({ ptyID: ctx.params.ptyID }) } + }), + ) + .handleRaw( + "persistentPty.connect", + Effect.fn("PersistentPtyHandler.connect")(function* (ctx) { + const exists = yield* pty.get(ctx.params.ptyID).pipe( + Effect.as(true), + Effect.catchTag("PersistentPty.NotFoundError", () => Effect.succeed(false)), + Effect.catchTag("PersistentPty.UnavailableError", () => Effect.succeed(false)), + ) + if (!exists) return HttpServerResponse.empty({ status: 404 }) + + const url = new URL(ctx.request.url, "http://localhost") + const ticket = url.searchParams.get(PTY_CONNECT_TICKET_QUERY) + if (ticket) { + const valid = isAllowedRequestOrigin(ctx.request.headers.origin, ctx.request.headers.host, cors) + ? yield* tickets.consume({ ticket, ptyID: ctx.params.ptyID }) + : false + if (!valid) return HttpServerResponse.empty({ status: 403 }) + } + + const cursor = Number(url.searchParams.get("cursor") ?? "0") + const role = url.searchParams.get("role") === "observer" ? "observer" : "controller" + const attachmentID = url.searchParams.get("attachment_id") ?? crypto.randomUUID() + if (!Number.isSafeInteger(cursor) || cursor < 0) return HttpServerResponse.empty({ status: 400 }) + + const socket = yield* Effect.orDie(ctx.request.upgrade) + const write = yield* socket.writer + const outbox = yield* Queue.unbounded() + const attachment = yield* pty + .attach(ctx.params.ptyID, { + cursor, + attachmentID, + role, + takeover: url.searchParams.get("takeover") === "true", + onEvent: (event) => { + if (event.type === "output") Queue.offerUnsafe(outbox, event.data) + if (event.type !== "output") Queue.offerUnsafe(outbox, JSON.stringify(event)) + }, + onEnd: () => Queue.offerUnsafe(outbox, new Socket.CloseEvent(1000)), + }) + .pipe( + Effect.catchTags({ + "PersistentPty.NotFoundError": () => Effect.succeed(undefined), + "PersistentPty.UnavailableError": () => Effect.succeed(undefined), + }), + ) + if (!attachment) return HttpServerResponse.empty({ status: 404 }) + + Queue.offerUnsafe( + outbox, + JSON.stringify({ + type: "attached", + attachmentID, + info: attachment.info, + role: attachment.role, + generation: attachment.generation, + replay: { + requestedOffset: attachment.replay.requestedOffset, + availableOffset: attachment.replay.availableOffset, + endOffset: attachment.replay.endOffset, + truncated: attachment.replay.truncated, + }, + }), + ) + if (attachment.replay.data.length > 0) Queue.offerUnsafe(outbox, attachment.replay.data) + attachment.activate() + + const drain = Effect.gen(function* () { + while (true) { + const item = yield* Queue.take(outbox) + yield* write(item) + if (item instanceof Socket.CloseEvent) return + } + }) + + yield* Effect.race( + drain, + socket.runRaw((message) => + role === "controller" + ? pty + .write( + ctx.params.ptyID, + typeof message === "string" ? message : Buffer.from(message).toString(), + attachmentID, + ) + .pipe(Effect.ignore) + : Effect.void, + ), + ).pipe( + Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void), + Effect.ensuring(Effect.sync(() => attachment.detach())), + Effect.orDie, + ) + return HttpServerResponse.empty() + }), + ) + }), +) + +const mapUnavailable = (effect: Effect.Effect) => + effect.pipe(Effect.catchTag("PersistentPty.UnavailableError", unavailable)) + +const mapTerminalError = ( + effect: Effect.Effect, +) => + effect.pipe( + Effect.catchTags({ + "PersistentPty.NotFoundError": (error) => + new PtyNotFoundError({ ptyID: error.ptyID, message: `PTY session not found: ${error.ptyID}` }), + "PersistentPty.UnavailableError": unavailable, + }), + ) + +const unavailable = (error: PersistentPty.UnavailableError) => + new ServiceUnavailableError({ message: error.message, service: "opencode-pty" }) diff --git a/packages/server/src/middleware/authorization.ts b/packages/server/src/middleware/authorization.ts index a004fa973d7..a4c60fb2152 100644 --- a/packages/server/src/middleware/authorization.ts +++ b/packages/server/src/middleware/authorization.ts @@ -3,6 +3,7 @@ import { UnauthorizedError } from "@opencode-ai/protocol/errors" import { Authorization } from "@opencode-ai/protocol/middleware/authorization" export { Authorization } from "@opencode-ai/protocol/middleware/authorization" import { hasPtyConnectTicketURL } from "@opencode-ai/protocol/groups/pty" +import { hasPersistentPtyConnectTicketURL } from "@opencode-ai/protocol/groups/persistent-pty" import { Effect, Encoding, Layer, Redacted } from "effect" import { HttpEffect, HttpServerRequest, HttpServerResponse } from "effect/unstable/http" @@ -49,7 +50,8 @@ export const authorizationLayer = Layer.effect( const request = yield* HttpServerRequest.HttpServerRequest // Browsers cannot set headers on WebSocket upgrades, so a ticketed PTY connect skips // credential checks here; the connect handler consumes and validates the ticket. - if (hasPtyConnectTicketURL(new URL(request.url, "http://localhost"))) return yield* effect + const url = new URL(request.url, "http://localhost") + if (hasPtyConnectTicketURL(url) || hasPersistentPtyConnectTicketURL(url)) return yield* effect if (yield* authorizedRequest(request, config)) return yield* effect yield* HttpEffect.appendPreResponseHandler((_request, response) => Effect.succeed(HttpServerResponse.setHeader(response, "www-authenticate", WWW_AUTHENTICATE)), diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index fd2f2a7549e..6afccb325c3 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -3,6 +3,7 @@ export * as ServerProcess from "./process" import { NodeHttpServer } from "@effect/platform-node" import { SessionRestart } from "@opencode-ai/core/session/execution/restart" import { hasPtyConnectTicketURL } from "@opencode-ai/protocol/groups/pty" +import { hasPersistentPtyConnectTicketURL } from "@opencode-ai/protocol/groups/persistent-pty" import { Cause, Context, Effect, Exit, Latch, Layer, Option, Ref, Scope } from "effect" import { HttpMiddleware, HttpRouter, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http" import { createServer } from "node:http" @@ -170,7 +171,11 @@ function dispatch( const state = yield* status.current const app = yield* Ref.get(application) const ready = state.type === "ready" && Option.isSome(app) - if ((!ready || !hasPtyConnectTicketURL(url)) && !(yield* authorizedRequest(request, auth))) return unauthorized() + if ( + (!ready || (!hasPtyConnectTicketURL(url) && !hasPersistentPtyConnectTicketURL(url))) && + !(yield* authorizedRequest(request, auth)) + ) + return unauthorized() if (ready) return yield* app.value return unavailable(state) }) diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index a38c830c654..ef2d8645b91 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -11,6 +11,7 @@ import { Credential } from "@opencode-ai/core/credential" import { Config } from "@opencode-ai/core/config" import { PermissionSaved } from "@opencode-ai/core/permission/saved" import { PtyTicket } from "@opencode-ai/core/pty/ticket" +import { Group, PersistentPty } from "@opencode-ai/core/persistent-pty" import { Project } from "@opencode-ai/core/project" import { Session } from "@opencode-ai/core/session" import { SessionTransfer } from "@opencode-ai/core/session/transfer" @@ -58,6 +59,8 @@ const applicationServiceNodes = [ SdkPlugins.node, PermissionSaved.node, PtyTicket.node, + Group.node, + PersistentPty.node, Credential.node, WellKnown.node, PtyEnvironment.node, diff --git a/packages/server/test/persistent-pty.test.ts b/packages/server/test/persistent-pty.test.ts new file mode 100644 index 00000000000..60354cb418e --- /dev/null +++ b/packages/server/test/persistent-pty.test.ts @@ -0,0 +1,164 @@ +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 { Group } from "@opencode-ai/schema/group" +import { PersistentPty } from "@opencode-ai/schema/persistent-pty" +import { Effect, Schema } from "effect" +import { HttpServer } from "effect/unstable/http" +import { it } from "../../core/test/lib/effect" +import { ServerProcess } from "../src/process" + +const binary = process.env.OPENCODE_PTY_BIN ?? "/root/projects/opencode-pty/target/debug/opencode-pty" +const smoke = existsSync(binary) ? it.live : it.live.skip + +smoke( + "creates a group with two persistent terminals 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, + } + 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 + 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) + expect(existsSync(path.join(fixture.directory, "service.json"))).toBeFalse() + const group = Schema.decodeUnknownSync(Group.Info)( + (yield* request(base, "POST", "/api/pty-group", { items: [] })).data, + ) + expect((yield* request(base, "GET", `/api/pty-group/${group.id}/terminal`)).data).toEqual([]) + expect(existsSync(path.join(fixture.directory, "service.json"))).toBeFalse() + const first = Schema.decodeUnknownSync(PersistentPty.Info)( + ( + yield* request(base, "POST", `/api/pty-group/${group.id}/terminal`, { + command: "/bin/sh", + args: ["-c", "printf terminal-one; sleep 30"], + cwd: process.cwd(), + title: "first", + env: {}, + }) + ).data, + ) + expect(existsSync(path.join(fixture.directory, "service.json"))).toBeTrue() + const second = Schema.decodeUnknownSync(PersistentPty.Info)( + ( + yield* request(base, "POST", `/api/pty-group/${group.id}/terminal`, { + command: "/bin/sh", + args: ["-c", "printf terminal-two; sleep 30"], + cwd: process.cwd(), + title: "second", + env: {}, + }) + ).data, + ) + + const updated = Schema.decodeUnknownSync(Group.Info)( + (yield* request(base, "GET", `/api/pty-group/${group.id}`)).data, + ) + expect(updated.items).toEqual([ + { type: "terminal", id: first.id }, + { type: "terminal", id: second.id }, + ]) + + const terminals = Schema.decodeUnknownSync(Schema.Array(PersistentPty.Info))( + (yield* request(base, "GET", `/api/pty-group/${group.id}/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* request(base, "DELETE", `/api/persistent-pty/${first.id}`) + yield* request(base, "DELETE", `/api/persistent-pty/${second.id}`) + expect((yield* request(base, "GET", `/api/pty-group/${group.id}`)).data).toMatchObject({ items: [] }) + yield* request(base, "DELETE", `/api/pty-group/${group.id}`) + }), + (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) + }), + ), + 20_000, +) + +function request(base: string, method: string, pathname: string, body?: unknown) { + return Effect.tryPromise({ + try: async () => { + const response = await fetch(new URL(pathname, base), { + method, + headers: { + authorization: `Basic ${btoa("opencode:secret")}`, + ...(body === undefined ? {} : { "content-type": "application/json" }), + }, + body: body === undefined ? undefined : JSON.stringify(body), + }) + if (!response.ok) throw new Error(`${method} ${pathname} failed (${response.status}): ${await response.text()}`) + if (response.status === 204) return {} + const value: unknown = await response.json() + if (!isRecord(value)) throw new Error(`${method} ${pathname} returned a non-object response`) + return value + }, + catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))), + }) +} + +function waitForText(base: string, ptyID: string, expected: string) { + return Effect.tryPromise({ + try: async () => { + for (let attempt = 0; attempt < 40; attempt++) { + const response = await Effect.runPromise(request(base, "GET", `/api/persistent-pty/${ptyID}/snapshot`)) + if (isRecord(response.data) && typeof response.data.text === "string" && response.data.text.includes(expected)) + return response.data.text + await Bun.sleep(50) + } + throw new Error(`Persistent PTY snapshot did not contain ${expected}`) + }, + catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))), + }) +} + +function restore(key: string, value: string | undefined) { + if (value === undefined) delete process.env[key] + if (value !== undefined) process.env[key] = value +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value) +}