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.gen(function* () {
179 const hub = yield* CommandOutputHub.CommandOutputHub;
180 const threadId = yield* seedThread;
181 yield* writeCommandItem(threadId, { status: "running" });
182 yield* hub.append({
183 threadId,
184 itemId: ITEM_ID,
185 chunk: "\u001b[32mbuilding\u001b[0m\n10%",
186 });
187 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "\r55%\u001b[3" });
188
189 const frames = yield* subscribeFrames(threadId);
190 const first = yield* takeFrame(frames);
191 assert.equal(first.kind, "replace");
192 assert.isTrue(first.running);
193 let client = applyFrame(EMPTY_TERMINAL_OUTPUT, first);
194 assert.equal(client.text, "\u001b[0;32mbuilding\u001b[0m\n55%");
195
196 yield* TestClock.adjust("1 second");
197 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "1m" });
198 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "\r100%\u001b[0m\ndone\n" });
199 const append = yield* takeFrame(frames);
200 assert.equal(append.kind, "append");
201 client = applyFrame(client, append);
202 assert.equal(
203 client.text,
204 normalizeTerminalOutput(
205 "\u001b[32mbuilding\u001b[0m\n10%\r55%\u001b[31m\r100%\u001b[0m\ndone\n",
206 ).text,
207 );
208
209 yield* TestClock.adjust("1 second");
210 yield* writeCommandItem(threadId, {
211 status: "completed",
212 exitCode: 0,
213 output: "building\n100%\ndone\n",
214 });
215 const final = yield* takeFrame(frames);
216 assert.deepEqual(final, {
217 kind: "replace",
218 text: "building\n100%\ndone\n",
219 truncated: false,
220 running: false,
221 });
222 assert.equal(yield* Queue.take(frames), "ended");
223 }),
224 );
225
226 it.effect("a row that is not a known command fails instead of waiting forever", () =>
227 Effect.gen(function* () {
228 const hub = yield* CommandOutputHub.CommandOutputHub;
229 const threadId = yield* seedThread;
230 const error = yield* hub
231 .subscribe({ threadId, itemId: ITEM_ID })
232 .pipe(Stream.runDrain, Effect.flip);
233 assert.equal(error._tag, "OrchestrationV2CommandOutputError");
234 assert.equal(error.message, "No such command.");
235 }),
236 );
237
238 it.effect("a viewer whose command was evicted for newer ones is resynced, not appended to", () =>
239 Effect.gen(function* () {
240 const hub = yield* CommandOutputHub.CommandOutputHub;
241 const threadId = yield* seedThread;
242 yield* writeCommandItem(threadId, { status: "running" });
243 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "abc" });
244 const frames = yield* subscribeFrames(threadId);
245 let client = applyFrame(EMPTY_TERMINAL_OUTPUT, yield* takeFrame(frames));
246 assert.equal(client.text, "abc");
247 for (let index = 0; index < 64; index += 1) {
248 yield* hub.append({
249 threadId,
250 itemId: TurnItemId.make(`turn-item:other-${index}`),
251 chunk: "x",
252 });
253 }
254 yield* TestClock.adjust("1 second");
255 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "\rX" });
256 const next = yield* takeFrame(frames);
257 assert.equal(next.kind, "replace");
258 client = applyFrame(client, next);
259 assert.equal(client.text, "X");
260 }),
261 );
262
263 it.effect("a finished command sends only its bounded final output", () =>
264 Effect.gen(function* () {
265 const threadId = yield* seedThread;
266 const huge = Array.from({ length: 400_000 }, (_, index) => `line ${index}`).join("\n");
267 yield* writeCommandItem(threadId, { status: "failed", exitCode: 1, output: huge });
268 const frames = yield* subscribeFrames(threadId);
269 const final = yield* takeFrame(frames);
270 assert.equal(final.kind, "replace");
271 assert.isFalse(final.running);
272 assert.isTrue(final.truncated);
273 assert.isAtMost(final.text.length, TERMINAL_OUTPUT_TAIL_CHARS);
274 assert.isTrue(final.text.endsWith("line 399999"));
275 assert.equal(yield* Queue.take(frames), "ended");
276 }),
277 );
278
279 it.effect("a finished command whose output is one huge line still shows its tail", () =>
280 Effect.gen(function* () {
281 const threadId = yield* seedThread;
282 const line = "x".repeat(TERMINAL_OUTPUT_TAIL_CHARS * 8);
283 yield* writeCommandItem(threadId, {
284 status: "completed",
285 exitCode: 0,
286 output: `${line}\n`,
287 });
288 const frames = yield* subscribeFrames(threadId);
289 const final = yield* takeFrame(frames);
290 assert.equal(final.kind, "replace");
291 assert.isTrue(final.truncated);
292 assert.isTrue(final.text.includes("xxxx"));
293 assert.isAtMost(final.text.length, TERMINAL_OUTPUT_TAIL_CHARS);
294 }),
295 );
296
297 it.effect("an old Claude row's raw Bash result shows as its stdout and stderr", () =>
298 Effect.gen(function* () {
299 const threadId = yield* seedThread;
300 yield* writeCommandItem(threadId, {
301 status: "completed",
302 exitCode: 0,
303 output: '{"stdout":"built","stderr":"1 warning","interrupted":false}',
304 });
305 const frames = yield* subscribeFrames(threadId);
306 const final = yield* takeFrame(frames);
307 assert.equal(final.text, "built\n1 warning");
308 }),
309 );
310
311 it.effect("providers that only report output in item snapshots still stream", () =>
312 Effect.gen(function* () {
313 const threadId = yield* seedThread;
314 yield* writeCommandItem(threadId, { status: "running", output: "step 1\n" });
315 const frames = yield* subscribeFrames(threadId);
316 assert.equal((yield* takeFrame(frames)).text, "step 1\n");
317 yield* TestClock.adjust("1 second");
318 yield* writeCommandItem(threadId, { status: "running", output: "step 1\nstep 2\n" });
319 const update = yield* takeFrame(frames);
320 assert.deepEqual(update, {
321 kind: "replace",
322 text: "step 1\nstep 2\n",
323 truncated: false,
324 running: true,
325 });
326 }),
327 );
328
329 it.effect("an interrupted command ends the stream with the output it printed", () =>
330 Effect.gen(function* () {
331 const hub = yield* CommandOutputHub.CommandOutputHub;
332 const threadId = yield* seedThread;
333 yield* writeCommandItem(threadId, { status: "running" });
334 yield* hub.append({ threadId, itemId: ITEM_ID, chunk: "waiting for server\n" });
335 const frames = yield* subscribeFrames(threadId);
336 yield* takeFrame(frames);
337 yield* TestClock.adjust("1 second");
338 // Some providers omit stopped-command output; the live tail is released before the settled item persists.
339 yield* hub.finish({ threadId, itemId: ITEM_ID });
340 yield* writeCommandItem(threadId, { status: "interrupted" });
341 const final = yield* takeFrame(frames);
342 assert.deepEqual(final, {
343 kind: "replace",
344 text: "waiting for server\n",
345 truncated: false,
346 running: false,
347 });
348 assert.equal(yield* Queue.take(frames), "ended");
349 }),
350 );
351});