fix(orchestration): preserve interrupts in thread deletion cleanup

Author: maria-rcks

apps/server/src/orchestration/Layers/ThreadDeletionReactor.test.ts

@@ -0,0 +1,36 @@
+import { ThreadId } from "@t3tools/contracts";
+import { Cause, Effect, Exit } from "effect";
+import { describe, expect, it } from "vitest";
+
+import { logCleanupCauseUnlessInterrupted } from "./ThreadDeletionReactor.ts";
+
+describe("logCleanupCauseUnlessInterrupted", () => {
+  const threadId = ThreadId.makeUnsafe("thread-deletion-reactor-test");
+
+  it("swallows ordinary cleanup failures", async () => {
+    const exit = await Effect.runPromiseExit(
+      logCleanupCauseUnlessInterrupted({
+        effect: Effect.fail("cleanup failed"),
+        message: "thread deletion cleanup skipped provider session stop",
+        threadId,
+      }),
+    );
+
+    expect(Exit.isSuccess(exit)).toBe(true);
+  });
+
+  it("preserves interrupt causes", async () => {
+    const exit = await Effect.runPromiseExit(
+      logCleanupCauseUnlessInterrupted({
+        effect: Effect.interrupt,
+        message: "thread deletion cleanup skipped provider session stop",
+        threadId,
+      }),
+    );
+
+    expect(Exit.isFailure(exit)).toBe(true);
+    if (Exit.isFailure(exit)) {
+      expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true);
+    }
+  });
+});

apps/server/src/orchestration/Layers/ThreadDeletionReactor.ts

@@ -12,30 +12,45 @@ import {
 
 type ThreadDeletedEvent = Extract<OrchestrationEvent, { type: "thread.deleted" }>;
 
+export const logCleanupCauseUnlessInterrupted = <R, E>({
+  effect,
+  message,
+  threadId,
+}: {
+  readonly effect: Effect.Effect<void, E, R>;
+  readonly message: string;
+  readonly threadId: ThreadDeletedEvent["payload"]["threadId"];
+}): Effect.Effect<void, E, R> =>
+  effect.pipe(
+    Effect.catchCause((cause) => {
+      if (Cause.hasInterruptsOnly(cause)) {
+        return Effect.failCause(cause);
+      }
+      return Effect.logDebug(message, {
+        threadId,
+        cause: Cause.pretty(cause),
+      });
+    }),
+  );
+
 const make = Effect.gen(function* () {
   const orchestrationEngine = yield* OrchestrationEngineService;
   const providerService = yield* ProviderService;
   const terminalManager = yield* TerminalManager;
 
   const stopProviderSession = (threadId: ThreadDeletedEvent["payload"]["threadId"]) =>
-    providerService.stopSession({ threadId }).pipe(
-      Effect.catchCause((cause) =>
-        Effect.logDebug("thread deletion cleanup skipped provider session stop", {
-          threadId,
-          cause: Cause.pretty(cause),
-        }),
-      ),
-    );
+    logCleanupCauseUnlessInterrupted({
+      effect: providerService.stopSession({ threadId }),
+      message: "thread deletion cleanup skipped provider session stop",
+      threadId,
+    });
 
   const closeThreadTerminals = (threadId: ThreadDeletedEvent["payload"]["threadId"]) =>
-    terminalManager.close({ threadId, deleteHistory: true }).pipe(
-      Effect.catchCause((cause) =>
-        Effect.logDebug("thread deletion cleanup skipped terminal close", {
-          threadId,
-          cause: Cause.pretty(cause),
-        }),
-      ),
-    );
+    logCleanupCauseUnlessInterrupted({
+      effect: terminalManager.close({ threadId, deleteHistory: true }),
+      message: "thread deletion cleanup skipped terminal close",
+      threadId,
+    });
 
   const processThreadDeleted = Effect.fn("processThreadDeleted")(function* (
     event: ThreadDeletedEvent,