1import { assert, describe, it } from "@effect/vitest";
2import { TerminalAttachStreamEvent, WS_METHODS } from "@t3tools/contracts";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Queue from "effect/Queue";
6import * as Schema from "effect/Schema";
7import * as Stream from "effect/Stream";
8import * as TestClock from "effect/testing/TestClock";
9import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/rpc";
10
11import { batchTerminalOutput, withTerminalOutputWindow } from "./OutputProtocol.ts";
12
13describe("terminal output window", () => {
14 it.effect.each([
15 { tag: WS_METHODS.terminalAttach, size: 1, limit: 8 },
16 { tag: WS_METHODS.terminalObserve, size: 1, limit: 8 },
17 { tag: WS_METHODS.subscribeTerminalEvents, size: 1, limit: 8 },
18 { tag: WS_METHODS.terminalAttach, size: 64 * 1024, limit: 1 },
19 { tag: WS_METHODS.terminalObserve, size: 64 * 1024, limit: 1 },
20 { tag: WS_METHODS.subscribeTerminalMetadata, size: 1, limit: 1 },
21 ])("limits $tag with $size-byte values to $limit pending chunks", ({ tag, size, limit }) =>
22 Effect.gen(function* () {
23 const group = RpcGroup.make(Rpc.make(tag, { success: Schema.String, stream: true }));
24 const output = yield* Queue.unbounded<string>();
25 const responses = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
26 const receive = yield* Deferred.make<Parameters<RpcServer.Protocol["Service"]["run"]>[0]>();
27 const protocol = yield* RpcServer.Protocol.make((write) =>
28 Effect.gen(function* () {
29 yield* Deferred.succeed(receive, write);
30 const serialization = yield* RpcSerialization.RpcSerialization;
31 return {
32 disconnects: yield* Queue.unbounded<number>(),
33 send: (_clientId, response) => Queue.offer(responses, response),
34 end: () => Effect.void,
35 clientIds: Effect.succeed(new Set([0])),
36 initialMessage: Effect.succeedNone,
37 supportsAck: true,
38 supportsTransferables: false,
39 supportsSpanPropagation: false,
40 supportsNotifications: true,
41 codecFor: serialization.codecFor,
42 };
43 }),
44 );
45 yield* RpcServer.make(group).pipe(
46 Effect.provide(group.toLayerHandler(tag, () => Stream.fromQueue(output))),
47 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
48 Effect.forkScoped,
49 );
50 const write = yield* Deferred.await(receive);
51 yield* write(0, { _tag: "Request", id: "1", tag, payload: null, headers: [] });
52
53 for (let index = 0; index < limit; index++) {
54 const value = String(index).repeat(size);
55 yield* Queue.offer(output, value);
56 yield* TestClock.adjust(0);
57 assert.equal(yield* Queue.size(responses), 1);
58 assert.deepEqual(yield* Queue.take(responses), {
59 _tag: "Chunk",
60 requestId: "1",
61 values: [value],
62 });
63 }
64 yield* Queue.offer(output, "i".repeat(size));
65 yield* TestClock.adjust(0);
66 assert.equal(yield* Queue.size(responses), 0);
67 yield* write(0, { _tag: "Ack", requestId: "1" });
68 const resumed = yield* Queue.take(responses);
69 assert.equal(resumed._tag, "Chunk");
70 if (resumed._tag === "Chunk") assert.deepEqual(resumed.values, ["i".repeat(size)]);
71
72 yield* Queue.offer(output, "j");
73 yield* TestClock.adjust(0);
74 assert.equal(yield* Queue.size(responses), 0);
75 yield* write(0, { _tag: "Interrupt", requestId: "1" });
76 assert.equal((yield* Queue.take(responses))._tag, "Exit");
77 }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped),
78 );
79});
80
81describe("terminal output batches", () => {
82 it.effect("bounds an acknowledged backlog without changing bytes or event order", () =>
83 Effect.gen(function* () {
84 const group = RpcGroup.make(
85 Rpc.make(WS_METHODS.terminalAttach, { success: TerminalAttachStreamEvent, stream: true }),
86 );
87 const output = yield* Queue.unbounded<TerminalAttachStreamEvent>();
88 const responses = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
89 const receive = yield* Deferred.make<Parameters<RpcServer.Protocol["Service"]["run"]>[0]>();
90 const protocol = yield* RpcServer.Protocol.make((write) =>
91 Effect.gen(function* () {
92 yield* Deferred.succeed(receive, write);
93 const serialization = yield* RpcSerialization.RpcSerialization;
94 return {
95 disconnects: yield* Queue.unbounded<number>(),
96 send: (_clientId, response) => Queue.offer(responses, response),
97 end: () => Effect.void,
98 clientIds: Effect.succeed(new Set([0])),
99 initialMessage: Effect.succeedNone,
100 supportsAck: true,
101 supportsTransferables: false,
102 supportsSpanPropagation: false,
103 supportsNotifications: true,
104 codecFor: serialization.codecFor,
105 };
106 }),
107 );
108 yield* RpcServer.make(group).pipe(
109 Effect.provide(
110 group.toLayerHandler(WS_METHODS.terminalAttach, () =>
111 Stream.fromQueue(output).pipe(batchTerminalOutput),
112 ),
113 ),
114 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
115 Effect.forkScoped,
116 );
117 const write = yield* Deferred.await(receive);
118 yield* write(0, {
119 _tag: "Request",
120 id: "1",
121 tag: WS_METHODS.terminalAttach,
122 payload: null,
123 headers: [],
124 });
125 const event = (sequence: number): TerminalAttachStreamEvent => ({
126 type: "output",
127 threadId: "thread-1",
128 terminalId: "term-1",
129 sequence,
130 data: `\u001b[31m${"🙂\\\n".repeat(400)}\u001b[0m`,
131 });
132 for (let index = 0; index < 8; index++) {
133 yield* Queue.offer(output, event(index));
134 const response = yield* Queue.take(responses);
135 assert.equal(response._tag, "Chunk");
136 }
137
138 const expected = Array.from({ length: 1_024 }, (_, index) => event(index + 8));
139 for (const value of expected) yield* Queue.offer(output, value);
140 yield* TestClock.adjust(0);
141 assert.equal(yield* Queue.size(output), expected.length);
142 assert.equal(yield* Queue.size(responses), 0);
143
144 for (let index = 0; index < 8; index++) {
145 yield* write(0, { _tag: "Ack", requestId: "1" });
146 }
147 const received: unknown[] = [];
148 let batchCount = 0;
149 while (received.length < expected.length) {
150 const response = yield* Queue.take(responses);
151 assert.equal(response._tag, "Chunk");
152 if (response._tag !== "Chunk") continue;
153
154 assert.isAtMost(Buffer.byteLength(JSON.stringify(response)), 64 * 1024);
155 received.push(...response.values);
156 batchCount += 1;
157 yield* write(0, { _tag: "Ack", requestId: "1" });
158 }
159 assert.isAbove(batchCount, 1);
160 assert.deepEqual(received, expected);
161 }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped),
162 );
163
164 it.effect("preserves large snapshots and oversized sequenced events as individual batches", () =>
165 Effect.gen(function* () {
166 const snapshot: TerminalAttachStreamEvent = {
167 type: "snapshot",
168 snapshot: {
169 threadId: "thread-1",
170 terminalId: "term-1",
171 cwd: "/tmp",
172 worktreePath: null,
173 status: "running",
174 pid: 123,
175 history: "x".repeat(8 * 1024 * 1024),
176 label: "sh",
177 exitCode: null,
178 exitSignal: null,
179 updatedAt: "2026-10-15T00:00:00.000Z",
180 },
181 };
182 const output: TerminalAttachStreamEvent = {
183 type: "output",
184 threadId: "thread-1",
185 terminalId: "term-1",
186 sequence: 12,
187 data: "🙂".repeat(32_768),
188 };
189 const exit: TerminalAttachStreamEvent = {
190 type: "exited",
191 threadId: "thread-1",
192 terminalId: "term-1",
193 sequence: 13,
194 exitCode: 0,
195 exitSignal: null,
196 };
197 const batches = yield* Stream.fromArray([snapshot, output, exit]).pipe(
198 batchTerminalOutput,
199 Stream.chunks,
200 Stream.runCollect,
201 );
202 assert.deepEqual(batches, [[snapshot], [output], [exit]]);
203 }),
204 );
205});