Preserve Claude stream interruption errors

- Split Claude stream failures from interrupts
- Keep interrupted turns from being reported as generic failures

Author: juliusmarminge

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

@@ -55,6 +55,7 @@ import {
   FileSystem,
   Fiber,
   Layer,
+  Option,
   Queue,
   Random,
   Ref,
@@ -146,7 +147,7 @@ interface ClaudeSessionContext {
   session: ProviderSession;
   readonly promptQueue: Queue.Queue<PromptQueueItem>;
   readonly query: ClaudeQueryRuntime;
-  streamFiber: Fiber.Fiber<void, Error> | undefined;
+  streamFiber: Fiber.Fiber<void, ClaudeStreamError> | undefined;
   readonly startedAt: string;
   readonly basePermissionMode: PermissionMode | undefined;
   currentApiModelId: string | undefined;
@@ -183,10 +184,22 @@ export interface ClaudeAdapterLiveOptions {
   readonly nativeEventLogger?: EventNdjsonLogger;
 }
 
-class ClaudeStreamError extends Schema.TaggedErrorClass<ClaudeStreamError>()("ClaudeStreamError", {
-  message: Schema.String,
-  cause: Schema.optional(Schema.Defect),
-}) {}
+class ClaudeStreamInterruptedError extends Schema.TaggedErrorClass<ClaudeStreamInterruptedError>()(
+  "ClaudeStreamInterruptedError",
+  {
+    message: Schema.String,
+    cause: Schema.optional(Schema.Defect),
+  },
+) {}
+class ClaudeStreamFailedError extends Schema.TaggedErrorClass<ClaudeStreamFailedError>()(
+  "ClaudeStreamFailedError",
+  {
+    message: Schema.String,
+    cause: Schema.optional(Schema.Defect),
+  },
+) {}
+const isClaudeStreamInterruptedError = Schema.is(ClaudeStreamInterruptedError);
+type ClaudeStreamError = ClaudeStreamInterruptedError | ClaudeStreamFailedError;
 
 function isUuid(value: string): boolean {
   return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value);
@@ -196,17 +209,6 @@ function isSyntheticClaudeThreadId(value: string): boolean {
   return value.startsWith("claude-thread-");
 }
 
-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;
-  }
-
-  return [Cause.pretty(cause)];
-}
-
 function getEffectiveClaudeCodeEffort(
   effort: ClaudeCodeEffort | null | undefined,
 ): Exclude<ClaudeCodeEffort, "ultrathink"> | null {
@@ -225,22 +227,6 @@ function isClaudeInterruptedMessage(message: string): boolean {
   );
 }
 
-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()
@@ -2221,39 +2207,51 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
   });
 
   const runSdkStream = (context: ClaudeSessionContext) =>
-    Stream.fromAsyncIterable(
-      context.query,
-      (cause) =>
-        new ClaudeStreamError({
-          message: "An error occurred while running the Claude SDK stream.",
-          cause,
-        }),
-    ).pipe(
+    Stream.fromAsyncIterable(context.query, (cause) => {
+      const message =
+        cause instanceof Error && cause.message.trim().length > 0
+          ? cause.message
+          : "Claude runtime stream failed.";
+      return isClaudeInterruptedMessage(message)
+        ? new ClaudeStreamInterruptedError({ message, cause })
+        : new ClaudeStreamFailedError({ message, cause });
+    }).pipe(
       Stream.takeWhile(() => !context.stopped),
       Stream.runForEach((message) => handleSdkMessage(context, message)),
     );
 
   const handleStreamExit = Effect.fn("handleStreamExit")(function* (
     context: ClaudeSessionContext,
-    exit: Exit.Exit<void, Error>,
+    exit: Exit.Exit<void, ClaudeStreamError>,
   ) {
     if (context.stopped) {
       return;
     }
 
     if (Exit.isFailure(exit)) {
-      if (isClaudeInterruptedCause(exit.cause)) {
+      if (Cause.hasInterruptsOnly(exit.cause)) {
         if (context.turnState) {
-          yield* completeTurn(
-            context,
-            "interrupted",
-            interruptionMessageFromClaudeCause(exit.cause),
-          );
+          yield* completeTurn(context, "interrupted", "Claude runtime interrupted.");
         }
       } else {
-        const message = messageFromClaudeStreamCause(exit.cause, "Claude runtime stream failed.");
-        yield* emitRuntimeError(context, message, Cause.pretty(exit.cause));
-        yield* completeTurn(context, "failed", message);
+        yield* Option.match(Cause.findErrorOption(exit.cause), {
+          onNone: () =>
+            context.turnState
+              ? completeTurn(context, "failed", "Claude runtime stream failed.")
+              : Effect.void,
+          onSome: (streamError) =>
+            isClaudeStreamInterruptedError(streamError)
+              ? context.turnState
+                ? completeTurn(context, "interrupted", "Claude runtime interrupted.")
+                : Effect.void
+              : Effect.gen(function* () {
+                  const message = streamError.message;
+                  yield* emitRuntimeError(context, message, Cause.pretty(exit.cause));
+                  if (context.turnState) {
+                    yield* completeTurn(context, "failed", message);
+                  }
+                }),
+        });
       }
     } else if (context.turnState) {
       yield* completeTurn(context, "interrupted", "Claude runtime stream ended.");