Skip to content

fix(server): Stop ends a Claude turn even when the CLI rejects the interrupt #16439

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yaacovcorcos wants to merge 2 commits into
base: main
Choose a base branch
from
+190 −2
Open

fix(server): Stop ends a Claude turn even when the CLI rejects the interrupt #16439

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
efdd9cb
fix(server): stopping a Claude turn no longer hangs without an acknow…
yaacovcorcos Oct 6, 2026
b161fe8
fix(server): Stop ends a Claude turn even when the CLI rejects the in…
yaacovcorcos Oct 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
178 changes: 177 additions & 1 deletion apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -1990,6 +1991,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
readonly subtype?: string;
readonly isError?: boolean;
readonly errors?: ReadonlyArray<string>;
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;
Expand All @@ -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
? {}
Expand Down Expand Up @@ -2076,7 +2079,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {

const makeWakeHarnessWithOptions = (options?: {
readonly close?: (sdkMessages: Queue.Queue<SDKMessage>) => Effect.Effect<void>;
readonly interrupt?: Effect.Effect<void>;
readonly interrupt?: Effect.Effect<void, ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError>;
readonly environment?: NodeJS.ProcessEnv;
}) =>
Effect.gen(function* () {
Expand Down Expand Up @@ -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<void>();
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* () {
Expand Down
14 changes: 13 additions & 1 deletion apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
Expand Down
Loading