1import {
2 DEFAULT_MAX_TERMINAL_BUFFER_BYTES,
3 type EnvironmentId,
4 type TerminalAttachInput,
5 type TerminalCloseInput,
6 type TerminalOpenInput,
7 type TerminalResizeInput,
8 type TerminalSummary,
9 type TerminalWriteInput,
10 type ThreadId,
11 WS_METHODS,
12} from "@t3tools/contracts";
13import * as Effect from "effect/Effect";
14import * as Stream from "effect/Stream";
15import { AsyncResult, Atom, type AtomRegistry } from "effect/reactivity";
16
17import {
18 type AtomCommand,
19 createAtomCommandScheduler,
20 createEnvironmentRpcCommand,
21 createEnvironmentSubscriptionAtomFamily,
22 runAtomCommand,
23} from "./runtime.ts";
24import type { EnvironmentRegistry } from "../connection/registry.ts";
25import { subscribe, subscribeDynamic, type EnvironmentRpcInput } from "../rpc/client.ts";
26import {
27 applyTerminalAttachStreamEvent,
28 applyTerminalMetadataStreamEvent,
29 EMPTY_TERMINAL_BUFFER_STATE,
30 nextTerminalAttachSeedState,
31 type TerminalBufferState,
32} from "./terminalSession.ts";
33import { terminalOutputText } from "./terminalOutput.ts";
34
35export function createTerminalEnvironmentAtoms<R, E>(
36 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
37) {
38 const lifecycleScheduler = createAtomCommandScheduler();
39 const resizeScheduler = createAtomCommandScheduler();
40 const terminalThreadKey = ({
41 environmentId,
42 input,
43 }: {
44 readonly environmentId: string;
45 readonly input: { readonly threadId: string; readonly terminalId?: string | undefined };
46 }) => JSON.stringify([environmentId, input.threadId]);
47 const terminalSessionKey = ({
48 environmentId,
49 input,
50 }: {
51 readonly environmentId: string;
52 readonly input: { readonly threadId: string; readonly terminalId?: string | undefined };
53 }) => JSON.stringify([environmentId, input.threadId, input.terminalId ?? null]);
54 const lifecycleConcurrency = { mode: "serial" as const, key: terminalThreadKey };
55 const atoms = {
56 observe: createEnvironmentSubscriptionAtomFamily(runtime, {
57 label: "environment-data:terminal:observe",
58 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.terminalObserve>) =>
59 Stream.suspend(() =>
60 subscribe(WS_METHODS.terminalObserve, input).pipe(
61 Stream.scan(nextTerminalAttachSeedState, applyTerminalAttachStreamEvent),
62 ),
63 ),
64 }),
65 attach: createEnvironmentSubscriptionAtomFamily(runtime, {
66 label: "environment-data:terminal:attach",
67 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.terminalAttach>) => {
68 // Retry an unanswered first attach unchanged so it can still open the shell.
69 let replied = false;
70 return subscribeDynamic(WS_METHODS.terminalAttach, () =>
71 Effect.sync(() => ({
72 ...input,
73 ...(replied ? { resume: true } : {}),
74 maxHistoryBytes: DEFAULT_MAX_TERMINAL_BUFFER_BYTES,
75 })),
76 ).pipe(
77 Stream.onFirst(() =>
78 Effect.sync(() => {
79 replied = true;
80 }),
81 ),
82 Stream.scan(nextTerminalAttachSeedState, applyTerminalAttachStreamEvent),
83 );
84 },
85 }),
86 metadata: createEnvironmentSubscriptionAtomFamily(runtime, {
87 label: "environment-data:terminal:metadata",
88 subscribe: (_input: null) =>
89 subscribe(WS_METHODS.subscribeTerminalMetadata, {}).pipe(
90 Stream.scan((): ReadonlyArray<TerminalSummary> => [], applyTerminalMetadataStreamEvent),
91 // The scan seed is not an authoritative empty server snapshot.
92 Stream.drop(1),
93 ),
94 }),
95 open: createEnvironmentRpcCommand(runtime, {
96 label: "environment-data:terminal:open",
97 tag: WS_METHODS.terminalOpen,
98 scheduler: lifecycleScheduler,
99 concurrency: lifecycleConcurrency,
100 }),
101 write: createEnvironmentRpcCommand(runtime, {
102 label: "environment-data:terminal:write",
103 tag: WS_METHODS.terminalWrite,
104 }),
105 resize: createEnvironmentRpcCommand(runtime, {
106 label: "environment-data:terminal:resize",
107 tag: WS_METHODS.terminalResize,
108 scheduler: resizeScheduler,
109 concurrency: { mode: "latest", key: terminalSessionKey },
110 }),
111 clear: createEnvironmentRpcCommand(runtime, {
112 label: "environment-data:terminal:clear",
113 tag: WS_METHODS.terminalClear,
114 scheduler: lifecycleScheduler,
115 concurrency: lifecycleConcurrency,
116 }),
117 close: createEnvironmentRpcCommand(runtime, {
118 label: "environment-data:terminal:close",
119 tag: WS_METHODS.terminalClose,
120 scheduler: lifecycleScheduler,
121 concurrency: lifecycleConcurrency,
122 }),
123 };
124 return {
125 ...atoms,
126 createSession: (options: TerminalSessionOptions) => createTerminalSession(atoms, options),
127 };
128}
129
130export interface TerminalLifecycleState {
131 readonly wasRunning: boolean;
132 /** The user has seen this terminal without operate permission. */
133 readonly observed: boolean;
134 readonly reopened: boolean;
135}
136
137export const INITIAL_TERMINAL_LIFECYCLE: TerminalLifecycleState = {
138 wasRunning: false,
139 observed: false,
140 reopened: false,
141};
142
143export type TerminalLifecycleEvent =
144 /** A viewer without operate permission is following this terminal. */
145 | { readonly type: "observed" }
146 | { readonly type: "detached"; readonly canOperate: boolean }
147 | {
148 readonly type: "status";
149 readonly status: TerminalBufferState["status"];
150 readonly version: number;
151 }
152 | { readonly type: "reopen-failed" };
153
154export type TerminalLifecycleCommand = "exit" | "reopen" | null;
155
156export function transitionTerminalLifecycle(
157 state: TerminalLifecycleState,
158 event: TerminalLifecycleEvent,
159): { readonly state: TerminalLifecycleState; readonly command: TerminalLifecycleCommand } {
160 switch (event.type) {
161 case "observed":
162 return { state: { ...state, observed: true }, command: null };
163 case "detached":
164 return {
165 state: {
166 ...state,
167 // Losing operate while the shell runs leaves the viewer reading its history.
168 observed: state.observed || (!event.canOperate && state.wasRunning),
169 wasRunning: false,
170 },
171 command: null,
172 };
173 case "reopen-failed":
174 return { state: { ...state, reopened: false }, command: null };
175 case "status": {
176 if (event.version === 0) return { state, command: null };
177 if (event.status === "running" || event.status === "starting") {
178 return { state: { ...state, wasRunning: true, reopened: false }, command: null };
179 }
180 if (event.status !== "closed" && event.status !== "exited") return { state, command: null };
181 if (state.wasRunning) {
182 return { state: { ...state, wasRunning: false, reopened: true }, command: "exit" };
183 }
184 // An explicit visit can reopen a cached ended session. Granting a viewer
185 // operate permission must leave the history they were reading intact.
186 if (state.observed || state.reopened) return { state, command: null };
187 return { state: { ...state, reopened: true }, command: "reopen" };
188 }
189 }
190}
191
192export interface TerminalGridSize {
193 readonly cols: number;
194 readonly rows: number;
195}
196
197export type TerminalSessionMode =
198 | { readonly kind: "observing" }
199 | { readonly kind: "attached"; readonly input: TerminalAttachInput };
200
201type EnvironmentCommand<Input> = AtomCommand<
202 { readonly environmentId: EnvironmentId; readonly input: Input },
203 unknown,
204 unknown
205>;
206
207export interface TerminalSessionAtoms {
208 readonly attach: (target: {
209 readonly environmentId: EnvironmentId;
210 readonly input: TerminalAttachInput;
211 }) => Atom.Atom<AsyncResult.AsyncResult<TerminalBufferState, unknown>>;
212 readonly observe: (target: {
213 readonly environmentId: EnvironmentId;
214 readonly input: { readonly threadId: ThreadId; readonly terminalId: string };
215 }) => Atom.Atom<AsyncResult.AsyncResult<TerminalBufferState, unknown>>;
216 readonly open: EnvironmentCommand<TerminalOpenInput>;
217 readonly resize: EnvironmentCommand<TerminalResizeInput>;
218 readonly write: EnvironmentCommand<TerminalWriteInput>;
219 readonly close: EnvironmentCommand<TerminalCloseInput>;
220}
221
222export interface TerminalSessionOptions {
223 readonly target: {
224 readonly environmentId: EnvironmentId;
225 readonly threadId: ThreadId;
226 readonly terminalId: string;
227 };
228 readonly gridSize: TerminalGridSize;
229 /** Written once, as soon as the first attachment snapshot lands. */
230 readonly initialInput?: string | undefined;
231 /** Read right before every command, so revoking operate stops them at once. */
232 readonly canOperate: () => boolean;
233 /** The shell ended while attached; the store has already closed it. */
234 readonly onExit: () => void;
235 /** Fires when the shell starts or stops accepting input, and on every restart. */
236 readonly onLiveChange?: (live: boolean) => void;
237 readonly debugLog?: (message: string, data: Record<string, unknown>) => void;
238}
239
240export interface TerminalSession {
241 /**
242 * Mount while the session follows a terminal: attached to operate it, or
243 * observing it read-only. `null` follows nothing. A session keeps the first
244 * attach input it is given; a different input needs a new session.
245 */
246 readonly driver: (mode: TerminalSessionMode | null) => Atom.Atom<void>;
247 /** The surface measured a new grid; the shell is resized while it runs. */
248 readonly reportGridSize: (size: TerminalGridSize) => void;
249}
250
251/**
252 * Owns one shown terminal's lifecycle: it reopens an ended shell on an explicit
253 * visit, closes and reports a shell that exits while attached, replays the
254 * measured grid whenever the server starts a new output generation, and writes
255 * the launch's initial input once.
256 */
257export function createTerminalSession(
258 atoms: TerminalSessionAtoms,
259 options: TerminalSessionOptions,
260): TerminalSession {
261 const { environmentId, threadId, terminalId } = options.target;
262 let lifecycle = INITIAL_TERMINAL_LIFECYCLE;
263 let gridSize = options.gridSize;
264 let generation: number | null = null;
265 let liveLifecycleVersion: number | null = null;
266 let initialInputPending = Boolean(options.initialInput);
267 let lastStatus: TerminalBufferState["status"] | null = null;
268 let loggedFirstOutput = false;
269 let resizeRegistry: AtomRegistry.AtomRegistry | null = null;
270
271 const sendResize = () => {
272 if (generation === null || resizeRegistry === null || !options.canOperate()) return;
273 void runAtomCommand(resizeRegistry, atoms.resize, {
274 environmentId,
275 input: { threadId, terminalId, cols: gridSize.cols, rows: gridSize.rows },
276 });
277 };
278
279 const setLive = (next: number | null) => {
280 if (next === liveLifecycleVersion) return;
281 liveLifecycleVersion = next;
282 options.onLiveChange?.(next !== null);
283 };
284
285 const apply = (
286 event: TerminalLifecycleEvent,
287 registry: AtomRegistry.AtomRegistry,
288 input: TerminalAttachInput,
289 ) => {
290 const step = transitionTerminalLifecycle(lifecycle, event);
291 lifecycle = step.state;
292 if (step.command === "exit") {
293 if (!options.canOperate()) return;
294 void runAtomCommand(registry, atoms.close, {
295 environmentId,
296 input: { threadId, terminalId },
297 });
298 options.onExit();
299 return;
300 }
301 if (step.command !== "reopen") return;
302 const reopen = async () => {
303 if (input.cwd === undefined || !options.canOperate()) return false;
304 const result = await runAtomCommand(registry, atoms.open, {
305 environmentId,
306 input: {
307 threadId,
308 terminalId,
309 cwd: input.cwd,
310 worktreePath: input.worktreePath,
311 cols: input.cols,
312 rows: input.rows,
313 ...(input.env ? { env: input.env } : {}),
314 },
315 });
316 return result._tag === "Success";
317 };
318 void reopen().then((succeeded) => {
319 if (!succeeded)
320 lifecycle = transitionTerminalLifecycle(lifecycle, { type: "reopen-failed" }).state;
321 });
322 };
323
324 const onAttachment = (
325 registry: AtomRegistry.AtomRegistry,
326 input: TerminalAttachInput,
327 result: AsyncResult.AsyncResult<TerminalBufferState, unknown>,
328 ) => {
329 resizeRegistry = registry;
330 const buffer = AsyncResult.getOrElse(result, () => EMPTY_TERMINAL_BUFFER_STATE);
331 const status = result._tag === "Failure" ? "error" : buffer.status;
332 if (status !== lastStatus) {
333 lastStatus = status;
334 options.debugLog?.("session:status", {
335 terminalId,
336 status,
337 error: buffer.error,
338 version: buffer.version,
339 });
340 }
341 if (!loggedFirstOutput && options.debugLog) {
342 const text = terminalOutputText(buffer.output);
343 if (text.length > 0) {
344 loggedFirstOutput = true;
345 options.debugLog("session:first-nonempty-buffer", {
346 terminalId,
347 length: text.length,
348 preview: text.slice(0, 160),
349 });
350 }
351 }
352
353 setLive(status === "running" || status === "starting" ? buffer.lifecycleVersion : null);
354
355 // Resize before the initial input so a launched command starts at the measured grid.
356 const nextGeneration =
357 buffer.version > 0 && status === "running" ? buffer.output.generation : null;
358 if (nextGeneration !== generation) {
359 generation = nextGeneration;
360 sendResize();
361 }
362
363 if (initialInputPending && buffer.version > 0 && options.initialInput && options.canOperate()) {
364 initialInputPending = false;
365 void runAtomCommand(registry, atoms.write, {
366 environmentId,
367 input: { threadId, terminalId, data: options.initialInput },
368 });
369 }
370
371 apply({ type: "status", status, version: buffer.version }, registry, input);
372 };
373
374 const label = `terminal-session:${environmentId}:${threadId}:${terminalId}`;
375 const idle = Atom.make<void>(undefined);
376 // Viewing read-only counts as observed before the first snapshot lands, so a
377 // grant that arrives early still leaves the ended shell's history alone.
378 const observing = Atom.make((get) => {
379 // `subscribe`, not `get`: a rebuild would resubscribe on every output chunk.
380 get.subscribe(
381 atoms.observe({ environmentId, input: { threadId, terminalId } }),
382 () => {
383 if (options.canOperate()) return;
384 lifecycle = transitionTerminalLifecycle(lifecycle, { type: "observed" }).state;
385 },
386 { immediate: true },
387 );
388 }).pipe(Atom.withLabel(`${label}:observing`));
389 const makeAttached = (input: TerminalAttachInput) =>
390 Atom.make((get) => {
391 const registry = get.registry;
392 get.addFinalizer(() => {
393 generation = null;
394 lastStatus = null;
395 setLive(null);
396 lifecycle = transitionTerminalLifecycle(lifecycle, {
397 type: "detached",
398 canOperate: options.canOperate(),
399 }).state;
400 });
401 get.subscribe(
402 atoms.attach({ environmentId, input }),
403 (result) => onAttachment(registry, input, result),
404 { immediate: true },
405 );
406 }).pipe(Atom.withLabel(`${label}:attached`));
407 let attached: Atom.Atom<void> | null = null;
408
409 return {
410 driver: (mode) => {
411 if (mode === null) return idle;
412 if (mode.kind === "observing") return observing;
413 attached ??= makeAttached(mode.input);
414 return attached;
415 },
416 reportGridSize: (size) => {
417 if (size.cols === gridSize.cols && size.rows === gridSize.rows) return;
418 gridSize = size;
419 sendResize();
420 },
421 };
422}
423
424export * from "./terminalSession.ts";