From 0a503d0cdeb4e6cd8a9952c930d6c61ce28104fa Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dean=20Ch=C3=A9rif?= <keyzou95@gmail.com>
Date: Tue, 24 Mar 2026 17:40:19 +0100
Subject: [PATCH] fix(provider,claude): handle prompt stream interrupt on
 session stop (#1365)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 .../src/provider/Layers/ClaudeAdapter.test.ts | 67 +++++++++++++++++++
 .../src/provider/Layers/ClaudeAdapter.ts      |  3 +
 2 files changed, 70 insertions(+)

diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts
index e23148973ea9f130f8882744cc81e798c3e93583..cf91bf2d52b5970a52f900ad6b4b96f19f9b24f6 100644
--- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts
+++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts
@@ -1153,6 +1153,73 @@ describe("ClaudeAdapterLive", () => {
     );
   });
 
+  it.effect("stopSession does not throw into the SDK prompt consumer", () => {
+    // The SDK consumes user messages via `for await (... of prompt)`.
+    // Stopping a session must end that loop cleanly — not throw an error.
+    //
+    // FakeClaudeQuery.close() masks this by resolving pending iterators
+    // before the shutdown propagates. Override it to match real SDK behavior
+    // where close() does not resolve the prompt consumer.
+    const query = new FakeClaudeQuery();
+    (query as { close: () => void }).close = () => {
+      query.closeCalls += 1;
+    };
+
+    let promptConsumerError: unknown = undefined;
+
+    const layer = makeClaudeAdapterLive({
+      createQuery: (input) => {
+        // Simulate the SDK consuming the prompt iterable
+        (async () => {
+          try {
+            for await (const _message of input.prompt) {
+              /* SDK processes user messages */
+            }
+          } catch (error) {
+            promptConsumerError = error;
+          }
+        })();
+        return query;
+      },
+    }).pipe(
+      Layer.provideMerge(ServerConfig.layerTest("/tmp/claude-adapter-test", "/tmp")),
+      Layer.provideMerge(NodeServices.layer),
+    );
+
+    return Effect.gen(function* () {
+      const adapter = yield* ClaudeAdapter;
+
+      const runtimeEventsFiber = Effect.runFork(
+        Stream.runForEach(adapter.streamEvents, () => Effect.void),
+      );
+
+      yield* adapter.startSession({
+        threadId: THREAD_ID,
+        provider: "claudeAgent",
+        runtimeMode: "full-access",
+      });
+
+      yield* adapter.stopSession(THREAD_ID);
+
+      yield* Effect.yieldNow;
+      yield* Effect.yieldNow;
+      yield* Effect.yieldNow;
+      yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 50)));
+
+      runtimeEventsFiber.interruptUnsafe();
+
+      assert.equal(
+        promptConsumerError,
+        undefined,
+        `Prompt consumer should not receive a thrown error on session stop, ` +
+          `but got: "${promptConsumerError instanceof Error ? promptConsumerError.message : String(promptConsumerError)}"`,
+      );
+    }).pipe(
+      Effect.provideService(Random.Random, makeDeterministicRandomService()),
+      Effect.provide(layer),
+    );
+  });
+
   it.effect("forwards Claude task progress summaries for subagent updates", () => {
     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 acbee860537de61cd4919645df247d1652632867..c1b320860daf9b4f00f15b3598582b6adbee36e9 100644
--- a/apps/server/src/provider/Layers/ClaudeAdapter.ts
+++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts
@@ -2258,6 +2258,9 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) {
         const prompt = Stream.fromQueue(promptQueue).pipe(
           Stream.filter((item) => item.type === "message"),
           Stream.map((item) => item.message),
+          Stream.catchCause((cause) =>
+            Cause.hasInterruptsOnly(cause) ? Stream.empty : Stream.failCause(cause),
+          ),
           Stream.toAsyncIterable,
         );
 
-- 
2.54.0