fix(server): deleting a thread closes its preview sessions

apps/server/src/orchestration-v2/EffectOutbox.ts

@@ -95,6 +95,9 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
   Schema.Struct({
     type: Schema.Literal("terminal.cleanup"),
   }),
+  Schema.Struct({
+    type: Schema.Literal("preview.cleanup"),
+  }),
   Schema.Struct({
     type: Schema.Literal("attachment.cleanup"),
     attachmentIds: Schema.Array(Schema.String),
@@ -120,6 +123,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [
   "provider-thread.rollback",
   "checkpoint.capture",
   "terminal.cleanup",
+  "preview.cleanup",
   "attachment.cleanup",
   "thread-title.generate",
   "delegated-tasks.stop",

apps/server/src/orchestration-v2/EffectWorker.test.ts

@@ -1,6 +1,9 @@
+import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
+import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, it } from "@effect/vitest";
 import {
   CommandId,
+  PreviewSessionLookupError,
   ProviderSessionId,
   ProviderThreadId,
   ProviderTurnId,
@@ -14,11 +17,18 @@ import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as Layer from "effect/Layer";
+import * as Logger from "effect/Logger";
 import * as Option from "effect/Option";
 import * as Queue from "effect/Queue";
 import * as Ref from "effect/Ref";
+import * as References from "effect/References";
 import * as TestClock from "effect/testing/TestClock";
 
+import * as ServerConfig from "../config.ts";
+import * as SqlitePersistence from "../persistence/Sqlite.ts";
+import * as PreviewManager from "../preview/Manager.ts";
+import * as TerminalManager from "../terminal/Manager.ts";
+import * as ResourceCleanupService from "./ResourceCleanupService.ts";
 import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
 import * as EffectOutbox from "./EffectOutbox.ts";
 import * as EffectWorker from "./EffectWorker.ts";
@@ -837,3 +847,98 @@ it.effect("settles a delegated child once its restart continuation fails for goo
     }).pipe(Effect.provide(layer));
   }),
 );
+
+function makePreviewWorkerLayer(previewLayer: Layer.Layer<PreviewManager.PreviewManager>) {
+  const cleanupLayer = ResourceCleanupService.layer.pipe(
+    Layer.provide(Layer.mock(TerminalManager.TerminalManager)({ close: () => Effect.void })),
+  );
+  const executor = Layer.unwrap(
+    Ref.make<ReadonlyArray<string>>([]).pipe(
+      Effect.map((events) => layerExecutorFor({ events }).pipe(Layer.provide(cleanupLayer))),
+    ),
+  );
+  return EffectWorker.layer.pipe(
+    Layer.provide(executor),
+    Layer.provideMerge(EffectOutbox.layer),
+    Layer.provide(SqlitePersistence.layerMemory),
+    Layer.provideMerge(previewLayer),
+    Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "preview-cleanup-test-" })),
+    Layer.provide(NodeServices.layer),
+  );
+}
+
+it.effect("drains preview cleanup for every tab of one thread and preserves another thread", () =>
+  Effect.gen(function* () {
+    const previews = yield* PreviewManager.PreviewManager;
+    const outbox = yield* EffectOutbox.EffectOutboxV2;
+    const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+    const threadA = ThreadId.make("thread:preview-cleanup:a");
+    const threadB = ThreadId.make("thread:preview-cleanup:b");
+    yield* previews.open({ threadId: threadA });
+    yield* previews.open({ threadId: threadA });
+    yield* previews.open({ threadId: threadB });
+    assert.lengthOf((yield* previews.list({ threadId: threadA })).sessions, 2);
+    const otherSessions = (yield* previews.list({ threadId: threadB })).sessions;
+    const commandId = CommandId.make("command:preview-cleanup");
+    yield* outbox.enqueue([
+      {
+        id: "effect:preview-cleanup",
+        commandId,
+        threadId: threadA,
+        request: { type: "preview.cleanup" },
+      },
+    ]);
+
+    assert.equal(yield* worker.drain(), 1);
+    assert.isEmpty((yield* previews.list({ threadId: threadA })).sessions);
+    assert.deepEqual((yield* previews.list({ threadId: threadB })).sessions, otherSessions);
+    const rows = yield* outbox.listByCommandId(commandId);
+    assert.lengthOf(rows, 1);
+    assert.equal(rows[0]?.status, "succeeded");
+    assert.equal(yield* worker.drain(), 0);
+  }).pipe(
+    Effect.provide(
+      makePreviewWorkerLayer(PreviewManager.layer.pipe(Layer.provide(NodeCrypto.layer))),
+    ),
+  ),
+);
+
+it.effect("warns and succeeds a preview cleanup outbox row when closing previews fails", () => {
+  const warnings: unknown[] = [];
+  const failure = new PreviewSessionLookupError({ threadId, tabId: "tab:failed-close" });
+  const logger = Logger.make(({ logLevel, message }) => {
+    if (logLevel === "Warn") warnings.push(message);
+  });
+  return Effect.gen(function* () {
+    const outbox = yield* EffectOutbox.EffectOutboxV2;
+    const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+    const commandId = CommandId.make("command:preview-cleanup-failure");
+    yield* outbox.enqueue([
+      {
+        id: "effect:preview-cleanup-failure",
+        commandId,
+        threadId,
+        request: { type: "preview.cleanup" },
+      },
+    ]);
+
+    assert.equal(yield* worker.drain(), 1);
+    const rows = yield* outbox.listByCommandId(commandId);
+    assert.lengthOf(rows, 1);
+    assert.equal(rows[0]?.status, "succeeded");
+    assert.equal(rows[0]?.attemptCount, 1);
+    assert.isNull(rows[0]?.lastError);
+    assert.deepEqual(warnings, [["Failed to close thread previews", { threadId, error: failure }]]);
+    assert.equal(yield* worker.drain(), 0);
+  }).pipe(
+    Effect.provide(
+      Layer.merge(
+        makePreviewWorkerLayer(
+          Layer.mock(PreviewManager.PreviewManager)({ close: () => Effect.fail(failure) }),
+        ),
+        Logger.layer([logger], { mergeWithExisting: false }),
+      ),
+    ),
+    Effect.provideService(References.MinimumLogLevel, "Warn"),
+  );
+});

apps/server/src/orchestration-v2/EffectWorker.ts

@@ -439,6 +439,8 @@ export const layerExecutor: Layer.Layer<
                   }),
               ),
             );
+          case "preview.cleanup":
+            return resourceCleanup.cleanupPreviews(effect.threadId);
           case "attachment.cleanup":
             return resourceCleanup.cleanupAttachments(effect.request.attachmentIds).pipe(
               Effect.mapError(

apps/server/src/orchestration-v2/ResourceCleanupService.ts

@@ -1,3 +1,4 @@
+import type { ThreadId } from "@t3tools/contracts";
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
@@ -6,6 +7,7 @@ import * as Schema from "effect/Schema";
 
 import { resolveAttachmentPathById } from "../attachmentStore.ts";
 import * as ServerConfig from "../config.ts";
+import * as PreviewManager from "../preview/Manager.ts";
 import * as TerminalManager from "../terminal/Manager.ts";
 
 export class ResourceCleanupError extends Schema.TaggedError<ResourceCleanupError>()(
@@ -20,12 +22,15 @@ export class ResourceCleanupError extends Schema.TaggedError<ResourceCleanupErro
 
 export class ResourceCleanupService extends Context.Reference<{
   readonly cleanupTerminals: (threadId: string) => Effect.Effect<void, ResourceCleanupError>;
+  /** Closes every preview session of the thread. Failures are logged, never retried. */
+  readonly cleanupPreviews: (threadId: ThreadId) => Effect.Effect<void>;
   readonly cleanupAttachments: (
     attachmentIds: ReadonlyArray<string>,
   ) => Effect.Effect<void, ResourceCleanupError>;
 }>("t3/orchestration-v2/ResourceCleanupService", {
   defaultValue: () => ({
     cleanupTerminals: () => Effect.void,
+    cleanupPreviews: () => Effect.void,
     cleanupAttachments: () => Effect.void,
   }),
 }) {}
@@ -34,6 +39,7 @@ export const layer = Layer.effect(
   ResourceCleanupService,
   Effect.gen(function* () {
     const terminals = yield* TerminalManager.TerminalManager;
+    const previews = yield* PreviewManager.PreviewManager;
     const fileSystem = yield* FileSystem.FileSystem;
     const config = yield* ServerConfig.ServerConfig;
     return {
@@ -45,6 +51,15 @@ export const layer = Layer.effect(
               (cause) => new ResourceCleanupError({ operation: "terminal", threadId, cause }),
             ),
           ),
+      // Preview sessions live in memory, so a retry cannot do better than this attempt.
+      cleanupPreviews: (threadId: ThreadId) =>
+        previews
+          .close({ threadId })
+          .pipe(
+            Effect.catch((error) =>
+              Effect.logWarning("Failed to close thread previews", { threadId, error }),
+            ),
+          ),
       cleanupAttachments: (attachmentIds: ReadonlyArray<string>) =>
         Effect.forEach(
           attachmentIds,

apps/server/src/orchestration-v2/ThreadDeletion.test.ts

@@ -291,8 +291,28 @@ it.effect("queues provider and resource cleanup and preserves an earlier deletio
           revokeMcpCredential: true,
         },
         { type: "terminal.cleanup" },
+        { type: "preview.cleanup" },
         { type: "attachment.cleanup", attachmentIds: ["shared_file"] },
       ],
     );
   }).pipe(Effect.provide(IdAllocator.layer)),
 );
+
+it.effect("queues preview cleanup without provider sessions or attachments", () =>
+  Effect.gen(function* () {
+    const projection = makeProjection();
+    assert.isEmpty(projection.providerSessions);
+    assert.isEmpty(projection.messages);
+    const plan = yield* planThreadDeletion({
+      command,
+      projection,
+      attachmentIds: [],
+      now: deletedAt,
+      idAllocator: yield* IdAllocator.IdAllocatorV2,
+    });
+    assert.deepEqual(
+      plan.effects.map((effect) => effect.request),
+      [{ type: "terminal.cleanup" }, { type: "preview.cleanup" }],
+    );
+  }).pipe(Effect.provide(IdAllocator.layer)),
+);

apps/server/src/orchestration-v2/ThreadDeletion.ts

@@ -215,6 +215,12 @@ export const planThreadDeletion = Effect.fn("ThreadDeletion.planThreadDeletion")
     threadId: command.threadId,
     request: { type: "terminal.cleanup" },
   });
+  effects.push({
+    id: `effect:${command.commandId}:preview.cleanup`,
+    commandId: command.commandId,
+    threadId: command.threadId,
+    request: { type: "preview.cleanup" },
+  });
   const attachmentIds = Array.from(new Set(input.attachmentIds));
   if (attachmentIds.length > 0) {
     effects.push({

apps/server/src/project/ProjectService.deletion.test.ts

@@ -1,3 +1,4 @@
+import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, it } from "@effect/vitest";
 import {
@@ -17,6 +18,20 @@ import { TestClock } from "effect/testing";
 import * as SqlClient from "effect/sql/SqlClient";
 
 import * as ServerConfig from "../config.ts";
+import * as PreviewManager from "../preview/Manager.ts";
+import * as TerminalManager from "../terminal/Manager.ts";
+import * as ServerSettings from "../serverSettings.ts";
+import * as CheckpointRollbackService from "../orchestration-v2/CheckpointRollbackService.ts";
+import * as EffectOutbox from "../orchestration-v2/EffectOutbox.ts";
+import * as EffectWorker from "../orchestration-v2/EffectWorker.ts";
+import * as ProviderSessionManager from "../orchestration-v2/ProviderSessionManager.ts";
+import * as ProviderTurnControlService from "../orchestration-v2/ProviderTurnControlService.ts";
+import * as ProviderTurnStartService from "../orchestration-v2/ProviderTurnStartService.ts";
+import * as ResourceCleanupService from "../orchestration-v2/ResourceCleanupService.ts";
+import * as RunFinalizationService from "../orchestration-v2/RunFinalizationService.ts";
+import * as RuntimeRequestService from "../orchestration-v2/RuntimeRequestService.ts";
+import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
+import * as ThreadTitleRegenerationService from "../orchestration-v2/ThreadTitleRegenerationService.ts";
 import { OrchestrationEffectRequestV2 } from "../orchestration-v2/EffectOutbox.ts";
 import * as EventSink from "../orchestration-v2/EventSink.ts";
 import * as EventStore from "../orchestration-v2/EventStore.ts";
@@ -197,9 +212,9 @@ it.effect("retries a partial project deletion without repeating child events or
       assert.lengthOf(partialEvents, 1);
       assert.equal(partialEvents[0]?.stream_id, firstThreadId);
       assert.equal(partialEvents[0]?.event_type, "thread.deleted");
-      assert.lengthOf(partialCleanup, 1);
+      assert.lengthOf(partialCleanup, 2);
       assert.equal(partialCleanup[0]?.thread_id, firstThreadId);
-      assert.equal(partialCleanup[0]?.effect_type, "terminal.cleanup");
+      assert.equal(partialCleanup[0]?.effect_type, "preview.cleanup");
 
       const deletedProject = yield* service.delete(input);
       assert.isNotNull(deletedProject.deletedAt);
@@ -220,7 +235,7 @@ it.effect("retries a partial project deletion without repeating child events or
       assert.deepEqual(finalEvents[0], partialEvents[0]);
       assert.equal(finalEvents[2]?.command_id, commandId);
       const finalCleanup = yield* readCleanup;
-      assert.lengthOf(finalCleanup, 2);
+      assert.lengthOf(finalCleanup, 4);
       assert.deepEqual(
         finalCleanup.filter((effect) => effect.thread_id === firstThreadId),
         partialCleanup,
@@ -230,6 +245,12 @@ it.effect("retries a partial project deletion without repeating child events or
         assert.deepEqual(
           finalCleanup.filter((effect) => effect.thread_id === threadId),
           [
+            {
+              effect_id: `effect:${expectedCommandId}:preview.cleanup`,
+              thread_id: threadId,
+              command_id: expectedCommandId,
+              effect_type: "preview.cleanup",
+            },
             {
               effect_id: `effect:${expectedCommandId}:terminal.cleanup`,
               thread_id: threadId,
@@ -484,3 +505,74 @@ it.effect("deletes a project without force once its imported threads were delete
     }).pipe(Effect.provide(layerServices));
   }).pipe(Effect.provide(layerDatabase)),
 );
+
+it.effect(
+  "forced project deletion closes previews for every child thread after draining cleanup",
+  () =>
+    Effect.gen(function* () {
+      const projectId = ProjectId.make("project:preview-deletion");
+      const threadIds = [
+        ThreadId.make("thread:preview-project:a"),
+        ThreadId.make("thread:preview-project:b"),
+      ];
+      yield* seedProject(projectId);
+      const eventSink = yield* EventSink.EventSinkV2;
+      const previews = yield* PreviewManager.PreviewManager;
+      const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+      const outbox = yield* EffectOutbox.EffectOutboxV2;
+      yield* eventSink.write({
+        events: threadIds.map((threadId) => nativeThreadCreated(projectId, threadId)),
+      });
+      for (const threadId of threadIds) {
+        yield* previews.open({ threadId });
+        assert.lengthOf((yield* previews.list({ threadId })).sessions, 1);
+      }
+      const commandId = CommandId.make("command:preview-project-delete");
+      const service = yield* ProjectService.make;
+      yield* service.delete({ commandId, projectId, force: true });
+
+      assert.equal(yield* worker.drain(), 4);
+      for (const threadId of threadIds) {
+        assert.isEmpty((yield* previews.list({ threadId })).sessions);
+        const rows = yield* outbox.listByCommandId(
+          CommandId.make(`${commandId}:delete-thread:${threadId}`),
+        );
+        assert.lengthOf(rows, 2);
+        assert.isTrue(rows.every((row) => row.status === "succeeded"));
+      }
+    }).pipe(
+      Effect.provide(
+        Layer.merge(
+          layerServices,
+          EffectWorker.layer.pipe(
+            Layer.provide(
+              EffectWorker.layerExecutor.pipe(
+                Layer.provide(
+                  Layer.mergeAll(
+                    Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}),
+                    Layer.mock(ProviderTurnControlService.ProviderTurnControlServiceV2)({}),
+                    Layer.mock(ProviderTurnStartService.ProviderTurnStartServiceV2)({}),
+                    Layer.mock(RunFinalizationService.RunFinalizationService)({}),
+                    Layer.mock(CheckpointRollbackService.CheckpointRollbackServiceV2)({}),
+                    Layer.mock(RuntimeRequestService.RuntimeRequestServiceV2)({}),
+                    Layer.mock(ThreadTitleRegenerationService.ThreadTitleRegenerationService)({}),
+                    Layer.mock(ThreadManagementService.ThreadManagementService)({}),
+                    ServerSettings.layerTest(),
+                    ResourceCleanupService.layer.pipe(
+                      Layer.provide(
+                        Layer.mock(TerminalManager.TerminalManager)({ close: () => Effect.void }),
+                      ),
+                    ),
+                  ),
+                ),
+              ),
+            ),
+            Layer.provideMerge(EffectOutbox.layer),
+          ),
+        ).pipe(
+          Layer.provideMerge(PreviewManager.layer.pipe(Layer.provide(NodeCrypto.layer))),
+          Layer.provideMerge(layerDatabase),
+        ),
+      ),
+    ),
+);