Handle Claude stream interruption and collapse duplicate work log update

- close Claude sessions cleanly when the runtime stream aborts mid-turn
- normalize interrupted stream exits into a completed interrupted turn
- collapse repeated tool lifecycle updates into a single work log entry

Author: juliusmarminge

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -10,7 +10,12 @@ import type {
   SDKMessage,
   SDKUserMessage,
 } from "@anthropic-ai/claude-agent-sdk";
-import { ApprovalRequestId, ProviderItemId, ThreadId } from "@t3tools/contracts";
+import {
+  ApprovalRequestId,
+  ProviderItemId,
+  ProviderRuntimeEvent,
+  ThreadId,
+} from "@t3tools/contracts";
 import { assert, describe, it } from "@effect/vitest";
 import { Effect, Fiber, Layer, Random, Stream } from "effect";
 
@@ -22,8 +27,12 @@ import { makeClaudeAdapterLive, type ClaudeAdapterLiveOptions } from "./ClaudeAd
 
 class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
   private readonly queue: Array<SDKMessage> = [];
-  private readonly resolvers: Array<(value: IteratorResult<SDKMessage>) => void> = [];
+  private readonly waiters: Array<{
+    readonly resolve: (value: IteratorResult<SDKMessage>) => void;
+    readonly reject: (reason: unknown) => void;
+  }> = [];
   private done = false;
+  private failure: unknown | undefined;
 
   public readonly interruptCalls: Array<void> = [];
   public readonly setModelCalls: Array<string | undefined> = [];
@@ -35,21 +44,33 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
     if (this.done) {
       return;
     }
-    const resolver = this.resolvers.shift();
-    if (resolver) {
-      resolver({ done: false, value: message });
+    const waiter = this.waiters.shift();
+    if (waiter) {
+      waiter.resolve({ done: false, value: message });
       return;
     }
     this.queue.push(message);
   }
 
+  fail(cause: unknown): void {
+    if (this.done) {
+      return;
+    }
+    this.done = true;
+    this.failure = cause;
+    for (const waiter of this.waiters.splice(0)) {
+      waiter.reject(cause);
+    }
+  }
+
   finish(): void {
     if (this.done) {
       return;
     }
     this.done = true;
-    for (const resolver of this.resolvers.splice(0)) {
-      resolver({ done: true, value: undefined });
+    this.failure = undefined;
+    for (const waiter of this.waiters.splice(0)) {
+      waiter.resolve({ done: true, value: undefined });
     }
   }
 
@@ -86,14 +107,22 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
             });
           }
         }
+        if (this.failure !== undefined) {
+          const failure = this.failure;
+          this.failure = undefined;
+          return Promise.reject(failure);
+        }
         if (this.done) {
           return Promise.resolve({
             done: true,
             value: undefined,
           });
         }
-        return new Promise((resolve) => {
-          this.resolvers.push(resolve);
+        return new Promise((resolve, reject) => {
+          this.waiters.push({
+            resolve,
+            reject,
+          });
         });
       },
     };
@@ -1037,6 +1066,71 @@ describe("ClaudeAdapterLive", () => {
     );
   });
 
+  it.effect("closes the session when the Claude stream aborts after a turn starts", () => {
+    const harness = makeHarness();
+    return Effect.gen(function* () {
+      const adapter = yield* ClaudeAdapter;
+      const runtimeEvents: Array<ProviderRuntimeEvent> = [];
+
+      const runtimeEventsFiber = Effect.runFork(
+        Stream.runForEach(adapter.streamEvents, (event) =>
+          Effect.sync(() => {
+            runtimeEvents.push(event);
+          }),
+        ),
+      );
+
+      yield* adapter.startSession({
+        threadId: THREAD_ID,
+        provider: "claudeAgent",
+        runtimeMode: "full-access",
+      });
+
+      const turn = yield* adapter.sendTurn({
+        threadId: THREAD_ID,
+        input: "hello",
+        attachments: [],
+      });
+
+      harness.query.fail(new Error("All fibers interrupted without error"));
+
+      yield* Effect.yieldNow;
+      yield* Effect.yieldNow;
+      yield* Effect.yieldNow;
+      runtimeEventsFiber.interruptUnsafe();
+      assert.deepEqual(
+        runtimeEvents.map((event) => event.type),
+        [
+          "session.started",
+          "session.configured",
+          "session.state.changed",
+          "turn.started",
+          "turn.completed",
+          "session.exited",
+        ],
+      );
+
+      const turnCompleted = runtimeEvents[4];
+      assert.equal(turnCompleted?.type, "turn.completed");
+      if (turnCompleted?.type === "turn.completed") {
+        assert.equal(String(turnCompleted.turnId), String(turn.turnId));
+        assert.equal(turnCompleted.payload.state, "interrupted");
+        assert.equal(turnCompleted.payload.errorMessage, "Claude runtime interrupted.");
+      }
+
+      const sessionExited = runtimeEvents[5];
+      assert.equal(sessionExited?.type, "session.exited");
+
+      assert.equal(yield* adapter.hasSession(THREAD_ID), false);
+      const sessions = yield* adapter.listSessions();
+      assert.equal(sessions.length, 0);
+      assert.equal(harness.query.closeCalls, 1);
+    }).pipe(
+      Effect.provideService(Random.Random, makeDeterministicRandomService()),
+      Effect.provide(harness.layer),
+    );
+  });
+
   it.effect("forwards Claude task progress summaries for subagent updates", () => {
     const harness = makeHarness();
     return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -51,7 +51,9 @@ import {
   DateTime,
   Deferred,
   Effect,
+  Exit,
   FileSystem,
+  Fiber,
   Layer,
   Queue,
   Random,
@@ -141,6 +143,7 @@ interface ClaudeSessionContext {
   session: ProviderSession;
   readonly promptQueue: Queue.Queue<PromptQueueItem>;
   readonly query: ClaudeQueryRuntime;
+  streamFiber: Fiber.Fiber<void, Error> | undefined;
   readonly startedAt: string;
   readonly basePermissionMode: PermissionMode | undefined;
   resumeSessionId: string | undefined;
@@ -189,6 +192,47 @@ function toMessage(cause: unknown, fallback: string): string {
   return fallback;
 }
 
+function toError(cause: unknown, fallback: string): Error {
+  return cause instanceof Error ? cause : new Error(toMessage(cause, fallback));
+}
+
+function normalizeClaudeStreamMessages(cause: Cause.Cause<Error>): ReadonlyArray<string> {
+  const errors = Cause.prettyErrors(cause)
+    .map((error) => error.message.trim())
+    .filter((message) => message.length > 0);
+  if (errors.length > 0) {
+    return errors;
+  }
+
+  const squashed = toMessage(Cause.squash(cause), "").trim();
+  return squashed.length > 0 ? [squashed] : [];
+}
+
+function isClaudeInterruptedMessage(message: string): boolean {
+  const normalized = message.toLowerCase();
+  return (
+    normalized.includes("all fibers interrupted without error") ||
+    normalized.includes("request was aborted") ||
+    normalized.includes("interrupted by user")
+  );
+}
+
+function isClaudeInterruptedCause(cause: Cause.Cause<Error>): boolean {
+  return (
+    Cause.hasInterruptsOnly(cause) ||
+    normalizeClaudeStreamMessages(cause).some(isClaudeInterruptedMessage)
+  );
+}
+
+function messageFromClaudeStreamCause(cause: Cause.Cause<Error>, fallback: string): string {
+  return normalizeClaudeStreamMessages(cause)[0] ?? fallback;
+}
+
+function interruptionMessageFromClaudeCause(cause: Cause.Cause<Error>): string {
+  const message = messageFromClaudeStreamCause(cause, "Claude runtime interrupted.");
+  return isClaudeInterruptedMessage(message) ? "Claude runtime interrupted." : message;
+}
+
 function resultErrorsText(result: SDKResultMessage): string {
   return "errors" in result && Array.isArray(result.errors)
     ? result.errors.join(" ").toLowerCase()
@@ -2045,21 +2089,48 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) {
         }
       });
 
-    const runSdkStream = (context: ClaudeSessionContext): Effect.Effect<void> =>
-      Stream.fromAsyncIterable(context.query, (cause) => cause).pipe(
+    const runSdkStream = (context: ClaudeSessionContext): Effect.Effect<void, Error> =>
+      Stream.fromAsyncIterable(context.query, (cause) =>
+        toError(cause, "Claude runtime stream failed."),
+      ).pipe(
         Stream.takeWhile(() => !context.stopped),
         Stream.runForEach((message) => handleSdkMessage(context, message)),
-        Effect.catchCause((cause) =>
-          Effect.gen(function* () {
-            if (Cause.hasInterruptsOnly(cause) || context.stopped) {
-              return;
+      );
+
+    const handleStreamExit = (
+      context: ClaudeSessionContext,
+      exit: Exit.Exit<void, Error>,
+    ): Effect.Effect<void> =>
+      Effect.gen(function* () {
+        if (context.stopped) {
+          return;
+        }
+
+        if (Exit.isFailure(exit)) {
+          if (isClaudeInterruptedCause(exit.cause)) {
+            if (context.turnState) {
+              yield* completeTurn(
+                context,
+                "interrupted",
+                interruptionMessageFromClaudeCause(exit.cause),
+              );
             }
-            const message = toMessage(Cause.squash(cause), "Claude runtime stream failed.");
-            yield* emitRuntimeError(context, message, cause);
+          } else {
+            const message = messageFromClaudeStreamCause(
+              exit.cause,
+              "Claude runtime stream failed.",
+            );
+            yield* emitRuntimeError(context, message, Cause.pretty(exit.cause));
             yield* completeTurn(context, "failed", message);
-          }),
-        ),
-      );
+          }
+        } else if (context.turnState) {
+          yield* completeTurn(context, "interrupted", "Claude runtime stream ended.");
+        }
+
+        yield* stopSessionInternal(context, {
+          emitExitEvent: true,
+        });
+      });
 
     const stopSessionInternal = (
       context: ClaudeSessionContext,
@@ -2096,7 +2167,18 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) {
 
         yield* Queue.shutdown(context.promptQueue);
 
-        context.query.close();
+        const streamFiber = context.streamFiber;
+        context.streamFiber = undefined;
+        if (streamFiber && streamFiber.pollUnsafe() === undefined) {
+          yield* Fiber.interrupt(streamFiber);
+        }
+
+        // @effect-diagnostics-next-line tryCatchInEffectGen:off
+        try {
+          context.query.close();
+        } catch (cause) {
+          yield* emitRuntimeError(context, "Failed to close Claude runtime query.", cause);
+        }
 
         const updatedAt = yield* nowIso;
         context.session = {
@@ -2536,6 +2618,7 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) {
           session,
           promptQueue,
           query: queryRuntime,
+          streamFiber: undefined,
           startedAt,
           basePermissionMode: permissionMode,
           resumeSessionId: sessionId,
@@ -2597,7 +2680,17 @@ function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) {
           providerRefs: {},
         });
 
-        Effect.runFork(runSdkStream(context));
+        const streamFiber = Effect.runFork(runSdkStream(context));
+        context.streamFiber = streamFiber;
+        streamFiber.addObserver((exit) => {
+          if (context.stopped) {
+            return;
+          }
+          if (context.streamFiber === streamFiber) {
+            context.streamFiber = undefined;
+          }
+          Effect.runFork(handleStreamExit(context, exit));
+        });
 
         return {
           ...session,