| author |  Aaron Abu Usama <50079365+AaronAbuUsama@users.noreply.github.com> 2026-08-15 12:27:04+00:00 |
| committer |  GitHub <noreply@github.com> 2026-08-15 12:27:04+00:00 |
| log | 3b54a2a574b061a1ce80d7130e0776f4f5c95e20 |
| tree | 9359e9c24141c01495d47cbee38c0ebbb0783f77 |
| parent | 39167eb1a586ee9e060d6ec3314617e14e0259c0 |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): settle pending user-input requests when a Claude session stops (#5127)
Co-authored-by: Capxul Agent <agent@capxul.dev>
Co-authored-by: Julius Marminge <julius0216@outlook.com>
Co-authored-by: codex <codex@users.noreply.github.com>2 files changed, 81 insertions(+), 6 deletions(-)
apps
server
src
provider
apps/server/src/provider/Layers/ClaudeAdapter.test.ts+61
| ... | @@ -4448,6 +4448,67 @@ describe("ClaudeAdapterLive", () => { | ... | @@ -4448,6 +4448,67 @@ describe("ClaudeAdapterLive", () => { |
| 4448 | ); | 4448 | ); |
| 4449 | }); | 4449 | }); |
| 4450 | 4450 | ||
| 4451 | it.effect("stopping a session settles pending user-input waits", () => { | ||
| 4452 | const harness = makeHarness(); | ||
| 4453 | return Effect.gen(function* () { | ||
| 4454 | const adapter = yield* ClaudeAdapter; | ||
| 4455 | |||
| 4456 | yield* adapter.startSession({ | ||
| 4457 | threadId: THREAD_ID, | ||
| 4458 | provider: ProviderDriverKind.make("claudeAgent"), | ||
| 4459 | runtimeMode: "approval-required", | ||
| 4460 | }); | ||
| 4461 | |||
| 4462 | yield* Stream.take(adapter.streamEvents, 3).pipe(Stream.runDrain); | ||
| 4463 | |||
| 4464 | const canUseTool = harness.getLastCreateQueryInput()?.options.canUseTool; | ||
| 4465 | assert.equal(typeof canUseTool, "function"); | ||
| 4466 | if (!canUseTool) { | ||
| 4467 | return; | ||
| 4468 | } | ||
| 4469 | |||
| 4470 | const permissionPromise = canUseTool( | ||
| 4471 | "AskUserQuestion", | ||
| 4472 | { | ||
| 4473 | questions: [ | ||
| 4474 | { | ||
| 4475 | question: "Continue?", | ||
| 4476 | header: "Continue", | ||
| 4477 | options: [{ label: "Yes", description: "Proceed" }], | ||
| 4478 | multiSelect: false, | ||
| 4479 | }, | ||
| 4480 | ], | ||
| 4481 | }, | ||
| 4482 | { signal: new AbortController().signal, toolUseID: "tool-ask-stop" }, | ||
| 4483 | ); | ||
| 4484 | |||
| 4485 | const requestedEvent = yield* Stream.runHead(adapter.streamEvents); | ||
| 4486 | if (requestedEvent._tag !== "Some" || requestedEvent.value.type !== "user-input.requested") { | ||
| 4487 | assert.fail("Expected user-input.requested event"); | ||
| 4488 | return; | ||
| 4489 | } | ||
| 4490 | |||
| 4491 | // The session dies while the question is still on screen. | ||
| 4492 | yield* adapter.stopSession(THREAD_ID); | ||
| 4493 | |||
| 4494 | const resolvedEvent = yield* Stream.runHead(adapter.streamEvents); | ||
| 4495 | if (resolvedEvent._tag !== "Some" || resolvedEvent.value.type !== "user-input.resolved") { | ||
| 4496 | assert.fail("Expected user-input.resolved event"); | ||
| 4497 | return; | ||
| 4498 | } | ||
| 4499 | assert.deepEqual(resolvedEvent.value.payload.answers, {}); | ||
| 4500 | |||
| 4501 | const permissionResult = yield* Effect.promise(() => permissionPromise); | ||
| 4502 | assert.deepEqual(permissionResult, { | ||
| 4503 | behavior: "deny", | ||
| 4504 | message: "User cancelled tool execution.", | ||
| 4505 | } satisfies PermissionResult); | ||
| 4506 | }).pipe( | ||
| 4507 | Effect.provideService(Random.Random, makeDeterministicRandomService()), | ||
| 4508 | Effect.provide(harness.layer), | ||
| 4509 | ); | ||
| 4510 | }); | ||
| 4511 | |||
| 4451 | it.effect("writes provider-native observability records when enabled", () => { | 4512 | it.effect("writes provider-native observability records when enabled", () => { |
| 4452 | const nativeEvents: Array<{ | 4513 | const nativeEvents: Array<{ |
| 4453 | event?: { | 4514 | event?: { |
apps/server/src/provider/Layers/ClaudeAdapter.ts+20-6
| ... | @@ -194,6 +194,8 @@ function toSessionPermissionUpdates( | ... | @@ -194,6 +194,8 @@ function toSessionPermissionUpdates( |
| 194 | interface PendingUserInput { | 194 | interface PendingUserInput { |
| 195 | readonly questions: ReadonlyArray<UserInputQuestion>; | 195 | readonly questions: ReadonlyArray<UserInputQuestion>; |
| 196 | readonly answers: Deferred.Deferred<ProviderUserInputAnswers>; | 196 | readonly answers: Deferred.Deferred<ProviderUserInputAnswers>; |
| 197 | /** Unparks the waiting handler as cancelled. Session teardown must run it. */ | ||
| 198 | readonly cancel: Effect.Effect<void>; | ||
| 197 | } | 199 | } |
| 198 | 200 | ||
| 199 | interface ToolInFlight { | 201 | interface ToolInFlight { |
| ... | @@ -3656,6 +3658,12 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( | ... | @@ -3656,6 +3658,12 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 3656 | } | 3658 | } |
| 3657 | context.pendingApprovals.clear(); | 3659 | context.pendingApprovals.clear(); |
| 3658 | 3660 | ||
| 3661 | // Same reason as the approvals above: a request nobody can answer any more | ||
| 3662 | // must not stay open, or the thread can never be settled. | ||
| 3663 | for (const pending of [...context.pendingUserInputs.values()]) { | ||
| 3664 | yield* pending.cancel; | ||
| 3665 | } | ||
| 3666 | |||
| 3659 | if (context.turnState) { | 3667 | if (context.turnState) { |
| 3660 | yield* completeTurn(context, "interrupted", "Session stopped."); | 3668 | yield* completeTurn(context, "interrupted", "Session stopped."); |
| 3661 | } | 3669 | } |
| ... | @@ -3837,9 +3845,20 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( | ... | @@ -3837,9 +3845,20 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 3837 | 3845 | ||
| 3838 | const answersDeferred = yield* Deferred.make<ProviderUserInputAnswers>(); | 3846 | const answersDeferred = yield* Deferred.make<ProviderUserInputAnswers>(); |
| 3839 | let aborted = false; | 3847 | let aborted = false; |
| 3848 | const settleAsAborted = Effect.suspend(() => { | ||
| 3849 | if (!pendingUserInputs.has(requestId)) { | ||
| 3850 | return Effect.void; | ||
| 3851 | } | ||
| 3852 | aborted = true; | ||
| 3853 | pendingUserInputs.delete(requestId); | ||
| 3854 | return Deferred.succeed(answersDeferred, {} as ProviderUserInputAnswers).pipe( | ||
| 3855 | Effect.ignore, | ||
| 3856 | ); | ||
| 3857 | }); | ||
| 3840 | const pendingInput: PendingUserInput = { | 3858 | const pendingInput: PendingUserInput = { |
| 3841 | questions, | 3859 | questions, |
| 3842 | answers: answersDeferred, | 3860 | answers: answersDeferred, |
| 3861 | cancel: settleAsAborted, | ||
| 3843 | }; | 3862 | }; |
| 3844 | 3863 | ||
| 3845 | // Emit user-input.requested so the UI can present the questions. | 3864 | // Emit user-input.requested so the UI can present the questions. |
| ... | @@ -3874,12 +3893,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( | ... | @@ -3874,12 +3893,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 3874 | 3893 | ||
| 3875 | // Handle abort (e.g. turn interrupted while waiting for user input). | 3894 | // Handle abort (e.g. turn interrupted while waiting for user input). |
| 3876 | const onAbort = () => { | 3895 | const onAbort = () => { |
| 3877 | if (!pendingUserInputs.has(requestId)) { | 3896 | runFork(settleAsAborted); |
| 3878 | return; | ||
| 3879 | } | ||
| 3880 | aborted = true; | ||
| 3881 | pendingUserInputs.delete(requestId); | ||
| 3882 | runFork(Deferred.succeed(answersDeferred, {} as ProviderUserInputAnswers)); | ||
| 3883 | }; | 3897 | }; |
| 3884 | callbackOptions.signal.addEventListener("abort", onAbort, { | 3898 | callbackOptions.signal.addEventListener("abort", onAbort, { |
| 3885 | once: true, | 3899 | once: true, |