authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-20 11:22:57-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-20 11:22:57-07:00
log5ca7676661325b13ee2a557b28f0d0d3ce8ab695
tree8c7b0848320fed54f1be378d516461f378726d5b
parentc3e3e26844a591f2bb21e617db3c5494e700fbcc
signature Signed by PGP key B5690EEEBB952194

[codex] Structure Claude adapter failures (#3406)

Co-authored-by: codex <codex@users.noreply.github.com>

2 files changed, 143 insertions(+), 68 deletions(-)

apps/server/src/provider/Layers/ClaudeAdapter.test.ts+99-17
...@@ -35,7 +35,7 @@ import * as TestClock from "effect/testing/TestClock";...@@ -35,7 +35,7 @@ import * as TestClock from "effect/testing/TestClock";
35import { attachmentRelativePath } from "../../attachmentStore.ts";35import { attachmentRelativePath } from "../../attachmentStore.ts";
36import { ServerConfig } from "../../config.ts";36import { ServerConfig } from "../../config.ts";
37import { ServerSettingsService } from "../../serverSettings.ts";37import { ServerSettingsService } from "../../serverSettings.ts";
38import { ProviderAdapterValidationError } from "../Errors.ts";38import { ProviderAdapterProcessError, ProviderAdapterValidationError } from "../Errors.ts";
39import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts";39import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts";
40import { makeClaudeAdapter, type ClaudeAdapterLiveOptions } from "./ClaudeAdapter.ts";40import { makeClaudeAdapter, type ClaudeAdapterLiveOptions } from "./ClaudeAdapter.ts";
41const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings);41const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings);
...@@ -298,6 +298,44 @@ describe("ClaudeAdapterLive", () => {...@@ -298,6 +298,44 @@ describe("ClaudeAdapterLive", () => {
298 );298 );
299 });299 });
300300
301 it.effect("retains Claude session startup causes without exposing their messages", () => {
302 const cause = new Error("credential material that must remain in the cause chain");
303 const layer = Layer.effect(
304 ClaudeAdapter,
305 Effect.gen(function* () {
306 const claudeConfig = decodeClaudeSettings({});
307 return yield* makeClaudeAdapter(claudeConfig, {
308 createQuery: () => {
309 throw cause;
310 },
311 });
312 }),
313 ).pipe(
314 Layer.provideMerge(ServerConfig.layerTest("/tmp/claude-adapter-test", "/tmp")),
315 Layer.provideMerge(ServerSettingsService.layerTest()),
316 Layer.provideMerge(NodeServices.layer),
317 );
318
319 return Effect.gen(function* () {
320 const adapter = yield* ClaudeAdapter;
321 const error = yield* adapter
322 .startSession({
323 threadId: THREAD_ID,
324 provider: ProviderDriverKind.make("claudeAgent"),
325 runtimeMode: "full-access",
326 })
327 .pipe(Effect.flip);
328
329 assert.instanceOf(error, ProviderAdapterProcessError);
330 assert.equal(error.detail, "Failed to start Claude runtime session.");
331 assert.strictEqual(error.cause, cause);
332 assert.notMatch(error.message, /credential material/u);
333 }).pipe(
334 Effect.provideService(Random.Random, makeDeterministicRandomService()),
335 Effect.provide(layer),
336 );
337 });
338
301 it.effect("derives bypass permission mode from full-access runtime policy", () => {339 it.effect("derives bypass permission mode from full-access runtime policy", () => {
302 const harness = makeHarness();340 const harness = makeHarness();
303 return Effect.gen(function* () {341 return Effect.gen(function* () {
...@@ -1365,19 +1403,14 @@ describe("ClaudeAdapterLive", () => {...@@ -1365,19 +1403,14 @@ describe("ClaudeAdapterLive", () => {
1365 it.effect("closes the session when the Claude stream aborts after a turn starts", () => {1403 it.effect("closes the session when the Claude stream aborts after a turn starts", () => {
1366 const harness = makeHarness();1404 const harness = makeHarness();
1367 return Effect.gen(function* () {1405 return Effect.gen(function* () {
1368 const context = yield* Effect.context<never>();
1369 const runFork = Effect.runForkWith(context);
1370
1371 const adapter = yield* ClaudeAdapter;1406 const adapter = yield* ClaudeAdapter;
1372 const runtimeEvents: Array<ProviderRuntimeEvent> = [];1407 const runtimeEvents: Array<ProviderRuntimeEvent> = [];
13731408
1374 const runtimeEventsFiber = runFork(1409 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
1375 Stream.runForEach(adapter.streamEvents, (event) =>1410 Effect.sync(() => {
1376 Effect.sync(() => {1411 runtimeEvents.push(event);
1377 runtimeEvents.push(event);1412 }),
1378 }),1413 ).pipe(Effect.forkChild);
1379 ),
1380 );
13811414
1382 yield* adapter.startSession({1415 yield* adapter.startSession({
1383 threadId: THREAD_ID,1416 threadId: THREAD_ID,
...@@ -1430,6 +1463,57 @@ describe("ClaudeAdapterLive", () => {...@@ -1430,6 +1463,57 @@ describe("ClaudeAdapterLive", () => {
1430 );1463 );
1431 });1464 });
14321465
1466 it.effect("keeps Claude stream failure events structural", () => {
1467 const harness = makeHarness();
1468 return Effect.gen(function* () {
1469 const adapter = yield* ClaudeAdapter;
1470 const runtimeEvents: Array<ProviderRuntimeEvent> = [];
1471 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
1472 Effect.sync(() => {
1473 runtimeEvents.push(event);
1474 }),
1475 ).pipe(Effect.forkChild);
1476
1477 yield* adapter.startSession({
1478 threadId: THREAD_ID,
1479 provider: ProviderDriverKind.make("claudeAgent"),
1480 runtimeMode: "full-access",
1481 });
1482 yield* adapter.sendTurn({
1483 threadId: THREAD_ID,
1484 input: "hello",
1485 attachments: [],
1486 });
1487
1488 harness.query.fail(new Error("credential material that must stay in the cause chain"));
1489
1490 yield* Effect.yieldNow;
1491 yield* Effect.yieldNow;
1492 yield* Effect.yieldNow;
1493 runtimeEventsFiber.interruptUnsafe();
1494
1495 const runtimeError = runtimeEvents.find((event) => event.type === "runtime.error");
1496 assert.equal(runtimeError?.type, "runtime.error");
1497 if (runtimeError?.type === "runtime.error") {
1498 assert.equal(runtimeError.payload.message, "Claude runtime stream failed.");
1499 assert.deepEqual(runtimeError.payload.detail, {
1500 failureCount: 1,
1501 failureTags: ["ProviderAdapterProcessError"],
1502 });
1503 }
1504
1505 const completed = runtimeEvents.find((event) => event.type === "turn.completed");
1506 assert.equal(completed?.type, "turn.completed");
1507 if (completed?.type === "turn.completed") {
1508 assert.equal(completed.payload.state, "failed");
1509 assert.equal(completed.payload.errorMessage, "Claude runtime stream failed.");
1510 }
1511 }).pipe(
1512 Effect.provideService(Random.Random, makeDeterministicRandomService()),
1513 Effect.provide(harness.layer),
1514 );
1515 });
1516
1433 it.effect("closes the previous session before replacing an existing thread session", () => {1517 it.effect("closes the previous session before replacing an existing thread session", () => {
1434 const queries: FakeClaudeQuery[] = [];1518 const queries: FakeClaudeQuery[] = [];
1435 const layer = Layer.effect(1519 const layer = Layer.effect(
...@@ -1542,14 +1626,12 @@ describe("ClaudeAdapterLive", () => {...@@ -1542,14 +1626,12 @@ describe("ClaudeAdapterLive", () => {
1542 );1626 );
15431627
1544 return Effect.gen(function* () {1628 return Effect.gen(function* () {
1545 const context = yield* Effect.context<never>();
1546 const runFork = Effect.runForkWith(context);
1547
1548 const adapter = yield* ClaudeAdapter;1629 const adapter = yield* ClaudeAdapter;
15491630
1550 const runtimeEventsFiber = runFork(1631 const runtimeEventsFiber = yield* Stream.runForEach(
1551 Stream.runForEach(adapter.streamEvents, () => Effect.void),1632 adapter.streamEvents,
1552 );1633 () => Effect.void,
1634 ).pipe(Effect.forkChild);
15531635
1554 yield* adapter.startSession({1636 yield* adapter.startSession({
1555 threadId: THREAD_ID,1637 threadId: THREAD_ID,
apps/server/src/provider/Layers/ClaudeAdapter.ts+44-51
...@@ -249,21 +249,8 @@ function toMessage(cause: unknown, fallback: string): string {...@@ -249,21 +249,8 @@ function toMessage(cause: unknown, fallback: string): string {
249 return fallback;249 return fallback;
250}250}
251251
252function toProcessError(
253 cause: unknown,
254 fallback: string,
255 threadId: ThreadId,
256): ProviderAdapterProcessError {
257 return new ProviderAdapterProcessError({
258 provider: PROVIDER,
259 threadId,
260 detail: toMessage(cause, fallback),
261 cause,
262 });
263}
264
265function normalizeClaudeStreamMessages(252function normalizeClaudeStreamMessages(
266 cause: Cause.Cause<{ readonly message: string }>,253 cause: Cause.Cause<ProviderAdapterProcessError>,
267): ReadonlyArray<string> {254): ReadonlyArray<string> {
268 const errors: Array<string> = [];255 const errors: Array<string> = [];
269 for (const error of Cause.prettyErrors(cause)) {256 for (const error of Cause.prettyErrors(cause)) {
...@@ -297,27 +284,17 @@ function isClaudeInterruptedMessage(message: string): boolean {...@@ -297,27 +284,17 @@ function isClaudeInterruptedMessage(message: string): boolean {
297 );284 );
298}285}
299286
300function isClaudeInterruptedCause(cause: Cause.Cause<{ readonly message: string }>): boolean {287function isClaudeInterruptedCause(cause: Cause.Cause<ProviderAdapterProcessError>): boolean {
301 return (288 return (
302 Cause.hasInterruptsOnly(cause) ||289 Cause.hasInterruptsOnly(cause) ||
303 normalizeClaudeStreamMessages(cause).some(isClaudeInterruptedMessage)290 normalizeClaudeStreamMessages(cause).some(isClaudeInterruptedMessage) ||
291 cause.reasons.some(
292 (reason) =>
293 Cause.isFailReason(reason) && isClaudeInterruptedMessage(toMessage(reason.error.cause, "")),
294 )
304 );295 );
305}296}
306297
307function messageFromClaudeStreamCause(
308 cause: Cause.Cause<{ readonly message: string }>,
309 fallback: string,
310): string {
311 return normalizeClaudeStreamMessages(cause)[0] ?? fallback;
312}
313
314function interruptionMessageFromClaudeCause(
315 cause: Cause.Cause<{ readonly message: string }>,
316): string {
317 const message = messageFromClaudeStreamCause(cause, "Claude runtime interrupted.");
318 return isClaudeInterruptedMessage(message) ? "Claude runtime interrupted." : message;
319}
320
321function resultErrorsText(result: SDKResultMessage): string {298function resultErrorsText(result: SDKResultMessage): string {
322 return "errors" in result && Array.isArray(result.errors)299 return "errors" in result && Array.isArray(result.errors)
323 ? result.errors.join(" ").toLowerCase()300 ? result.errors.join(" ").toLowerCase()
...@@ -1004,7 +981,7 @@ const buildUserMessageEffect = Effect.fn("buildUserMessageEffect")(function* (...@@ -1004,7 +981,7 @@ const buildUserMessageEffect = Effect.fn("buildUserMessageEffect")(function* (
1004 new ProviderAdapterRequestError({981 new ProviderAdapterRequestError({
1005 provider: PROVIDER,982 provider: PROVIDER,
1006 method: "turn/start",983 method: "turn/start",
1007 detail: toMessage(cause, "Failed to read attachment file."),984 detail: "Failed to read attachment file.",
1008 cause,985 cause,
1009 }),986 }),
1010 ),987 ),
...@@ -1242,7 +1219,7 @@ function toRequestError(threadId: ThreadId, method: string, cause: unknown): Pro...@@ -1242,7 +1219,7 @@ function toRequestError(threadId: ThreadId, method: string, cause: unknown): Pro
1242 return new ProviderAdapterRequestError({1219 return new ProviderAdapterRequestError({
1243 provider: PROVIDER,1220 provider: PROVIDER,
1244 method,1221 method,
1245 detail: toMessage(cause, `${method} failed`),1222 detail: `${method} failed`,
1246 cause,1223 cause,
1247 });1224 });
1248}1225}
...@@ -2910,18 +2887,27 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2910,18 +2887,27 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2910 const runSdkStream = (2887 const runSdkStream = (
2911 context: ClaudeSessionContext,2888 context: ClaudeSessionContext,
2912 ): Effect.Effect<void, ProviderAdapterProcessError> =>2889 ): Effect.Effect<void, ProviderAdapterProcessError> =>
2913 Stream.fromAsyncIterable(context.query, (cause) =>2890 Stream.fromAsyncIterable(
2914 toProcessError(cause, "Claude runtime stream failed.", context.session.threadId),2891 context.query,
2892 (cause) =>
2893 new ProviderAdapterProcessError({
2894 provider: PROVIDER,
2895 threadId: context.session.threadId,
2896 detail: "Claude runtime stream failed.",
2897 cause,
2898 }),
2915 ).pipe(2899 ).pipe(
2916 Stream.takeWhile(() => !context.stopped),2900 Stream.takeWhile(() => !context.stopped),
2917 Stream.runForEach((message) =>2901 Stream.runForEach((message) =>
2918 handleSdkMessage(context, message).pipe(2902 handleSdkMessage(context, message).pipe(
2919 Effect.mapError((cause) =>2903 Effect.mapError(
2920 toProcessError(2904 (cause) =>
2921 cause,2905 new ProviderAdapterProcessError({
2922 "Failed to process Claude runtime event.",2906 provider: PROVIDER,
2923 context.session.threadId,2907 threadId: context.session.threadId,
2924 ),2908 detail: "Failed to process Claude runtime event.",
2909 cause,
2910 }),
2925 ),2911 ),
2926 ),2912 ),
2927 ),2913 ),
...@@ -2938,15 +2924,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2938,15 +2924,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2938 if (Exit.isFailure(exit)) {2924 if (Exit.isFailure(exit)) {
2939 if (isClaudeInterruptedCause(exit.cause)) {2925 if (isClaudeInterruptedCause(exit.cause)) {
2940 if (context.turnState) {2926 if (context.turnState) {
2941 yield* completeTurn(2927 yield* completeTurn(context, "interrupted", "Claude runtime interrupted.");
2942 context,
2943 "interrupted",
2944 interruptionMessageFromClaudeCause(exit.cause),
2945 );
2946 }2928 }
2947 } else {2929 } else {
2948 const message = messageFromClaudeStreamCause(exit.cause, "Claude runtime stream failed.");2930 const failures = exit.cause.reasons.flatMap((reason) =>
2949 yield* emitRuntimeError(context, message, Cause.pretty(exit.cause));2931 Cause.isFailReason(reason) ? [reason.error] : [],
2932 );
2933 const message = failures[0]?.detail ?? "Claude runtime stream failed.";
2934 yield* emitRuntimeError(context, message, {
2935 failureCount: failures.length,
2936 failureTags: failures.map((failure) => failure._tag),
2937 });
2950 yield* completeTurn(context, "failed", message);2938 yield* completeTurn(context, "failed", message);
2951 }2939 }
2952 } else if (context.turnState) {2940 } else if (context.turnState) {
...@@ -3004,12 +2992,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -3004,12 +2992,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
3004 new ProviderAdapterProcessError({2992 new ProviderAdapterProcessError({
3005 provider: PROVIDER,2993 provider: PROVIDER,
3006 threadId: context.session.threadId,2994 threadId: context.session.threadId,
3007 detail: toMessage(cause, "Failed to close Claude runtime query."),2995 detail: "Failed to close Claude runtime query.",
3008 cause,2996 cause,
3009 }),2997 }),
3010 }).pipe(2998 }).pipe(
3011 Effect.catch((cause) =>2999 Effect.catch((error) =>
3012 emitRuntimeError(context, "Failed to close Claude runtime query.", cause),3000 emitRuntimeError(context, "Failed to close Claude runtime query.", {
3001 errorTag: error._tag,
3002 provider: error.provider,
3003 threadId: error.threadId,
3004 detail: error.detail,
3005 }),
3013 ),3006 ),
3014 );3007 );
30153008
...@@ -3522,7 +3515,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -3522,7 +3515,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
3522 new ProviderAdapterProcessError({3515 new ProviderAdapterProcessError({
3523 provider: PROVIDER,3516 provider: PROVIDER,
3524 threadId,3517 threadId,
3525 detail: toMessage(cause, "Failed to start Claude runtime session."),3518 detail: "Failed to start Claude runtime session.",
3526 cause,3519 cause,
3527 }),3520 }),
3528 });3521 });