authorgravatar for saiansh2525@gmail.comYash Singh <saiansh2525@gmail.com> 2026-09-27 15:37:31-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-27 15:37:31-05:00
log94f92a7a386a26c98892b24fabdd0ea9fa804ce3
tree7a0e25f860ae0dbbf760df8207df7c261a7228b6
parentde251fc2971a884cb5b1305ba4daf309dc8cccb0
signature Signed by PGP key B5690EEEBB952194

fix(server): let Claude abort turns before closing sessions (#13999)

2 files changed, 109 insertions(+), 0 deletions(-)

apps/server/src/provider/Layers/ClaudeAdapter.test.ts+79
...@@ -73,6 +73,8 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {...@@ -73,6 +73,8 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
73 public readonly setMaxThinkingTokensCalls: Array<number | null> = [];73 public readonly setMaxThinkingTokensCalls: Array<number | null> = [];
74 public closeCalls = 0;74 public closeCalls = 0;
75 public closeError: unknown | undefined;75 public closeError: unknown | undefined;
76 /** Set by tests that exercise Claude's graceful interrupt. */
77 public interrupt?: () => Promise<unknown>;
7678
77 emit(message: SDKMessage): void {79 emit(message: SDKMessage): void {
78 if (this.done) {80 if (this.done) {
...@@ -3255,6 +3257,83 @@ describe("ClaudeAdapterLive", () => {...@@ -3255,6 +3257,83 @@ describe("ClaudeAdapterLive", () => {
3255 );3257 );
3256 });3258 });
32573259
3260 it.effect("interruptTurn lets Claude abort the turn before closing the session", () => {
3261 const harness = makeHarness();
3262 return Effect.gen(function* () {
3263 const adapter = yield* ClaudeAdapter;
3264 const session = yield* adapter.startSession({
3265 threadId: THREAD_ID,
3266 provider: ProviderDriverKind.make("claudeAgent"),
3267 runtimeMode: "full-access",
3268 });
3269 yield* adapter.sendTurn({
3270 threadId: session.threadId,
3271 input: "hello",
3272 attachments: [],
3273 });
3274
3275 const turnCompletedFiber = yield* adapter.streamEvents.pipe(
3276 Stream.filter((event) => event.type === "turn.completed"),
3277 Stream.take(1),
3278 Stream.runCollect,
3279 Effect.forkChild,
3280 );
3281 let closeCallsAtInterrupt: number | undefined;
3282 harness.query.interrupt = async () => {
3283 closeCallsAtInterrupt = harness.query.closeCalls;
3284 harness.query.emit({
3285 type: "result",
3286 subtype: "error_during_execution",
3287 is_error: false,
3288 errors: ["Error: Request was aborted."],
3289 session_id: "sdk-session",
3290 uuid: "result-interrupted",
3291 } as unknown as SDKMessage);
3292 };
3293
3294 yield* adapter.interruptTurn(session.threadId);
3295
3296 assert.equal(closeCallsAtInterrupt, 0);
3297 assert.equal(harness.query.closeCalls, 1);
3298 const [turnCompleted] = Array.from(yield* Fiber.join(turnCompletedFiber));
3299 assert.equal(turnCompleted?.type, "turn.completed");
3300 if (turnCompleted?.type === "turn.completed") {
3301 assert.equal(turnCompleted.payload.state, "interrupted");
3302 }
3303 }).pipe(
3304 Effect.provideService(Random.Random, makeDeterministicRandomService()),
3305 Effect.provide(harness.layer),
3306 );
3307 });
3308
3309 it.effect("interruptTurn closes the session when Claude never aborts the turn", () => {
3310 const harness = makeHarness();
3311 return Effect.gen(function* () {
3312 const adapter = yield* ClaudeAdapter;
3313 const session = yield* adapter.startSession({
3314 threadId: THREAD_ID,
3315 provider: ProviderDriverKind.make("claudeAgent"),
3316 runtimeMode: "full-access",
3317 });
3318 yield* adapter.sendTurn({
3319 threadId: session.threadId,
3320 input: "hello",
3321 attachments: [],
3322 });
3323 harness.query.interrupt = () => new Promise(() => {});
3324
3325 const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
3326 yield* TestClock.adjust("3 seconds");
3327 yield* Fiber.join(interruptFiber);
3328
3329 assert.equal(harness.query.closeCalls, 1);
3330 assert.equal(yield* adapter.hasSession(session.threadId), false);
3331 }).pipe(
3332 Effect.provideService(Random.Random, makeDeterministicRandomService()),
3333 Effect.provide(harness.layer),
3334 );
3335 });
3336
3258 it.effect("keeps the session available when process close fails", () => {3337 it.effect("keeps the session available when process close fails", () => {
3259 const harness = makeHarness();3338 const harness = makeHarness();
3260 return Effect.gen(function* () {3339 return Effect.gen(function* () {
apps/server/src/provider/Layers/ClaudeAdapter.ts+30
...@@ -385,6 +385,8 @@ interface ClaudeTaskAgentState {...@@ -385,6 +385,8 @@ interface ClaudeTaskAgentState {
385 * lifetime; oldest entries evict first.385 * lifetime; oldest entries evict first.
386 */386 */
387const PENDING_TASK_MODEL_CAP = 64;387const PENDING_TASK_MODEL_CAP = 64;
388/** How long Stop waits for Claude to abort a turn before killing the process. */
389const CLAUDE_INTERRUPT_GRACE = "3 seconds";
388390
389/**391/**
390 * Buffers a subagent snapshot's authoritative model under its392 * Buffers a subagent snapshot's authoritative model under its
...@@ -453,10 +455,14 @@ interface ClaudeSessionContext {...@@ -453,10 +455,14 @@ interface ClaudeSessionContext {
453 lastThreadStartedId: string | undefined;455 lastThreadStartedId: string | undefined;
454 /** Limits already announced for the running turn, keyed `window:resetsAt`. */456 /** Limits already announced for the running turn, keyed `window:resetsAt`. */
455 announcedUsageLimits: { turnId: string; keys: Set<string> } | undefined;457 announcedUsageLimits: { turnId: string; keys: Set<string> } | undefined;
458 /** Resolved by completeTurn while Stop waits for Claude to abort the turn. */
459 interruptedTurnSettled: Deferred.Deferred<void> | undefined;
456 stopped: boolean;460 stopped: boolean;
457}461}
458462
459interface ClaudeQueryRuntime extends AsyncIterable<SDKMessage> {463interface ClaudeQueryRuntime extends AsyncIterable<SDKMessage> {
464 /** SDK Query.interrupt — present on real queries; optional for test doubles. */
465 readonly interrupt?: () => Promise<unknown>;
460 readonly setModel: (model?: string) => Promise<void>;466 readonly setModel: (model?: string) => Promise<void>;
461 readonly setPermissionMode: (mode: PermissionMode) => Promise<void>;467 readonly setPermissionMode: (mode: PermissionMode) => Promise<void>;
462 readonly setMaxThinkingTokens: (maxThinkingTokens: number | null) => Promise<void>;468 readonly setMaxThinkingTokens: (maxThinkingTokens: number | null) => Promise<void>;
...@@ -2836,6 +2842,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2836,6 +2842,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
28362842
2837 const updatedAt = yield* nowIso;2843 const updatedAt = yield* nowIso;
2838 context.turnState = undefined;2844 context.turnState = undefined;
2845 if (context.interruptedTurnSettled) {
2846 yield* Deferred.succeed(context.interruptedTurnSettled, undefined);
2847 }
2839 context.session = {2848 context.session = {
2840 ...context.session,2849 ...context.session,
2841 status: "ready",2850 status: "ready",
...@@ -5045,6 +5054,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -5045,6 +5054,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
5045 lastAssistantUuid: resumeState?.resumeSessionAt,5054 lastAssistantUuid: resumeState?.resumeSessionAt,
5046 lastThreadStartedId: undefined,5055 lastThreadStartedId: undefined,
5047 announcedUsageLimits: undefined,5056 announcedUsageLimits: undefined,
5057 interruptedTurnSettled: undefined,
5048 stopped: false,5058 stopped: false,
5049 };5059 };
5050 yield* Ref.set(contextRef, context);5060 yield* Ref.set(contextRef, context);
...@@ -5276,6 +5286,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -5276,6 +5286,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
5276 const interruptTurn: ClaudeAdapterShape["interruptTurn"] = Effect.fn("interruptTurn")(5286 const interruptTurn: ClaudeAdapterShape["interruptTurn"] = Effect.fn("interruptTurn")(
5277 function* (threadId, _turnId) {5287 function* (threadId, _turnId) {
5278 const context = yield* requireSession(threadId);5288 const context = yield* requireSession(threadId);
5289 yield* settleInterruptedTurn(context);
5279 // interrupt() can acknowledge while resumed background tasks keep the5290 // interrupt() can acknowledge while resumed background tasks keep the
5280 // CLI alive. Stop is a hard session boundary for Claude, so close the5291 // CLI alive. Stop is a hard session boundary for Claude, so close the
5281 // query and let the SDK escalate to SIGKILL when graceful exit fails.5292 // query and let the SDK escalate to SIGKILL when graceful exit fails.
...@@ -5283,6 +5294,25 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -5283,6 +5294,25 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
5283 },5294 },
5284 );5295 );
52855296
5297 // Lets Claude abort the running turn through its own path before Stop kills
5298 // the process, so the prompt reaches the transcript. Killing a first turn
5299 // before Claude writes it leaves a resume cursor for a session Claude never
5300 // saved, and every later message fails with "No conversation found".
5301 const settleInterruptedTurn = Effect.fn("settleInterruptedTurn")(function* (
5302 context: ClaudeSessionContext,
5303 ) {
5304 const interrupt = context.query.interrupt?.bind(context.query);
5305 if (context.stopped || !context.turnState || !interrupt) return;
5306 const settled = yield* Deferred.make<void>();
5307 context.interruptedTurnSettled = settled;
5308 yield* Effect.tryPromise(interrupt).pipe(
5309 Effect.ignore,
5310 Effect.andThen(Deferred.await(settled)),
5311 Effect.timeoutOption(CLAUDE_INTERRUPT_GRACE),
5312 );
5313 context.interruptedTurnSettled = undefined;
5314 });
5315
5286 const readThread: ClaudeAdapterShape["readThread"] = Effect.fn("readThread")(5316 const readThread: ClaudeAdapterShape["readThread"] = Effect.fn("readThread")(
5287 function* (threadId) {5317 function* (threadId) {
5288 const context = yield* requireSession(threadId);5318 const context = yield* requireSession(threadId);