Open History to see the last commit
1import { assert, describe, it } from "@effect/vitest";
2import {
3 ORCHESTRATION_V2_WS_METHODS,
4 type TerminalAttachStreamEvent,
5 TerminalAttachStreamEvent as TerminalAttachStreamEventSchema,
6 WS_METHODS,
7} from "@t3tools/contracts";
8import * as Deferred from "effect/Deferred";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as Queue from "effect/Queue";
12import * as Stream from "effect/Stream";
13import * as TestClock from "effect/testing/TestClock";
14import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/rpc";
15
16import { terminalOutputStream, withOutputWindow } from "./OutputProtocol.ts";
17
18const eventWith = (
19 field: "data" | "message" | "threadId",
20 text: string,
21): TerminalAttachStreamEvent =>
22 field === "message"
23 ? { threadId: "thread-1", terminalId: "default", type: "error", message: text }
24 : {
25 threadId: field === "threadId" ? text : "thread-1",
26 terminalId: "default",
27 type: "output",
28 data: field === "data" ? text : "x",
29 };
30
31describe("terminal output window", () => {
32 it.effect.each([
33 { tag: WS_METHODS.terminalAttach, field: "data", size: 1, limit: 8 },
34 { tag: WS_METHODS.terminalObserve, field: "data", size: 1, limit: 8 },
35 { tag: WS_METHODS.terminalAttach, field: "data", size: 64 * 1024, limit: 1 },
36 { tag: WS_METHODS.terminalObserve, field: "data", size: 64 * 1024, limit: 1 },
37 { tag: WS_METHODS.terminalAttach, field: "message", size: 64 * 1024, limit: 1 },
38 { tag: WS_METHODS.terminalAttach, field: "threadId", size: 64 * 1024, limit: 1 },
39 { tag: WS_METHODS.subscribeServerConfig, field: "data", size: 1, limit: 8 },
40 { tag: ORCHESTRATION_V2_WS_METHODS.subscribeShell, field: "data", size: 1, limit: 8 },
41 { tag: ORCHESTRATION_V2_WS_METHODS.subscribeThread, field: "data", size: 64 * 1024, limit: 8 },
42 { tag: WS_METHODS.subscribeTerminalMetadata, field: "data", size: 1, limit: 1 },
43 ] as const)(
44 "limits $tag with $size-byte $field values to $limit pending chunks",
45 ({ tag, field, size, limit }) =>
46 Effect.gen(function* () {
47 const group = RpcGroup.make(
48 Rpc.make(tag, { success: TerminalAttachStreamEventSchema, stream: true }),
49 );
50 const output = yield* Queue.unbounded<TerminalAttachStreamEvent>();
51 const responses = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
52 const receive = yield* Deferred.make<Parameters<RpcServer.Protocol["Service"]["run"]>[0]>();
53 const protocol = yield* RpcServer.Protocol.make((write) =>
54 Effect.gen(function* () {
55 yield* Deferred.succeed(receive, write);
56 const serialization = yield* RpcSerialization.RpcSerialization;
57 return {
58 disconnects: yield* Queue.unbounded<number>(),
59 send: (_clientId, response) => Queue.offer(responses, response),
60 end: () => Effect.void,
61 clientIds: Effect.succeed(new Set([0])),
62 initialMessage: Effect.succeedNone,
63 supportsAck: true,
64 supportsTransferables: false,
65 supportsSpanPropagation: false,
66 supportsNotifications: true,
67 codecFor: serialization.codecFor,
68 };
69 }),
70 );
71 yield* RpcServer.make(group).pipe(
72 Effect.provide(group.toLayerHandler(tag, () => Stream.fromQueue(output))),
73 Effect.provideService(RpcServer.Protocol, withOutputWindow(protocol)),
74 Effect.forkScoped,
75 );
76 const write = yield* Deferred.await(receive);
77 yield* write(0, { _tag: "Request", id: "1", tag, payload: null, headers: [] });
78
79 for (let index = 0; index < limit; index++) {
80 const value = eventWith(field, String(index).repeat(size));
81 yield* Queue.offer(output, value);
82 yield* TestClock.adjust(0);
83 assert.equal(yield* Queue.size(responses), 1);
84 assert.deepEqual(yield* Queue.take(responses), {
85 _tag: "Chunk",
86 requestId: "1",
87 values: [value],
88 });
89 }
90 const resumedValue = eventWith(field, "i".repeat(size));
91 yield* Queue.offer(output, resumedValue);
92 yield* TestClock.adjust(0);
93 assert.equal(yield* Queue.size(responses), 0);
94 yield* write(0, { _tag: "Ack", requestId: "1" });
95 const resumed = yield* Queue.take(responses);
96 assert.equal(resumed._tag, "Chunk");
97 if (resumed._tag === "Chunk") assert.deepEqual(resumed.values, [resumedValue]);
98
99 yield* Queue.offer(output, eventWith(field, "j"));
100 yield* TestClock.adjust(0);
101 assert.equal(yield* Queue.size(responses), 0);
102 yield* write(0, { _tag: "Interrupt", requestId: "1" });
103 assert.equal((yield* Queue.take(responses))._tag, "Exit");
104 }).pipe(Effect.provide(RpcSerialization.layerJson)),
105 );
106});
107
108const output = (data: string): TerminalAttachStreamEvent => ({
109 type: "output",
110 threadId: "thread-1",
111 terminalId: "default",
112 data,
113});
114
115const snapshot = (history: string): TerminalAttachStreamEvent => ({
116 type: "snapshot",
117 snapshot: {
118 threadId: "thread-1",
119 terminalId: "default",
120 cwd: "/repo",
121 worktreePath: null,
122 status: "running",
123 pid: 1,
124 history,
125 exitCode: null,
126 exitSignal: null,
127 label: "bash",
128 updatedAt: "2026-01-01T00:00:00.000Z",
129 },
130});
131
132// Stands in for TerminalManager's fanout: listeners run in publish order and
133// may release themselves while it iterates.
134const makePublisher = () => {
135 const listeners = new Set<(event: TerminalAttachStreamEvent) => Effect.Effect<void>>();
136 return {
137 listeners,
138 subscribe: (
139 listener: (event: TerminalAttachStreamEvent) => Effect.Effect<void>,
140 registerCleanup?: (stop: () => void) => void,
141 ) =>
142 Effect.sync(() => {
143 listeners.add(listener);
144 const stop = () => {
145 listeners.delete(listener);
146 };
147 registerCleanup?.(stop);
148 return stop;
149 }),
150 publish: (event: TerminalAttachStreamEvent) =>
151 Effect.forEach(listeners, (listener) => listener(event), { discard: true }),
152 };
153};
154
155// Like TerminalManager.attachStream: a snapshot, then live events.
156const attach =
157 (publisher: ReturnType<typeof makePublisher>, history: string) =>
158 (
159 listener: (event: TerminalAttachStreamEvent) => Effect.Effect<void>,
160 registerCleanup: (stop: () => void) => void,
161 ) =>
162 publisher
163 .subscribe(listener, registerCleanup)
164 .pipe(Effect.tap(() => listener(snapshot(history))));
165
166const chunkBytes = (message: RpcMessage.FromServerEncoded) =>
167 Buffer.byteLength(JSON.stringify(message));
168
169describe("terminal output stream", () => {
170 it.effect.each([WS_METHODS.terminalAttach, WS_METHODS.terminalObserve])(
171 "releases only the ACK-stalled subscriber on %s and resumes safely",
172 (tag) =>
173 Effect.gen(function* () {
174 const group = RpcGroup.make(
175 Rpc.make(tag, {
176 success: TerminalAttachStreamEventSchema,
177 stream: true,
178 }),
179 );
180 const publisher = makePublisher();
181 const stalled = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
182 const healthy = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
183 const receive = yield* Deferred.make<Parameters<RpcServer.Protocol["Service"]["run"]>[0]>();
184 const protocol = yield* RpcServer.Protocol.make((write) =>
185 Effect.gen(function* () {
186 yield* Deferred.succeed(receive, write);
187 const serialization = yield* RpcSerialization.RpcSerialization;
188 return {
189 disconnects: yield* Queue.unbounded<number>(),
190 send: (clientId, response) =>
191 Queue.offer(clientId === 0 ? stalled : healthy, response),
192 end: () => Effect.void,
193 clientIds: Effect.succeed(new Set([0, 1])),
194 initialMessage: Effect.succeedNone,
195 supportsAck: true,
196 supportsTransferables: false,
197 supportsSpanPropagation: false,
198 supportsNotifications: true,
199 codecFor: serialization.codecFor,
200 };
201 }),
202 );
203 const initial = snapshot("");
204 yield* RpcServer.make(group).pipe(
205 Effect.provide(
206 group.toLayerHandler(tag, () =>
207 terminalOutputStream(
208 (listener, registerCleanup) =>
209 publisher
210 .subscribe(listener, registerCleanup)
211 .pipe(Effect.tap(() => listener(initial))),
212 attach(publisher, "fresh history"),
213 256 * 1024,
214 ),
215 ),
216 ),
217 Effect.provideService(RpcServer.Protocol, withOutputWindow(protocol)),
218 Effect.forkScoped,
219 );
220 const write = yield* Deferred.await(receive);
221 // Each event arrives after the batching window, so it is sent on its own.
222 const publish = (event: TerminalAttachStreamEvent) =>
223 TestClock.adjust("2 millis").pipe(Effect.andThen(publisher.publish(event)));
224 const takeChunk = Effect.fnUntraced(function* (queue: typeof stalled) {
225 const response = yield* Queue.take(queue);
226 assert.equal(response._tag, "Chunk");
227 if (response._tag !== "Chunk") throw new Error("Expected an RPC chunk");
228 assert.isAtMost(chunkBytes(response), 70 * 1024);
229 return response.values;
230 });
231 yield* write(0, { _tag: "Request", id: "1", tag, payload: null, headers: [] });
232 assert.deepEqual(yield* takeChunk(stalled), [initial]);
233 yield* write(1, { _tag: "Request", id: "2", tag, payload: null, headers: [] });
234 assert.deepEqual(yield* takeChunk(healthy), [initial]);
235 yield* write(1, { _tag: "Ack", requestId: "2" });
236
237 // Eight unacknowledged chunks fill only client 0's output window.
238 for (let index = 0; index < 7; index++) {
239 const event = output("before-stall-" + index);
240 yield* publish(event);
241 assert.deepEqual(yield* takeChunk(stalled), [event]);
242 assert.deepEqual(yield* takeChunk(healthy), [event]);
243 yield* write(1, { _tag: "Ack", requestId: "2" });
244 }
245 assert.equal(publisher.listeners.size, 2);
246 for (let index = 0; index < 100; index++) {
247 const event = output(index + ":" + "界".repeat(1365));
248 yield* publish(event);
249 assert.deepEqual(yield* takeChunk(healthy), [event]);
250 yield* write(1, { _tag: "Ack", requestId: "2" });
251 if (index === 31) assert.equal(publisher.listeners.size, 2);
252 }
253 assert.equal(yield* Queue.size(stalled), 0);
254 assert.equal(publisher.listeners.size, 1);
255
256 yield* write(0, { _tag: "Ack", requestId: "1" });
257 assert.deepEqual(yield* takeChunk(stalled), [snapshot("fresh history")]);
258 assert.equal(publisher.listeners.size, 2);
259 for (let index = 0; index < 7; index++) {
260 yield* write(0, { _tag: "Ack", requestId: "1" });
261 }
262 const after = output("unique output after recovery");
263 yield* publish(after);
264 assert.deepEqual(yield* takeChunk(healthy), [after]);
265 assert.deepEqual(yield* takeChunk(stalled), [after]);
266 assert.equal(yield* Queue.size(stalled), 0);
267 }).pipe(Effect.provide(RpcSerialization.layerJson)),
268 );
269
270 it.effect("interrupts recovery while its subscription is still waiting", () =>
271 Effect.gen(function* () {
272 const publisher = makePublisher();
273 const recovering = yield* Deferred.make<void>();
274 const interrupted = yield* Deferred.make<void>();
275 const pull = yield* Stream.toPull(
276 terminalOutputStream(
277 attach(publisher, ""),
278 () =>
279 Deferred.succeed(recovering, undefined).pipe(
280 Effect.andThen(Effect.never),
281 Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined)),
282 ),
283 1024,
284 ),
285 );
286 yield* pull;
287 yield* publisher.publish(output("x".repeat(1024)));
288 yield* publisher.publish(output("x".repeat(1024)));
289 assert.equal(publisher.listeners.size, 0);
290
291 const recovery = yield* pull.pipe(Effect.forkChild({ startImmediately: true }));
292 yield* Deferred.await(recovering);
293 yield* Fiber.interrupt(recovery);
294 yield* Deferred.await(interrupted);
295 assert.equal(publisher.listeners.size, 0);
296 }),
297 );
298
299 it.effect("cleans listeners when cancelled during a successful acquisition handoff", () =>
300 Effect.gen(function* () {
301 const publisher = makePublisher();
302 for (const recovering of [false, true, false, true]) {
303 const handoff = yield* Deferred.make<void>();
304 const complete = yield* Deferred.make<void>();
305 const acquire: ReturnType<typeof attach> = (listener, registerCleanup) =>
306 attach(publisher, "")(listener, registerCleanup).pipe(
307 // Hold the successful exit before its cleanup handle reaches the caller.
308 Effect.onExit(() =>
309 Deferred.succeed(handoff, undefined).pipe(Effect.andThen(Deferred.await(complete))),
310 ),
311 );
312 const consumer = yield* Effect.gen(function* () {
313 const pull = yield* Stream.toPull(
314 terminalOutputStream(recovering ? attach(publisher, "") : acquire, acquire, 1024),
315 );
316 yield* pull;
317 yield* publisher.publish(output("x".repeat(1024)));
318 yield* publisher.publish(output("x".repeat(1024)));
319 yield* pull;
320 }).pipe(Effect.scoped, Effect.forkChild({ startImmediately: true }));
321 yield* Deferred.await(handoff);
322 assert.equal(publisher.listeners.size, 1);
323 const cancellation = yield* Fiber.interrupt(consumer).pipe(
324 Effect.forkChild({ startImmediately: true }),
325 );
326 yield* Deferred.succeed(complete, undefined);
327 yield* Fiber.join(cancellation);
328 assert.equal(publisher.listeners.size, 0);
329 }
330 }),
331 );
332
333 it.effect("emits idle output immediately and batches a burst until a fixed deadline", () =>
334 Effect.gen(function* () {
335 const publisher = makePublisher();
336 const pull = yield* Stream.toPull(terminalOutputStream(attach(publisher, "")));
337 assert.deepEqual(yield* pull, [snapshot("")]);
338
339 yield* TestClock.adjust("10 millis");
340 yield* publisher.publish(output("a"));
341 assert.deepEqual(yield* pull, [output("a")]);
342
343 yield* publisher.publish(output("b"));
344 const burst = yield* pull.pipe(Effect.forkChild({ startImmediately: true }));
345 yield* TestClock.adjust("1 millis");
346 assert.isUndefined(burst.pollUnsafe());
347 // Arrivals inside the window never push the deadline back.
348 yield* publisher.publish(output("c"));
349 yield* publisher.publish(output("d"));
350 yield* TestClock.adjust("1 millis");
351 assert.deepEqual(yield* Fiber.join(burst), [output("b"), output("c"), output("d")]);
352 }).pipe(Effect.scoped),
353 );
354
355 it.effect("flushes pending output early, in order, when a lifecycle event arrives", () =>
356 Effect.gen(function* () {
357 const publisher = makePublisher();
358 const pull = yield* Stream.toPull(terminalOutputStream(attach(publisher, "")));
359 assert.deepEqual(yield* pull, [snapshot("")]);
360
361 yield* publisher.publish(output("last words"));
362 const waiting = yield* pull.pipe(Effect.forkChild({ startImmediately: true }));
363 assert.isUndefined(waiting.pollUnsafe());
364 const closed: TerminalAttachStreamEvent = {
365 type: "closed",
366 threadId: "thread-1",
367 terminalId: "default",
368 };
369 yield* publisher.publish(closed);
370 assert.deepEqual(yield* Fiber.join(waiting), [output("last words"), closed]);
371 }).pipe(Effect.scoped),
372 );
373
374 it.effect("flushes full batches early and caps each at the output window in UTF-8 bytes", () =>
375 Effect.gen(function* () {
376 const publisher = makePublisher();
377 const pull = yield* Stream.toPull(terminalOutputStream(attach(publisher, "")));
378 assert.deepEqual(yield* pull, [snapshot("")]);
379 // 1,024 code units but 3 KiB of UTF-8.
380 const data = "界".repeat(1024);
381 for (let index = 0; index < 1024; index++) yield* publisher.publish(output(data));
382
383 // Twenty 3 KiB events fill a 64 KiB batch, so full batches skip the deadline.
384 let delivered = 0;
385 while (delivered < 1020) {
386 const batch = yield* pull;
387 assert.isAtMost(
388 batch.reduce(
389 (bytes, event) => bytes + (event.type === "output" ? Buffer.byteLength(event.data) : 0),
390 0,
391 ),
392 64 * 1024,
393 );
394 delivered += batch.length;
395 }
396 assert.equal(delivered, 1020);
397 const tail = yield* pull.pipe(Effect.forkChild({ startImmediately: true }));
398 assert.isUndefined(tail.pollUnsafe());
399 yield* TestClock.adjust("2 millis");
400 assert.lengthOf(yield* Fiber.join(tail), 4);
401 }),
402 );
403
404 it.effect("charges snapshot history but never overflows on one snapshot", () =>
405 Effect.gen(function* () {
406 const publisher = makePublisher();
407 const history = "界".repeat(16 * 1024);
408 const events = terminalOutputStream(attach(publisher, history), undefined, 32 * 1024);
409 const pull = yield* Stream.toPull(events);
410 // A snapshot larger than the budget is still delivered.
411 assert.deepEqual(yield* pull, [snapshot(history)]);
412 assert.equal(publisher.listeners.size, 1);
413
414 // Queued snapshots count their history: a second one overflows.
415 yield* publisher.publish(snapshot(history));
416 assert.equal(publisher.listeners.size, 1);
417 yield* publisher.publish(snapshot(history));
418 assert.equal(publisher.listeners.size, 0);
419 }),
420 );
421});