Open History to see the last commit
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Fiber from "effect/Fiber";
6import * as FileSystem from "effect/FileSystem";
7import * as Path from "effect/Path";
8import * as Queue from "effect/Queue";
9
10import { claudeTaskOutputRoot, tailClaudeTaskOutput } from "./ClaudeTaskOutputTail.ts";
11
12const SESSION_ID = "0c91bb28-9e89-4889-bd22-3de9880233d3";
13const TASK_ID = "bk6xreeye";
14
15const makeTaskFile = Effect.gen(function* () {
16 const fileSystem = yield* FileSystem.FileSystem;
17 const path = yield* Path.Path;
18 const root = yield* fileSystem.makeTempDirectoryScoped({ prefix: "claude-task-output-" });
19 const tasks = path.join(root, "-private-tmp-project", SESSION_ID, "tasks");
20 return { root, tasks, file: path.join(tasks, `${TASK_ID}.output`) };
21});
22
23// TestClock cannot interleave with real filesystem I/O; tests that wait on a poll use the live clock.
24describe("tailClaudeTaskOutput", () => {
25 it.live("follows a file that appears after the task starts, across split characters", () =>
26 Effect.scoped(
27 Effect.gen(function* () {
28 const fileSystem = yield* FileSystem.FileSystem;
29 const { root, tasks, file } = yield* makeTaskFile;
30 const chunks = yield* Queue.unbounded<string>();
31 yield* tailClaudeTaskOutput({
32 root,
33 sessionId: SESSION_ID,
34 taskId: TASK_ID,
35 stop: yield* Deferred.make<void>(),
36 onChunk: (chunk) => Queue.offer(chunks, chunk),
37 }).pipe(Effect.forkScoped);
38
39 yield* fileSystem.makeDirectory(tasks, { recursive: true });
40 // Split the two-byte é across writes.
41 const encoded = new TextEncoder().encode("café 1\n");
42 yield* fileSystem.writeFile(file, encoded.subarray(0, 4));
43 assert.equal(yield* Queue.take(chunks), "caf");
44
45 yield* fileSystem.writeFile(file, encoded);
46 assert.equal(yield* Queue.take(chunks), "é 1\n");
47 }),
48 ).pipe(Effect.provide(NodeServices.layer)),
49 );
50
51 it.effect("drops a split character when the file is truncated and rewritten", () =>
52 Effect.scoped(
53 Effect.gen(function* () {
54 const fileSystem = yield* FileSystem.FileSystem;
55 const { root, tasks, file } = yield* makeTaskFile;
56 yield* fileSystem.makeDirectory(tasks, { recursive: true });
57 // End the first read inside the two-byte é.
58 yield* fileSystem.writeFile(file, new TextEncoder().encode("café").subarray(0, 4));
59 const chunks = yield* Queue.unbounded<string>();
60 const stop = yield* Deferred.make<void>();
61 yield* tailClaudeTaskOutput({
62 root,
63 sessionId: SESSION_ID,
64 taskId: TASK_ID,
65 stop,
66 onChunk: (chunk) => Queue.offer(chunks, chunk),
67 }).pipe(Effect.forkScoped);
68 assert.equal(yield* Queue.take(chunks), "caf");
69
70 yield* fileSystem.writeFileString(file, "ok\n");
71 yield* Deferred.succeed(stop, undefined);
72 assert.equal(yield* Queue.take(chunks), "ok\n");
73 }),
74 ).pipe(Effect.provide(NodeServices.layer)),
75 );
76
77 it.live("skips ahead when the command writes faster than it is read", () =>
78 Effect.scoped(
79 Effect.gen(function* () {
80 const fileSystem = yield* FileSystem.FileSystem;
81 const { root, tasks, file } = yield* makeTaskFile;
82 yield* fileSystem.makeDirectory(tasks, { recursive: true });
83 yield* fileSystem.writeFileString(file, "start\n");
84 const chunks = yield* Queue.unbounded<string>();
85 yield* tailClaudeTaskOutput({
86 root,
87 sessionId: SESSION_ID,
88 taskId: TASK_ID,
89 stop: yield* Deferred.make<void>(),
90 onChunk: (chunk) => Queue.offer(chunks, chunk),
91 }).pipe(Effect.forkScoped);
92 assert.equal(yield* Queue.take(chunks), "start\n");
93
94 // The skip lands on the second byte of a two-byte é.
95 yield* fileSystem.writeFileString(file, `start\n${"é\n".repeat(1_000_000)}end.\n`);
96 const flooded = yield* Queue.take(chunks);
97 assert.isAtMost(flooded.length, 256 * 1024 + 1);
98 assert.equal(flooded.slice(0, 3), "\n\né");
99 assert.isTrue(flooded.endsWith("é\nend.\n"));
100 }),
101 ).pipe(Effect.provide(NodeServices.layer)),
102 );
103
104 it.effect("reads what was written last when the tool result stops it", () =>
105 Effect.gen(function* () {
106 const fileSystem = yield* FileSystem.FileSystem;
107 const { root, tasks, file } = yield* makeTaskFile;
108 yield* fileSystem.makeDirectory(tasks, { recursive: true });
109 yield* fileSystem.writeFileString(file, "one\n");
110 const chunks: Array<string> = [];
111 const stop = yield* Deferred.make<void>();
112 const first = yield* Deferred.make<void>();
113 const fiber = yield* tailClaudeTaskOutput({
114 root,
115 sessionId: SESSION_ID,
116 taskId: TASK_ID,
117 stop,
118 onChunk: (chunk) =>
119 Effect.sync(() => chunks.push(chunk)).pipe(
120 Effect.andThen(Deferred.succeed(first, undefined)),
121 ),
122 }).pipe(Effect.forkScoped);
123 yield* Deferred.await(first);
124 yield* fileSystem.writeFileString(file, "one\ntwo\n");
125 yield* Deferred.succeed(stop, undefined);
126 yield* Fiber.join(fiber);
127 assert.deepEqual(chunks, ["one\n", "two\n"]);
128 }).pipe(Effect.provide(NodeServices.layer)),
129 );
130});
131
132describe("claudeTaskOutputRoot", () => {
133 it.effect.skipIf(process.getuid === undefined)(
134 "uses the temp dir Claude Code resolves from its environment on each platform",
135 () =>
136 Effect.gen(function* () {
137 const path = yield* Path.Path;
138 const user = `claude-${process.getuid?.()}`;
139 const root = (
140 environment: Record<string, string>,
141 platform: NodeJS.Platform,
142 cwd: string | null = null,
143 ) => claudeTaskOutputRoot(environment, path, cwd, platform);
144 assert.equal(root({ TMPDIR: "/var/folders/x/T/" }, "linux"), `/var/folders/x/T/${user}`);
145 assert.equal(root({ TMP: "/tmp/user/1000" }, "linux"), `/tmp/user/1000/${user}`);
146 assert.equal(root({}, "linux"), `/tmp/${user}`);
147 assert.equal(root({ TMPDIR: "/var/folders/x/T/" }, "darwin"), `/tmp/${user}`);
148 assert.equal(
149 root({ CLAUDE_CODE_TMPDIR: "/srv/claude", TMPDIR: "/var/folders/x/T/" }, "linux"),
150 `/srv/claude/${user}`,
151 );
152 assert.equal(
153 root({ CLAUDE_CODE_TMPDIR: "/srv/claude", TMPDIR: "/var/folders/x/T/" }, "darwin"),
154 `/srv/claude/${user}`,
155 );
156 assert.equal(
157 root({ TMPDIR: "tmp" }, "linux", "/work/project"),
158 `/work/project/tmp/${user}`,
159 );
160 assert.equal(
161 root({ CLAUDE_CODE_TMPDIR: "../claude" }, "darwin", "/work/project"),
162 `/work/claude/${user}`,
163 );
164 }).pipe(Effect.provide(NodeServices.layer)),
165 );
166});