diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 1c6339ffb4aa0..cb4df51c7d4c7 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -43,6 +43,7 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { Tool } from "effect/ai"; import { formatClaudeResumeCompactionQuestion } from "@t3tools/shared/claudeCompaction"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; @@ -1990,6 +1991,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { readonly subtype?: string; readonly isError?: boolean; readonly errors?: ReadonlyArray; + readonly userMessageUuid?: string; readonly apiErrorStatus?: number; // null omits the field, as the CLI does on a zero-turn result. readonly terminalReason?: SDKResultMessage["terminal_reason"] | null; @@ -2016,6 +2018,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { session_id: WAKE_NATIVE_SESSION, ...(input.origin === undefined ? {} : { origin: input.origin }), ...(input.errors === undefined ? {} : { errors: input.errors }), + ...(input.userMessageUuid === undefined ? {} : { user_message_uuid: input.userMessageUuid }), ...(input.apiErrorStatus === undefined ? {} : { api_error_status: input.apiErrorStatus }), ...(input.terminalReason === null ? {} @@ -2076,7 +2079,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { const makeWakeHarnessWithOptions = (options?: { readonly close?: (sdkMessages: Queue.Queue) => Effect.Effect; - readonly interrupt?: Effect.Effect; + readonly interrupt?: Effect.Effect; readonly environment?: NodeJS.ProcessEnv; }) => Effect.gen(function* () { @@ -3778,6 +3781,179 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("lets the native SDK acknowledge Stop before closing its own query", () => + Effect.scoped( + Effect.gen(function* () { + let closes = 0; + let acknowledge = Effect.void; + const harness = yield* makeWakeHarnessWithOptions({ + interrupt: Effect.suspend(() => { + assert.equal(closes, 0); + return acknowledge; + }), + close: (messages) => + Effect.sync(() => { + closes++; + }).pipe(Effect.andThen(Queue.shutdown(messages))), + }); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("graceful-stop"), + text: "Work until stopped", + attachments: [], + }), + ); + const userMessageUuid = harness.offeredMessages.at(-1)?.uuid; + if (userMessageUuid === undefined) return yield* Effect.die("Missing native user message"); + acknowledge = harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000620", + result: "", + subtype: "error_during_execution", + errors: ["Error: Request was aborted."], + terminalReason: "aborted_streaming", + userMessageUuid, + }), + ); + yield* awaitUntil( + () => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ), + "running native turn", + ); + const active = harness.events.find( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ); + if (active?.type !== "provider_turn.updated") + return yield* Effect.die("Missing running native turn"); + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: active.providerTurn.id, + requestRuntimeRestart: true, + }); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + assert.equal(closes, 1); + assert.lengthOf(harness.terminalEvents(), 1); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + + it.effect("bounds an unacknowledged native Stop and settles its turn once", () => + Effect.scoped( + Effect.gen(function* () { + let closes = 0; + const interruptStarted = yield* Deferred.make(); + const harness = yield* makeWakeHarnessWithOptions({ + interrupt: Deferred.succeed(interruptStarted, undefined).pipe( + Effect.andThen(Effect.never), + ), + close: (messages) => + Effect.sync(() => { + closes++; + }).pipe(Effect.andThen(Queue.shutdown(messages))), + }); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("unacknowledged-stop"), + text: "Work until stopped", + attachments: [], + }), + ); + yield* awaitUntil( + () => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ), + "running native turn", + ); + const active = harness.events.find( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ); + if (active?.type !== "provider_turn.updated") + return yield* Effect.die("Missing running native turn"); + const stop = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: active.providerTurn.id, + requestRuntimeRestart: true, + }) + .pipe(Effect.forkChild); + yield* Deferred.await(interruptStarted); + assert.equal(closes, 0); + yield* TestClock.adjust("11 seconds"); + assert.isDefined( + stop.pollUnsafe(), + "Stop must finish despite a missing SDK acknowledgement", + ); + yield* Fiber.join(stop); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + assert.equal(closes, 1); + assert.lengthOf(harness.terminalEvents(), 1); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + + it.effect("closes the query when the native SDK rejects Stop", () => + Effect.scoped( + Effect.gen(function* () { + let closes = 0; + const harness = yield* makeWakeHarnessWithOptions({ + interrupt: new ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError({ + method: "interrupt", + cause: new Error("Claude Code rejected the interrupt"), + }), + close: (messages) => + Effect.sync(() => { + closes++; + }).pipe(Effect.andThen(Queue.shutdown(messages))), + }); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("rejected-stop"), + text: "Work until stopped", + attachments: [], + }), + ); + yield* awaitUntil( + () => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ), + "running native turn", + ); + const active = harness.events.find( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ); + if (active?.type !== "provider_turn.updated") + return yield* Effect.die("Missing running native turn"); + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: active.providerTurn.id, + requestRuntimeRestart: true, + }); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + assert.equal(closes, 1); + assert.lengthOf(harness.terminalEvents(), 1); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("stops background work after the turn settled", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 9aa6163f2adb1..468df5e45b092 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -7466,7 +7466,19 @@ export function makeClaudeAdapterV2( next.add(turnInput.providerTurnId); return next; }); - yield* existing.query.interrupt; + // Closing the query is what stops the CLI, so a rejected or + // unanswered interrupt still falls through to the close. + yield* existing.query.interrupt.pipe( + Effect.timeoutOption("10 seconds"), + Effect.catch((cause) => + Effect.logWarning("orchestration-v2.claude-query-interrupt-failed", { + providerSessionId: input.providerSessionId, + providerThreadId: turnInput.providerThread.id, + providerTurnId: turnInput.providerTurnId, + cause, + }), + ), + ); yield* existing.query.close.pipe(Effect.ignore); const closed = yield* Deferred.await(existing.closed).pipe( Effect.timeoutOption("10 seconds"),