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