From 5ca7676661325b13ee2a557b28f0d0d3ce8ab695 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 20 Jun 2026 11:22:57 -0700 Subject: [PATCH] [codex] Structure Claude adapter failures (#3406) Co-authored-by: codex --- .../src/provider/Layers/ClaudeAdapter.test.ts | 116 +++++++++++++++--- .../src/provider/Layers/ClaudeAdapter.ts | 95 +++++++------- 2 files changed, 143 insertions(+), 68 deletions(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 4d22a2c1f8d925b66a70bd26a1e4cd5503d33ccf..191bf8e27db9a688ab0c9848c88c48b8e4634fef 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -35,7 +35,7 @@ import * as TestClock from "effect/testing/TestClock"; import { attachmentRelativePath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; -import { ProviderAdapterValidationError } from "../Errors.ts"; +import { ProviderAdapterProcessError, ProviderAdapterValidationError } from "../Errors.ts"; import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; import { makeClaudeAdapter, type ClaudeAdapterLiveOptions } from "./ClaudeAdapter.ts"; const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings); @@ -298,6 +298,44 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("retains Claude session startup causes without exposing their messages", () => { + const cause = new Error("credential material that must remain in the cause chain"); + const layer = Layer.effect( + ClaudeAdapter, + Effect.gen(function* () { + const claudeConfig = decodeClaudeSettings({}); + return yield* makeClaudeAdapter(claudeConfig, { + createQuery: () => { + throw cause; + }, + }); + }), + ).pipe( + Layer.provideMerge(ServerConfig.layerTest("/tmp/claude-adapter-test", "/tmp")), + Layer.provideMerge(ServerSettingsService.layerTest()), + Layer.provideMerge(NodeServices.layer), + ); + + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const error = yield* adapter + .startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }) + .pipe(Effect.flip); + + assert.instanceOf(error, ProviderAdapterProcessError); + assert.equal(error.detail, "Failed to start Claude runtime session."); + assert.strictEqual(error.cause, cause); + assert.notMatch(error.message, /credential material/u); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(layer), + ); + }); + it.effect("derives bypass permission mode from full-access runtime policy", () => { const harness = makeHarness(); return Effect.gen(function* () { @@ -1365,19 +1403,14 @@ describe("ClaudeAdapterLive", () => { it.effect("closes the session when the Claude stream aborts after a turn starts", () => { const harness = makeHarness(); return Effect.gen(function* () { - const context = yield* Effect.context(); - const runFork = Effect.runForkWith(context); - const adapter = yield* ClaudeAdapter; const runtimeEvents: Array = []; - const runtimeEventsFiber = runFork( - Stream.runForEach(adapter.streamEvents, (event) => - Effect.sync(() => { - runtimeEvents.push(event); - }), - ), - ); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }), + ).pipe(Effect.forkChild); yield* adapter.startSession({ threadId: THREAD_ID, @@ -1430,6 +1463,57 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("keeps Claude stream failure events structural", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEvents: Array = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: THREAD_ID, + input: "hello", + attachments: [], + }); + + harness.query.fail(new Error("credential material that must stay in the cause chain")); + + yield* Effect.yieldNow; + yield* Effect.yieldNow; + yield* Effect.yieldNow; + runtimeEventsFiber.interruptUnsafe(); + + const runtimeError = runtimeEvents.find((event) => event.type === "runtime.error"); + assert.equal(runtimeError?.type, "runtime.error"); + if (runtimeError?.type === "runtime.error") { + assert.equal(runtimeError.payload.message, "Claude runtime stream failed."); + assert.deepEqual(runtimeError.payload.detail, { + failureCount: 1, + failureTags: ["ProviderAdapterProcessError"], + }); + } + + const completed = runtimeEvents.find((event) => event.type === "turn.completed"); + assert.equal(completed?.type, "turn.completed"); + if (completed?.type === "turn.completed") { + assert.equal(completed.payload.state, "failed"); + assert.equal(completed.payload.errorMessage, "Claude runtime stream failed."); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("closes the previous session before replacing an existing thread session", () => { const queries: FakeClaudeQuery[] = []; const layer = Layer.effect( @@ -1542,14 +1626,12 @@ describe("ClaudeAdapterLive", () => { ); return Effect.gen(function* () { - const context = yield* Effect.context(); - const runFork = Effect.runForkWith(context); - const adapter = yield* ClaudeAdapter; - const runtimeEventsFiber = runFork( - Stream.runForEach(adapter.streamEvents, () => Effect.void), - ); + const runtimeEventsFiber = yield* Stream.runForEach( + adapter.streamEvents, + () => Effect.void, + ).pipe(Effect.forkChild); yield* adapter.startSession({ threadId: THREAD_ID, diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index c91f305b174c6565160f5eb795c96492a39f0ff2..97a93f85829cd61feec493405e7f14e91ba9bcd7 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -249,21 +249,8 @@ function toMessage(cause: unknown, fallback: string): string { return fallback; } -function toProcessError( - cause: unknown, - fallback: string, - threadId: ThreadId, -): ProviderAdapterProcessError { - return new ProviderAdapterProcessError({ - provider: PROVIDER, - threadId, - detail: toMessage(cause, fallback), - cause, - }); -} - function normalizeClaudeStreamMessages( - cause: Cause.Cause<{ readonly message: string }>, + cause: Cause.Cause, ): ReadonlyArray { const errors: Array = []; for (const error of Cause.prettyErrors(cause)) { @@ -297,27 +284,17 @@ function isClaudeInterruptedMessage(message: string): boolean { ); } -function isClaudeInterruptedCause(cause: Cause.Cause<{ readonly message: string }>): boolean { +function isClaudeInterruptedCause(cause: Cause.Cause): boolean { return ( Cause.hasInterruptsOnly(cause) || - normalizeClaudeStreamMessages(cause).some(isClaudeInterruptedMessage) + normalizeClaudeStreamMessages(cause).some(isClaudeInterruptedMessage) || + cause.reasons.some( + (reason) => + Cause.isFailReason(reason) && isClaudeInterruptedMessage(toMessage(reason.error.cause, "")), + ) ); } -function messageFromClaudeStreamCause( - cause: Cause.Cause<{ readonly message: string }>, - fallback: string, -): string { - return normalizeClaudeStreamMessages(cause)[0] ?? fallback; -} - -function interruptionMessageFromClaudeCause( - cause: Cause.Cause<{ readonly message: string }>, -): string { - const message = messageFromClaudeStreamCause(cause, "Claude runtime interrupted."); - return isClaudeInterruptedMessage(message) ? "Claude runtime interrupted." : message; -} - function resultErrorsText(result: SDKResultMessage): string { return "errors" in result && Array.isArray(result.errors) ? result.errors.join(" ").toLowerCase() @@ -1004,7 +981,7 @@ const buildUserMessageEffect = Effect.fn("buildUserMessageEffect")(function* ( new ProviderAdapterRequestError({ provider: PROVIDER, method: "turn/start", - detail: toMessage(cause, "Failed to read attachment file."), + detail: "Failed to read attachment file.", cause, }), ), @@ -1242,7 +1219,7 @@ function toRequestError(threadId: ThreadId, method: string, cause: unknown): Pro return new ProviderAdapterRequestError({ provider: PROVIDER, method, - detail: toMessage(cause, `${method} failed`), + detail: `${method} failed`, cause, }); } @@ -2910,18 +2887,27 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const runSdkStream = ( context: ClaudeSessionContext, ): Effect.Effect => - Stream.fromAsyncIterable(context.query, (cause) => - toProcessError(cause, "Claude runtime stream failed.", context.session.threadId), + Stream.fromAsyncIterable( + context.query, + (cause) => + new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: context.session.threadId, + detail: "Claude runtime stream failed.", + cause, + }), ).pipe( Stream.takeWhile(() => !context.stopped), Stream.runForEach((message) => handleSdkMessage(context, message).pipe( - Effect.mapError((cause) => - toProcessError( - cause, - "Failed to process Claude runtime event.", - context.session.threadId, - ), + Effect.mapError( + (cause) => + new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: context.session.threadId, + detail: "Failed to process Claude runtime event.", + cause, + }), ), ), ), @@ -2938,15 +2924,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( if (Exit.isFailure(exit)) { if (isClaudeInterruptedCause(exit.cause)) { if (context.turnState) { - yield* completeTurn( - context, - "interrupted", - interruptionMessageFromClaudeCause(exit.cause), - ); + yield* completeTurn(context, "interrupted", "Claude runtime interrupted."); } } else { - const message = messageFromClaudeStreamCause(exit.cause, "Claude runtime stream failed."); - yield* emitRuntimeError(context, message, Cause.pretty(exit.cause)); + const failures = exit.cause.reasons.flatMap((reason) => + Cause.isFailReason(reason) ? [reason.error] : [], + ); + const message = failures[0]?.detail ?? "Claude runtime stream failed."; + yield* emitRuntimeError(context, message, { + failureCount: failures.length, + failureTags: failures.map((failure) => failure._tag), + }); yield* completeTurn(context, "failed", message); } } else if (context.turnState) { @@ -3004,12 +2992,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( new ProviderAdapterProcessError({ provider: PROVIDER, threadId: context.session.threadId, - detail: toMessage(cause, "Failed to close Claude runtime query."), + detail: "Failed to close Claude runtime query.", cause, }), }).pipe( - Effect.catch((cause) => - emitRuntimeError(context, "Failed to close Claude runtime query.", cause), + Effect.catch((error) => + emitRuntimeError(context, "Failed to close Claude runtime query.", { + errorTag: error._tag, + provider: error.provider, + threadId: error.threadId, + detail: error.detail, + }), ), ); @@ -3522,7 +3515,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( new ProviderAdapterProcessError({ provider: PROVIDER, threadId, - detail: toMessage(cause, "Failed to start Claude runtime session."), + detail: "Failed to start Claude runtime session.", cause, }), }); -- 2.54.0