diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 760f0e7fbab..70dcf80a72f 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -1445,6 +1445,56 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("treats an error result as interrupted when the user requested a stop", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const runtimeEventsFiber = yield* Stream.takeUntil( + adapter.streamEvents, + (event) => event.type === "turn.completed", + ).pipe(Stream.runCollect, Effect.forkChild); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + const turn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "hello", + attachments: [], + }); + + yield* adapter.interruptTurn(THREAD_ID); + + harness.query.emit({ + type: "result", + subtype: "error_during_execution", + is_error: true, + errors: ["[ede_diagnostic] result_type=user last_content_type=n/a stop_reason=null"], + session_id: "sdk-session-ede", + uuid: "result-ede", + } as unknown as SDKMessage); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + + const runtimeError = runtimeEvents.find((event) => event.type === "runtime.error"); + assert.equal(runtimeError, undefined); + + const turnCompleted = runtimeEvents.find((event) => event.type === "turn.completed"); + assert.equal(turnCompleted?.type, "turn.completed"); + if (turnCompleted?.type === "turn.completed") { + assert.equal(String(turnCompleted.turnId), String(turn.turnId)); + assert.equal(turnCompleted.payload.state, "interrupted"); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("closes the session when the Claude stream aborts after a turn starts", () => { const harness = makeHarness(); return Effect.gen(function* () { @@ -1508,6 +1558,53 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("treats a stream failure as an interruption when the user requested a stop", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const runtimeEventsFiber = yield* Stream.takeUntil( + adapter.streamEvents, + (event) => event.type === "session.exited", + ).pipe(Stream.runCollect, Effect.forkChild); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + const turn = yield* adapter.sendTurn({ + threadId: THREAD_ID, + input: "hello", + attachments: [], + }); + + yield* adapter.interruptTurn(THREAD_ID, turn.turnId); + assert.equal(harness.query.interruptCalls.length, 1); + + harness.query.fail(new Error("Claude Code process exited with code 143")); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + + const runtimeError = runtimeEvents.find((event) => event.type === "runtime.error"); + assert.equal(runtimeError, undefined); + + const turnCompleted = runtimeEvents.find((event) => event.type === "turn.completed"); + assert.equal(turnCompleted?.type, "turn.completed"); + if (turnCompleted?.type === "turn.completed") { + assert.equal(String(turnCompleted.turnId), String(turn.turnId)); + assert.equal(turnCompleted.payload.state, "interrupted"); + assert.equal(turnCompleted.payload.errorMessage, "Claude runtime interrupted."); + } + + assert.equal(yield* adapter.hasSession(THREAD_ID), false); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("keeps Claude stream failure events structural", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index f87d5be7446..05bcea7a606 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -202,6 +202,7 @@ interface ClaudeSessionContext { lastAssistantUuid: string | undefined; lastThreadStartedId: string | undefined; stopped: boolean; + readonly interruptedTurnIds: Set; } interface ClaudeQueryRuntime extends AsyncIterable { @@ -2058,6 +2059,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const updatedAt = yield* nowIso; context.turnState = undefined; + context.interruptedTurnIds.delete(turnState.turnId); context.session = { ...context.session, status: "ready", @@ -2553,7 +2555,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( return; } - const status = turnStatusFromResult(message); + const interrupted = + context.turnState !== undefined && context.interruptedTurnIds.has(context.turnState.turnId); + const status = + interrupted && message.subtype !== "success" ? "interrupted" : turnStatusFromResult(message); const errorMessage = message.subtype === "success" ? undefined : message.errors[0]; if (status === "failed") { @@ -3001,8 +3006,11 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( return; } + const interrupted = + context.turnState !== undefined && context.interruptedTurnIds.has(context.turnState.turnId); + if (Exit.isFailure(exit)) { - if (isClaudeInterruptedCause(exit.cause)) { + if (interrupted || isClaudeInterruptedCause(exit.cause)) { if (context.turnState) { yield* completeTurn(context, "interrupted", "Claude runtime interrupted."); } @@ -3641,6 +3649,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( lastAssistantUuid: resumeState?.resumeSessionAt, lastThreadStartedId: undefined, stopped: false, + interruptedTurnIds: new Set(), }; yield* Ref.set(contextRef, context); sessions.set(threadId, context); @@ -3823,8 +3832,12 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); const interruptTurn: ClaudeAdapterShape["interruptTurn"] = Effect.fn("interruptTurn")( - function* (threadId, _turnId) { + function* (threadId, turnId) { const context = yield* requireSession(threadId); + const targetTurnId = turnId ?? context.turnState?.turnId; + if (targetTurnId !== undefined) { + context.interruptedTurnIds.add(targetTurnId); + } yield* Effect.tryPromise({ try: () => context.query.interrupt(), catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),