fix(tui): stabilize persistent terminal workspaces

This commit is contained in:
James Long
2026-08-14 21:12:04 +00:00
parent bb77765056
commit ca1064bf80
18 changed files with 571 additions and 60 deletions
@@ -4,11 +4,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.restart,
Effect.fn("cli.service.restart")(function* () {
const options = yield* ServiceConfig.options()
yield* ServerConnection.shutdownPersistentPty(options).pipe(Effect.ignore)
yield* Service.stop(options)
const transport = yield* Service.ensure(options)
process.stdout.write(transport.url + EOL)
@@ -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)
}),
)
@@ -56,12 +56,22 @@ function managedService(options: EnsureOptions) {
reconnect: () => Service.ensure(reconnectOptions),
restart: () =>
Effect.gen(function* () {
yield* shutdownPersistentPty(options).pipe(Effect.ignore)
yield* Service.stop(options)
yield* Service.ensure(reconnectOptions)
}),
}
}
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["server.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 })
@@ -195,6 +195,7 @@ import type {
ServerPersistentPtyListOutput,
ServerPersistentPtyCreateInput,
ServerPersistentPtyCreateOutput,
ServerPersistentPtyShutdownOutput,
ServerPersistentPtyGetInput,
ServerPersistentPtyGetOutput,
ServerPersistentPtyUpdateInput,
@@ -1686,6 +1687,17 @@ export function make(options: ClientOptions) {
},
requestOptions,
).then((value) => value.data),
shutdown: (requestOptions?: RequestOptions) =>
request<ServerPersistentPtyShutdownOutput>(
{
method: "POST",
path: `/api/persistent-pty/shutdown`,
successStatus: 204,
declaredStatuses: [503, 401, 400],
empty: true,
},
requestOptions,
),
get: (input: ServerPersistentPtyGetInput, requestOptions?: RequestOptions) =>
request<{ readonly data: ServerPersistentPtyGetOutput }>(
{
@@ -351,6 +351,7 @@ export type PersistentPtyInfo = {
pid: number
exitCode?: number
groupID: string
size: { cols: number; rows: number }
output: { head: number; tail: number }
}
@@ -1492,6 +1493,15 @@ export type FormMultiselectField1 = {
default?: Array<string>
}
export type GroupItemAdded = {
id: string
created: number
metadata?: { [x: string]: any }
type: "group.item.added"
location?: LocationRef
data: { groupID: string; item: GroupItem }
}
export type GroupItemRemoved = {
id: string
created: number
@@ -2110,6 +2120,7 @@ export type V2Event =
| FormCreated
| FormReplied
| FormCancelled
| GroupItemAdded
| GroupItemRemoved
| WebsearchUpdated
| SessionStatus2
@@ -5566,6 +5577,8 @@ export type ServerPersistentPtyCreateInput = {
export type ServerPersistentPtyCreateOutput = { data: PersistentPtyInfo }["data"]
export type ServerPersistentPtyShutdownOutput = void
export type ServerPersistentPtyGetInput = { readonly ptyID: { readonly ptyID: string }["ptyID"] }
export type ServerPersistentPtyGetOutput = { data: PersistentPtyInfo }["data"]
@@ -64,6 +64,13 @@ const layer = Layer.effect(
)
const previous = groups[index]
if (!previous) return
yield* Effect.forEach(
group.items.filter(
(item) => !previous.items.some((current) => current.type === item.type && current.id === item.id),
),
(item) => bus.publish(Event.ItemAdded, { groupID: group.id, item }),
{ discard: true },
)
yield* Effect.forEach(
previous.items.filter(
(item) => !group.items.some((next) => next.type === item.type && next.id === item.id),
+106 -6
View File
@@ -13,7 +13,7 @@ import { Group } from "./group.js"
import { Database } from "../database/database.js"
import { Pty } from "@opencode-ai/schema/pty"
const ProtocolVersion = 2
const ProtocolVersion = 4
const MaxFrameBytes = 8 * 1024 * 1024
const Lifecycle = Schema.Union([
@@ -78,6 +78,7 @@ const Response = Schema.Union([
cols: Schema.Number,
rows: Schema.Number,
generation: Schema.Number,
checkpoint_base64: Schema.String,
}),
Schema.Struct({
type: Schema.Literal("exited"),
@@ -100,6 +101,7 @@ export type Role = "controller" | "observer"
export type Info = Pty.Info & {
readonly groupID: Group.ID
readonly size: { readonly cols: number; readonly rows: number }
readonly output: { readonly head: number; readonly tail: number }
}
@@ -112,7 +114,13 @@ export type Snapshot = {
export type StreamEvent =
| { readonly type: "output"; readonly start: number; readonly end: number; readonly data: Uint8Array }
| { readonly type: "resized"; readonly cols: number; readonly rows: number; readonly generation: number }
| {
readonly type: "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 }
@@ -170,8 +178,22 @@ export interface Interface {
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: {
@@ -282,6 +304,44 @@ export const layer = Layer.effect(
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(client, {
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(client, {
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(client, { op: "snapshot", id: fromID(id) })
@@ -309,6 +369,12 @@ export const layer = Layer.effect(
return undefined
})
const shutdown = Effect.fn("PersistentPty.shutdown")(function* () {
const response = yield* Effect.tryPromise({ try: () => client.shutdown(), catch: 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)
@@ -349,7 +415,7 @@ export const layer = Layer.effect(
})
})
return Service.of({ list, get, create, write, resize, snapshot, remove, attach })
return Service.of({ list, get, create, write, resize, control, input, snapshot, remove, shutdown, attach })
}),
)
@@ -360,14 +426,37 @@ class Client {
constructor(private readonly directory: string) {}
request(value: object, start = false) {
return this.connect(start).then((registration) => oneShot(registration, value))
request(value: object, start = false): Promise<WireResponse> {
return this.connect(start)
.then((registration) => oneShot(registration, value))
.catch((error) => {
if (!(error instanceof ConnectError)) throw error
this.registration = undefined
if (!start) throw error
return this.connect(true).then((registration) => oneShot(registration, value))
})
}
requestIfRunning(value: object) {
return this.request(value).catch(() => undefined)
}
async shutdown() {
const response = await this.requestIfRunning({ op: "shutdown" })
this.registration = undefined
if (!response) return
const deadline = Date.now() + 5_000
while (Date.now() < deadline) {
const running = await discover(this.directory).then(
() => true,
() => false,
)
if (!running) return response
await setTimeout(50)
}
throw new Error("opencode-pty did not stop")
}
async subscribe(
id: number,
input: {
@@ -422,6 +511,7 @@ class Client {
cols: event.cols,
rows: event.rows,
generation: event.generation,
checkpoint: Buffer.from(event.checkpoint_base64, "base64"),
})
if (event.type === "controller_changed")
input.onEvent({
@@ -560,7 +650,10 @@ async function discover(directory: string) {
async function oneShot(registration: Registration, request: object) {
const socket = net.createConnection(registration.socket)
const frames = decoder(socket)
await connected(socket)
await connected(socket).catch((cause) => {
socket.destroy()
throw new ConnectError(cause)
})
socket.write(encode({ token: registration.token, request }))
const first = await frames.next()
socket.end()
@@ -570,6 +663,12 @@ async function oneShot(registration: Registration, request: object) {
return response
}
class ConnectError extends Error {
constructor(cause: unknown) {
super(cause instanceof Error ? cause.message : String(cause))
}
}
function connected(socket: net.Socket) {
return new Promise<void>((resolve, reject) => {
socket.once("connect", resolve)
@@ -620,6 +719,7 @@ function toInfo(value: WireTerminal): Info {
...(status === "exited" ? { exitCode: value.lifecycle.exit_code ?? undefined } : {}),
}),
groupID: Group.ID.make(value.group_id),
size: { cols: value.cols, rows: value.rows },
output: { head: value.output_head, tail: value.output_tail },
}
}
+16
View File
@@ -72,4 +72,20 @@ describe("Group", () => {
])
}),
)
it.effect("publishes every added group item", () =>
Effect.gen(function* () {
const groups = yield* Group.Service
const bus = yield* Bus.Service
const session = { type: "session" as const, id: Session.ID.make("ses_one") }
const terminal = { type: "terminal" as const, id: Pty.ID.make("pty_one") }
const group = yield* groups.create([session])
const event = yield* bus.subscribe(Group.Event.ItemAdded).pipe(Stream.runHead, Effect.forkScoped)
yield* Effect.yieldNow
yield* groups.set(Group.Info.make({ id: group.id, items: [session, terminal] }))
expect((yield* Fiber.join(event)).valueOrUndefined?.data).toEqual({ groupID: group.id, item: terminal })
}),
)
})
@@ -78,6 +78,12 @@ export const PersistentPtyGroup = HttpApiGroup.make("server.persistentPty")
error: errors,
}),
)
.add(
HttpApiEndpoint.post("persistentPty.shutdown", "/api/persistent-pty/shutdown", {
success: HttpApiSchema.NoContent,
error: [ServiceUnavailableError],
}),
)
.add(
HttpApiEndpoint.get("persistentPty.get", "/api/persistent-pty/:ptyID", {
params: { ptyID: Pty.ID },
+2 -1
View File
@@ -38,5 +38,6 @@ export const Info = Schema.Struct({
}).annotate({ identifier: "Group.Info" })
export interface Info extends Schema.Schema.Type<typeof Info> {}
const ItemAdded = ephemeral({ type: "group.item.added", schema: { groupID: ID, item: Item } })
const ItemRemoved = ephemeral({ type: "group.item.removed", schema: { groupID: ID, item: Item } })
export const Event = { ItemRemoved, Definitions: inventory(ItemRemoved) }
export const Event = { ItemAdded, ItemRemoved, Definitions: inventory(ItemAdded, ItemRemoved) }
+1
View File
@@ -8,6 +8,7 @@ import { NonNegativeInt, PositiveInt, optional } from "./schema.js"
export const Info = Schema.Struct({
...Pty.Info.fields,
groupID: Group.ID,
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> {}
+1 -1
View File
@@ -65,7 +65,7 @@ describe("public event manifest", () => {
expect(Integration.Event.Definitions).toEqual([Integration.Event.Updated, Integration.Event.ConnectionUpdated])
expect(Permission.Event.Definitions).toEqual([Permission.Event.Asked, Permission.Event.Replied])
expect(Form.Event.Definitions).toEqual([Form.Event.Created, Form.Event.Replied, Form.Event.Cancelled])
expect(Group.Event.Definitions).toEqual([Group.Event.ItemRemoved])
expect(Group.Event.Definitions).toEqual([Group.Event.ItemAdded, Group.Event.ItemRemoved])
expect(Reference.Event.Definitions).toEqual([Reference.Event.Updated])
expect(Plugin.Event.Definitions).toEqual([Plugin.Event.Added, Plugin.Event.Updated])
expect(McpEvent.Definitions).toEqual([McpEvent.ToolsChanged, McpEvent.ResourcesChanged, McpEvent.StatusChanged])
+42 -12
View File
@@ -98,6 +98,13 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.persistent
}
}),
)
.handle(
"persistentPty.shutdown",
Effect.fn(function* () {
yield* pty.shutdown().pipe(mapUnavailable)
return HttpApiSchema.NoContent.make()
}),
)
.handle(
"persistentPty.get",
Effect.fn(function* (ctx) {
@@ -165,6 +172,7 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.persistent
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 })
@@ -179,7 +187,13 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.persistent
takeover: url.searchParams.get("takeover") === "true",
onEvent: (event) => {
if (event.type === "output") Queue.offerUnsafe(outbox, event.data)
if (event.type !== "output") Queue.offerUnsafe(outbox, JSON.stringify(event))
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)),
})
@@ -196,6 +210,7 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.persistent
JSON.stringify({
type: "attached",
attachmentID,
inputProtocol: framedInput ? 1 : 0,
info: attachment.info,
role: attachment.role,
generation: attachment.generation,
@@ -208,6 +223,10 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.persistent
}),
)
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* () {
@@ -220,17 +239,28 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.persistent
yield* Effect.race(
drain,
socket.runRaw((message) =>
role === "controller"
? pty
.write(
ctx.params.ptyID,
typeof message === "string" ? message : Buffer.from(message).toString(),
attachmentID,
)
.pipe(Effect.ignore)
: Effect.void,
),
socket.runRaw((message) => {
if (role !== "controller") return Effect.void
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())),
+124 -1
View File
@@ -62,13 +62,14 @@ smoke(
(
yield* request(base, "POST", `/api/pty-group/${group.id}/terminal`, {
command: "/bin/sh",
args: ["-c", "printf terminal-one; sleep 30"],
args: ["-c", "stty -echo; printf terminal-one; cat"],
cwd: process.cwd(),
title: "first",
env: {},
})
).data,
)
expect(first.size).toEqual({ cols: 80, rows: 24 })
expect(existsSync(path.join(fixture.directory, "service.json"))).toBeTrue()
const second = Schema.decodeUnknownSync(PersistentPty.Info)(
(
@@ -96,6 +97,7 @@ smoke(
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/persistent-pty/${first.id}/snapshot`)
if (
!isRecord(snapshot.data) ||
@@ -112,6 +114,8 @@ smoke(
yield* request(base, "DELETE", `/api/persistent-pty/${second.id}`)
expect((yield* request(base, "GET", `/api/pty-group/${group.id}`)).data).toMatchObject({ items: [] })
yield* request(base, "POST", "/api/persistent-pty/shutdown")
const unattended = Schema.decodeUnknownSync(PersistentPty.Info)(
(
yield* request(base, "POST", `/api/pty-group/${group.id}/terminal`, {
@@ -250,6 +254,125 @@ function attachAndExit(base: string, ptyID: string) {
})
}
async function verifySharedControl(base: string, ptyID: string) {
const first = await openTerminalSocket(base, ptyID, "first")
const second = await openTerminalSocket(base, ptyID, "second")
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")
second.socket.send(inputFrame(70, 20, "x".repeat(1024 * 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) {
const response = await Effect.runPromise(
request(base, "POST", `/api/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/persistent-pty/${ptyID}/connect`, base)
url.protocol = "ws:"
url.searchParams.set("ticket", response.data.ticket)
url.searchParams.set("attachment_id", attachmentID)
url.searchParams.set("takeover", "true")
url.searchParams.set("input_protocol", "1")
const state = { socket: new WebSocket(url), output: "", closed: false, resizes: 0 }
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 === "attached") {
if (message.inputProtocol === 1) {
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
}
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 waitForGroupItems(base: string, groupID: string, expected: unknown[]) {
return Effect.tryPromise({
try: async () => {
@@ -29,6 +29,7 @@ export function PaneWorkspace(props: { sessionID?: string; groupID?: string; ver
}
function PaneNode(props: { node: PaneLayoutNode; rootSessionID?: string; verticalTabsWidth: number }) {
const panes = usePaneLayout()
const theme = useTheme()
return (
<Switch>
@@ -42,7 +43,11 @@ function PaneNode(props: { node: PaneLayoutNode; rootSessionID?: string; vertica
<UnavailablePane label={`Session ${item().id}`} />
</Match>
<Match when={item().type === "terminal"}>
<PersistentTerminalPane ptyID={item().id} autoFocus={!props.rootSessionID} />
<PersistentTerminalPane
ptyID={item().id}
autoFocus={!props.rootSessionID || panes.shouldFocus(item().id)}
onAutoFocus={() => panes.clearFocus(item().id)}
/>
</Match>
</Switch>
)}
@@ -1,6 +1,6 @@
import { EmbeddedTerminalRenderable } from "@opentui/core"
import { extend } from "@opentui/solid"
import { createSignal, onCleanup, onMount, Show } from "solid-js"
import { extend, useRenderer } from "@opentui/solid"
import { createEffect, createSignal, onCleanup, onMount, Show } from "solid-js"
import { useClient } from "../context/client"
import { Keymap } from "../context/keymap"
import { useTheme } from "../context/theme"
@@ -14,45 +14,105 @@ declare module "@opentui/solid" {
extend({ embeddedTerminal: EmbeddedTerminalRenderable })
export function PersistentTerminalPane(props: { ptyID: string; autoFocus?: boolean }) {
type TerminalSize = { cols: number; rows: number }
type StreamItem =
| { type: "output"; data: Uint8Array }
| { type: "resize"; size: TerminalSize; checkpoint?: Uint8Array }
| { type: "ready" }
export function PersistentTerminalPane(props: { ptyID: string; autoFocus?: boolean; onAutoFocus?: () => void }) {
const client = useClient()
const keymap = Keymap.use()
const theme = useTheme()
const renderer = useRenderer()
const [failure, setFailure] = createSignal<string>()
const attachmentID = crypto.randomUUID()
const pending: Uint8Array[] = []
const output: Uint8Array[] = []
const stream: StreamItem[] = []
const pendingInput: Uint8Array[] = []
let terminal: EmbeddedTerminalRenderable | undefined
let socket: WebSocket | undefined
let outputTimer: ReturnType<typeof setTimeout> | undefined
let attached = false
let controller = false
let restored = false
let wantsControl = false
let disposed = false
let size: { cols: number; rows: number } | undefined
let size: TerminalSize | undefined
let canonicalSize: TerminalSize | undefined
let terminalSize: TerminalSize | undefined
let lastIntermediateRender = 0
let waitingSize: { size: TerminalSize; resolve: () => void } | undefined
const setCanonicalSize = (value: TerminalSize) => {
canonicalSize = value
if (!terminal) return
terminal.width = value.cols
terminal.height = value.rows
}
const send = (data: Uint8Array) => {
if (attached && socket?.readyState === WebSocket.OPEN) {
socket.send(data)
if (attached && socket?.readyState === WebSocket.OPEN) socket.send(data)
}
const interact = () => {
if (!restored) {
wantsControl = true
return
}
pending.push(data)
if (!size) return
send(interactionFrame(size))
}
const resize = () => {
if (!attached || !size) return
void client.api["server.persistentPty"]
.update({ ptyID: props.ptyID, attachmentID, size })
.catch((error) => setFailure(errorMessage(error)))
const sendInput = (data: Uint8Array) => {
if (!restored) {
pendingInput.push(data)
return
}
if (size) send(interactionFrame(size, data))
}
const writeOutput = (data: Uint8Array) => {
output.push(data)
if (outputTimer) return
outputTimer = setTimeout(() => {
outputTimer = undefined
if (disposed) return
terminal?.write(output.length === 1 ? output[0] : Buffer.concat(output))
output.length = 0
}, 16)
const processStream = () => {
if (disposed || !terminal || !sameSize(canonicalSize, terminalSize)) return
while (stream.length > 0) {
const item = stream[0]!
if (item.type === "output") {
stream.shift()
const output = [item.data]
while (true) {
const next = stream[0]
if (!next || next.type !== "output") break
output.push(next.data)
stream.shift()
}
terminal.write(output.length === 1 ? output[0] : Buffer.concat(output))
continue
}
if (item.type === "resize") {
setCanonicalSize(item.size)
if (!sameSize(canonicalSize, terminalSize)) return
stream.shift()
if (item.checkpoint)
terminal.write(Buffer.concat([Buffer.from("\x1bc"), Buffer.from(item.checkpoint)]))
continue
}
stream.shift()
restored = true
const input = pendingInput.splice(0)
if (input.length > 0) input.forEach(sendInput)
if (input.length === 0 && (controller || wantsControl)) interact()
wantsControl = false
}
}
const enqueue = (item: StreamItem) => {
stream.push(item)
processStream()
}
const waitForTerminalSize = (value: TerminalSize) => {
if (sameSize(value, terminalSize)) return Promise.resolve()
return new Promise<void>((resolve) => {
waitingSize = { size: value, resolve }
})
}
const offKeys = keymap.intercept(
@@ -66,13 +126,19 @@ export function PersistentTerminalPane(props: { ptyID: string; autoFocus?: boole
{ priority: 100 },
)
createEffect(() => {
if (!props.autoFocus || !terminal) return
terminal.focus()
props.onAutoFocus?.()
})
onMount(() => {
void connect().catch((error) => setFailure(errorMessage(error)))
})
onCleanup(() => {
disposed = true
if (outputTimer) clearTimeout(outputTimer)
waitingSize?.resolve()
socket?.close()
offKeys()
})
@@ -82,6 +148,9 @@ export function PersistentTerminalPane(props: { ptyID: string; autoFocus?: boole
if (!endpoint) throw new Error("Persistent terminal server endpoint is unavailable")
const snapshot = await client.api["server.persistentPty"].snapshot({ ptyID: props.ptyID })
if (disposed) return
setCanonicalSize(snapshot.info.size)
await waitForTerminalSize(snapshot.info.size)
if (disposed) return
terminal?.write(Buffer.from(snapshot.checkpoint, "base64"))
const token = await client.api["server.persistentPty"].connectToken(
{ ptyID: props.ptyID },
@@ -93,21 +162,76 @@ export function PersistentTerminalPane(props: { ptyID: string; autoFocus?: boole
url.searchParams.set("ticket", token.ticket)
url.searchParams.set("cursor", String(snapshot.info.output.tail))
url.searchParams.set("attachment_id", attachmentID)
url.searchParams.set("takeover", "true")
url.searchParams.set("input_protocol", "1")
const next = new WebSocket(url)
next.binaryType = "arraybuffer"
next.addEventListener("message", (event) => {
if (disposed) return
if (event.data instanceof ArrayBuffer) {
writeOutput(new Uint8Array(event.data))
enqueue({ type: "output", data: new Uint8Array(event.data) })
const now = performance.now()
if (now - lastIntermediateRender >= 16) {
lastIntermediateRender = now
renderer.intermediateRender()
}
return
}
if (typeof event.data !== "string") return
const message: unknown = JSON.parse(event.data)
if (!message || typeof message !== "object" || !("type" in message) || message.type !== "attached") return
if (!message || typeof message !== "object" || !("type" in message)) return
if (
message.type === "resized" &&
"cols" in message &&
typeof message.cols === "number" &&
"rows" in message &&
typeof message.rows === "number" &&
"checkpoint" in message &&
typeof message.checkpoint === "string"
) {
enqueue({
type: "resize",
size: { cols: message.cols, rows: message.rows },
checkpoint: Buffer.from(message.checkpoint, "base64"),
})
return
}
if (message.type === "replay_complete") {
enqueue({ type: "ready" })
return
}
if (
message.type === "controller_changed" &&
"attachmentID" in message &&
(typeof message.attachmentID === "string" || message.attachmentID === undefined)
) {
const previous = controller
controller = message.attachmentID === attachmentID
if (controller && !previous && restored) interact()
return
}
if (message.type !== "attached") return
if (!("inputProtocol" in message) || message.inputProtocol !== 1) {
setFailure("Persistent terminal server is out of date; restart OpenCode")
next.close()
return
}
if (
"info" in message &&
message.info &&
typeof message.info === "object" &&
"size" in message.info &&
message.info.size &&
typeof message.info.size === "object" &&
"cols" in message.info.size &&
typeof message.info.size.cols === "number" &&
"rows" in message.info.size &&
typeof message.info.size.rows === "number"
)
enqueue({ type: "resize", size: { cols: message.info.size.cols, rows: message.info.size.rows } })
controller = "role" in message && message.role === "controller"
attached = true
pending.splice(0).forEach((data) => next.send(data))
resize()
})
next.addEventListener("error", () => {
if (!disposed) setFailure("Terminal connection failed")
@@ -119,24 +243,60 @@ export function PersistentTerminalPane(props: { ptyID: string; autoFocus?: boole
}
return (
<box flexGrow={1} minWidth={0} minHeight={0}>
<box
flexGrow={1}
minWidth={0}
minHeight={0}
overflow="hidden"
onSizeChange={function () {
size = { cols: this.width, rows: this.height }
if (controller && restored) interact()
}}
// TODO: Revisit when embedded terminal mouse handlers can compose without replacing its internal focus handler.
onMouseDown={() => interact()}
>
<Show when={!failure()} fallback={<text fg={theme.text.feedback.error.default}>{failure()}</text>}>
<embeddedTerminal
ref={(value) => {
terminal = value
if (props.autoFocus) value.focus()
terminalSize = { cols: 80, rows: 24 }
if (canonicalSize) {
value.width = canonicalSize.cols
value.height = canonicalSize.rows
}
}}
width="100%"
height="100%"
position="absolute"
left={0}
top={0}
width={80}
height={24}
onData={(data, source) => {
if (source === "input") send(data)
if (source === "input") sendInput(data)
}}
onTerminalResize={(cols, rows) => {
size = { cols, rows }
resize()
terminalSize = { cols, rows }
if (waitingSize && sameSize(waitingSize.size, terminalSize)) {
waitingSize.resolve()
waitingSize = undefined
}
processStream()
}}
/>
</Show>
</box>
)
}
function sameSize(first: TerminalSize | undefined, second: TerminalSize | undefined) {
return !!first && !!second && first.cols === second.cols && first.rows === second.rows
}
function interactionFrame(size: { cols: number; rows: number }, data?: Uint8Array) {
const frame = new Uint8Array(5 + (data?.byteLength ?? 0))
const view = new DataView(frame.buffer)
frame[0] = data ? 1 : 0
view.setUint16(1, size.cols)
view.setUint16(3, size.rows)
if (data) frame.set(data, 5)
return frame
}
+22 -1
View File
@@ -5,7 +5,7 @@ import { useData } from "./data"
import { useStorage } from "./storage"
import { reconcilePaneLayout, removePaneLayoutItem, type PaneLayoutNode } from "./pane-layout-model"
import { useEvent } from "./event"
import { onCleanup } from "solid-js"
import { createSignal, onCleanup } from "solid-js"
type PaneWorkspace = {
sessionID?: string
@@ -24,6 +24,7 @@ export const { use: usePaneLayout, provider: PaneLayoutProvider } = createSimple
const client = useClient()
const data = useData()
const event = useEvent()
const [focus, setFocus] = createSignal<string>()
const [store, update] = useStorage().store<PaneLayoutState>("pane-layout-v1", {
initial: { workspaces: {} },
})
@@ -43,6 +44,19 @@ export const { use: usePaneLayout, provider: PaneLayoutProvider } = createSimple
}
})
onCleanup(
event.on("group.item.added", (evt) => {
void update((draft) => {
Object.values(draft.workspaces).forEach((workspace) => {
if (workspace.groupID !== evt.data.groupID) return
if (workspace.items.some((item) => item.type === evt.data.item.type && item.id === evt.data.item.id)) return
workspace.items.push(evt.data.item)
workspace.layout = reconcilePaneLayout(workspace.layout, workspace.items) ?? workspace.layout
})
}).catch((error) => console.error("Failed to add pane layout item", error))
}),
)
onCleanup(
event.on("group.item.removed", (evt) => {
void update((draft) => {
@@ -110,6 +124,7 @@ export const { use: usePaneLayout, provider: PaneLayoutProvider } = createSimple
env: {},
})
const next = await api.group.get({ groupID: group.id })
setFocus(terminal.id)
await save(sessionID, next, sessionID)
return terminal
},
@@ -132,6 +147,12 @@ export const { use: usePaneLayout, provider: PaneLayoutProvider } = createSimple
await save(group.id, await api.group.get({ groupID: group.id }))
return { group, terminal }
},
shouldFocus(ptyID: string) {
return focus() === ptyID
},
clearFocus(ptyID: string) {
setFocus((current) => (current === ptyID ? undefined : current))
},
}
},
})
@@ -440,6 +440,7 @@ test("closing a terminal-only workspace tab terminates its terminals and removes
status: "running" as const,
pid: 123,
groupID,
size: { cols: 80, rows: 24 },
output: { head: 0, tail: 0 },
}
const setup = await renderSessionTabs("first", {