Fix detached fiber in ClaudeCodeAdapter to use Effect.forkDetach with explicit interruption

cursoragent · cursoragent · commit 4e4055c7f0d6 · 2026-02-27T20:49:38.000Z
Replace Effect.runFork(runSdkStream(context)) with yield* Effect.forkDetach()
to properly inherit provided services (e.g. deterministic Random in tests)
and enable deterministic cleanup.

Store the fiber reference in ClaudeSessionContext and interrupt it in
stopSessionInternal before other cleanup, eliminating the race window
where the stream fiber could offer events to runtimeEventQueue after
it has been shut down by the scope finalizer.
diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts
@@ -31,7 +31,18 @@ import {
   ProviderThreadId,
   ProviderTurnId,
 } from "@t3tools/contracts";
-import { Cause, DateTime, Deferred, Effect, Layer, Queue, Random, Ref, Stream } from "effect";
+import {
+  Cause,
+  DateTime,
+  Deferred,
+  Effect,
+  Fiber,
+  Layer,
+  Queue,
+  Random,
+  Ref,
+  Stream,
+} from "effect";
 
 import {
   ProviderAdapterProcessError,
@@ -99,6 +110,7 @@ interface ClaudeSessionContext {
   lastAssistantUuid: string | undefined;
   lastThreadStartedId: ProviderThreadId | undefined;
   stopped: boolean;
+  streamFiber: Fiber.RuntimeFiber<void> | undefined;
 }
 
 interface ClaudeQueryRuntime extends AsyncIterable<SDKMessage> {
@@ -117,9 +129,7 @@ export interface ClaudeCodeAdapterLiveOptions {
 }
 
 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,
-  );
+  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);
 }
 
 function toMessage(cause: unknown, fallback: string): string {
@@ -695,6 +705,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
 
         context.stopped = true;
 
+        if (context.streamFiber) {
+          yield* Fiber.interrupt(context.streamFiber);
+          context.streamFiber = undefined;
+        }
+
         for (const [requestId, pending] of context.pendingApprovals) {
           yield* Deferred.succeed(pending.decision, "cancel");
           const stamp = yield* makeEventStamp();
@@ -967,6 +982,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
           lastAssistantUuid: resumeState?.resumeSessionAt,
           lastThreadStartedId: undefined,
           stopped: false,
+          streamFiber: undefined,
         };
         yield* Ref.set(contextRef, context);
         sessions.set(sessionId, context);
@@ -981,7 +997,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
           threadId,
         });
 
-        Effect.runFork(runSdkStream(context));
+        context.streamFiber = yield* Effect.forkDetach(runSdkStream(context));
 
         return {
           ...session,
