diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index d7d03c779b4bb..86f82d6e2f3d0 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -2634,6 +2634,10 @@ function makeFakeClaudeCli(answerSetServers = true) { const spawns: Array = []; const controlRequests: Array = []; const stdins: Array = []; + const processes: Array<{ + readonly stdout: NodeStream.PassThrough; + readonly exit: (code: number, endStdout: boolean) => void; + }> = []; const spawn = (options: SpawnOptions): SpawnedProcess => { spawns.push(options); const stdin = new NodeStream.PassThrough(); @@ -2650,6 +2654,14 @@ function makeFakeClaudeCli(answerSetServers = true) { return true; }, }); + processes.push({ + stdout, + exit: (code, endStdout) => { + child.exitCode = code; + child.emit("exit", code, null); + if (endStdout) stdout.end(); + }, + }); let pending = ""; stdin.on("data", (chunk: Buffer) => { pending += chunk.toString("utf8"); @@ -2675,10 +2687,10 @@ function makeFakeClaudeCli(answerSetServers = true) { return child; }; const closed = () => stdins.length > 0 && stdins.every((stdin) => stdin.writableEnded); - return { spawn, spawns, controlRequests, closed }; + return { spawn, spawns, controlRequests, closed, processes }; } -describe("ClaudeAdapterV2 MCP credential channel", () => { +describe("ClaudeAdapterV2 SDK query runner", () => { const t3McpServer = { type: "http" as const, url: "http://127.0.0.1:43123/mcp", @@ -2716,6 +2728,55 @@ describe("ClaudeAdapterV2 MCP credential channel", () => { Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), ); + it.effect.each([0, 1])( + "fails the message stream when the child exits %s but stdout stays open", + (code) => + Effect.gen(function* () { + const cli = makeFakeClaudeCli(); + const session = yield* openWith(cli); + const completed = + yield* Deferred.make>(); + yield* session.messages.pipe( + Stream.runDrain, + Effect.exit, + Effect.flatMap((exit) => Deferred.succeed(completed, exit)), + Effect.forkChild, + ); + cli.processes[0]!.exit(code, false); + yield* TestClock.adjust("2 seconds"); + + assert.isTrue(yield* Deferred.isDone(completed)); + const exit = yield* Deferred.await(completed); + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + assert.equal( + makeProviderFailure({ cause: exit.cause, class: "transport_error" }).message, + "Claude Code exited before its message stream closed. Retry the turn.", + ); + } + yield* session.close; + }), + ); + + it.effect("drains buffered messages when the child exits before stream consumption", () => + Effect.gen(function* () { + const cli = makeFakeClaudeCli(); + const session = yield* openWith(cli); + const message: SDKMessage = { + type: "system", + subtype: "status", + status: null, + uuid: "00000000-0000-4000-8000-000000000001", + session_id: "native-thread-claude-mcp-channel", + }; + cli.processes[0]!.stdout.write(`${JSON.stringify(message)}\n`); + cli.processes[0]!.exit(0, true); + + assert.deepEqual(yield* session.messages.pipe(Stream.runCollect), [message]); + yield* session.close; + }), + ); + it.effect("commits queued delivery before interruptible protocol logging", () => Effect.gen(function* () { const logging = yield* Deferred.make(); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index df80f1316fb77..3b2a5c2fa674b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -352,6 +352,8 @@ const CLAUDE_STOP_CONFIRM_TIMEOUT_MS = 10_000; const CLAUDE_STDERR_TAIL_LENGTH = 2_048; // The process group is killed on exit, so its stderr closes promptly. const CLAUDE_STDERR_DRAIN_TIMEOUT = Duration.seconds(1); +// Process exit can precede buffered SDK messages, but descendants can keep stdout open forever. +const CLAUDE_QUERY_EXIT_DRAIN_TIMEOUT = Duration.seconds(1); // Claude Code's refusal of bypassPermissions as root, which it explains only on stderr. const CLAUDE_ROOT_REFUSAL = "--dangerously-skip-permissions cannot be used with root/sudo privileges for security reasons"; @@ -610,6 +612,15 @@ export class ClaudeCodeRootRefusalError extends Schema.TaggedError()( + "ClaudeProcessExitError", + {}, +) { + override get message(): string { + return "Claude Code exited before its message stream closed. Retry the turn."; + } +} + class ClaudeMcpRegistrationTimeoutError extends Schema.TaggedError()( "ClaudeMcpRegistrationTimeoutError", {}, @@ -1252,7 +1263,7 @@ export const layerQueryRunner: Layer.Layer< }); let spawnedProcess: ClaudeSpawnedProcess | undefined; let processExited = false; - let processExit: Promise | undefined; + const processExit = Promise.withResolvers(); const spawnClaudeCodeProcess = (spawnOptions: ClaudeSpawnOptions) => { const child = spawnProcess(spawnOptions); spawnedProcess = child; @@ -1260,15 +1271,20 @@ export const layerQueryRunner: Layer.Layer< const releaseAgent = portDiscovery.trackAgent(); child.once("exit", releaseAgent); child.on("error", releaseAgent); - processExited = child.exitCode !== null || child.signalCode != null; - processExit = new Promise((resolve) => { - if (processExited) resolve(); - else - child.once("exit", () => { - processExited = true; - resolve(); - }); + child.once("error", () => { + // A failed spawn emits no exit event and never receives a pid. + if ("pid" in child && child.pid === undefined) { + processExited = true; + processExit.resolve(); + } }); + processExited = child.exitCode !== null || child.signalCode != null; + if (processExited) processExit.resolve(); + else + child.once("exit", () => { + processExited = true; + processExit.resolve(); + }); return child; }; const queryRuntime = yield* Effect.uninterruptibleMask((restore) => @@ -1335,10 +1351,10 @@ export const layerQueryRunner: Layer.Layer< Effect.andThen( Effect.tryPromise({ try: async () => { - if (spawnedProcess !== undefined && !processExited) { - spawnedProcess.kill("SIGKILL"); + if (spawnedProcess !== undefined) { + if (!processExited) spawnedProcess.kill("SIGKILL"); + await processExit.promise; } - await processExit; }, catch: (cause) => queryRunnerError(cause, "close"), }).pipe(Effect.timeout("10 seconds")), @@ -1362,6 +1378,14 @@ export const layerQueryRunner: Layer.Layer< claudeQueryMessages(queryRuntime), (cause) => cause, ).pipe( + Stream.interruptWhen( + Effect.promise(() => processExit.promise).pipe( + Effect.andThen(Effect.sleep(CLAUDE_QUERY_EXIT_DRAIN_TIMEOUT)), + Effect.andThen( + Effect.fail(queryRunnerError(new ClaudeProcessExitError(), "processExit")), + ), + ), + ), Stream.catch((cause) => Stream.fromEffect( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts index 1ea0a62e168ce..5ad70c26897a6 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts @@ -340,6 +340,7 @@ it.effect.each([true, false])("only retries after confirmed process exit: %s", ( const child = { stdin, stdout, + pid: 1234, killed: false, exitCode: null, kill: vi.fn(() => { @@ -349,7 +350,7 @@ it.effect.each([true, false])("only retries after confirmed process exit: %s", ( on: events.on.bind(events), once: events.once.bind(events), off: events.off.bind(events), - } satisfies ClaudeSdk.SpawnedProcess; + } satisfies ClaudeSdk.SpawnedProcess & { readonly pid: number }; const signal = yield* Effect.abortSignal; const disposed = vi.fn(async () => {}); const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(({ options }) => { @@ -385,6 +386,7 @@ it.effect.each([true, false])("only retries after confirmed process exit: %s", ( }) .pipe(Effect.exit, Effect.forkChild); yield* Effect.promise(() => registrationStarted.promise); + events.emit("error", new Error("running child error")); yield* TestClock.adjust("90 seconds"); assert.equal(yield* Effect.promise(() => cleanupProgress.promise), "kill"); @@ -401,45 +403,72 @@ it.effect.each([true, false])("only retries after confirmed process exit: %s", ( ), ); -it.effect("stops retrying MCP registration after two timeouts and reports the cause", () => - Effect.scoped( - Effect.gen(function* () { - const firstRegistration = Promise.withResolvers(); - const secondRegistration = Promise.withResolvers(); - const disposed = vi.fn(async () => {}); - const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(() => { - const started = querySpy.mock.calls.length === 1 ? firstRegistration : secondRegistration; - return { - setMcpServers: () => { - started.resolve(); - return new Promise(() => {}); - }, - [Symbol.asyncDispose]: disposed, - } as unknown as ClaudeSdk.Query; - }); - yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore())); - const runner = yield* ClaudeAgentSdkQueryRunner; - const opening = yield* runner - .open(resumedMcpQueryInput) - .pipe(Effect.result, Effect.forkChild); - yield* Effect.promise(() => firstRegistration.promise); - yield* TestClock.adjust("90 seconds"); - yield* Effect.promise(() => secondRegistration.promise); - yield* TestClock.adjust("90 seconds"); - const result = yield* Fiber.join(opening); - - assert.equal(result._tag, "Failure"); - if (result._tag !== "Failure") return; - assert.equal(querySpy.mock.calls.length, 2); - assert.equal(disposed.mock.calls.length, 2); - assert.equal( - makeProviderFailure({ - cause: { _tag: "ProviderAdapterTurnStartError", cause: result.failure }, - }).message, - "Claude timed out while registering its MCP tools. Retry the turn; if it keeps failing, check Claude and the server logs.", - ); - }).pipe(Effect.provide(realQueryRunner)), - ), +it.effect.each([false, true])( + "reports MCP registration timeouts even when spawning fails: %s", + (spawnFails) => + Effect.scoped( + Effect.gen(function* () { + const firstRegistration = Promise.withResolvers(); + const secondRegistration = Promise.withResolvers(); + const disposed = vi.fn(async () => {}); + const signal = yield* Effect.abortSignal; + const kill = vi.fn(() => false); + let spawnedChild: ClaudeSdk.SpawnedProcess; + const querySpy = vi.spyOn(ClaudeSdk, "query").mockImplementation(({ options }) => { + const child = Object.assign(new NodeEvents.EventEmitter(), { + stdin: new NodeStream.PassThrough(), + stdout: new NodeStream.PassThrough(), + pid: undefined, + killed: false, + exitCode: null, + kill, + }); + spawnedChild = child; + if (spawnFails) { + options?.spawnClaudeCodeProcess?.({ command: "claude", args: [], env: {}, signal }); + } + const started = querySpy.mock.calls.length === 1 ? firstRegistration : secondRegistration; + return { + setMcpServers: () => { + if (spawnFails) child.emit("error", new Error("spawn ENOENT")); + started.resolve(); + return new Promise(() => {}); + }, + [Symbol.asyncDispose]: disposed, + } as unknown as ClaudeSdk.Query; + }); + yield* Effect.addFinalizer(() => Effect.sync(() => querySpy.mockRestore())); + const runner = yield* ClaudeAgentSdkQueryRunner; + const opening = yield* runner + .open({ + ...resumedMcpQueryInput, + options: { + ...resumedMcpQueryInput.options, + spawnClaudeCodeProcess: () => spawnedChild, + }, + }) + .pipe(Effect.result, Effect.forkChild); + yield* Effect.promise(() => firstRegistration.promise); + yield* TestClock.adjust("90 seconds"); + yield* TestClock.adjust("10 seconds"); + assert.equal(querySpy.mock.calls.length, 2); + yield* Effect.promise(() => secondRegistration.promise); + yield* TestClock.adjust("90 seconds"); + const result = yield* Fiber.join(opening); + + assert.equal(result._tag, "Failure"); + if (result._tag !== "Failure") return; + assert.equal(querySpy.mock.calls.length, 2); + assert.equal(disposed.mock.calls.length, 2); + assert.equal(kill.mock.calls.length, 0); + assert.equal( + makeProviderFailure({ + cause: { _tag: "ProviderAdapterTurnStartError", cause: result.failure }, + }).message, + "Claude timed out while registering its MCP tools. Retry the turn; if it keeps failing, check Claude and the server logs.", + ); + }).pipe(Effect.provide(realQueryRunner)), + ), ); it.effect("cancels a stalled registration without retrying or retaining its prompt queue", () => diff --git a/packages/provider-core/src/server/failure.ts b/packages/provider-core/src/server/failure.ts index edaf0aa97a743..6411cb5e22931 100644 --- a/packages/provider-core/src/server/failure.ts +++ b/packages/provider-core/src/server/failure.ts @@ -112,6 +112,7 @@ function curatedCauseMessages(cause: unknown): { case "ClaudeSettingsLaunchArgError": case "ClaudeMcpRegistrationTimeoutError": case "ClaudeCodeRootRefusalError": + case "ClaudeProcessExitError": case "CursorModelUnavailableError": case "CodexGoalObjectiveTooLongError": case "ProviderSetupError":