openAgent

1export * as Pty from "./pty"
2 
3import { makeLocationNode } from "./effect/app-node"
4import type { Disp, Proc } from "#pty"
5import { Context, Effect, Layer, Schema, Types } from "effect"
6import { Pty } from "@opencode-ai/schema/pty"
7import { Config } from "./config"
8import { EventV2 } from "./event"
9import { Location } from "./location"
10import { PtyID } from "./pty/schema"
11import { Shell } from "./shell"
12import { lazy } from "./util/lazy"
13 
14const BUFFER_LIMIT = 1024 * 1024 * 2
15// Exited sessions stay observable (status, exit code, retained output) until removed explicitly.
16// Cap retention so abandoned terminals do not accumulate unbounded buffers.
17const EXITED_LIMIT = 25
18const pty = lazy(() => import("#pty"))
19 
20type Subscriber = {
21  readonly onData: (chunk: string) => void
22  readonly onEnd: (event: { exitCode?: number }) => void
23  active: boolean
24  detached: boolean
25  pending: string[]
26  end?: { exitCode?: number }
27}
28 
29type Active = {
30  info: Info
31  process: Proc
32  buffer: string
33  bufferCursor: number
34  cursor: number
35  subscribers: Map<object, Subscriber>
36  listeners: Disp[]
37}
38 
39export const Info = Pty.Info
40export type Info = Types.DeepMutable<typeof Info.Type>
41 
42export const CreateInput = Pty.CreateInput
43 
44export type CreateInput = Types.DeepMutable<typeof CreateInput.Type>
45 
46export const UpdateInput = Pty.UpdateInput
47 
48export type UpdateInput = Types.DeepMutable<typeof UpdateInput.Type>
49 
50export const Event = Pty.Event
51 
52export type AttachInput = {
53  // Absolute output cursor to replay from. -1 tails from the current end; omitted replays the full retained buffer.
54  readonly cursor?: number
55  // Callbacks fire synchronously from the native PTY data path; keep them non-blocking.
56  readonly onData: (chunk: string) => void
57  // Fired once when the session stops producing output: process exit (exitCode set), removal, or service teardown.
58  readonly onEnd: (event: { exitCode?: number }) => void
59}
60 
61export type Attachment = {
62  // Retained output from the requested cursor to the current end.
63  readonly replay: string
64  // Absolute output cursor after replay.
65  readonly cursor: number
66  readonly write: (data: string) => void
67  // Starts live delivery after the caller has applied replay and cursor metadata.
68  readonly activate: () => void
69  readonly detach: () => void
70}
71 
72export class NotFoundError extends Schema.TaggedErrorClass<NotFoundError>()("Pty.NotFoundError", {
73  ptyID: PtyID,
74}) {}
75 
76export class ExitedError extends Schema.TaggedErrorClass<ExitedError>()("Pty.ExitedError", {
77  ptyID: PtyID,
78}) {}
79 
80export interface Interface {
81  readonly list: () => Effect.Effect<Info[]>
82  readonly get: (id: PtyID) => Effect.Effect<Info, NotFoundError>
83  readonly create: (input: CreateInput) => Effect.Effect<Info>
84  readonly update: (id: PtyID, input: UpdateInput) => Effect.Effect<Info, NotFoundError>
85  readonly remove: (id: PtyID) => Effect.Effect<void, NotFoundError>
86  readonly write: (id: PtyID, data: string) => Effect.Effect<void, NotFoundError>
87  readonly attach: (id: PtyID, input: AttachInput) => Effect.Effect<Attachment, NotFoundError | ExitedError>
88}
89 
90export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Pty") {}
91 
92const layer = Layer.effect(
93  Service,
94  Effect.gen(function* () {
95    const events = yield* EventV2.Service
96    const location = yield* Location.Service
97    const config = yield* Config.Service
98    const context = yield* Effect.context()
99    const runFork = Effect.runForkWith(context)
100    const sessions = new Map<PtyID, Active>()
101    const exitOrder: PtyID[] = []
102 
103    function notifyEnd(session: Active, event: { exitCode?: number }) {
104      for (const subscriber of session.subscribers.values()) {
105        if (!subscriber.active) {
106          subscriber.end = event
107          continue
108        }
109        try {
110          subscriber.onEnd(event)
111        } catch {}
112      }
113      session.subscribers.clear()
114    }
115 
116    function teardown(session: Active) {
117      for (const listener of session.listeners) listener.dispose()
118      session.listeners.length = 0
119      if (session.info.status === "running") {
120        try {
121          session.process.kill()
122        } catch {}
123      }
124      notifyEnd(session, {})
125    }
126 
127    yield* Effect.addFinalizer(() =>
128      Effect.sync(() => {
129        for (const session of sessions.values()) teardown(session)
130        sessions.clear()
131        exitOrder.length = 0
132      }),
133    )
134 
135    const requireSession = Effect.fn("Pty.requireSession")(function* (id: PtyID) {
136      const session = sessions.get(id)
137      if (!session) return yield* new NotFoundError({ ptyID: id })
138      return session
139    })
140 
141    const removeSession = Effect.fnUntraced(function* (id: PtyID) {
142      const session = sessions.get(id)
143      if (!session) return
144      sessions.delete(id)
145      const index = exitOrder.indexOf(id)
146      if (index !== -1) exitOrder.splice(index, 1)
147      yield* Effect.logInfo("removing session", { id })
148      teardown(session)
149      yield* events.publish(Event.Deleted, { id: session.info.id })
150    })
151 
152    const remove = Effect.fn("Pty.remove")(function* (id: PtyID) {
153      yield* requireSession(id)
154      yield* removeSession(id)
155    })
156 
157    const list = Effect.fn("Pty.list")(function* () {
158      return Array.from(sessions.values()).map((session) => session.info)
159    })
160 
161    const get = Effect.fn("Pty.get")(function* (id: PtyID) {
162      return (yield* requireSession(id)).info
163    })
164 
165    const create = Effect.fn("Pty.create")(function* (input: CreateInput) {
166      const id = PtyID.ascending()
167      const command = input.command || Shell.preferred(Config.latest(yield* config.entries(), "shell"))
168      const args = Shell.login(command) ? [...(input.args ?? []), "-l"] : [...(input.args ?? [])]
169      const cwd = input.cwd || location.directory
170      const env = {
171        ...process.env,
172        ...input.env,
173        TERM: "xterm-256color",
174        OPENCODE_TERMINAL: "1",
175      } as Record<string, string>
176      if (process.platform === "win32") {
177        env.LC_ALL = "C.UTF-8"
178        env.LC_CTYPE = "C.UTF-8"
179        env.LANG = "C.UTF-8"
180      }
181      yield* Effect.logInfo("creating session", { id, cmd: command, args, cwd })
182      const { spawn } = yield* Effect.promise(() => pty())
183      const proc = yield* Effect.sync(() => spawn(command, args, { name: "xterm-256color", cwd, env }))
184      const info: Info = {
185        id,
186        title: input.title || `Terminal ${id.slice(-4)}`,
187        command,
188        args,
189        cwd,
190        status: "running",
191        pid: proc.pid,
192      }
193      const session: Active = {
194        info,
195        process: proc,
196        buffer: "",
197        bufferCursor: 0,
198        cursor: 0,
199        subscribers: new Map(),
200        listeners: [],
201      }
202      sessions.set(id, session)
203      session.listeners.push(
204        proc.onData((chunk) => {
205          session.cursor += chunk.length
206          for (const [token, subscriber] of session.subscribers.entries()) {
207            if (!subscriber.active) {
208              subscriber.pending.push(chunk)
209              continue
210            }
211            try {
212              subscriber.onData(chunk)
213            } catch {
214              session.subscribers.delete(token)
215            }
216          }
217          session.buffer += chunk
218          if (session.buffer.length <= BUFFER_LIMIT) return
219          const excess = session.buffer.length - BUFFER_LIMIT
220          session.buffer = session.buffer.slice(excess)
221          session.bufferCursor += excess
222        }),
223        proc.onExit(({ exitCode }) => {
224          if (session.info.status === "exited") return
225          session.info.status = "exited"
226          session.info.exitCode = exitCode
227          notifyEnd(session, { exitCode })
228          exitOrder.push(id)
229          runFork(
230            Effect.gen(function* () {
231              yield* Effect.logInfo("session exited", { id, exitCode })
232              yield* events.publish(Event.Exited, { id, exitCode })
233              while (exitOrder.length > EXITED_LIMIT) {
234                const oldest = exitOrder[0]
235                if (!oldest) break
236                yield* removeSession(oldest)
237              }
238            }),
239          )
240        }),
241      )
242      yield* events.publish(Event.Created, { info })
243      return info
244    })
245 
246    const update = Effect.fn("Pty.update")(function* (id: PtyID, input: UpdateInput) {
247      const session = yield* requireSession(id)
248      if (input.title) session.info.title = input.title
249      if (input.size && session.info.status === "running") session.process.resize(input.size.cols, input.size.rows)
250      yield* events.publish(Event.Updated, { info: session.info })
251      return session.info
252    })
253 
254    const write = Effect.fn("Pty.write")(function* (id: PtyID, data: string) {
255      const session = yield* requireSession(id)
256      if (session.info.status === "running") session.process.write(data)
257    })
258 
259    const attach = Effect.fn("Pty.attach")(function* (id: PtyID, input: AttachInput) {
260      const session = yield* requireSession(id)
261      if (session.info.status !== "running") return yield* new ExitedError({ ptyID: id })
262      yield* Effect.logInfo("client attached to session", { id, directory: location.directory })
263      const token = {}
264      const subscriber: Subscriber = {
265        onData: input.onData,
266        onEnd: input.onEnd,
267        active: false,
268        detached: false,
269        pending: [],
270      }
271      session.subscribers.set(token, subscriber)
272      const start = session.bufferCursor
273      const end = session.cursor
274      const from =
275        input.cursor === -1
276          ? end
277          : typeof input.cursor === "number" && Number.isSafeInteger(input.cursor)
278            ? Math.max(0, input.cursor)
279            : 0
280      const replay = (() => {
281        if (!session.buffer || from >= end) return ""
282        const offset = Math.max(0, from - start)
283        if (offset >= session.buffer.length) return ""
284        return session.buffer.slice(offset)
285      })()
286      return {
287        replay,
288        cursor: end,
289        write: (data: string) => {
290          if (session.info.status === "running") session.process.write(data)
291        },
292        activate: () => {
293          if (subscriber.active || subscriber.detached) return
294          subscriber.active = true
295          try {
296            for (const chunk of subscriber.pending) subscriber.onData(chunk)
297            subscriber.pending.length = 0
298            if (subscriber.end) subscriber.onEnd(subscriber.end)
299          } catch {
300            session.subscribers.delete(token)
301          }
302        },
303        detach: () => {
304          subscriber.detached = true
305          subscriber.pending.length = 0
306          subscriber.end = undefined
307          session.subscribers.delete(token)
308        },
309      }
310    })
311 
312    return Service.of({ list, get, create, update, remove, write, attach })
313  }),
314)
315 
316export const locationLayer = layer.pipe(Layer.provide(Config.locationLayer))
317 
318export const node = makeLocationNode({ service: Service, layer, deps: [EventV2.node, Location.node, Config.node] })
319