1import { assert, it } from "@effect/vitest";
2import {
3 type ModelSelection,
4 type OrchestrationV2AppThread,
5 type OrchestrationV2CommandOutputFrame,
6 type OrchestrationV2TurnItem,
7 ProviderDriverKind,
8 ProviderInstanceId,
9 type ThreadId,
10 TurnItemId,
11} from "@t3tools/contracts";
12import {
13 appendTerminalOutput,
14 EMPTY_TERMINAL_OUTPUT,
15 normalizeTerminalOutput,
16 TERMINAL_OUTPUT_TAIL_CHARS,
17 type TerminalOutputState,
18} from "@t3tools/shared/terminalOutput";
19import * as DateTime from "effect/DateTime";
20import * as Effect from "effect/Effect";
21import * as Layer from "effect/Layer";
22import * as Queue from "effect/Queue";
23import * as Stream from "effect/Stream";
24import * as TestClock from "effect/testing/TestClock";
25
26import * as SqlitePersistence from "../persistence/Sqlite.ts";
27import * as CommandOutputHub from "./CommandOutputHub.ts";
28import * as EventSink from "./EventSink.ts";
29import * as EventStore from "./EventStore.ts";
30import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
31import * as ProjectionStore from "./ProjectionStore.ts";
32
33const layerTestDatabase = SqlitePersistence.layerMemory;
34const layerTestStores = Layer.merge(EventStore.layer, ProjectionStore.layer).pipe(
35 Layer.provide(layerTestDatabase),
36);
37const layerTestEventSink = EventSink.layer.pipe(
38 Layer.provide(Layer.mergeAll(layerTestStores, layerTestDatabase)),
39);
40const layerTest = Layer.mergeAll(
41 layerTestEventSink,
42 IdAllocator.layer,
43 CommandOutputHub.layer.pipe(Layer.provide(Layer.merge(layerTestEventSink, layerTestStores))),
44);
45
46const modelSelection = {
47 instanceId: ProviderInstanceId.make("codex"),
48 model: "gpt-5.4",
49} satisfies ModelSelection;
50const CODEX_DRIVER = ProviderDriverKind.make("codex");
51const ITEM_ID = TurnItemId.make("turn-item:command");
52
53function applyFrame(
54 state: TerminalOutputState,
55 frame: OrchestrationV2CommandOutputFrame,
56): TerminalOutputState {
57 return frame.kind === "replace"
58 ? appendTerminalOutput({ ...EMPTY_TERMINAL_OUTPUT, truncated: frame.truncated }, frame.text)
59 : appendTerminalOutput(state, frame.text);
60}
61
62const seedThread = Effect.gen(function* () {
63 const idAllocator = yield* IdAllocator.IdAllocatorV2;
64 const eventSink = yield* EventSink.EventSinkV2;
65 const now = yield* DateTime.now;
66 const projectId = yield* idAllocator.allocate.project({ fixtureName: "command-output" });
67 const threadId = yield* idAllocator.allocate.thread({ fixtureName: "command-output", projectId });
68 const thread: OrchestrationV2AppThread = {
69 createdBy: "user",
70 creationSource: "web",
71 id: threadId,
72 projectId,
73 title: "Command output",
74 providerInstanceId: modelSelection.instanceId,
75 modelSelection,
76 runtimeMode: "full-access",
77 interactionMode: "default",
78 branch: null,
79 worktreePath: null,
80 branchPullRequest: null,
81 activeOrderKey: null,
82 activeProviderThreadId: null,
83 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
84 forkedFrom: null,
85 createdAt: now,
86 updatedAt: now,
87 archivedAt: null,
88 settledOverride: null,
89 settledAt: null,
90 lastVisitedAt: null,
91 deletedAt: null,
92 };
93 yield* eventSink.write({
94 events: [
95 {
96 id: yield* idAllocator.allocate.event({ threadId }),
97 type: "thread.created",
98 threadId,
99 occurredAt: now,
100 payload: thread,
101 },
102 ],
103 });
104 return threadId;
105});
106
107const writeCommandItem = (
108 threadId: ThreadId,
109 fields: Pick<
110 Extract<OrchestrationV2TurnItem, { type: "command_execution" }>,
111 "status" | "output" | "exitCode"
112 >,
113) =>
114 Effect.gen(function* () {
115 const idAllocator = yield* IdAllocator.IdAllocatorV2;
116 const eventSink = yield* EventSink.EventSinkV2;
117 const now = yield* DateTime.now;
118 const settled = fields.status !== "running";
119 const item: OrchestrationV2TurnItem = {
120 id: ITEM_ID,
121 threadId,
122 runId: null,
123 nodeId: null,
124 providerThreadId: null,
125 providerTurnId: null,
126 nativeItemRef: null,
127 parentItemId: null,
128 ordinal: 1,
129 title: null,
130 startedAt: now,
131 completedAt: settled ? now : null,
132 updatedAt: now,
133 type: "command_execution",
134 input: "yes | head -n 1000000",
135 ...fields,
136 };
137 yield* eventSink.write({
138 events: [
139 {
140 id: yield* idAllocator.allocate.event({ threadId }),
141 type: "turn-item.updated",
142 threadId,
143 driver: CODEX_DRIVER,
144 providerInstanceId: modelSelection.instanceId,
145 occurredAt: now,
146 payload: item,
147 },
148 ],
149 });
150 });
151
152const subscribeFrames = (threadId: ThreadId) =>
153 Effect.gen(function* () {
154 const hub = yield* CommandOutputHub.CommandOutputHub;
155 const frames = yield* Queue.unbounded<OrchestrationV2CommandOutputFrame | "ended">();
156 yield* hub.subscribe({ threadId, itemId: ITEM_ID }).pipe(
157 Stream.runForEach((frame) => Queue.offer(frames, frame)),
158 Effect.andThen(Queue.offer(frames, "ended")),
159 Effect.forkScoped,
160 );
161 return frames;
162 });
163
164const takeFrame = (frames: Queue.Queue<OrchestrationV2CommandOutputFrame | "ended">) =>
165 Queue.take(frames).pipe(
166 Effect.filterOrElse(
167 (frame) => !(frame === "ended"),
168 (_frame) => Effect.die("command output stream ended early"),
169 ),
170 );
171
172const layer = it.layer(layerTest);
173
174layer("CommandOutputHub", (it) => {
175 it.effect(
176 "a late subscriber gets the running tail, then live appends, then the final output",
177 () =>
178 Effect.scoped(
179 Effect.gen(function* () {
180 const hub = yield* CommandOutputHub.CommandOutputHub;
181 const threadId = yield* seedThread;
182 yield* writeCommandItem(threadId, { status: "running" });
183 yield* hub.append({
184 threadId,
185 itemId: ITEM_ID,
186 chunk: "\u001b[32mbuilding\u001b[0m\n10%",
187 });
188 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "\r55%\u001b[3" });
189
190 const frames = yield* subscribeFrames(threadId);
191 const first = yield* takeFrame(frames);
192 assert.equal(first.kind, "replace");
193 assert.isTrue(first.running);
194 let client = applyFrame(EMPTY_TERMINAL_OUTPUT, first);
195 assert.equal(client.text, "\u001b[0;32mbuilding\u001b[0m\n55%");
196
197 yield* TestClock.adjust("1 second");
198 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "1m" });
199 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "\r100%\u001b[0m\ndone\n" });
200 const append = yield* takeFrame(frames);
201 assert.equal(append.kind, "append");
202 client = applyFrame(client, append);
203 assert.equal(
204 client.text,
205 normalizeTerminalOutput(
206 "\u001b[32mbuilding\u001b[0m\n10%\r55%\u001b[31m\r100%\u001b[0m\ndone\n",
207 ).text,
208 );
209
210 yield* TestClock.adjust("1 second");
211 yield* writeCommandItem(threadId, {
212 status: "completed",
213 exitCode: 0,
214 output: "building\n100%\ndone\n",
215 });
216 const final = yield* takeFrame(frames);
217 assert.deepEqual(final, {
218 kind: "replace",
219 text: "building\n100%\ndone\n",
220 truncated: false,
221 running: false,
222 });
223 assert.equal(yield* Queue.take(frames), "ended");
224 }),
225 ),
226 );
227
228 it.effect("a row that is not a known command fails instead of waiting forever", () =>
229 Effect.gen(function* () {
230 const hub = yield* CommandOutputHub.CommandOutputHub;
231 const threadId = yield* seedThread;
232 const error = yield* hub
233 .subscribe({ threadId, itemId: ITEM_ID })
234 .pipe(Stream.runDrain, Effect.flip);
235 assert.equal(error._tag, "OrchestrationV2CommandOutputError");
236 assert.equal(error.message, "No such command.");
237 }),
238 );
239
240 it.effect("a viewer whose command was evicted for newer ones is resynced, not appended to", () =>
241 Effect.scoped(
242 Effect.gen(function* () {
243 const hub = yield* CommandOutputHub.CommandOutputHub;
244 const threadId = yield* seedThread;
245 yield* writeCommandItem(threadId, { status: "running" });
246 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "abc" });
247 const frames = yield* subscribeFrames(threadId);
248 let client = applyFrame(EMPTY_TERMINAL_OUTPUT, yield* takeFrame(frames));
249 assert.equal(client.text, "abc");
250 for (let index = 0; index < 64; index += 1) {
251 yield* hub.append({
252 threadId,
253 itemId: TurnItemId.make(`turn-item:other-${index}`),
254 chunk: "x",
255 });
256 }
257 yield* TestClock.adjust("1 second");
258 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "\rX" });
259 const next = yield* takeFrame(frames);
260 assert.equal(next.kind, "replace");
261 client = applyFrame(client, next);
262 assert.equal(client.text, "X");
263 }),
264 ),
265 );
266
267 it.effect("a finished command sends only its bounded final output", () =>
268 Effect.scoped(
269 Effect.gen(function* () {
270 const threadId = yield* seedThread;
271 const huge = Array.from({ length: 400_000 }, (_, index) => `line ${index}`).join("\n");
272 yield* writeCommandItem(threadId, { status: "failed", exitCode: 1, output: huge });
273 const frames = yield* subscribeFrames(threadId);
274 const final = yield* takeFrame(frames);
275 assert.equal(final.kind, "replace");
276 assert.isFalse(final.running);
277 assert.isTrue(final.truncated);
278 assert.isAtMost(final.text.length, TERMINAL_OUTPUT_TAIL_CHARS);
279 assert.isTrue(final.text.endsWith("line 399999"));
280 assert.equal(yield* Queue.take(frames), "ended");
281 }),
282 ),
283 );
284
285 it.effect("a finished command whose output is one huge line still shows its tail", () =>
286 Effect.scoped(
287 Effect.gen(function* () {
288 const threadId = yield* seedThread;
289 const line = "x".repeat(TERMINAL_OUTPUT_TAIL_CHARS * 8);
290 yield* writeCommandItem(threadId, {
291 status: "completed",
292 exitCode: 0,
293 output: `${line}\n`,
294 });
295 const frames = yield* subscribeFrames(threadId);
296 const final = yield* takeFrame(frames);
297 assert.equal(final.kind, "replace");
298 assert.isTrue(final.truncated);
299 assert.isTrue(final.text.includes("xxxx"));
300 assert.isAtMost(final.text.length, TERMINAL_OUTPUT_TAIL_CHARS);
301 }),
302 ),
303 );
304
305 it.effect("an old Claude row's raw Bash result shows as its stdout and stderr", () =>
306 Effect.scoped(
307 Effect.gen(function* () {
308 const threadId = yield* seedThread;
309 yield* writeCommandItem(threadId, {
310 status: "completed",
311 exitCode: 0,
312 output: '{"stdout":"built","stderr":"1 warning","interrupted":false}',
313 });
314 const frames = yield* subscribeFrames(threadId);
315 const final = yield* takeFrame(frames);
316 assert.equal(final.text, "built\n1 warning");
317 }),
318 ),
319 );
320
321 it.effect("providers that only report output in item snapshots still stream", () =>
322 Effect.scoped(
323 Effect.gen(function* () {
324 const threadId = yield* seedThread;
325 yield* writeCommandItem(threadId, { status: "running", output: "step 1\n" });
326 const frames = yield* subscribeFrames(threadId);
327 assert.equal((yield* takeFrame(frames)).text, "step 1\n");
328 yield* TestClock.adjust("1 second");
329 yield* writeCommandItem(threadId, { status: "running", output: "step 1\nstep 2\n" });
330 const update = yield* takeFrame(frames);
331 assert.deepEqual(update, {
332 kind: "replace",
333 text: "step 1\nstep 2\n",
334 truncated: false,
335 running: true,
336 });
337 }),
338 ),
339 );
340
341 it.effect("an interrupted command ends the stream with the output it printed", () =>
342 Effect.scoped(
343 Effect.gen(function* () {
344 const hub = yield* CommandOutputHub.CommandOutputHub;
345 const threadId = yield* seedThread;
346 yield* writeCommandItem(threadId, { status: "running" });
347 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "waiting for server\n" });
348 const frames = yield* subscribeFrames(threadId);
349 yield* takeFrame(frames);
350 yield* TestClock.adjust("1 second");
351 // Some providers omit stopped-command output; the live tail is released before the settled item persists.
352 yield* hub.finish({ threadId, itemId: ITEM_ID });
353 yield* writeCommandItem(threadId, { status: "interrupted" });
354 const final = yield* takeFrame(frames);
355 assert.deepEqual(final, {
356 kind: "replace",
357 text: "waiting for server\n",
358 truncated: false,
359 running: false,
360 });
361 assert.equal(yield* Queue.take(frames), "ended");
362 }),
363 ),
364 );
365});