ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): canonicalize pty service (#32182)

Browse source
This commit is contained in:
Shoubhit Dash 2026-06-14 16:16:39 +05:30 • committed by GitHub
commit f2cf607376
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
30 changed files with 1132 additions and 504 deletions
Download patch file Download diff file Expand all files Collapse all files

248
packages/core/src/pty.ts
View file

@ -2,22 +2,27 @@ export * as Pty from "./pty"
import type { Disp, Proc } from "#pty"
import { Context, Effect, Layer, Schema, Types } from "effect"
import { Config } from "./config"
import { EventV2 } from "./event"
import { Location } from "./location"
import { NonNegativeInt, PositiveInt } from "./schema"
import { PtyID } from "./pty/schema"
import { Shell } from "./shell"
import { lazy } from "./util/lazy"
const BUFFER_LIMIT = 1024 * 1024 * 2
const BUFFER_CHUNK = 64 * 1024
const encoder = new TextEncoder()
// Exited sessions stay observable (status, exit code, retained output) until removed explicitly.
// Cap retention so abandoned terminals do not accumulate unbounded buffers.
const EXITED_LIMIT = 25
const pty = lazy(() => import("#pty"))
type Socket = {
readyState: number
data?: unknown
send: (data: string | Uint8Array | ArrayBuffer) => void
close: (code?: number, reason?: string) => void
type Subscriber = {
readonly onData: (chunk: string) => void
readonly onEnd: (event: { exitCode?: number }) => void
active: boolean
detached: boolean
pending: string[]
end?: { exitCode?: number }
}
type Active = {
@ -26,22 +31,10 @@ type Active = {
buffer: string
bufferCursor: number
cursor: number
subscribers: Map<unknown, Socket>
subscribers: Map<object, Subscriber>
listeners: Disp[]
}
const sock = (ws: Socket) => (ws.data && typeof ws.data === "object" ? ws.data : ws)
// WebSocket control frame: 0x00 + UTF-8 JSON.
const meta = (cursor: number) => {
const json = JSON.stringify({ cursor })
const bytes = encoder.encode(json)
const out = new Uint8Array(bytes.length + 1)
out[0] = 0
out.set(bytes, 1)
return out
}
export const Info = Schema.Struct({
id: PtyID,
title: Schema.String,
@ -51,6 +44,8 @@ export const Info = Schema.Struct({
status: Schema.Literals(["running", "exited"]),
// Windows ConPTY assigns the child pid asynchronously, so 0 is valid at spawn time.
pid: NonNegativeInt,
// Present once status is "exited".
exitCode: Schema.optional(NonNegativeInt),
}).annotate({ identifier: "Pty" })
export type Info = Types.DeepMutable<typeof Info.Type>
@ -65,14 +60,6 @@ export const CreateInput = Schema.Struct({
export type CreateInput = Types.DeepMutable<typeof CreateInput.Type>
export type PreparedCreate = {
readonly command: string
readonly args: string[]
readonly cwd: string
readonly title?: string
readonly env: Record<string, string>
}
export const UpdateInput = Schema.Struct({
title: Schema.optional(Schema.String),
size: Schema.optional(
@ -85,10 +72,34 @@ export const UpdateInput = Schema.Struct({
export type UpdateInput = Types.DeepMutable<typeof UpdateInput.Type>
export type AttachInput = {
// Absolute output cursor to replay from. -1 tails from the current end; omitted replays the full retained buffer.
readonly cursor?: number
// Callbacks fire synchronously from the native PTY data path; keep them non-blocking.
readonly onData: (chunk: string) => void
// Fired once when the session stops producing output: process exit (exitCode set), removal, or service teardown.
readonly onEnd: (event: { exitCode?: number }) => void
}
export type Attachment = {
// Retained output from the requested cursor to the current end.
readonly replay: string
// Absolute output cursor after replay.
readonly cursor: number
readonly write: (data: string) => void
// Starts live delivery after the caller has applied replay and cursor metadata.
readonly activate: () => void
readonly detach: () => void
}
export class NotFoundError extends Schema.TaggedErrorClass<NotFoundError>()("Pty.NotFoundError", {
ptyID: PtyID,
}) {}
export class ExitedError extends Schema.TaggedErrorClass<ExitedError>()("Pty.ExitedError", {
ptyID: PtyID,
}) {}
export const Event = {
Created: EventV2.define({ type: "pty.created", schema: { info: Info } }),
Updated: EventV2.define({ type: "pty.updated", schema: { info: Info } }),
@ -99,19 +110,11 @@ export const Event = {
export interface Interface {
readonly list: () => Effect.Effect<Info[]>
readonly get: (id: PtyID) => Effect.Effect<Info, NotFoundError>
readonly create: (input: PreparedCreate) => Effect.Effect<Info>
readonly create: (input: CreateInput) => Effect.Effect<Info>
readonly update: (id: PtyID, input: UpdateInput) => Effect.Effect<Info, NotFoundError>
readonly remove: (id: PtyID) => Effect.Effect<void, NotFoundError>
readonly resize: (id: PtyID, cols: number, rows: number) => Effect.Effect<void, NotFoundError>
readonly write: (id: PtyID, data: string) => Effect.Effect<void, NotFoundError>
readonly connect: (
id: PtyID,
ws: Socket,
cursor?: number,
) => Effect.Effect<
{ onMessage: (message: string | ArrayBuffer) => void; onClose: () => void } | undefined,
NotFoundError
>
readonly attach: (id: PtyID, input: AttachInput) => Effect.Effect<Attachment, NotFoundError | ExitedError>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Pty") {}
@ -121,28 +124,41 @@ export const layer = Layer.effect(
Effect.gen(function* () {
const events = yield* EventV2.Service
const location = yield* Location.Service
const config = yield* Config.Service
const context = yield* Effect.context()
const runFork = Effect.runForkWith(context)
const sessions = new Map<PtyID, Active>()
const exitOrder: PtyID[] = []
function notifyEnd(session: Active, event: { exitCode?: number }) {
for (const subscriber of session.subscribers.values()) {
if (!subscriber.active) {
subscriber.end = event
continue
}
try {
subscriber.onEnd(event)
} catch {}
}
session.subscribers.clear()
}
function teardown(session: Active) {
for (const listener of session.listeners) listener.dispose()
session.listeners.length = 0
try {
session.process.kill()
} catch {}
for (const [sub, ws] of session.subscribers.entries()) {
if (session.info.status === "running") {
try {
if (sock(ws) === sub) ws.close()
session.process.kill()
} catch {}
}
session.subscribers.clear()
notifyEnd(session, {})
}
yield* Effect.addFinalizer(() =>
Effect.sync(() => {
for (const session of sessions.values()) teardown(session)
sessions.clear()
exitOrder.length = 0
}),
)
@ -154,12 +170,13 @@ export const layer = Layer.effect(
const removeSession = Effect.fnUntraced(function* (id: PtyID) {
const session = sessions.get(id)
if (!session) return false
if (!session) return
sessions.delete(id)
const index = exitOrder.indexOf(id)
if (index !== -1) exitOrder.splice(index, 1)
yield* Effect.logInfo("removing session", { id })
teardown(session)
yield* events.publish(Event.Deleted, { id: session.info.id })
return true
})
const remove = Effect.fn("Pty.remove")(function* (id: PtyID) {
@ -175,26 +192,36 @@ export const layer = Layer.effect(
return (yield* requireSession(id)).info
})
const create = Effect.fn("Pty.create")(function* (input: PreparedCreate) {
const create = Effect.fn("Pty.create")(function* (input: CreateInput) {
const id = PtyID.ascending()
yield* Effect.logInfo("creating session", { id, cmd: input.command, args: input.args, cwd: input.cwd })
const command = input.command || Shell.preferred(Config.latest(yield* config.entries(), "shell"))
const args = Shell.login(command) ? [...(input.args ?? []), "-l"] : [...(input.args ?? [])]
const cwd = input.cwd || location.directory
// TODO: Apply plugin shell.env environment augmentation once V2 plugin hooks exist; legacy
// routes merge plugin-provided values into input.env at the boundary.
const env = {
...process.env,
...input.env,
TERM: "xterm-256color",
OPENCODE_TERMINAL: "1",
} as Record<string, string>
if (process.platform === "win32") {
env.LC_ALL = "C.UTF-8"
env.LC_CTYPE = "C.UTF-8"
env.LANG = "C.UTF-8"
}
yield* Effect.logInfo("creating session", { id, cmd: command, args, cwd })
const { spawn } = yield* Effect.promise(() => pty())
const proc = yield* Effect.sync(() =>
spawn(input.command, input.args, {
name: "xterm-256color",
cwd: input.cwd,
env: input.env,
}),
)
const info = {
const proc = yield* Effect.sync(() => spawn(command, args, { name: "xterm-256color", cwd, env }))
const info: Info = {
id,
title: input.title || `Terminal ${id.slice(-4)}`,
command: input.command,
args: input.args,
cwd: input.cwd,
command,
args,
cwd,
status: "running",
pid: proc.pid,
} as const
}
const session: Active = {
info,
process: proc,
@ -208,15 +235,15 @@ export const layer = Layer.effect(
session.listeners.push(
proc.onData((chunk) => {
session.cursor += chunk.length
for (const [key, ws] of session.subscribers.entries()) {
if (ws.readyState !== 1 || sock(ws) !== key) {
session.subscribers.delete(key)
for (const [token, subscriber] of session.subscribers.entries()) {
if (!subscriber.active) {
subscriber.pending.push(chunk)
continue
}
try {
ws.send(chunk)
subscriber.onData(chunk)
} catch {
session.subscribers.delete(key)
session.subscribers.delete(token)
}
}
session.buffer += chunk
@ -227,12 +254,19 @@ export const layer = Layer.effect(
}),
proc.onExit(({ exitCode }) => {
if (session.info.status === "exited") return
session.info.status = "exited"
session.info.exitCode = exitCode
notifyEnd(session, { exitCode })
exitOrder.push(id)
runFork(
Effect.gen(function* () {
yield* Effect.logInfo("session exited", { id, exitCode })
session.info.status = "exited"
yield* events.publish(Event.Exited, { id, exitCode })
yield* removeSession(id)
while (exitOrder.length > EXITED_LIMIT) {
const oldest = exitOrder[0]
if (!oldest) break
yield* removeSession(oldest)
}
}),
)
}),
@ -244,66 +278,72 @@ export const layer = Layer.effect(
const update = Effect.fn("Pty.update")(function* (id: PtyID, input: UpdateInput) {
const session = yield* requireSession(id)
if (input.title) session.info.title = input.title
if (input.size) session.process.resize(input.size.cols, input.size.rows)
if (input.size && session.info.status === "running")
session.process.resize(input.size.cols, input.size.rows)
yield* events.publish(Event.Updated, { info: session.info })
return session.info
})
const resize = Effect.fn("Pty.resize")(function* (id: PtyID, cols: number, rows: number) {
const session = yield* requireSession(id)
if (session.info.status === "running") session.process.resize(cols, rows)
})
const write = Effect.fn("Pty.write")(function* (id: PtyID, data: string) {
const session = yield* requireSession(id)
if (session.info.status === "running") session.process.write(data)
})
const connect = Effect.fn("Pty.connect")(function* (id: PtyID, ws: Socket, cursor?: number) {
const session = yield* requireSession(id).pipe(Effect.tapError(() => Effect.sync(() => ws.close())))
yield* Effect.logInfo("client connected to session", { id, directory: location.directory })
const sub = sock(ws)
session.subscribers.delete(sub)
session.subscribers.set(sub, ws)
const cleanup = () => session.subscribers.delete(sub)
const attach = Effect.fn("Pty.attach")(function* (id: PtyID, input: AttachInput) {
const session = yield* requireSession(id)
if (session.info.status !== "running") return yield* new ExitedError({ ptyID: id })
yield* Effect.logInfo("client attached to session", { id, directory: location.directory })
const token = {}
const subscriber: Subscriber = {
onData: input.onData,
onEnd: input.onEnd,
active: false,
detached: false,
pending: [],
}
session.subscribers.set(token, subscriber)
const start = session.bufferCursor
const end = session.cursor
const from =
cursor === -1 ? end : typeof cursor === "number" && Number.isSafeInteger(cursor) ? Math.max(0, cursor) : 0
const data = (() => {
input.cursor === -1
? end
: typeof input.cursor === "number" && Number.isSafeInteger(input.cursor)
? Math.max(0, input.cursor)
: 0
const replay = (() => {
if (!session.buffer || from >= end) return ""
const offset = Math.max(0, from - start)
if (offset >= session.buffer.length) return ""
return session.buffer.slice(offset)
})()
if (data) {
try {
for (let i = 0; i < data.length; i += BUFFER_CHUNK) ws.send(data.slice(i, i + BUFFER_CHUNK))
} catch {
cleanup()
ws.close()
return
}
}
try {
ws.send(meta(end))
} catch {
cleanup()
ws.close()
return
}
return {
onMessage: (message: string | ArrayBuffer) => {
session.process.write(typeof message === "string" ? message : new TextDecoder().decode(message))
replay,
cursor: end,
write: (data: string) => {
if (session.info.status === "running") session.process.write(data)
},
onClose: () => {
cleanup()
activate: () => {
if (subscriber.active || subscriber.detached) return
subscriber.active = true
try {
for (const chunk of subscriber.pending) subscriber.onData(chunk)
subscriber.pending.length = 0
if (subscriber.end) subscriber.onEnd(subscriber.end)
} catch {
session.subscribers.delete(token)
}
},
detach: () => {
subscriber.detached = true
subscriber.pending.length = 0
subscriber.end = undefined
session.subscribers.delete(token)
},
}
})
return Service.of({ list, get, create, update, remove, resize, write, connect })
return Service.of({ list, get, create, update, remove, write, attach })
}),
)
export const locationLayer = layer
export const locationLayer = layer.pipe(Layer.provide(Config.locationLayer))