mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-26 13:22:15 +00:00
feat(server): add persistent PTY daemon API (#44969)
This commit is contained in:
James Long
GitHub
parent
cce86ac166
commit
004b647311
24 changed files with 3633 additions and 5 deletions
|
|
@ -3,10 +3,13 @@ import { Service } from "@opencode-ai/client/effect/service"
|
|||
import { Commands } from "../../commands"
|
||||
import { Runtime } from "../../../framework/runtime"
|
||||
import { ServiceConfig } from "../../../services/service-config"
|
||||
import { ServerConnection } from "../../../services/server-connection"
|
||||
|
||||
export default Runtime.handler(
|
||||
Commands.commands.service.commands.stop,
|
||||
Effect.fn("cli.service.stop")(function* () {
|
||||
yield* Service.stop(yield* ServiceConfig.options())
|
||||
const options = yield* ServiceConfig.options()
|
||||
yield* ServerConnection.shutdownPersistentPty(options).pipe(Effect.ignore)
|
||||
yield* Service.stop(options)
|
||||
}),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -62,6 +62,15 @@ function managedService(options: EnsureOptions) {
|
|||
}
|
||||
}
|
||||
|
||||
export const shutdownPersistentPty = Effect.fn("cli.server-connection.shutdown-persistent-pty")(function* (
|
||||
options: EnsureOptions,
|
||||
) {
|
||||
const endpoint = yield* Service.discover({ ...options, version: undefined })
|
||||
if (!endpoint) return
|
||||
const client = OpenCode.make({ baseUrl: endpoint.url, headers: Service.headers(endpoint) })
|
||||
yield* Effect.tryPromise(() => client.experimental.persistentPty.shutdown())
|
||||
})
|
||||
|
||||
const resolveManaged = Effect.fnUntraced(function* (options: EnsureOptions, mismatch: NonNullable<Args["mismatch"]>) {
|
||||
if (mismatch === "replace") return yield* Service.ensure(options)
|
||||
if (mismatch === "ignore") return yield* Service.ensure({ ...options, version: undefined })
|
||||
|
|
|
|||
|
|
@ -1592,6 +1592,152 @@ export interface PtyApi<E = never> {
|
|||
readonly connect: { readonly token: PtyConnectTokenOperation<E> }
|
||||
}
|
||||
|
||||
export type ExperimentalPersistentPtyListInput = { readonly sessionID: Session.ID }
|
||||
export type ExperimentalPersistentPtyListOutput = ReadonlyArray<{
|
||||
readonly id: Pty.ID
|
||||
readonly title: string
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly status: "running" | "exited"
|
||||
readonly pid: number
|
||||
readonly exitCode?: number | undefined
|
||||
readonly sessionID: Session.ID
|
||||
readonly foregroundProcess: string | null
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
readonly output: { readonly head: number; readonly tail: number }
|
||||
}>
|
||||
export type ExperimentalPersistentPtyListOperation<E = never> = (
|
||||
input: ExperimentalPersistentPtyListInput,
|
||||
) => Effect.Effect<ExperimentalPersistentPtyListOutput, E>
|
||||
|
||||
export type ExperimentalPersistentPtyCreateInput = {
|
||||
readonly sessionID: Session.ID
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly title: string
|
||||
readonly env: { readonly [x: string]: string }
|
||||
readonly size?: { readonly cols: number; readonly rows: number } | undefined
|
||||
}
|
||||
export type ExperimentalPersistentPtyCreateOutput = {
|
||||
readonly id: Pty.ID
|
||||
readonly title: string
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly status: "running" | "exited"
|
||||
readonly pid: number
|
||||
readonly exitCode?: number | undefined
|
||||
readonly sessionID: Session.ID
|
||||
readonly foregroundProcess: string | null
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
readonly output: { readonly head: number; readonly tail: number }
|
||||
}
|
||||
export type ExperimentalPersistentPtyCreateOperation<E = never> = (
|
||||
input: ExperimentalPersistentPtyCreateInput,
|
||||
) => Effect.Effect<ExperimentalPersistentPtyCreateOutput, E>
|
||||
|
||||
export type ExperimentalPersistentPtyShutdownOutput = void
|
||||
export type ExperimentalPersistentPtyShutdownOperation<E = never> = () => Effect.Effect<
|
||||
ExperimentalPersistentPtyShutdownOutput,
|
||||
E
|
||||
>
|
||||
|
||||
export type ExperimentalPersistentPtyGetInput = { readonly ptyID: Pty.ID }
|
||||
export type ExperimentalPersistentPtyGetOutput = {
|
||||
readonly id: Pty.ID
|
||||
readonly title: string
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly status: "running" | "exited"
|
||||
readonly pid: number
|
||||
readonly exitCode?: number | undefined
|
||||
readonly sessionID: Session.ID
|
||||
readonly foregroundProcess: string | null
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
readonly output: { readonly head: number; readonly tail: number }
|
||||
}
|
||||
export type ExperimentalPersistentPtyGetOperation<E = never> = (
|
||||
input: ExperimentalPersistentPtyGetInput,
|
||||
) => Effect.Effect<ExperimentalPersistentPtyGetOutput, E>
|
||||
|
||||
export type ExperimentalPersistentPtyUpdateInput = {
|
||||
readonly ptyID: Pty.ID
|
||||
readonly attachmentID?: string | undefined
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
}
|
||||
export type ExperimentalPersistentPtyUpdateOutput = {
|
||||
readonly id: Pty.ID
|
||||
readonly title: string
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly status: "running" | "exited"
|
||||
readonly pid: number
|
||||
readonly exitCode?: number | undefined
|
||||
readonly sessionID: Session.ID
|
||||
readonly foregroundProcess: string | null
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
readonly output: { readonly head: number; readonly tail: number }
|
||||
}
|
||||
export type ExperimentalPersistentPtyUpdateOperation<E = never> = (
|
||||
input: ExperimentalPersistentPtyUpdateInput,
|
||||
) => Effect.Effect<ExperimentalPersistentPtyUpdateOutput, E>
|
||||
|
||||
export type ExperimentalPersistentPtySnapshotInput = { readonly ptyID: Pty.ID }
|
||||
export type ExperimentalPersistentPtySnapshotOutput = {
|
||||
readonly info: {
|
||||
readonly id: Pty.ID
|
||||
readonly title: string
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly status: "running" | "exited"
|
||||
readonly pid: number
|
||||
readonly exitCode?: number | undefined
|
||||
readonly sessionID: Session.ID
|
||||
readonly foregroundProcess: string | null
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
readonly output: { readonly head: number; readonly tail: number }
|
||||
}
|
||||
readonly text: string
|
||||
readonly checkpoint: globalThis.Uint8Array
|
||||
readonly cursor: { readonly x: number; readonly y: number }
|
||||
}
|
||||
export type ExperimentalPersistentPtySnapshotOperation<E = never> = (
|
||||
input: ExperimentalPersistentPtySnapshotInput,
|
||||
) => Effect.Effect<ExperimentalPersistentPtySnapshotOutput, E>
|
||||
|
||||
export type ExperimentalPersistentPtyRemoveInput = { readonly ptyID: Pty.ID }
|
||||
export type ExperimentalPersistentPtyRemoveOutput = void
|
||||
export type ExperimentalPersistentPtyRemoveOperation<E = never> = (
|
||||
input: ExperimentalPersistentPtyRemoveInput,
|
||||
) => Effect.Effect<ExperimentalPersistentPtyRemoveOutput, E>
|
||||
|
||||
export type ExperimentalPersistentPtyConnectTokenInput = {
|
||||
readonly ptyID: Pty.ID
|
||||
readonly "x-opencode-ticket"?: string | undefined
|
||||
}
|
||||
export type ExperimentalPersistentPtyConnectTokenOutput = PtyTicket.ConnectToken
|
||||
export type ExperimentalPersistentPtyConnectTokenOperation<E = never> = (
|
||||
input: ExperimentalPersistentPtyConnectTokenInput,
|
||||
) => Effect.Effect<ExperimentalPersistentPtyConnectTokenOutput, E>
|
||||
|
||||
export interface ExperimentalApi<E = never> {
|
||||
readonly persistentPty: {
|
||||
readonly list: ExperimentalPersistentPtyListOperation<E>
|
||||
readonly create: ExperimentalPersistentPtyCreateOperation<E>
|
||||
readonly shutdown: ExperimentalPersistentPtyShutdownOperation<E>
|
||||
readonly get: ExperimentalPersistentPtyGetOperation<E>
|
||||
readonly update: ExperimentalPersistentPtyUpdateOperation<E>
|
||||
readonly snapshot: ExperimentalPersistentPtySnapshotOperation<E>
|
||||
readonly remove: ExperimentalPersistentPtyRemoveOperation<E>
|
||||
readonly connectToken: ExperimentalPersistentPtyConnectTokenOperation<E>
|
||||
}
|
||||
}
|
||||
|
||||
export type ShellListInput = {
|
||||
readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined
|
||||
}
|
||||
|
|
@ -1842,6 +1988,7 @@ export interface AppApi<E = never> {
|
|||
readonly skill: SkillApi<E>
|
||||
readonly event: EventApi<E>
|
||||
readonly pty: PtyApi<E>
|
||||
readonly experimental: ExperimentalApi<E>
|
||||
readonly shell: ShellApi<E>
|
||||
readonly reference: ReferenceApi<E>
|
||||
readonly worktree: WorktreeApi<E>
|
||||
|
|
|
|||
|
|
@ -194,6 +194,21 @@ import type {
|
|||
PtyRemoveOutput,
|
||||
PtyConnectTokenInput,
|
||||
PtyConnectTokenOutput,
|
||||
ExperimentalPersistentPtyListInput,
|
||||
ExperimentalPersistentPtyListOutput,
|
||||
ExperimentalPersistentPtyCreateInput,
|
||||
ExperimentalPersistentPtyCreateOutput,
|
||||
ExperimentalPersistentPtyShutdownOutput,
|
||||
ExperimentalPersistentPtyGetInput,
|
||||
ExperimentalPersistentPtyGetOutput,
|
||||
ExperimentalPersistentPtyUpdateInput,
|
||||
ExperimentalPersistentPtyUpdateOutput,
|
||||
ExperimentalPersistentPtySnapshotInput,
|
||||
ExperimentalPersistentPtySnapshotOutput,
|
||||
ExperimentalPersistentPtyRemoveInput,
|
||||
ExperimentalPersistentPtyRemoveOutput,
|
||||
ExperimentalPersistentPtyConnectTokenInput,
|
||||
ExperimentalPersistentPtyConnectTokenOutput,
|
||||
ShellListInput,
|
||||
ShellListOutput,
|
||||
ShellCreateInput,
|
||||
|
|
@ -1192,6 +1207,100 @@ const adaptGroupPty = (raw: RawClient["server.pty"]) => ({
|
|||
connect: { token: EndpointPtyConnectToken(raw) },
|
||||
})
|
||||
|
||||
const EndpointExperimentalPersistentPtyList =
|
||||
(raw: RawClient["server.experimental"]) => (input: ExperimentalPersistentPtyListInput) =>
|
||||
preserveEffect<ExperimentalPersistentPtyListOutput>()(
|
||||
raw["persistentPty.list"]({ params: { sessionID: input["sessionID"] } }).pipe(
|
||||
Effect.mapError(mapClientError),
|
||||
Effect.map((value) => value.data),
|
||||
),
|
||||
)
|
||||
|
||||
const EndpointExperimentalPersistentPtyCreate =
|
||||
(raw: RawClient["server.experimental"]) => (input: ExperimentalPersistentPtyCreateInput) =>
|
||||
preserveEffect<ExperimentalPersistentPtyCreateOutput>()(
|
||||
raw["persistentPty.create"]({
|
||||
params: { sessionID: input["sessionID"] },
|
||||
payload: {
|
||||
command: input["command"],
|
||||
args: input["args"],
|
||||
cwd: input["cwd"],
|
||||
title: input["title"],
|
||||
env: input["env"],
|
||||
size: input["size"],
|
||||
},
|
||||
}).pipe(
|
||||
Effect.mapError(mapClientError),
|
||||
Effect.map((value) => value.data),
|
||||
),
|
||||
)
|
||||
|
||||
const EndpointExperimentalPersistentPtyShutdown = (raw: RawClient["server.experimental"]) => () =>
|
||||
preserveEffect<ExperimentalPersistentPtyShutdownOutput>()(
|
||||
raw["persistentPty.shutdown"]({}).pipe(Effect.mapError(mapClientError)),
|
||||
)
|
||||
|
||||
const EndpointExperimentalPersistentPtyGet =
|
||||
(raw: RawClient["server.experimental"]) => (input: ExperimentalPersistentPtyGetInput) =>
|
||||
preserveEffect<ExperimentalPersistentPtyGetOutput>()(
|
||||
raw["persistentPty.get"]({ params: { ptyID: input["ptyID"] } }).pipe(
|
||||
Effect.mapError(mapClientError),
|
||||
Effect.map((value) => value.data),
|
||||
),
|
||||
)
|
||||
|
||||
const EndpointExperimentalPersistentPtyUpdate =
|
||||
(raw: RawClient["server.experimental"]) => (input: ExperimentalPersistentPtyUpdateInput) =>
|
||||
preserveEffect<ExperimentalPersistentPtyUpdateOutput>()(
|
||||
raw["persistentPty.update"]({
|
||||
params: { ptyID: input["ptyID"] },
|
||||
payload: { attachmentID: input["attachmentID"], size: input["size"] },
|
||||
}).pipe(
|
||||
Effect.mapError(mapClientError),
|
||||
Effect.map((value) => value.data),
|
||||
),
|
||||
)
|
||||
|
||||
const EndpointExperimentalPersistentPtySnapshot =
|
||||
(raw: RawClient["server.experimental"]) => (input: ExperimentalPersistentPtySnapshotInput) =>
|
||||
preserveEffect<ExperimentalPersistentPtySnapshotOutput>()(
|
||||
raw["persistentPty.snapshot"]({ params: { ptyID: input["ptyID"] } }).pipe(
|
||||
Effect.mapError(mapClientError),
|
||||
Effect.map((value) => value.data),
|
||||
),
|
||||
)
|
||||
|
||||
const EndpointExperimentalPersistentPtyRemove =
|
||||
(raw: RawClient["server.experimental"]) => (input: ExperimentalPersistentPtyRemoveInput) =>
|
||||
preserveEffect<ExperimentalPersistentPtyRemoveOutput>()(
|
||||
raw["persistentPty.remove"]({ params: { ptyID: input["ptyID"] } }).pipe(Effect.mapError(mapClientError)),
|
||||
)
|
||||
|
||||
const EndpointExperimentalPersistentPtyConnectToken =
|
||||
(raw: RawClient["server.experimental"]) => (input: ExperimentalPersistentPtyConnectTokenInput) =>
|
||||
preserveEffect<ExperimentalPersistentPtyConnectTokenOutput>()(
|
||||
raw["persistentPty.connectToken"]({
|
||||
params: { ptyID: input["ptyID"] },
|
||||
headers: { "x-opencode-ticket": input["x-opencode-ticket"] },
|
||||
}).pipe(
|
||||
Effect.mapError(mapClientError),
|
||||
Effect.map((value) => value.data),
|
||||
),
|
||||
)
|
||||
|
||||
const adaptGroupExperimental = (raw: RawClient["server.experimental"]) => ({
|
||||
persistentPty: {
|
||||
list: EndpointExperimentalPersistentPtyList(raw),
|
||||
create: EndpointExperimentalPersistentPtyCreate(raw),
|
||||
shutdown: EndpointExperimentalPersistentPtyShutdown(raw),
|
||||
get: EndpointExperimentalPersistentPtyGet(raw),
|
||||
update: EndpointExperimentalPersistentPtyUpdate(raw),
|
||||
snapshot: EndpointExperimentalPersistentPtySnapshot(raw),
|
||||
remove: EndpointExperimentalPersistentPtyRemove(raw),
|
||||
connectToken: EndpointExperimentalPersistentPtyConnectToken(raw),
|
||||
},
|
||||
})
|
||||
|
||||
const EndpointShellList = (raw: RawClient["server.shell"]) => (input?: ShellListInput) =>
|
||||
preserveEffect<ShellListOutput>()(
|
||||
raw["shell.list"]({ query: { location: input?.["location"] } }).pipe(Effect.mapError(mapClientError)),
|
||||
|
|
@ -1404,6 +1513,7 @@ const adaptClient = (raw: RawClient) => ({
|
|||
skill: adaptGroupSkill(raw["server.skill"]),
|
||||
event: adaptGroupEvent(raw["server.event"]),
|
||||
pty: adaptGroupPty(raw["server.pty"]),
|
||||
experimental: adaptGroupExperimental(raw["server.experimental"]),
|
||||
shell: adaptGroupShell(raw["server.shell"]),
|
||||
reference: adaptGroupReference(raw["server.reference"]),
|
||||
worktree: adaptGroupWorktree(raw["server.worktree"]),
|
||||
|
|
|
|||
|
|
@ -190,6 +190,21 @@ import type {
|
|||
PtyRemoveOutput,
|
||||
PtyConnectTokenInput,
|
||||
PtyConnectTokenOutput,
|
||||
ExperimentalPersistentPtyListInput,
|
||||
ExperimentalPersistentPtyListOutput,
|
||||
ExperimentalPersistentPtyCreateInput,
|
||||
ExperimentalPersistentPtyCreateOutput,
|
||||
ExperimentalPersistentPtyShutdownOutput,
|
||||
ExperimentalPersistentPtyGetInput,
|
||||
ExperimentalPersistentPtyGetOutput,
|
||||
ExperimentalPersistentPtyUpdateInput,
|
||||
ExperimentalPersistentPtyUpdateOutput,
|
||||
ExperimentalPersistentPtySnapshotInput,
|
||||
ExperimentalPersistentPtySnapshotOutput,
|
||||
ExperimentalPersistentPtyRemoveInput,
|
||||
ExperimentalPersistentPtyRemoveOutput,
|
||||
ExperimentalPersistentPtyConnectTokenInput,
|
||||
ExperimentalPersistentPtyConnectTokenOutput,
|
||||
ShellListInput,
|
||||
ShellListOutput,
|
||||
ShellCreateInput,
|
||||
|
|
@ -1637,6 +1652,108 @@ export function make(options: ClientOptions) {
|
|||
),
|
||||
},
|
||||
},
|
||||
experimental: {
|
||||
persistentPty: {
|
||||
list: (input: ExperimentalPersistentPtyListInput, requestOptions?: RequestOptions) =>
|
||||
request<{ readonly data: ExperimentalPersistentPtyListOutput }>(
|
||||
{
|
||||
method: "GET",
|
||||
path: `/api/experimental/session/${encodeURIComponent(input.sessionID)}/terminal`,
|
||||
successStatus: 200,
|
||||
declaredStatuses: [400, 503, 401],
|
||||
empty: false,
|
||||
},
|
||||
requestOptions,
|
||||
).then((value) => value.data),
|
||||
create: (input: ExperimentalPersistentPtyCreateInput, requestOptions?: RequestOptions) =>
|
||||
request<{ readonly data: ExperimentalPersistentPtyCreateOutput }>(
|
||||
{
|
||||
method: "POST",
|
||||
path: `/api/experimental/session/${encodeURIComponent(input.sessionID)}/terminal`,
|
||||
body: {
|
||||
command: input["command"],
|
||||
args: input["args"],
|
||||
cwd: input["cwd"],
|
||||
title: input["title"],
|
||||
env: input["env"],
|
||||
size: input["size"],
|
||||
},
|
||||
successStatus: 200,
|
||||
declaredStatuses: [400, 503, 401],
|
||||
empty: false,
|
||||
},
|
||||
requestOptions,
|
||||
).then((value) => value.data),
|
||||
shutdown: (requestOptions?: RequestOptions) =>
|
||||
request<ExperimentalPersistentPtyShutdownOutput>(
|
||||
{
|
||||
method: "POST",
|
||||
path: `/api/experimental/persistent-pty/shutdown`,
|
||||
successStatus: 204,
|
||||
declaredStatuses: [503, 401, 400],
|
||||
empty: true,
|
||||
},
|
||||
requestOptions,
|
||||
),
|
||||
get: (input: ExperimentalPersistentPtyGetInput, requestOptions?: RequestOptions) =>
|
||||
request<{ readonly data: ExperimentalPersistentPtyGetOutput }>(
|
||||
{
|
||||
method: "GET",
|
||||
path: `/api/experimental/persistent-pty/${encodeURIComponent(input.ptyID)}`,
|
||||
successStatus: 200,
|
||||
declaredStatuses: [404, 503, 401, 400],
|
||||
empty: false,
|
||||
},
|
||||
requestOptions,
|
||||
).then((value) => value.data),
|
||||
update: (input: ExperimentalPersistentPtyUpdateInput, requestOptions?: RequestOptions) =>
|
||||
request<{ readonly data: ExperimentalPersistentPtyUpdateOutput }>(
|
||||
{
|
||||
method: "PUT",
|
||||
path: `/api/experimental/persistent-pty/${encodeURIComponent(input.ptyID)}`,
|
||||
body: { attachmentID: input["attachmentID"], size: input["size"] },
|
||||
successStatus: 200,
|
||||
declaredStatuses: [404, 503, 401, 400],
|
||||
empty: false,
|
||||
},
|
||||
requestOptions,
|
||||
).then((value) => value.data),
|
||||
snapshot: (input: ExperimentalPersistentPtySnapshotInput, requestOptions?: RequestOptions) =>
|
||||
request<{ readonly data: ExperimentalPersistentPtySnapshotOutput }>(
|
||||
{
|
||||
method: "GET",
|
||||
path: `/api/experimental/persistent-pty/${encodeURIComponent(input.ptyID)}/snapshot`,
|
||||
successStatus: 200,
|
||||
declaredStatuses: [404, 503, 401, 400],
|
||||
empty: false,
|
||||
},
|
||||
requestOptions,
|
||||
).then((value) => value.data),
|
||||
remove: (input: ExperimentalPersistentPtyRemoveInput, requestOptions?: RequestOptions) =>
|
||||
request<ExperimentalPersistentPtyRemoveOutput>(
|
||||
{
|
||||
method: "DELETE",
|
||||
path: `/api/experimental/persistent-pty/${encodeURIComponent(input.ptyID)}`,
|
||||
successStatus: 204,
|
||||
declaredStatuses: [404, 503, 401, 400],
|
||||
empty: true,
|
||||
},
|
||||
requestOptions,
|
||||
),
|
||||
connectToken: (input: ExperimentalPersistentPtyConnectTokenInput, requestOptions?: RequestOptions) =>
|
||||
request<{ readonly data: ExperimentalPersistentPtyConnectTokenOutput }>(
|
||||
{
|
||||
method: "POST",
|
||||
path: `/api/experimental/persistent-pty/${encodeURIComponent(input.ptyID)}/connect-token`,
|
||||
headers: { "x-opencode-ticket": input["x-opencode-ticket"] },
|
||||
successStatus: 200,
|
||||
declaredStatuses: [403, 404, 503, 401, 400],
|
||||
empty: false,
|
||||
},
|
||||
requestOptions,
|
||||
).then((value) => value.data),
|
||||
},
|
||||
},
|
||||
shell: {
|
||||
list: (input?: ShellListInput, requestOptions?: RequestOptions) =>
|
||||
request<ShellListOutput>(
|
||||
|
|
|
|||
|
|
@ -338,6 +338,21 @@ export type Pty = {
|
|||
exitCode?: number
|
||||
}
|
||||
|
||||
export type PersistentPtyInfo = {
|
||||
id: string
|
||||
title: string
|
||||
command: string
|
||||
args: Array<string>
|
||||
cwd: string
|
||||
status: "running" | "exited"
|
||||
pid: number
|
||||
exitCode?: number
|
||||
sessionID: string
|
||||
foregroundProcess: string | null
|
||||
size: { cols: number; rows: number }
|
||||
output: { head: number; tail: number }
|
||||
}
|
||||
|
||||
export type FormMetadata1 = { [x: string]: any }
|
||||
|
||||
export type FormWhen1 = { key: string; op: "eq" | "neq"; value: string | number | boolean }
|
||||
|
|
@ -998,6 +1013,15 @@ export type PtyDeleted = {
|
|||
data: { id: string }
|
||||
}
|
||||
|
||||
export type PersistentPtyRemoved = {
|
||||
id: string
|
||||
created: number
|
||||
metadata?: { [x: string]: any }
|
||||
type: "persistent-pty.removed"
|
||||
location?: LocationRef
|
||||
data: { sessionID: string; ptyID: string }
|
||||
}
|
||||
|
||||
export type ShellExited = {
|
||||
id: string
|
||||
created: number
|
||||
|
|
@ -1439,6 +1463,22 @@ export type PtyUpdated = {
|
|||
data: { info: Pty }
|
||||
}
|
||||
|
||||
export type PersistentPtyAdded = {
|
||||
id: string
|
||||
created: number
|
||||
metadata?: { [x: string]: any }
|
||||
type: "persistent-pty.added"
|
||||
location?: LocationRef
|
||||
data: { sessionID: string; terminal: PersistentPtyInfo }
|
||||
}
|
||||
|
||||
export type PersistentPtySnapshot = {
|
||||
info: PersistentPtyInfo
|
||||
text: string
|
||||
checkpoint: string
|
||||
cursor: { x: number; y: number }
|
||||
}
|
||||
|
||||
export type FormStringField1 = {
|
||||
key: string
|
||||
title?: string
|
||||
|
|
@ -2133,6 +2173,8 @@ export type V2Event =
|
|||
| PtyUpdated
|
||||
| PtyExited
|
||||
| PtyDeleted
|
||||
| PersistentPtyAdded
|
||||
| PersistentPtyRemoved
|
||||
| ShellCreated
|
||||
| ShellExited
|
||||
| ShellDeleted
|
||||
|
|
@ -5500,6 +5542,99 @@ export type PtyConnectTokenOutput = {
|
|||
data: PtyTicketConnectToken
|
||||
}
|
||||
|
||||
export type ExperimentalPersistentPtyListInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] }
|
||||
|
||||
export type ExperimentalPersistentPtyListOutput = { data: Array<PersistentPtyInfo> }["data"]
|
||||
|
||||
export type ExperimentalPersistentPtyCreateInput = {
|
||||
readonly sessionID: { readonly sessionID: string }["sessionID"]
|
||||
readonly command: {
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly title: string
|
||||
readonly env: { readonly [x: string]: string }
|
||||
readonly size?: { readonly cols: number; readonly rows: number }
|
||||
}["command"]
|
||||
readonly args: {
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly title: string
|
||||
readonly env: { readonly [x: string]: string }
|
||||
readonly size?: { readonly cols: number; readonly rows: number }
|
||||
}["args"]
|
||||
readonly cwd: {
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly title: string
|
||||
readonly env: { readonly [x: string]: string }
|
||||
readonly size?: { readonly cols: number; readonly rows: number }
|
||||
}["cwd"]
|
||||
readonly title: {
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly title: string
|
||||
readonly env: { readonly [x: string]: string }
|
||||
readonly size?: { readonly cols: number; readonly rows: number }
|
||||
}["title"]
|
||||
readonly env: {
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly title: string
|
||||
readonly env: { readonly [x: string]: string }
|
||||
readonly size?: { readonly cols: number; readonly rows: number }
|
||||
}["env"]
|
||||
readonly size?: {
|
||||
readonly command: string
|
||||
readonly args: ReadonlyArray<string>
|
||||
readonly cwd: string
|
||||
readonly title: string
|
||||
readonly env: { readonly [x: string]: string }
|
||||
readonly size?: { readonly cols: number; readonly rows: number }
|
||||
}["size"]
|
||||
}
|
||||
|
||||
export type ExperimentalPersistentPtyCreateOutput = { data: PersistentPtyInfo }["data"]
|
||||
|
||||
export type ExperimentalPersistentPtyShutdownOutput = void
|
||||
|
||||
export type ExperimentalPersistentPtyGetInput = { readonly ptyID: { readonly ptyID: string }["ptyID"] }
|
||||
|
||||
export type ExperimentalPersistentPtyGetOutput = { data: PersistentPtyInfo }["data"]
|
||||
|
||||
export type ExperimentalPersistentPtyUpdateInput = {
|
||||
readonly ptyID: { readonly ptyID: string }["ptyID"]
|
||||
readonly attachmentID?: {
|
||||
readonly attachmentID?: string
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
}["attachmentID"]
|
||||
readonly size: {
|
||||
readonly attachmentID?: string
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
}["size"]
|
||||
}
|
||||
|
||||
export type ExperimentalPersistentPtyUpdateOutput = { data: PersistentPtyInfo }["data"]
|
||||
|
||||
export type ExperimentalPersistentPtySnapshotInput = { readonly ptyID: { readonly ptyID: string }["ptyID"] }
|
||||
|
||||
export type ExperimentalPersistentPtySnapshotOutput = { data: PersistentPtySnapshot }["data"]
|
||||
|
||||
export type ExperimentalPersistentPtyRemoveInput = { readonly ptyID: { readonly ptyID: string }["ptyID"] }
|
||||
|
||||
export type ExperimentalPersistentPtyRemoveOutput = void
|
||||
|
||||
export type ExperimentalPersistentPtyConnectTokenInput = {
|
||||
readonly ptyID: { readonly ptyID: string }["ptyID"]
|
||||
readonly "x-opencode-ticket"?: { readonly "x-opencode-ticket"?: string | undefined }["x-opencode-ticket"]
|
||||
}
|
||||
|
||||
export type ExperimentalPersistentPtyConnectTokenOutput = { data: PtyTicketConnectToken }["data"]
|
||||
|
||||
export type ShellListInput = {
|
||||
readonly location?: {
|
||||
readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined
|
||||
|
|
|
|||
1
packages/core/src/persistent-pty.ts
Normal file
1
packages/core/src/persistent-pty.ts
Normal file
|
|
@ -0,0 +1 @@
|
|||
export { PersistentPty } from "./persistent-pty/index.js"
|
||||
535
packages/core/src/persistent-pty/daemon.ts
Normal file
535
packages/core/src/persistent-pty/daemon.ts
Normal file
|
|
@ -0,0 +1,535 @@
|
|||
import { spawn } from "node:child_process"
|
||||
import { readFile } from "node:fs/promises"
|
||||
import net from "node:net"
|
||||
import path from "node:path"
|
||||
import { Data, Duration, Effect, Schema, Semaphore } from "effect"
|
||||
|
||||
const ProtocolVersion = 6
|
||||
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 }),
|
||||
])
|
||||
|
||||
export const WireTerminal = Schema.Struct({
|
||||
id: Schema.Number,
|
||||
pid: Schema.NullOr(Schema.Number),
|
||||
title: Schema.String,
|
||||
foreground_process: Schema.NullOr(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,
|
||||
})
|
||||
export type WireTerminal = typeof WireTerminal.Type
|
||||
|
||||
const Registration = Schema.Struct({
|
||||
instance_id: Schema.String,
|
||||
pid: Schema.Number,
|
||||
protocol: Schema.Number,
|
||||
socket: Schema.String,
|
||||
token: Schema.String,
|
||||
})
|
||||
type Registration = typeof Registration.Type
|
||||
|
||||
export const WireResponse = 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("resized"),
|
||||
cols: Schema.Number,
|
||||
rows: Schema.Number,
|
||||
generation: Schema.Number,
|
||||
checkpoint_base64: Schema.String,
|
||||
}),
|
||||
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("title_changed"), title: Schema.String }),
|
||||
Schema.Struct({ type: Schema.Literal("foreground_process_changed"), process: Schema.NullOr(Schema.String) }),
|
||||
Schema.Struct({ type: Schema.Literal("error"), message: Schema.String }),
|
||||
])
|
||||
export type WireResponse = typeof WireResponse.Type
|
||||
|
||||
export type Role = "controller" | "observer"
|
||||
|
||||
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 checkpoint: Uint8Array
|
||||
}
|
||||
| { readonly type: "exited"; readonly exitCode?: number; readonly finalOffset: number }
|
||||
| { readonly type: "controller_changed"; readonly attachmentID?: string; readonly generation: number }
|
||||
| { readonly type: "title_changed"; readonly title: string }
|
||||
| { readonly type: "foreground_process_changed"; readonly process: string | null }
|
||||
|
||||
export type DaemonAttachment = {
|
||||
readonly terminal: WireTerminal
|
||||
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 DaemonError extends Data.TaggedError("PersistentPty.DaemonError")<{
|
||||
readonly kind: "connect" | "response" | "registration" | "protocol" | "spawn"
|
||||
readonly message: string
|
||||
readonly pid?: number
|
||||
}> {}
|
||||
|
||||
export interface DaemonTransport {
|
||||
readonly request: (value: object, start?: boolean) => Effect.Effect<WireResponse, DaemonError>
|
||||
readonly requestIfRunning: (value: object) => Effect.Effect<WireResponse | undefined, DaemonError>
|
||||
readonly shutdown: Effect.Effect<WireResponse | undefined, DaemonError>
|
||||
readonly subscribe: (
|
||||
id: number,
|
||||
input: {
|
||||
readonly cursor: number
|
||||
readonly attachmentID: string
|
||||
readonly role: Role
|
||||
readonly takeover?: boolean
|
||||
readonly onEvent: (event: StreamEvent) => void
|
||||
readonly onEnd: () => void
|
||||
},
|
||||
) => Effect.Effect<DaemonAttachment, DaemonError>
|
||||
}
|
||||
|
||||
export const makeDaemonTransport = Effect.fn("PersistentPty.makeDaemonTransport")(function* (
|
||||
directory: string,
|
||||
binary: () => Promise<string> = () => Promise.resolve(process.env.OPENCODE_PTY_BIN || "opencode-pty"),
|
||||
) {
|
||||
const startup = Semaphore.makeUnsafe(1)
|
||||
let registration: Registration | undefined
|
||||
|
||||
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),
|
||||
})
|
||||
if (decoded.protocol !== ProtocolVersion)
|
||||
return yield* Effect.fail(
|
||||
new DaemonError({
|
||||
kind: "protocol",
|
||||
message: `opencode-pty protocol mismatch: daemon=${decoded.protocol}, client=${ProtocolVersion}`,
|
||||
pid: decoded.pid,
|
||||
}),
|
||||
)
|
||||
const response = yield* oneShot(decoded, { op: "ping" })
|
||||
if (
|
||||
response.type !== "pong" ||
|
||||
response.instance_id !== decoded.instance_id ||
|
||||
response.pid !== decoded.pid ||
|
||||
response.protocol !== ProtocolVersion
|
||||
)
|
||||
return yield* Effect.fail(new DaemonError({ kind: "protocol", message: "opencode-pty registration mismatch" }))
|
||||
return decoded
|
||||
})
|
||||
|
||||
const start = Effect.fn("PersistentPty.daemon.start")(function* () {
|
||||
const executable = yield* Effect.tryPromise({ try: binary, catch: (cause) => failure("spawn", cause) })
|
||||
yield* Effect.tryPromise({
|
||||
try: () =>
|
||||
new Promise<void>((resolve, reject) => {
|
||||
const child = spawn(executable, ["daemon"], {
|
||||
detached: true,
|
||||
stdio: "ignore",
|
||||
env: { ...process.env, OPENCODE_PTY_RUNTIME_DIR: directory },
|
||||
})
|
||||
child.once("spawn", () => {
|
||||
child.unref()
|
||||
resolve()
|
||||
})
|
||||
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)
|
||||
}),
|
||||
)
|
||||
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" }),
|
||||
)
|
||||
})
|
||||
|
||||
const connect = Effect.fn("PersistentPty.daemon.connect")(function* (shouldStart: boolean) {
|
||||
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()))
|
||||
}),
|
||||
)
|
||||
registration = found
|
||||
return found
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
const attempt = Effect.fn("PersistentPty.daemon.request-attempt")(function* (value: object, shouldStart: boolean) {
|
||||
const current = yield* connect(shouldStart)
|
||||
return yield* oneShot(current, value).pipe(
|
||||
Effect.catch((error) => {
|
||||
if (error.kind === "registration" && registration === current) registration = undefined
|
||||
return Effect.fail(error)
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
const request = Effect.fn("PersistentPty.daemon.request")(function* (value: object, shouldStart = false) {
|
||||
return yield* attempt(value, shouldStart).pipe(
|
||||
Effect.catch((error) => {
|
||||
if (error.kind === "registration") return attempt(value, shouldStart)
|
||||
if (error.kind !== "connect") return Effect.fail(error)
|
||||
registration = undefined
|
||||
if (!shouldStart) return Effect.fail(error)
|
||||
return attempt(value, true)
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
const requestIfRunning = (value: object) =>
|
||||
request(value).pipe(
|
||||
Effect.catch((error) => (error.kind === "connect" ? Effect.succeed(undefined) : Effect.fail(error))),
|
||||
)
|
||||
|
||||
const shutdown = Effect.gen(function* () {
|
||||
const response = yield* requestIfRunning({ op: "shutdown" })
|
||||
registration = undefined
|
||||
if (!response) return undefined
|
||||
const deadline = Date.now() + 5_000
|
||||
while (Date.now() < deadline) {
|
||||
const running = yield* discover().pipe(
|
||||
Effect.as(true),
|
||||
Effect.orElseSucceed(() => false),
|
||||
)
|
||||
if (!running) return response
|
||||
yield* Effect.sleep(50)
|
||||
}
|
||||
return yield* Effect.fail(new DaemonError({ kind: "connect", message: "opencode-pty did not stop" }))
|
||||
})
|
||||
|
||||
const subscribe = Effect.fn("PersistentPty.daemon.subscribe")(function* (
|
||||
id: number,
|
||||
input: Parameters<DaemonTransport["subscribe"]>[1],
|
||||
) {
|
||||
const attempt = Effect.gen(function* () {
|
||||
const current = yield* connect(false)
|
||||
return yield* Effect.tryPromise({
|
||||
try: () => subscribePromise(current, id, input),
|
||||
catch: (cause) => (cause instanceof DaemonError ? cause : failure("connect", cause)),
|
||||
}).pipe(
|
||||
Effect.catch((error) => {
|
||||
if (error.kind === "registration" && registration === current) registration = undefined
|
||||
return Effect.fail(error)
|
||||
}),
|
||||
)
|
||||
})
|
||||
return yield* attempt.pipe(Effect.catch((error) => (error.kind === "registration" ? attempt : Effect.fail(error))))
|
||||
})
|
||||
|
||||
return { request, requestIfRunning, shutdown, subscribe } satisfies DaemonTransport
|
||||
})
|
||||
|
||||
const oneShot = Effect.fn("PersistentPty.daemon.oneShot")(function* (registration: Registration, request: object) {
|
||||
const payload = yield* Effect.try({
|
||||
try: () => encode({ token: registration.token, request }),
|
||||
catch: (cause) => failure("protocol", cause),
|
||||
})
|
||||
let dispatched = false
|
||||
return yield* Effect.acquireUseRelease(
|
||||
Effect.tryPromise({
|
||||
try: (signal) =>
|
||||
new Promise<net.Socket>((resolve, reject) => {
|
||||
const socket = net.createConnection({ path: registration.socket, signal })
|
||||
socket.once("connect", () => resolve(socket))
|
||||
socket.once("error", reject)
|
||||
}),
|
||||
catch: (cause) => failure("connect", cause),
|
||||
}),
|
||||
(socket) =>
|
||||
Effect.gen(function* () {
|
||||
yield* Effect.try({
|
||||
try: () => {
|
||||
dispatched = true
|
||||
socket.write(payload)
|
||||
},
|
||||
catch: (cause) => failure("response", cause),
|
||||
})
|
||||
const first = yield* Effect.tryPromise({
|
||||
try: async (signal) => {
|
||||
const frames = decoder(socket)
|
||||
const abort = () => socket.destroy()
|
||||
signal.addEventListener("abort", abort, { once: true })
|
||||
try {
|
||||
const frame = await frames.next()
|
||||
if (frame.done) throw new Error("opencode-pty closed without response")
|
||||
return frame.value
|
||||
} finally {
|
||||
signal.removeEventListener("abort", abort)
|
||||
}
|
||||
},
|
||||
catch: (cause) => failure("response", cause),
|
||||
})
|
||||
const response = yield* Effect.try({
|
||||
try: () => decode(first),
|
||||
catch: (cause) => failure("protocol", cause),
|
||||
})
|
||||
if (response.type === "error")
|
||||
return yield* Effect.fail(
|
||||
new DaemonError({
|
||||
kind: response.message === "authentication failed" ? "registration" : "protocol",
|
||||
message: response.message,
|
||||
}),
|
||||
)
|
||||
return response
|
||||
}),
|
||||
(socket) => Effect.sync(() => socket.destroy()),
|
||||
).pipe(
|
||||
Effect.timeoutOrElse({
|
||||
duration: Duration.seconds(5),
|
||||
orElse: () =>
|
||||
Effect.fail(
|
||||
new DaemonError({
|
||||
kind: dispatched ? "response" : "connect",
|
||||
message: "opencode-pty request timed out",
|
||||
}),
|
||||
),
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
async function subscribePromise(
|
||||
registration: Registration,
|
||||
id: number,
|
||||
input: Parameters<DaemonTransport["subscribe"]>[1],
|
||||
): Promise<DaemonAttachment> {
|
||||
const socket = net.createConnection(registration.socket)
|
||||
const frames = decoder(socket)
|
||||
try {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
socket.once("connect", resolve)
|
||||
socket.once("error", reject)
|
||||
})
|
||||
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 DaemonError({
|
||||
kind: response.message === "authentication failed" ? "registration" : "protocol",
|
||||
message: 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) {
|
||||
if (frame[0] === 0) {
|
||||
if (frame.length < 17) throw new Error("invalid opencode-pty output frame")
|
||||
input.onEvent({
|
||||
type: "output",
|
||||
start: Number(frame.readBigUInt64BE(1)),
|
||||
end: Number(frame.readBigUInt64BE(9)),
|
||||
data: frame.subarray(17),
|
||||
})
|
||||
continue
|
||||
}
|
||||
const event = decode(frame)
|
||||
if (event.type === "resized")
|
||||
input.onEvent({
|
||||
type: "resized",
|
||||
cols: event.cols,
|
||||
rows: event.rows,
|
||||
generation: event.generation,
|
||||
checkpoint: Buffer.from(event.checkpoint_base64, "base64"),
|
||||
})
|
||||
if (event.type === "controller_changed")
|
||||
input.onEvent({
|
||||
type: "controller_changed",
|
||||
attachmentID: event.attachment_id ?? undefined,
|
||||
generation: event.generation,
|
||||
})
|
||||
if (event.type === "title_changed") input.onEvent({ type: "title_changed", title: event.title })
|
||||
if (event.type === "foreground_process_changed")
|
||||
input.onEvent({ type: "foreground_process_changed", process: event.process })
|
||||
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 {
|
||||
terminal: 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()
|
||||
},
|
||||
}
|
||||
} catch (error) {
|
||||
socket.destroy()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
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(WireResponse)(JSON.parse(Buffer.from(payload).toString("utf8")))
|
||||
}
|
||||
|
||||
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")
|
||||
}
|
||||
386
packages/core/src/persistent-pty/index.ts
Normal file
386
packages/core/src/persistent-pty/index.ts
Normal file
|
|
@ -0,0 +1,386 @@
|
|||
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 { 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 {
|
||||
makeDaemonTransport,
|
||||
type DaemonTransport,
|
||||
type Role,
|
||||
type StreamEvent,
|
||||
type WireResponse,
|
||||
type WireTerminal,
|
||||
} from "./daemon.js"
|
||||
|
||||
export type { Role, StreamEvent } from "./daemon.js"
|
||||
|
||||
export type Info = Pty.Info & {
|
||||
readonly sessionID: Session.ID
|
||||
readonly foregroundProcess: string | null
|
||||
readonly size: { readonly cols: number; readonly rows: number }
|
||||
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 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.TaggedError<UnavailableError>()("PersistentPty.UnavailableError", {
|
||||
message: Schema.String,
|
||||
}) {}
|
||||
|
||||
export class NotFoundError extends Schema.TaggedError<NotFoundError>()("PersistentPty.NotFoundError", {
|
||||
ptyID: Pty.ID,
|
||||
}) {}
|
||||
|
||||
export interface Interface {
|
||||
readonly list: (sessionID?: Session.ID) => Effect.Effect<Info[], UnavailableError>
|
||||
readonly get: (id: Pty.ID) => Effect.Effect<Info, NotFoundError | UnavailableError>
|
||||
readonly create: (
|
||||
sessionID: Session.ID,
|
||||
input: {
|
||||
readonly command: string
|
||||
readonly args: readonly string[]
|
||||
readonly cwd: string
|
||||
readonly title: string
|
||||
readonly env: Readonly<Record<string, string>>
|
||||
readonly cols?: number
|
||||
readonly rows?: number
|
||||
},
|
||||
) => Effect.Effect<Info, UnavailableError>
|
||||
readonly write: (
|
||||
id: Pty.ID,
|
||||
data: string,
|
||||
attachmentID?: string,
|
||||
) => Effect.Effect<void, NotFoundError | UnavailableError>
|
||||
readonly resize: (
|
||||
id: Pty.ID,
|
||||
cols: number,
|
||||
rows: number,
|
||||
attachmentID?: string,
|
||||
) => Effect.Effect<void, NotFoundError | UnavailableError>
|
||||
readonly control: (
|
||||
id: Pty.ID,
|
||||
attachmentID: string,
|
||||
cols: number,
|
||||
rows: number,
|
||||
) => Effect.Effect<void, NotFoundError | UnavailableError>
|
||||
readonly input: (
|
||||
id: Pty.ID,
|
||||
attachmentID: string,
|
||||
cols: number,
|
||||
rows: number,
|
||||
data: Uint8Array,
|
||||
) => Effect.Effect<void, NotFoundError | UnavailableError>
|
||||
readonly snapshot: (id: Pty.ID) => Effect.Effect<Snapshot, NotFoundError | UnavailableError>
|
||||
readonly remove: (id: Pty.ID) => Effect.Effect<void, NotFoundError | UnavailableError>
|
||||
readonly shutdown: () => Effect.Effect<void, UnavailableError>
|
||||
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<Attachment, NotFoundError | UnavailableError>
|
||||
}
|
||||
|
||||
export class Service extends Context.Service<Service, Interface>()("@opencode/PersistentPty") {}
|
||||
|
||||
export const layer = Layer.effect(
|
||||
Service,
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const database = yield* Database.Service
|
||||
const context = yield* Effect.context()
|
||||
const runFork = Effect.runForkWith(context)
|
||||
const daemon = yield* makeDaemonTransport(runtimeDirectory(databasePath(database.db)))
|
||||
const removing = new Set<Pty.ID>()
|
||||
|
||||
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 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<Record<string, string>>
|
||||
readonly cols?: number
|
||||
readonly rows?: number
|
||||
},
|
||||
) {
|
||||
const response = yield* request(
|
||||
daemon,
|
||||
{
|
||||
op: "create",
|
||||
program: input.command,
|
||||
args: input.args,
|
||||
cwd: input.cwd,
|
||||
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"),
|
||||
})
|
||||
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)
|
||||
},
|
||||
})
|
||||
.pipe(Effect.mapError(unavailable))
|
||||
return {
|
||||
info: toInfo(attachment.terminal),
|
||||
role: attachment.role,
|
||||
generation: attachment.generation,
|
||||
replay: attachment.replay,
|
||||
activate: attachment.activate,
|
||||
detach: attachment.detach,
|
||||
}
|
||||
})
|
||||
|
||||
return Service.of({ list, get, create, write, resize, control, input, snapshot, remove, shutdown, attach })
|
||||
}),
|
||||
)
|
||||
|
||||
export const node = makeGlobalNode({ service: Service, layer, deps: [Bus.node, Database.node] })
|
||||
|
||||
const request = (daemon: DaemonTransport, value: object, start = false) =>
|
||||
daemon.request(value, start).pipe(Effect.mapError(unavailable))
|
||||
|
||||
const optionalRequest = (daemon: DaemonTransport, value: object) =>
|
||||
daemon.requestIfRunning(value).pipe(Effect.mapError(unavailable))
|
||||
|
||||
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))
|
||||
}
|
||||
|
||||
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 } : {}),
|
||||
}),
|
||||
sessionID: Session.ID.make(value.group_id),
|
||||
foregroundProcess: value.foreground_process,
|
||||
size: { cols: value.cols, rows: value.rows },
|
||||
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
|
||||
}
|
||||
277
packages/core/test/persistent-pty-daemon.test.ts
Normal file
277
packages/core/test/persistent-pty-daemon.test.ts
Normal file
|
|
@ -0,0 +1,277 @@
|
|||
import { expect, test } from "bun:test"
|
||||
import { spawn } from "node:child_process"
|
||||
import { mkdtemp, rm, writeFile } from "node:fs/promises"
|
||||
import net from "node:net"
|
||||
import os from "node:os"
|
||||
import path from "node:path"
|
||||
import { Effect } from "effect"
|
||||
import { makeDaemonTransport } from "../src/persistent-pty/daemon"
|
||||
|
||||
const pong = { type: "pong", instance_id: "test", pid: process.pid, protocol: 6 }
|
||||
|
||||
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" }))
|
||||
|
||||
token = "new-token"
|
||||
instance = "new-instance"
|
||||
await writeRegistration(directory, socketPath, instance, token)
|
||||
|
||||
const running = await Effect.runPromise(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))
|
||||
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" }))
|
||||
|
||||
token = "new-token"
|
||||
instance = "new-instance"
|
||||
await writeRegistration(directory, socketPath, instance, token)
|
||||
|
||||
const attachment = await Effect.runPromise(
|
||||
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)
|
||||
|
||||
let creates = 0
|
||||
const second = await 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 })
|
||||
}
|
||||
})
|
||||
|
||||
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)))
|
||||
|
||||
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<void>((resolve) => existing.once("exit", () => resolve()))
|
||||
try {
|
||||
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")),
|
||||
)
|
||||
|
||||
const optional = await Effect.runPromise(Effect.flip(daemon.requestIfRunning({ op: "list" })))
|
||||
expect(optional).toMatchObject({
|
||||
kind: "protocol",
|
||||
message: "opencode-pty protocol mismatch: daemon=5, client=6",
|
||||
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 })
|
||||
}
|
||||
})
|
||||
|
||||
function listen(
|
||||
socketPath: string,
|
||||
handle: (socket: net.Socket, request: Record<string, unknown>, 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))
|
||||
})
|
||||
.catch(() => socket.destroy())
|
||||
})
|
||||
return new Promise<net.Server>((resolve, reject) => {
|
||||
server.once("error", reject)
|
||||
server.listen(socketPath, () => resolve(server))
|
||||
})
|
||||
}
|
||||
|
||||
function readRequest(socket: net.Socket) {
|
||||
return new Promise<Buffer>((resolve, reject) => {
|
||||
let pending = Buffer.alloc(0)
|
||||
const cleanup = () => {
|
||||
socket.off("data", onData)
|
||||
socket.off("end", onEnd)
|
||||
socket.off("error", onError)
|
||||
}
|
||||
const onData = (value: Buffer) => {
|
||||
pending = Buffer.concat([pending, value])
|
||||
if (pending.length < 4) return
|
||||
const length = pending.readUInt32BE(0)
|
||||
if (pending.length < length + 4) return
|
||||
cleanup()
|
||||
resolve(pending.subarray(4, length + 4))
|
||||
}
|
||||
const onEnd = () => {
|
||||
cleanup()
|
||||
reject(new Error("connection closed before request"))
|
||||
}
|
||||
const onError = (error: Error) => {
|
||||
cleanup()
|
||||
reject(error)
|
||||
}
|
||||
socket.on("data", onData)
|
||||
socket.on("end", onEnd)
|
||||
socket.on("error", onError)
|
||||
}).then((payload) => {
|
||||
const envelope: unknown = JSON.parse(payload.toString("utf8"))
|
||||
if (!isRecord(envelope) || typeof envelope.token !== "string" || !isRecord(envelope.request))
|
||||
throw new Error("Invalid test daemon envelope")
|
||||
return { token: envelope.token, request: envelope.request }
|
||||
})
|
||||
}
|
||||
|
||||
function frame(value: object) {
|
||||
const payload = Buffer.from(JSON.stringify(value))
|
||||
const output = Buffer.allocUnsafe(payload.length + 4)
|
||||
output.writeUInt32BE(payload.length)
|
||||
payload.copy(output, 4)
|
||||
return output
|
||||
}
|
||||
|
||||
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 }),
|
||||
)
|
||||
}
|
||||
|
||||
function terminal(id: number) {
|
||||
return {
|
||||
id,
|
||||
pid: null,
|
||||
title: "test",
|
||||
foreground_process: null,
|
||||
group_id: "test",
|
||||
command: ["test"],
|
||||
cwd: "/tmp",
|
||||
cols: 80,
|
||||
rows: 24,
|
||||
lifecycle: { status: "running" },
|
||||
output_head: 0,
|
||||
output_tail: 0,
|
||||
}
|
||||
}
|
||||
|
||||
function close(server: net.Server) {
|
||||
if (!server.listening) return Promise.resolve()
|
||||
return new Promise<void>((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value)
|
||||
}
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -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"
|
||||
|
|
@ -88,6 +89,7 @@ type ApiGroups<
|
|||
| typeof WorktreeGroup
|
||||
| typeof WorkspaceGroup
|
||||
| typeof GenerateGroup
|
||||
| typeof PersistentPtyGroup
|
||||
| LocationGroups<LocationId>
|
||||
| FormGroups<LocationId, LocationService, FormLocationId, FormLocationService>
|
||||
| SessionGroups<SessionLocationId, SessionLocationService>
|
||||
|
|
@ -168,6 +170,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)
|
||||
|
|
|
|||
|
|
@ -55,6 +55,7 @@ export const groupNames = {
|
|||
"server.skill": "skill",
|
||||
"server.event": "event",
|
||||
"server.pty": "pty",
|
||||
"server.experimental": "experimental",
|
||||
"server.shell": "shell",
|
||||
"server.mcp": "mcp",
|
||||
"server.reference": "reference",
|
||||
|
|
@ -65,5 +66,5 @@ export const groupNames = {
|
|||
"server.config": "config",
|
||||
} as const
|
||||
|
||||
export const promiseOmitEndpoints = new Set(["pty.connect"])
|
||||
export const effectOmitEndpoints = new Set(["fs.read", "pty.connect"])
|
||||
export const promiseOmitEndpoints = new Set(["pty.connect", "persistentPty.connect"])
|
||||
export const effectOmitEndpoints = new Set(["fs.read", "pty.connect", "persistentPty.connect"])
|
||||
|
|
|
|||
103
packages/protocol/src/groups/persistent-pty.ts
Normal file
103
packages/protocol/src/groups/persistent-pty.ts
Normal file
|
|
@ -0,0 +1,103 @@
|
|||
import { PersistentPty } from "@opencode-ai/schema/persistent-pty"
|
||||
import { Pty } from "@opencode-ai/schema/pty"
|
||||
import { PtyTicket } from "@opencode-ai/schema/pty-ticket"
|
||||
import { Session } from "@opencode-ai/schema/session"
|
||||
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\/experimental\/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.experimental")
|
||||
.add(
|
||||
HttpApiEndpoint.get("persistentPty.list", "/api/experimental/session/:sessionID/terminal", {
|
||||
params: { sessionID: Session.ID },
|
||||
success: Schema.Struct({ data: Schema.Array(PersistentPty.Info) }),
|
||||
error: errors,
|
||||
}),
|
||||
)
|
||||
.add(
|
||||
HttpApiEndpoint.post("persistentPty.create", "/api/experimental/session/:sessionID/terminal", {
|
||||
params: { sessionID: Session.ID },
|
||||
payload: PersistentPty.CreateInput,
|
||||
success: Schema.Struct({ data: PersistentPty.Info }),
|
||||
error: errors,
|
||||
}),
|
||||
)
|
||||
.add(
|
||||
HttpApiEndpoint.post("persistentPty.shutdown", "/api/experimental/persistent-pty/shutdown", {
|
||||
success: HttpApiSchema.NoContent,
|
||||
error: [ServiceUnavailableError],
|
||||
}),
|
||||
)
|
||||
.add(
|
||||
HttpApiEndpoint.get("persistentPty.get", "/api/experimental/persistent-pty/:ptyID", {
|
||||
params: { ptyID: Pty.ID },
|
||||
success: Schema.Struct({ data: PersistentPty.Info }),
|
||||
error: terminalErrors,
|
||||
}),
|
||||
)
|
||||
.add(
|
||||
HttpApiEndpoint.put("persistentPty.update", "/api/experimental/persistent-pty/:ptyID", {
|
||||
params: { ptyID: Pty.ID },
|
||||
payload: PersistentPty.UpdateInput,
|
||||
success: Schema.Struct({ data: PersistentPty.Info }),
|
||||
error: terminalErrors,
|
||||
}),
|
||||
)
|
||||
.add(
|
||||
HttpApiEndpoint.get("persistentPty.snapshot", "/api/experimental/persistent-pty/:ptyID/snapshot", {
|
||||
params: { ptyID: Pty.ID },
|
||||
success: Schema.Struct({ data: PersistentPty.Snapshot }),
|
||||
error: terminalErrors,
|
||||
}),
|
||||
)
|
||||
.add(
|
||||
HttpApiEndpoint.delete("persistentPty.remove", "/api/experimental/persistent-pty/:ptyID", {
|
||||
params: { ptyID: Pty.ID },
|
||||
success: HttpApiSchema.NoContent,
|
||||
error: terminalErrors,
|
||||
}),
|
||||
)
|
||||
.add(
|
||||
HttpApiEndpoint.post("persistentPty.connectToken", "/api/experimental/persistent-pty/:ptyID/connect-token", {
|
||||
params: { ptyID: Pty.ID },
|
||||
headers: Schema.Struct({ [PTY_CONNECT_TOKEN_HEADER]: Schema.optional(Schema.String) }),
|
||||
success: Schema.Struct({ data: PtyTicket.ConnectToken }),
|
||||
error: [ForbiddenError, PtyNotFoundError, ServiceUnavailableError],
|
||||
}),
|
||||
)
|
||||
.add(
|
||||
HttpApiEndpoint.get("persistentPty.connect", "/api/experimental/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,
|
||||
parameters: [
|
||||
...(operation.parameters ?? []),
|
||||
...["cursor", "role", "attachment_id", "takeover", "input_protocol", PTY_CONNECT_TICKET_QUERY].map(
|
||||
(name) => ({ in: "query", name, schema: { type: "string" } }),
|
||||
),
|
||||
],
|
||||
}),
|
||||
}),
|
||||
),
|
||||
)
|
||||
.annotateMerge(OpenApi.annotations({ title: "persistentPty", description: "Prototype persistent PTY routes." }))
|
||||
|
|
@ -17,6 +17,7 @@ import { LspEvent } from "./lsp-event.js"
|
|||
import { McpEvent } from "./mcp-event.js"
|
||||
import { ModelsDev } from "./models-dev.js"
|
||||
import { Permission } from "./permission.js"
|
||||
import { PersistentPty } from "./persistent-pty.js"
|
||||
import { Plugin } from "./plugin.js"
|
||||
import { Project } from "./project.js"
|
||||
import { Worktree } from "./worktree.js"
|
||||
|
|
@ -54,6 +55,7 @@ const featureDefinitions = Event.inventory(
|
|||
...Config.Event.Definitions,
|
||||
...Skill.Event.Definitions,
|
||||
...Pty.Event.Definitions,
|
||||
...PersistentPty.Event.Definitions,
|
||||
...Shell.Event.Definitions,
|
||||
...Form.Event.Definitions,
|
||||
...WebSearch.Event.Definitions,
|
||||
|
|
|
|||
|
|
@ -31,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"
|
||||
|
|
|
|||
44
packages/schema/src/persistent-pty.ts
Normal file
44
packages/schema/src/persistent-pty.ts
Normal file
|
|
@ -0,0 +1,44 @@
|
|||
export * as PersistentPty from "./persistent-pty.js"
|
||||
|
||||
import { Schema } from "effect"
|
||||
import { ephemeral, inventory } from "./event.js"
|
||||
import { Pty } from "./pty.js"
|
||||
import { NonNegativeInt, PositiveInt, optional } from "./schema.js"
|
||||
import { Session } from "./session.js"
|
||||
|
||||
export const Info = Schema.Struct({
|
||||
...Pty.Info.fields,
|
||||
sessionID: Session.ID,
|
||||
foregroundProcess: Schema.NullOr(Schema.String),
|
||||
size: Schema.Struct({ cols: PositiveInt, rows: PositiveInt }),
|
||||
output: Schema.Struct({ head: NonNegativeInt, tail: NonNegativeInt }),
|
||||
}).annotate({ identifier: "PersistentPty.Info" })
|
||||
export interface Info extends Schema.Schema.Type<typeof Info> {}
|
||||
|
||||
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<typeof CreateInput> {}
|
||||
|
||||
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<typeof UpdateInput> {}
|
||||
|
||||
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<typeof Snapshot> {}
|
||||
|
||||
export const Added = ephemeral({ type: "persistent-pty.added", schema: { sessionID: Session.ID, terminal: Info } })
|
||||
export const Removed = ephemeral({ type: "persistent-pty.removed", schema: { sessionID: Session.ID, ptyID: Pty.ID } })
|
||||
export const Event = { Added, Removed, Definitions: inventory(Added, Removed) }
|
||||
|
|
@ -6,6 +6,7 @@ import {
|
|||
Form,
|
||||
Integration,
|
||||
Permission,
|
||||
PersistentPty,
|
||||
Project,
|
||||
Reference,
|
||||
Session,
|
||||
|
|
@ -63,6 +64,7 @@ describe("public event manifest", () => {
|
|||
expect(FileSystemV1.Event.Definitions).toEqual([FileSystemV1.Event.Edited])
|
||||
expect(Integration.Event.Definitions).toEqual([Integration.Event.Updated, Integration.Event.ConnectionUpdated])
|
||||
expect(Permission.Event.Definitions).toEqual([Permission.Event.Asked, Permission.Event.Replied])
|
||||
expect(PersistentPty.Event.Definitions).toEqual([PersistentPty.Event.Added, PersistentPty.Event.Removed])
|
||||
expect(Form.Event.Definitions).toEqual([Form.Event.Created, Form.Event.Replied, Form.Event.Cancelled])
|
||||
expect(Reference.Event.Definitions).toEqual([Reference.Event.Updated])
|
||||
expect(Plugin.Event.Definitions).toEqual([Plugin.Event.Added, Plugin.Event.Updated])
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
@ -56,6 +57,7 @@ export const handlers = Layer.mergeAll(
|
|||
SkillHandler,
|
||||
EventHandler.pipe(Layer.provide(EventFeed.layer)),
|
||||
PtyHandler,
|
||||
PersistentPtyHandler,
|
||||
ShellHandler,
|
||||
ReferenceHandler,
|
||||
WorktreeHandler,
|
||||
|
|
|
|||
224
packages/server/src/handlers/persistent-pty.ts
Normal file
224
packages/server/src/handlers/persistent-pty.ts
Normal file
|
|
@ -0,0 +1,224 @@
|
|||
import { PersistentPty } from "@opencode-ai/core/persistent-pty"
|
||||
import { PtyTicket } from "@opencode-ai/core/pty/ticket"
|
||||
import { ForbiddenError, 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, Semaphore } 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.experimental", (handlers) =>
|
||||
Effect.gen(function* () {
|
||||
const tickets = yield* PtyTicket.Service
|
||||
const cors = yield* CorsConfig
|
||||
const pty = yield* PersistentPty.Service
|
||||
|
||||
return handlers
|
||||
.handle(
|
||||
"persistentPty.list",
|
||||
Effect.fn(function* (ctx) {
|
||||
return { data: yield* pty.list(ctx.params.sessionID).pipe(mapUnavailable) }
|
||||
}),
|
||||
)
|
||||
.handle(
|
||||
"persistentPty.create",
|
||||
Effect.fn(function* (ctx) {
|
||||
return {
|
||||
data: yield* pty
|
||||
.create(ctx.params.sessionID, {
|
||||
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.catchTag("PersistentPty.UnavailableError", unavailable)),
|
||||
}
|
||||
}),
|
||||
)
|
||||
.handle(
|
||||
"persistentPty.shutdown",
|
||||
Effect.fn(function* () {
|
||||
yield* pty.shutdown().pipe(mapUnavailable)
|
||||
return HttpApiSchema.NoContent.make()
|
||||
}),
|
||||
)
|
||||
.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 framedInput = url.searchParams.get("input_protocol") === "1"
|
||||
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<string | Uint8Array | Socket.CloseEvent>()
|
||||
const input = yield* Semaphore.make(1)
|
||||
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 === "resized")
|
||||
Queue.offerUnsafe(
|
||||
outbox,
|
||||
JSON.stringify({ ...event, checkpoint: Buffer.from(event.checkpoint).toString("base64") }),
|
||||
)
|
||||
if (event.type !== "output" && event.type !== "resized")
|
||||
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,
|
||||
inputProtocol: framedInput ? 1 : 0,
|
||||
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)
|
||||
Queue.offerUnsafe(outbox, JSON.stringify({ type: "replay_complete", endOffset: attachment.replay.endOffset }))
|
||||
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) =>
|
||||
input.withPermit(
|
||||
Effect.suspend(() => {
|
||||
const data = typeof message === "string" ? Buffer.from(message) : message
|
||||
if (!framedInput)
|
||||
return pty
|
||||
.input(ctx.params.ptyID, attachmentID, attachment.info.size.cols, attachment.info.size.rows, data)
|
||||
.pipe(Effect.ignore)
|
||||
if (data.byteLength < 5) return Effect.void
|
||||
const view = new DataView(data.buffer, data.byteOffset, data.byteLength)
|
||||
const type = data[0]
|
||||
const cols = view.getUint16(1)
|
||||
const rows = view.getUint16(3)
|
||||
if ((type !== 0 && type !== 1) || cols === 0 || rows === 0) return Effect.void
|
||||
if (type === 0) return pty.control(ctx.params.ptyID, attachmentID, cols, rows).pipe(Effect.ignore)
|
||||
return pty.input(ctx.params.ptyID, attachmentID, cols, rows, data.subarray(5)).pipe(Effect.ignore)
|
||||
}),
|
||||
),
|
||||
),
|
||||
).pipe(
|
||||
Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void),
|
||||
Effect.ensuring(Effect.sync(() => attachment.detach())),
|
||||
Effect.orDie,
|
||||
)
|
||||
return HttpServerResponse.empty()
|
||||
}),
|
||||
)
|
||||
}),
|
||||
)
|
||||
|
||||
const mapUnavailable = <A>(effect: Effect.Effect<A, PersistentPty.UnavailableError>) =>
|
||||
effect.pipe(Effect.catchTag("PersistentPty.UnavailableError", unavailable))
|
||||
|
||||
const mapTerminalError = <A>(effect: Effect.Effect<A, PersistentPty.NotFoundError | PersistentPty.UnavailableError>) =>
|
||||
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" })
|
||||
|
|
@ -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)),
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -182,7 +183,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)
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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 { 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"
|
||||
|
|
@ -59,6 +60,7 @@ const applicationServiceNodes = [
|
|||
SdkPlugins.node,
|
||||
PermissionSaved.node,
|
||||
PtyTicket.node,
|
||||
PersistentPty.node,
|
||||
Credential.node,
|
||||
WellKnown.node,
|
||||
PtyEnvironment.node,
|
||||
|
|
|
|||
457
packages/server/test/persistent-pty.test.ts
Normal file
457
packages/server/test/persistent-pty.test.ts
Normal file
|
|
@ -0,0 +1,457 @@
|
|||
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 { 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 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,
|
||||
}
|
||||
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<never, never>({
|
||||
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")
|
||||
const events = yield* Effect.promise(() => openEventStream(base))
|
||||
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 first = Schema.decodeUnknownSync(PersistentPty.Info)(
|
||||
(yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, {
|
||||
command: "/bin/sh",
|
||||
args: ["-c", "stty -echo; printf terminal-one; cat"],
|
||||
cwd: process.cwd(),
|
||||
title: "first",
|
||||
env: {},
|
||||
})).data,
|
||||
)
|
||||
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,
|
||||
)
|
||||
|
||||
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, "DELETE", `/api/experimental/persistent-pty/${first.id}`)
|
||||
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([])
|
||||
|
||||
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)
|
||||
}),
|
||||
),
|
||||
20_000,
|
||||
)
|
||||
|
||||
function request(base: string, method: string, pathname: string, body?: unknown, headers?: Record<string, string>) {
|
||||
return Effect.tryPromise({
|
||||
try: async () => {
|
||||
const response = await fetch(new URL(pathname, base), {
|
||||
method,
|
||||
headers: {
|
||||
authorization: `Basic ${btoa("opencode:secret")}`,
|
||||
...headers,
|
||||
...(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))),
|
||||
})
|
||||
}
|
||||
|
||||
async function openEventStream(base: string) {
|
||||
const response = await fetch(new URL("/api/event", base), {
|
||||
headers: { authorization: `Basic ${btoa("opencode:secret")}` },
|
||||
})
|
||||
if (!response.ok || !response.body) throw new Error(`Persistent PTY event stream failed (${response.status})`)
|
||||
const reader = response.body.getReader()
|
||||
const decoder = new TextDecoder()
|
||||
let pending = ""
|
||||
return {
|
||||
async next(type: string) {
|
||||
while (true) {
|
||||
const boundary = pending.indexOf("\n\n")
|
||||
if (boundary !== -1) {
|
||||
const frame = pending.slice(0, boundary)
|
||||
pending = pending.slice(boundary + 2)
|
||||
const data = frame
|
||||
.split("\n")
|
||||
.find((line) => line.startsWith("data: "))
|
||||
?.slice(6)
|
||||
if (!data) continue
|
||||
const event: unknown = JSON.parse(data)
|
||||
if (isRecord(event) && event.type === type) return event
|
||||
continue
|
||||
}
|
||||
const chunk = await reader.read()
|
||||
if (chunk.done) throw new Error(`Persistent PTY event stream closed before ${type}`)
|
||||
pending += decoder.decode(chunk.value, { stream: true })
|
||||
}
|
||||
},
|
||||
close: () => reader.cancel(),
|
||||
}
|
||||
}
|
||||
|
||||
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/experimental/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 waitForStatus(base: string, ptyID: string, status: string) {
|
||||
return Effect.tryPromise({
|
||||
try: async () => {
|
||||
for (let attempt = 0; attempt < 40; attempt++) {
|
||||
const response = await Effect.runPromise(request(base, "GET", `/api/experimental/persistent-pty/${ptyID}`))
|
||||
if (isRecord(response.data) && response.data.status === status) return
|
||||
await Bun.sleep(50)
|
||||
}
|
||||
throw new Error(`Persistent PTY ${ptyID} did not reach status ${status}`)
|
||||
},
|
||||
catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
|
||||
})
|
||||
}
|
||||
|
||||
function attachAndExit(base: string, ptyID: string) {
|
||||
return Effect.tryPromise({
|
||||
try: async () => {
|
||||
const response = await Effect.runPromise(
|
||||
request(base, "POST", `/api/experimental/persistent-pty/${ptyID}/connect-token`, undefined, {
|
||||
"x-opencode-ticket": "1",
|
||||
}),
|
||||
)
|
||||
if (!isRecord(response.data) || typeof response.data.ticket !== "string")
|
||||
throw new Error("Persistent PTY connect token response was invalid")
|
||||
const url = new URL(`/api/experimental/persistent-pty/${ptyID}/connect`, base)
|
||||
url.protocol = "ws:"
|
||||
url.searchParams.set("ticket", response.data.ticket)
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const socket = new WebSocket(url)
|
||||
const timeout = setTimeout(() => {
|
||||
socket.close()
|
||||
reject(new Error("Persistent PTY did not exit while attached"))
|
||||
}, 5_000)
|
||||
socket.addEventListener("message", (event) => {
|
||||
if (typeof event.data !== "string") return
|
||||
const message: unknown = JSON.parse(event.data)
|
||||
if (!isRecord(message)) return
|
||||
if (message.type === "attached") socket.send(new Uint8Array([4]))
|
||||
if (message.type !== "exited") return
|
||||
clearTimeout(timeout)
|
||||
socket.close()
|
||||
resolve()
|
||||
})
|
||||
socket.addEventListener("error", () => {
|
||||
clearTimeout(timeout)
|
||||
reject(new Error("Persistent PTY WebSocket failed"))
|
||||
})
|
||||
})
|
||||
},
|
||||
catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
|
||||
})
|
||||
}
|
||||
|
||||
async function verifySharedControl(base: string, ptyID: string) {
|
||||
const first = await openTerminalSocket(base, ptyID, "first")
|
||||
const second = await openTerminalSocket(base, ptyID, "second", "observer")
|
||||
try {
|
||||
first.socket.send(controlFrame(90, 25))
|
||||
first.socket.send(inputFrame(90, 25, "from-first\n"))
|
||||
await waitForSocketOutput([first, second], "from-first")
|
||||
|
||||
second.socket.send(inputFrame(70, 20, "from-second\n"))
|
||||
await waitForSocketOutput([first, second], "from-second")
|
||||
|
||||
await waitForForegroundProcess([first, second], "cat")
|
||||
|
||||
for (const character of "printf abc | rev\n") second.socket.send(inputFrame(70, 20, character))
|
||||
await waitForSocketOutput([first, second], "printf abc | rev")
|
||||
|
||||
second.socket.send(inputFrame(70, 20, "x".repeat(1024)))
|
||||
second.socket.send(inputFrame(70, 20, "after-burst\n"))
|
||||
await waitForSocketOutput([first, second], "after-burst")
|
||||
expect(first.closed).toBeFalse()
|
||||
expect(second.closed).toBeFalse()
|
||||
expect(first.resizes).toBeGreaterThan(0)
|
||||
expect(second.resizes).toBeGreaterThan(0)
|
||||
expect(first.output).not.toContain("\0")
|
||||
expect(second.output).not.toContain("\0")
|
||||
} finally {
|
||||
first.socket.close()
|
||||
second.socket.close()
|
||||
}
|
||||
}
|
||||
|
||||
async function openTerminalSocket(
|
||||
base: string,
|
||||
ptyID: string,
|
||||
attachmentID: string,
|
||||
role: "controller" | "observer" = "controller",
|
||||
) {
|
||||
const response = await Effect.runPromise(
|
||||
request(base, "POST", `/api/experimental/persistent-pty/${ptyID}/connect-token`, undefined, {
|
||||
"x-opencode-ticket": "1",
|
||||
}),
|
||||
)
|
||||
if (!isRecord(response.data) || typeof response.data.ticket !== "string")
|
||||
throw new Error("Persistent PTY connect token response was invalid")
|
||||
const url = new URL(`/api/experimental/persistent-pty/${ptyID}/connect`, base)
|
||||
url.protocol = "ws:"
|
||||
url.searchParams.set("ticket", response.data.ticket)
|
||||
url.searchParams.set("attachment_id", attachmentID)
|
||||
url.searchParams.set("role", role)
|
||||
url.searchParams.set("takeover", "true")
|
||||
url.searchParams.set("input_protocol", "1")
|
||||
const state = {
|
||||
socket: new WebSocket(url),
|
||||
output: "",
|
||||
closed: false,
|
||||
resizes: 0,
|
||||
foregroundProcess: null as string | null,
|
||||
}
|
||||
state.socket.binaryType = "arraybuffer"
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const timeout = setTimeout(() => reject(new Error("Persistent PTY WebSocket did not attach")), 5_000)
|
||||
let attached = false
|
||||
state.socket.addEventListener("message", (event) => {
|
||||
if (event.data instanceof ArrayBuffer) {
|
||||
state.output += new TextDecoder().decode(event.data)
|
||||
return
|
||||
}
|
||||
if (typeof event.data !== "string") return
|
||||
const message: unknown = JSON.parse(event.data)
|
||||
if (!isRecord(message)) return
|
||||
if (message.type === "resized") {
|
||||
if (typeof message.checkpoint !== "string") {
|
||||
clearTimeout(timeout)
|
||||
reject(new Error("Persistent PTY resize omitted its checkpoint"))
|
||||
return
|
||||
}
|
||||
state.resizes++
|
||||
return
|
||||
}
|
||||
if (message.type === "foreground_process_changed") {
|
||||
state.foregroundProcess = typeof message.process === "string" ? message.process : null
|
||||
return
|
||||
}
|
||||
if (message.type === "attached") {
|
||||
if (message.inputProtocol === 1) {
|
||||
if (isRecord(message.info) && typeof message.info.foregroundProcess === "string")
|
||||
state.foregroundProcess = message.info.foregroundProcess
|
||||
attached = true
|
||||
return
|
||||
}
|
||||
clearTimeout(timeout)
|
||||
reject(new Error("Persistent PTY WebSocket did not negotiate framed input"))
|
||||
return
|
||||
}
|
||||
if (message.type !== "replay_complete" || !attached) return
|
||||
clearTimeout(timeout)
|
||||
resolve()
|
||||
})
|
||||
state.socket.addEventListener("close", () => {
|
||||
state.closed = true
|
||||
})
|
||||
state.socket.addEventListener("error", () => {
|
||||
clearTimeout(timeout)
|
||||
reject(new Error("Persistent PTY WebSocket failed"))
|
||||
})
|
||||
})
|
||||
return state
|
||||
}
|
||||
|
||||
async function waitForForegroundProcess(
|
||||
sockets: Array<{ foregroundProcess: string | null; closed: boolean }>,
|
||||
expected: string,
|
||||
) {
|
||||
for (let attempt = 0; attempt < 100; attempt++) {
|
||||
if (sockets.every((socket) => socket.foregroundProcess === expected)) return
|
||||
if (sockets.some((socket) => socket.closed)) throw new Error("Persistent PTY observer disconnected")
|
||||
await Bun.sleep(20)
|
||||
}
|
||||
throw new Error(
|
||||
`Persistent PTY sockets did not both report ${expected}: ${JSON.stringify(sockets.map((socket) => socket.foregroundProcess))}`,
|
||||
)
|
||||
}
|
||||
|
||||
function inputFrame(cols: number, rows: number, input: string) {
|
||||
const data = new TextEncoder().encode(input)
|
||||
const frame = new Uint8Array(5 + data.byteLength)
|
||||
const view = new DataView(frame.buffer)
|
||||
frame[0] = 1
|
||||
view.setUint16(1, cols)
|
||||
view.setUint16(3, rows)
|
||||
frame.set(data, 5)
|
||||
return frame
|
||||
}
|
||||
|
||||
function controlFrame(cols: number, rows: number) {
|
||||
const frame = new Uint8Array(5)
|
||||
const view = new DataView(frame.buffer)
|
||||
view.setUint16(1, cols)
|
||||
view.setUint16(3, rows)
|
||||
return frame
|
||||
}
|
||||
|
||||
async function waitForSocketOutput(sockets: Array<{ output: string; closed: boolean }>, expected: string) {
|
||||
for (let attempt = 0; attempt < 100; attempt++) {
|
||||
if (sockets.every((socket) => socket.output.includes(expected))) return
|
||||
if (sockets.some((socket) => socket.closed)) throw new Error("Persistent PTY observer disconnected")
|
||||
await Bun.sleep(20)
|
||||
}
|
||||
throw new Error(
|
||||
`Persistent PTY sockets did not both receive ${expected}: ${JSON.stringify(sockets.map((socket) => socket.output))}`,
|
||||
)
|
||||
}
|
||||
|
||||
function waitForTerminals(base: string, sessionID: string, expected: unknown[]) {
|
||||
return Effect.tryPromise({
|
||||
try: async () => {
|
||||
for (let attempt = 0; attempt < 40; attempt++) {
|
||||
const response = await Effect.runPromise(
|
||||
request(base, "GET", `/api/experimental/session/${sessionID}/terminal`),
|
||||
)
|
||||
if (JSON.stringify(response.data) === JSON.stringify(expected)) return
|
||||
await Bun.sleep(50)
|
||||
}
|
||||
throw new Error(`Persistent PTYs for ${sessionID} did not reconcile`)
|
||||
},
|
||||
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<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value)
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue