Open History to see the last commit
1// @effect-diagnostics nodeBuiltinImport:off -- the fake CLI and its leftover child are real processes.
2import * as ClaudeSdk from "@anthropic-ai/claude-agent-sdk";
3import * as NodeChildProcess from "node:child_process";
4import * as NodeEvents from "node:events";
5import * as NodeFS from "node:fs";
6import * as NodeOS from "node:os";
7import * as NodePath from "node:path";
8import * as NodeStream from "node:stream";
9import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
10import * as NodeServices from "@effect/platform-node/NodeServices";
11import { assert, describe, it } from "@effect/vitest";
12import { ProviderSessionId, ThreadId } from "@t3tools/contracts";
13import * as Effect from "effect/Effect";
14import * as Exit from "effect/Exit";
15import * as Deferred from "effect/Deferred";
16import * as Fiber from "effect/Fiber";
17import * as FileSystem from "effect/FileSystem";
18import * as Layer from "effect/Layer";
19import * as Path from "effect/Path";
20import * as Stream from "effect/Stream";
21import * as TestClock from "effect/testing/TestClock";
22import { vi } from "vite-plus/test";
23
24import * as HostProcess from "@t3tools/shared/HostProcess";
25import { AgentScope } from "@t3tools/shared/AgentScope";
26import { ProviderEventLoggers } from "@t3tools/provider-core/server/ProviderEventLoggers";
27import { makeProviderFailure } from "@t3tools/provider-core/server/failure";
28import { signalProcessGroup } from "@t3tools/provider-core/server/processGroup";
29import * as PortScanner from "../../preview/PortScanner.ts";
30import {
31 ClaudeAgentSdkQueryRunner,
32 type ClaudeAgentSdkQueryOpenInput,
33 layerQueryRunner,
34 spawnClaudeProcessGroup,
35} from "./ClaudeAdapterV2.ts";
36
37vi.mock("@anthropic-ai/claude-agent-sdk", { spy: true });
38
39const FORK_SESSION_ID = "11111111-1111-4111-8111-111111111111";
40
41const sessionEntry = (uuid: string, parentUuid: string | null, message: unknown) => ({
42 type: parentUuid === null ? "user" : "assistant",
43 uuid,
44 parentUuid,
45 sessionId: FORK_SESSION_ID,
46 cwd: "/repo",
47 isSidechain: false,
48 timestamp: "2026-10-01T00:00:00.000Z",
49 message,
50});
51
52const writeClaudeSession = Effect.fnUntraced(function* (
53 configDir: string,
54 entries: ReadonlyArray<unknown> = [
55 sessionEntry("aaaaaaaa-0000-4000-8000-000000000001", null, { role: "user", content: "hello" }),
56 sessionEntry("aaaaaaaa-0000-4000-8000-000000000002", "aaaaaaaa-0000-4000-8000-000000000001", {
57 role: "assistant",
58 content: [{ type: "text", text: "hi" }],
59 }),
60 ],
61) {
62 const fileSystem = yield* FileSystem.FileSystem;
63 const path = yield* Path.Path;
64 const projectDir = path.join(configDir, "projects", "-repo");
65 yield* fileSystem.makeDirectory(projectDir, { recursive: true });
66 yield* fileSystem.writeFileString(
67 path.join(projectDir, `${FORK_SESSION_ID}.jsonl`),
68 entries.map((line) => JSON.stringify(line)).join("\n"),
69 );
70 return projectDir;
71});
72
73const forkRunnerLayer = layerQueryRunner.pipe(
74 Layer.provide(PortScanner.layerTest),
75 Layer.provide(Layer.succeed(ProviderEventLoggers, { native: undefined, canonical: undefined })),
76);
77
78it.effect("forks a session stored under the instance's own CLAUDE_CONFIG_DIR", () =>
79 Effect.scoped(
80 Effect.gen(function* () {
81 const fileSystem = yield* FileSystem.FileSystem;
82 const path = yield* Path.Path;
83 const configDir = yield* fileSystem.makeTempDirectoryScoped({
84 prefix: "t3-claude-fork-config-",
85 });
86 const projectDir = yield* writeClaudeSession(configDir);
87 const hostConfigDir = (yield* HostProcess.Environment).CLAUDE_CONFIG_DIR;
88 const inProcessFork = vi.spyOn(ClaudeSdk, "forkSession");
89 yield* Effect.addFinalizer(() => Effect.sync(() => inProcessFork.mockRestore()));
90
91 const runner = yield* ClaudeAgentSdkQueryRunner;
92 const forked = yield* runner.forkSession({
93 sessionId: FORK_SESSION_ID,
94 options: { dir: "/repo" },
95 environment: { ...process.env, CLAUDE_CONFIG_DIR: configDir },
96 threadId: ThreadId.make("thread-fork-config-dir"),
97 providerSessionId: ProviderSessionId.make("session-fork-config-dir"),
98 });
99
100 assert.notEqual(forked.sessionId, FORK_SESSION_ID);
101 const transcript = yield* fileSystem.readFileString(
102 path.join(projectDir, `${forked.sessionId}.jsonl`),
103 );
104 assert.include(transcript, '"content":"hello"');
105 assert.include(transcript, '"text":"hi"');
106 const inheritedHead = transcript
107 .trim()
108 .split("\n")
109 .findLast((line) => line.includes('"type":"assistant"'))
110 ?.match(/"uuid"\s*:\s*"([^"]+)"/)?.[1];
111 assert.isString(inheritedHead);
112 assert.equal(forked.headMessageId, inheritedHead);
113 assert.equal((yield* HostProcess.Environment).CLAUDE_CONFIG_DIR, hostConfigDir);
114 assert.equal(inProcessFork.mock.calls.length, 0);
115 }).pipe(Effect.provide(forkRunnerLayer)),
116 ).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))),
117);
118it.effect("forked Claude sessions keep their compaction chain", () =>
119 Effect.scoped(
120 Effect.gen(function* () {
121 const fileSystem = yield* FileSystem.FileSystem;
122 const path = yield* Path.Path;
123 const configDir = yield* fileSystem.makeTempDirectoryScoped({
124 prefix: "t3-claude-fork-compact-",
125 });
126 const uuid = (n: number) => `bbbbbbbb-0000-4000-8000-00000000000${n}`;
127 const turn = (id: number, parent: string | null, role: "user" | "assistant") => ({
128 ...sessionEntry(uuid(id), parent, {
129 role,
130 content: role === "user" ? `question ${id}` : [{ type: "text", text: `answer ${id}` }],
131 }),
132 type: role,
133 });
134 const projectDir = yield* writeClaudeSession(configDir, [
135 turn(1, null, "user"),
136 turn(2, uuid(1), "assistant"),
137 turn(3, uuid(2), "user"),
138 turn(4, uuid(3), "assistant"),
139 {
140 ...sessionEntry(uuid(5), null, undefined),
141 type: "system",
142 subtype: "compact_boundary",
143 content: "Conversation compacted",
144 logicalParentUuid: uuid(4),
145 compactMetadata: {
146 trigger: "manual",
147 preTokens: 19979,
148 preservedSegment: { headUuid: uuid(3), anchorUuid: uuid(6), tailUuid: uuid(4) },
149 preservedMessages: {
150 anchorUuid: uuid(6),
151 uuids: [uuid(3), uuid(4)],
152 allUuids: [uuid(3), uuid(4)],
153 },
154 postTokens: 3589,
155 },
156 },
157 {
158 ...turn(6, uuid(5), "user"),
159 isCompactSummary: true,
160 isVisibleInTranscriptOnly: true,
161 },
162 turn(7, uuid(4), "user"),
163 turn(8, uuid(7), "assistant"),
164 ]);
165
166 const runner = yield* ClaudeAgentSdkQueryRunner;
167 const forked = yield* runner.forkSession({
168 sessionId: FORK_SESSION_ID,
169 options: { dir: "/repo" },
170 environment: { ...process.env, CLAUDE_CONFIG_DIR: configDir },
171 threadId: ThreadId.make("thread-fork-compaction"),
172 providerSessionId: ProviderSessionId.make("session-fork-compaction"),
173 });
174
175 const entries: ReadonlyArray<{
176 readonly uuid?: string;
177 readonly subtype?: string;
178 readonly compactMetadata?: {
179 readonly preservedSegment: Record<"headUuid" | "anchorUuid" | "tailUuid", string>;
180 readonly preservedMessages: {
181 readonly anchorUuid: string;
182 readonly uuids: string[];
183 readonly allUuids: string[];
184 };
185 };
186 }> = (yield* fileSystem.readFileString(path.join(projectDir, `${forked.sessionId}.jsonl`)))
187 .trim()
188 .split("\n")
189 .map((line) => JSON.parse(line));
190 const forkedUuids = new Set(entries.map((entry) => entry.uuid));
191 const boundary = entries.find((entry) => entry.subtype === "compact_boundary");
192 assert.isDefined(boundary?.compactMetadata);
193 const { preservedSegment, preservedMessages } = boundary!.compactMetadata!;
194 const danglingRefs = [
195 ...Object.values(preservedSegment),
196 preservedMessages.anchorUuid,
197 ...preservedMessages.uuids,
198 ...preservedMessages.allUuids,
199 ].filter((ref) => !forkedUuids.has(ref));
200 assert.deepEqual(danglingRefs, [], "compaction refs missing from the forked transcript");
201 }).pipe(Effect.provide(forkRunnerLayer)),
202 ).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))),
203);
204
205const realQueryRunner = layerQueryRunner.pipe(
206 Layer.provide(PortScanner.layerTest),
207 Layer.provide(
208 Layer.succeed(ProviderEventLoggers, {
209 native: undefined,
210 canonical: undefined,
211 }),
212 ),
213 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
214);
215
216const resumedMcpQueryInput = {
217 threadId: ThreadId.make("thread-restart-registration"),
218 providerSessionId: ProviderSessionId.make("session-restart-registration"),
219 options: {
220 resume: FORK_SESSION_ID,
221 model: "claude-sonnet-4-6",
222 tools: [],
223 permissionMode: "default",
224 mcpServers: {
225 "t3-code": {
226 type: "http",
227 url: "http://localhost:3773/mcp",
228 headers: { Authorization: "Bearer test-mcp-credential" },
229 },
230 },
231 },
232} satisfies ClaudeAgentSdkQueryOpenInput;
233
234it.effect("reopens a stalled MCP registration before delivering the resumed prompt", () =>
235 Effect.scoped(
236 Effect.gen(function* () {
237 const registrationStarted = Promise.withResolvers<void>();
238 const disposalStarted = Promise.withResolvers<void>();
239 const finishDisposal = Promise.withResolvers<void>();
240 const prompts: AsyncIterable<ClaudeSdk.SDKUserMessage>[] = [];
241 const firstClose = vi.fn();
242 const secondClose = vi.fn();
243 const firstRegistration = vi.fn(() => {
244 registrationStarted.resolve();
245 return new Promise<ClaudeSdk.McpSetServersResult>(() => {});
246 });
247 const secondRegistration = vi.fn(async () => ({
248 added: ["t3-code"],
249 removed: [],
250 errors: {},
251 }));
252 const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(({ prompt }) => {
253 assert.notTypeOf(prompt, "string");
254 if (typeof prompt === "string") throw new Error("Expected streaming input");
255 prompts.push(prompt);
256 return {
257 initializationResult: async () => ({}),
258 setMcpServers: prompts.length === 1 ? firstRegistration : secondRegistration,
259 close: prompts.length === 1 ? firstClose : secondClose,
260 [Symbol.asyncDispose]: async () => {
261 firstClose();
262 disposalStarted.resolve();
263 await finishDisposal.promise;
264 },
265 } as unknown as ClaudeSdk.Query;
266 });
267 yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore()));
268 const runner = yield* ClaudeAgentSdkQueryRunner;
269 const opening = yield* runner.open(resumedMcpQueryInput).pipe(Effect.forkChild);
270 yield* Effect.promise(() => registrationStarted.promise);
271 yield* TestClock.adjust("90 seconds");
272 yield* Effect.promise(() => disposalStarted.promise);
273 assert.equal(querySpy.mock.calls.length, 1);
274 finishDisposal.resolve();
275 const session = yield* Fiber.join(opening);
276 yield* Effect.addFinalizer(() => session.close.pipe(Effect.orDie));
277
278 assert.equal(firstClose.mock.calls.length, 1);
279 assert.equal(secondClose.mock.calls.length, 0);
280 assert.equal(querySpy.mock.calls.length, 2);
281 for (const [call] of querySpy.mock.calls) {
282 assert.equal(call.options?.resume, FORK_SESSION_ID);
283 assert.isUndefined(call.options?.mcpServers);
284 }
285 assert.deepEqual(secondRegistration.mock.calls, [[resumedMcpQueryInput.options.mcpServers]]);
286 const message = {
287 type: "user",
288 message: { role: "user", content: "Continue where you left off." },
289 parent_tool_use_id: null,
290 session_id: FORK_SESSION_ID,
291 } satisfies ClaudeSdk.SDKUserMessage;
292 yield* session.offer(message);
293 assert.isTrue((yield* Effect.promise(() => prompts[0]![Symbol.asyncIterator]().next())).done);
294 const delivered = yield* Effect.promise(() => prompts[1]![Symbol.asyncIterator]().next());
295 assert.deepEqual(delivered.value, message);
296 assert.isFalse(delivered.done);
297 }).pipe(Effect.provide(realQueryRunner)),
298 ),
299);
300
301it.effect.each([
302 { case: "a readable rule list", errors: undefined, readable: true },
303 { case: "a settings file Claude skipped", errors: [{ file: "settings.json" }], readable: false },
304 { case: "a reply without a rule list", response: {}, readable: false },
305])("reads Claude's permission rules only when no settings file was skipped: $case", (testCase) =>
306 Effect.scoped(
307 Effect.gen(function* () {
308 const rules = [{ behavior: "deny" as const, rule: "mcp__probe__redeploy" }];
309 const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(
310 () =>
311 ({
312 close: vi.fn(),
313 request: async () => ({
314 response:
315 "response" in testCase
316 ? testCase.response
317 : { state: { rules, errors: testCase.errors } },
318 }),
319 }) as unknown as ClaudeSdk.Query,
320 );
321 yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore()));
322 const runner = yield* ClaudeAgentSdkQueryRunner;
323 const { mcpServers: _mcpServers, ...options } = resumedMcpQueryInput.options;
324 const session = yield* runner.open({ ...resumedMcpQueryInput, options });
325 yield* Effect.addFinalizer(() => session.close.pipe(Effect.orDie));
326 const exit = yield* Effect.exit(session.listPermissionRules);
327 if (testCase.readable) assert.deepEqual(exit, Exit.succeed(rules));
328 else assert.isTrue(Exit.isFailure(exit));
329 }).pipe(Effect.provide(realQueryRunner)),
330 ),
331);
332
333it.effect.each([true, false])("only retries after confirmed process exit: %s", (exits) =>
334 Effect.scoped(
335 Effect.gen(function* () {
336 const registrationStarted = Promise.withResolvers<void>();
337 const cleanupProgress = Promise.withResolvers<"kill" | "retry">();
338 const events = new NodeEvents.EventEmitter();
339 const stdin = new NodeStream.PassThrough();
340 const stdout = new NodeStream.PassThrough();
341 const child = {
342 stdin,
343 stdout,
344 killed: false,
345 exitCode: null,
346 kill: vi.fn(() => {
347 cleanupProgress.resolve("kill");
348 return true;
349 }),
350 on: events.on.bind(events),
351 once: events.once.bind(events),
352 off: events.off.bind(events),
353 } satisfies ClaudeSdk.SpawnedProcess;
354 const signal = yield* Effect.abortSignal;
355 const disposed = vi.fn(async () => {});
356 const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(({ options }) => {
357 const first = querySpy.mock.calls.length === 1;
358 if (first) {
359 options?.spawnClaudeCodeProcess?.({ command: "claude", args: [], env: {}, signal });
360 } else {
361 cleanupProgress.resolve("retry");
362 }
363 return {
364 initializationResult: async () => ({}),
365 setMcpServers: () => {
366 registrationStarted.resolve();
367 return first
368 ? new Promise<ClaudeSdk.McpSetServersResult>(() => {})
369 : Promise.resolve({ added: ["t3-code"], removed: [], errors: {} });
370 },
371 close: vi.fn(),
372 [Symbol.asyncDispose]: disposed,
373 } as unknown as ClaudeSdk.Query;
374 });
375 yield* Effect.addFinalizer(() =>
376 Effect.sync(() => {
377 querySpy.mockRestore();
378 stdin.destroy();
379 stdout.destroy();
380 }),
381 );
382 const runner = yield* ClaudeAgentSdkQueryRunner;
383 const opening = yield* runner
384 .open({
385 ...resumedMcpQueryInput,
386 options: { ...resumedMcpQueryInput.options, spawnClaudeCodeProcess: () => child },
387 })
388 .pipe(Effect.exit, Effect.forkChild);
389 yield* Effect.promise(() => registrationStarted.promise);
390 yield* TestClock.adjust("90 seconds");
391
392 assert.equal(yield* Effect.promise(() => cleanupProgress.promise), "kill");
393 assert.equal(querySpy.mock.calls.length, 1);
394 assert.deepEqual(child.kill.mock.calls, [["SIGKILL"]]);
395 if (exits) events.emit("exit", null, "SIGKILL");
396 else yield* TestClock.adjust("10 seconds");
397 const result = yield* Fiber.join(opening);
398 assert.equal(Exit.isSuccess(result), exits);
399 if (Exit.isSuccess(result)) yield* result.value.close;
400 assert.equal(querySpy.mock.calls.length, exits ? 2 : 1);
401 assert.equal(disposed.mock.calls.length, 1);
402 }).pipe(Effect.provide(realQueryRunner)),
403 ),
404);
405
406it.effect("stops retrying MCP registration after two timeouts and reports the cause", () =>
407 Effect.scoped(
408 Effect.gen(function* () {
409 const firstRegistration = Promise.withResolvers<void>();
410 const secondRegistration = Promise.withResolvers<void>();
411 const disposed = vi.fn(async () => {});
412 const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(() => {
413 const started = querySpy.mock.calls.length === 1 ? firstRegistration : secondRegistration;
414 return {
415 initializationResult: async () => ({}),
416 setMcpServers: () => {
417 started.resolve();
418 return new Promise<ClaudeSdk.McpSetServersResult>(() => {});
419 },
420 [Symbol.asyncDispose]: disposed,
421 } as unknown as ClaudeSdk.Query;
422 });
423 yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore()));
424 const runner = yield* ClaudeAgentSdkQueryRunner;
425 const opening = yield* runner
426 .open(resumedMcpQueryInput)
427 .pipe(Effect.result, Effect.forkChild);
428 yield* Effect.promise(() => firstRegistration.promise);
429 yield* TestClock.adjust("90 seconds");
430 yield* Effect.promise(() => secondRegistration.promise);
431 yield* TestClock.adjust("90 seconds");
432 const result = yield* Fiber.join(opening);
433
434 assert.equal(result._tag, "Failure");
435 if (result._tag !== "Failure") return;
436 assert.equal(querySpy.mock.calls.length, 2);
437 assert.equal(disposed.mock.calls.length, 2);
438 assert.equal(
439 makeProviderFailure({
440 cause: { _tag: "ProviderAdapterTurnStartError", cause: result.failure },
441 }).message,
442 "Claude timed out while registering its MCP tools. Retry the turn; if it keeps failing, check Claude and the server logs.",
443 );
444 }).pipe(Effect.provide(realQueryRunner)),
445 ),
446);
447
448it.effect("cancels a stalled registration without retrying or retaining its prompt queue", () =>
449 Effect.scoped(
450 Effect.gen(function* () {
451 const registrationStarted = Promise.withResolvers<void>();
452 const disposed = vi.fn(async () => {});
453 const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(
454 () =>
455 ({
456 initializationResult: async () => ({}),
457 setMcpServers: () => {
458 registrationStarted.resolve();
459 return new Promise<ClaudeSdk.McpSetServersResult>(() => {});
460 },
461 [Symbol.asyncDispose]: disposed,
462 }) as unknown as ClaudeSdk.Query,
463 );
464 yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore()));
465 const runner = yield* ClaudeAgentSdkQueryRunner;
466 const opening = yield* runner.open(resumedMcpQueryInput).pipe(Effect.forkChild);
467 yield* Effect.promise(() => registrationStarted.promise);
468 yield* Fiber.interrupt(opening);
469
470 assert.equal(querySpy.mock.calls.length, 1);
471 assert.equal(disposed.mock.calls.length, 1);
472 const prompt = querySpy.mock.calls[0]![0].prompt;
473 if (typeof prompt === "string") throw new Error("Expected streaming input");
474 assert.isTrue((yield* Effect.promise(() => prompt[Symbol.asyncIterator]().next())).done);
475 }).pipe(Effect.provide(realQueryRunner)),
476 ),
477);
478
479it.effect("does not reuse a fresh session ID after MCP registration times out", () =>
480 Effect.scoped(
481 Effect.gen(function* () {
482 const registrationStarted = Promise.withResolvers<void>();
483 const disposed = vi.fn(async () => {});
484 const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(
485 () =>
486 ({
487 initializationResult: async () => ({}),
488 setMcpServers: () => {
489 registrationStarted.resolve();
490 return new Promise<ClaudeSdk.McpSetServersResult>(() => {});
491 },
492 [Symbol.asyncDispose]: disposed,
493 }) as unknown as ClaudeSdk.Query,
494 );
495 yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore()));
496 const runner = yield* ClaudeAgentSdkQueryRunner;
497 const { resume, ...options } = resumedMcpQueryInput.options;
498 const opening = yield* runner
499 .open({
500 ...resumedMcpQueryInput,
501 options: { ...options, sessionId: resume },
502 })
503 .pipe(Effect.result, Effect.forkChild);
504 yield* Effect.promise(() => registrationStarted.promise);
505 yield* TestClock.adjust("90 seconds");
506 const result = yield* Fiber.join(opening);
507
508 assert.equal(result._tag, "Failure");
509 assert.equal(querySpy.mock.calls.length, 1);
510 assert.equal(disposed.mock.calls.length, 1);
511 }).pipe(Effect.provide(realQueryRunner)),
512 ),
513);
514
515it.effect.each([
516 ["opens after a slow CLI boot", "120 seconds", true],
517 ["times out a CLI that never boots", "5 minutes", false],
518] as const)("%s", ([, bootWait, boots]) =>
519 Effect.scoped(
520 Effect.gen(function* () {
521 const waiting = Promise.withResolvers<void>();
522 const booted = Promise.withResolvers<void>();
523 const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(
524 () =>
525 ({
526 initializationResult: () => {
527 waiting.resolve();
528 return booted.promise;
529 },
530 // The CLI answers control requests only after it has booted.
531 setMcpServers: async () => {
532 waiting.resolve();
533 await booted.promise;
534 return { added: ["t3-code"], removed: [], errors: {} };
535 },
536 close: vi.fn(),
537 [Symbol.asyncDispose]: async () => {},
538 }) as unknown as ClaudeSdk.Query,
539 );
540 yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore()));
541 const runner = yield* ClaudeAgentSdkQueryRunner;
542 const { resume, ...options } = resumedMcpQueryInput.options;
543 const opening = yield* runner
544 .open({ ...resumedMcpQueryInput, options: { ...options, sessionId: resume } })
545 .pipe(Effect.result, Effect.forkChild);
546 yield* Effect.promise(() => waiting.promise);
547 yield* TestClock.adjust(bootWait);
548 if (boots) booted.resolve();
549 const result = yield* Fiber.join(opening);
550
551 assert.equal(result._tag, boots ? "Success" : "Failure");
552 if (result._tag === "Success") yield* result.success.close;
553 else
554 assert.equal(
555 makeProviderFailure({
556 cause: { _tag: "ProviderAdapterTurnStartError", cause: result.failure },
557 }).message,
558 "Claude timed out while starting up. Retry the turn; if it keeps failing, check Claude and the server logs.",
559 );
560 }).pipe(Effect.provide(realQueryRunner)),
561 ),
562);
563
564function fakeClaudeCli(script: string): string {
565 const dir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-claude-cli-"));
566 const cli = NodePath.join(dir, "claude");
567 NodeFS.writeFileSync(cli, `#!/bin/sh\n${script}\n`, { mode: 0o755 });
568 return cli;
569}
570
571const openFakeCli = (cli: string) =>
572 Effect.gen(function* () {
573 const runner = yield* ClaudeAgentSdkQueryRunner;
574 return yield* runner.open({
575 threadId: ThreadId.make("thread-fake-cli"),
576 providerSessionId: ProviderSessionId.make("session-fake-cli"),
577 options: {
578 sessionId: "native-session-fake-cli",
579 model: "claude-sonnet-4-6",
580 tools: [],
581 permissionMode: "default",
582 pathToClaudeCodeExecutable: cli,
583 },
584 });
585 });
586
587describe.skipIf(HostProcess.Platform.defaultValue() === "win32")("Claude CLI process", () => {
588 // The leftover child holds a FIFO open; its death produces EOF without polling.
589 it.effect("stops what the CLI left running when the session closes", () =>
590 Effect.gen(function* () {
591 const cli = fakeClaudeCli(
592 [
593 'dir=$(dirname "$0")',
594 'sleep 600 > "$dir/alive" &',
595 'echo $! > "$dir/leftover.pid"',
596 "cat > /dev/null",
597 ].join("\n"),
598 );
599 const dir = NodePath.dirname(cli);
600 NodeChildProcess.execFileSync("mkfifo", [NodePath.join(dir, "alive")]);
601 yield* Effect.addFinalizer(() =>
602 Effect.sync(() => {
603 try {
604 process.kill(Number(NodeFS.readFileSync(NodePath.join(dir, "leftover.pid"), "utf8")));
605 } catch {}
606 NodeFS.rmSync(dir, { recursive: true, force: true });
607 }),
608 );
609 const alive = NodeFS.createReadStream(NodePath.join(dir, "alive"));
610 const leftoverStarted = new Promise<void>((resolve) => alive.once("open", () => resolve()));
611 const leftoverStopped = new Promise<void>((resolve) => alive.once("close", () => resolve()));
612 alive.resume();
613
614 const stopped = yield* Deferred.make<string>();
615 const session = yield* openFakeCli(cli).pipe(
616 Effect.provide(
617 realQueryRunner.pipe(
618 Layer.provide(
619 Layer.succeed(AgentScope, {
620 ...AgentScope.defaultValue(),
621 wrap: ({ command, args }) =>
622 Effect.succeed({ command, args, unit: "t3code-claude-test.scope" }),
623 stop: (unit) => Deferred.succeed(stopped, unit).pipe(Effect.asVoid),
624 }),
625 ),
626 ),
627 ),
628 );
629 yield* Effect.promise(() => leftoverStarted);
630 yield* session.close;
631 yield* Effect.promise(() => leftoverStopped);
632 assert.strictEqual(yield* Deferred.await(stopped), "t3code-claude-test.scope");
633 }),
634 );
635
636 // The default AgentScope is what macOS gets: no systemd scope to catch a job outside the CLI's group.
637 it.effect("stops a job the CLI started in its own process group when the session closes", () =>
638 Effect.gen(function* () {
639 const cli = fakeClaudeCli(
640 [
641 'dir=$(dirname "$0")',
642 `perl -e 'setpgrp(0, 0); exec @ARGV' sleep 600 > "$dir/alive" &`,
643 'echo $! > "$dir/leftover.pid"',
644 "cat > /dev/null",
645 ].join("\n"),
646 );
647 const dir = NodePath.dirname(cli);
648 NodeChildProcess.execFileSync("mkfifo", [NodePath.join(dir, "alive")]);
649 yield* Effect.addFinalizer(() =>
650 Effect.sync(() => {
651 try {
652 process.kill(Number(NodeFS.readFileSync(NodePath.join(dir, "leftover.pid"), "utf8")));
653 } catch {}
654 NodeFS.rmSync(dir, { recursive: true, force: true });
655 }),
656 );
657 const alive = NodeFS.createReadStream(NodePath.join(dir, "alive"));
658 const leftoverStarted = new Promise<void>((resolve) => alive.once("open", () => resolve()));
659 const leftoverStopped = new Promise<void>((resolve) => alive.once("close", () => resolve()));
660 alive.resume();
661
662 const session = yield* openFakeCli(cli).pipe(Effect.provide(realQueryRunner));
663 yield* Effect.promise(() => leftoverStarted);
664 yield* session.close;
665 yield* Effect.promise(() => leftoverStopped);
666 }),
667 );
668
669 // Live clock: the job is found by polling while the CLI runs, and the CLI exits on its own.
670 it.live("stops a job the CLI started in its own process group when the CLI exits", () =>
671 Effect.gen(function* () {
672 const cli = fakeClaudeCli(
673 [
674 'dir=$(dirname "$0")',
675 `perl -e 'setpgrp(0, 0); exec @ARGV' sleep 600 > "$dir/alive" &`,
676 'echo $! > "$dir/leftover.pid"',
677 "sleep 3",
678 ].join("\n"),
679 );
680 const dir = NodePath.dirname(cli);
681 NodeChildProcess.execFileSync("mkfifo", [NodePath.join(dir, "alive")]);
682 yield* Effect.addFinalizer(() =>
683 Effect.sync(() => {
684 try {
685 process.kill(Number(NodeFS.readFileSync(NodePath.join(dir, "leftover.pid"), "utf8")));
686 } catch {}
687 NodeFS.rmSync(dir, { recursive: true, force: true });
688 }),
689 );
690 const alive = NodeFS.createReadStream(NodePath.join(dir, "alive"));
691 const leftoverStopped = new Promise<void>((resolve) => alive.once("close", () => resolve()));
692 alive.resume();
693
694 yield* openFakeCli(cli).pipe(Effect.provide(realQueryRunner));
695 yield* Effect.promise(() => leftoverStopped);
696 }),
697 );
698
699 const rootRefusal =
700 "--dangerously-skip-permissions cannot be used with root/sudo privileges for security reasons";
701
702 const startupFailureMessage = (stderr: string) =>
703 Effect.gen(function* () {
704 const cli = fakeClaudeCli(`printf '%s\\n' '${stderr}' >&2\nexit 1`);
705 yield* Effect.addFinalizer(() =>
706 Effect.sync(() => NodeFS.rmSync(NodePath.dirname(cli), { recursive: true, force: true })),
707 );
708 const session = yield* openFakeCli(cli);
709 const exit = yield* Stream.runDrain(session.messages).pipe(Effect.exit);
710 assert.isTrue(Exit.isFailure(exit));
711 return Exit.isFailure(exit)
712 ? makeProviderFailure({ cause: exit.cause, class: "transport_error" }).message
713 : "";
714 });
715
716 // Startup stderr can carry paths, tokens, or settings, so only the known
717 // refusal reaches the failure the client shows.
718 it.effect("keeps other startup stderr out of the turn failure", () =>
719 Effect.gen(function* () {
720 const surrounded = yield* startupFailureMessage(
721 `token=sk-private-123\n${rootRefusal}\nat /home/private/settings.json`,
722 );
723 assert.include(surrounded, rootRefusal);
724 assert.notInclude(surrounded, "sk-private-123");
725 assert.notInclude(surrounded, "/home/private");
726
727 const unknown = yield* startupFailureMessage("token=sk-private-123");
728 assert.notInclude(unknown, "sk-private-123");
729 }).pipe(Effect.provide(realQueryRunner)),
730 );
731
732 // The CLI keeps running its turn after its pipes close, so a server that
733 // crashed would leave it working next to the session it resumes.
734 it.effect("stops the CLI when the server that started it goes away", () =>
735 Effect.gen(function* () {
736 const cli = fakeClaudeCli(['dir=$(dirname "$0")', 'sleep 600 > "$dir/alive"'].join("\n"));
737 const dir = NodePath.dirname(cli);
738 NodeChildProcess.execFileSync("mkfifo", [NodePath.join(dir, "alive")]);
739 const child = spawnClaudeProcessGroup({
740 command: cli,
741 args: [],
742 cwd: dir,
743 env: process.env,
744 // @effect-diagnostics-next-line abortControllerInEffect:off - This crash fixture closes pipes itself; aborting on test teardown raises an unrelated process error.
745 signal: new AbortController().signal,
746 }) as NodeChildProcess.ChildProcess;
747 yield* Effect.addFinalizer(() =>
748 Effect.sync(() => {
749 try {
750 signalProcessGroup(child.pid ?? 0, "SIGKILL");
751 } catch {}
752 NodeFS.rmSync(dir, { recursive: true, force: true });
753 }),
754 );
755 const alive = NodeFS.createReadStream(NodePath.join(dir, "alive"));
756 const turnStarted = new Promise<void>((resolve) => alive.once("open", () => resolve()));
757 const turnStopped = new Promise<void>((resolve) => alive.once("close", () => resolve()));
758 alive.resume();
759 yield* Effect.promise(() => turnStarted);
760
761 // A dying server's descriptors close; the kernel would do this for it.
762 for (const pipe of child.stdio) pipe?.destroy();
763 yield* Effect.promise(() => turnStopped);
764 }),
765 );
766
767 it.effect("marks the CLI's jobs with its thread and counts it as an agent until it exits", () => {
768 let runningAgents = 0;
769 let releaseAgent!: () => void;
770 const agentReleased = new Promise<void>((resolve) => (releaseAgent = resolve));
771 const runner = layerQueryRunner.pipe(
772 Layer.provide(
773 Layer.mock(PortScanner.PortDiscovery)({
774 trackAgent: () => {
775 runningAgents += 1;
776 return () => {
777 runningAgents -= 1;
778 releaseAgent();
779 };
780 },
781 }),
782 ),
783 Layer.provide(
784 Layer.succeed(ProviderEventLoggers, { native: undefined, canonical: undefined }),
785 ),
786 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
787 );
788 return Effect.gen(function* () {
789 const cli = fakeClaudeCli(
790 [
791 'dir=$(dirname "$0")',
792 `printf '%s' "$${PortScanner.AGENT_THREAD_ENV}" > "$dir/thread"`,
793 "cat > /dev/null",
794 ].join("\n"),
795 );
796 const dir = NodePath.dirname(cli);
797 NodeChildProcess.execFileSync("mkfifo", [NodePath.join(dir, "thread")]);
798 yield* Effect.addFinalizer(() =>
799 Effect.sync(() => NodeFS.rmSync(dir, { recursive: true, force: true })),
800 );
801 const threadRead = new Promise<string>((resolve) => {
802 let thread = "";
803 NodeFS.createReadStream(NodePath.join(dir, "thread"), "utf8")
804 .on("data", (chunk) => (thread += chunk))
805 .once("end", () => resolve(thread));
806 });
807
808 const session = yield* openFakeCli(cli);
809 assert.equal(yield* Effect.promise(() => threadRead), "thread-fake-cli");
810 assert.equal(runningAgents, 1);
811 yield* session.close;
812 yield* Effect.promise(() => agentReleased);
813 assert.equal(runningAgents, 0);
814 }).pipe(Effect.provide(runner));
815 });
816});