fix(server): preserve attachments referenced by surviving threads

apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts

@@ -1,5 +1,6 @@
 import { assert, it } from "@effect/vitest";
 import {
+  ChatAttachmentId,
   CommandId,
   MessageId,
   EventId,
@@ -1231,3 +1232,66 @@ it.effect("explicit settle clears snooze atomically and stays idempotent after r
     assert.deepInclude((yield* projections.getThreadProjection(threadId)).thread, active);
   }).pipe(Effect.provide(layerTest)),
 );
+
+it.effect("cleans the last attachment reference when threads are deleted concurrently", () =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    const outbox = yield* EffectOutbox.EffectOutboxV2;
+    const ids = [ThreadId.make("concurrent-delete-a"), ThreadId.make("concurrent-delete-b")];
+    const attachmentId = ChatAttachmentId.make("legacy_shared");
+    for (const threadId of ids) {
+      yield* orchestrator.dispatch({
+        type: "thread.create",
+        commandId: CommandId.make(`create:${threadId}`),
+        threadId,
+        projectId: ProjectId.make("shared-attachment"),
+        title: "Shared",
+        modelSelection,
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        branch: null,
+        worktreePath: null,
+        createdBy: "user",
+        creationSource: "web",
+      });
+      yield* orchestrator.dispatch({
+        type: "message.dispatch",
+        commandId: CommandId.make(`message:${threadId}`),
+        threadId,
+        messageId: MessageId.make(`message:${threadId}`),
+        text: "Shared file",
+        attachments: [
+          {
+            type: "file",
+            id: attachmentId,
+            name: "file.pdf",
+            mimeType: "application/pdf",
+            sizeBytes: 10,
+          },
+        ],
+        dispatchMode: { type: "defer_start" },
+        createdBy: "user",
+        creationSource: "web",
+      });
+    }
+    yield* Effect.forEach(
+      ids,
+      (threadId) =>
+        orchestrator.dispatch({
+          type: "thread.delete",
+          commandId: CommandId.make(`delete:${threadId}`),
+          threadId,
+        }),
+      { concurrency: "unbounded" },
+    );
+    const effects = (yield* Effect.forEach(ids, (threadId) =>
+      outbox.listByCommandId(CommandId.make(`delete:${threadId}`)),
+    )).flat();
+    assert.deepEqual(
+      effects.flatMap(({ request }) =>
+        request.type === "attachment.cleanup" ? request.attachmentIds : [],
+      ),
+      [attachmentId],
+    );
+  }).pipe(Effect.provide(layerTest)),
+);

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

@@ -71,6 +71,7 @@ import {
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
+import * as Semaphore from "effect/Semaphore";
 import * as Exit from "effect/Exit";
 import * as FileSystem from "effect/FileSystem";
 import * as Path from "effect/Path";
@@ -897,6 +898,7 @@ function lastDeliveredRunForProviderThread(
 }
 
 const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(function* () {
+  const attachmentDeletionLock = yield* Semaphore.make(1);
   const checkpointService = yield* CheckpointServiceV2;
   const commandPolicy = yield* CommandPolicyV2;
   const contextHandoffService = yield* ContextHandoffServiceV2;
@@ -11782,7 +11784,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   const dispatchWithReceipt = Effect.fnUntraced(function* (command: OrchestrationV2ServerCommand) {
     const dispatch = threadDispatch.withLock(
       commandThreadId(command),
-      dispatchWithReceiptEffect(command),
+      // Shared attachment references cross thread boundaries. Hold this through planning and
+      // commit so simultaneous deletions cannot both leave cleanup to the other thread.
+      command.type === "thread.delete"
+        ? attachmentDeletionLock.withPermits(1)(dispatchWithReceiptEffect(command))
+        : dispatchWithReceiptEffect(command),
     );
     if (
       command.type !== "message.dispatch" ||

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

@@ -1,5 +1,6 @@
 import { assert, it } from "@effect/vitest";
 import {
+  ChatAttachmentId,
   EnvironmentId,
   EventId,
   CommandId,
@@ -8,6 +9,7 @@ import {
   CheckpointScopeId,
   MessageId,
   type ModelSelection,
+  type OrchestrationV2AppThread,
   type OrchestrationV2ProviderThread,
   type OrchestrationV2TurnItem,
   type OrchestrationV2ThreadShellSnapshot,
@@ -5644,3 +5646,106 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
     }),
   );
 });
+
+for (const storage of ["sqlite", "memory"] as const) {
+  it.effect.each(["source", "relaunch"] as const)(
+    `preserves shared attachments when deleting %s first in ${storage}`,
+    (deletedFirst) =>
+      Effect.gen(function* () {
+        const store = yield* ProjectionStore.ProjectionStoreV2;
+        const now = yield* DateTime.now;
+        const instanceId = ProviderInstanceId.make("codex");
+        const shared = [
+          "source-00000000-0000-4000-8000-000000000001",
+          "source-00000000-0000-4000-8000-000000000002-pdf",
+          "legacy_screenshot",
+        ];
+        const threads = new Map<string, OrchestrationV2AppThread>();
+        for (const name of ["source", "relaunch"] as const) {
+          const threadId = ThreadId.make(name);
+          const thread: OrchestrationV2AppThread = {
+            id: threadId,
+            projectId: ProjectId.make("attachment-cleanup"),
+            title: name,
+            providerInstanceId: instanceId,
+            modelSelection: { instanceId, model: "test" },
+            runtimeMode: "full-access",
+            interactionMode: "default",
+            branch: null,
+            worktreePath: null,
+            activeProviderThreadId: null,
+            lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
+            forkedFrom: null,
+            createdBy: "user",
+            creationSource: "mobile",
+            createdAt: now,
+            updatedAt: now,
+            archivedAt: name === "source" ? now : null,
+            settledOverride: null,
+            settledAt: null,
+            lastVisitedAt: null,
+            deletedAt: null,
+          };
+          threads.set(name, thread);
+          yield* store.apply({
+            id: EventId.make(`thread:${name}`),
+            type: "thread.created",
+            threadId,
+            occurredAt: now,
+            payload: thread,
+          });
+          yield* store.apply({
+            id: EventId.make(`message:${name}`),
+            type: "message.updated",
+            threadId,
+            occurredAt: now,
+            payload: {
+              id: MessageId.make(`message:${name}`),
+              threadId,
+              runId: null,
+              nodeId: null,
+              role: "user",
+              text: "Shared uploads",
+              attachments: [...shared, `legacy_${name}_only`].map((id) => ({
+                type: "file" as const,
+                id: ChatAttachmentId.make(id),
+                name: "attachment.pdf",
+                mimeType: "application/pdf",
+                sizeBytes: 10,
+              })),
+              streaming: false,
+              createdBy: "user",
+              creationSource: "mobile",
+              createdAt: now,
+              updatedAt: now,
+            },
+          });
+        }
+
+        for (const name of threads.keys()) {
+          assert.deepEqual(yield* store.getThreadAttachmentIds(ThreadId.make(name)), [
+            `legacy_${name}_only`,
+          ]);
+        }
+        const deleted = threads.get(deletedFirst)!;
+        yield* store.apply({
+          id: EventId.make(`delete:${deletedFirst}`),
+          type: "thread.deleted",
+          threadId: deleted.id,
+          occurredAt: now,
+          payload: { ...deleted, deletedAt: now },
+        });
+        const survivor = deletedFirst === "source" ? "relaunch" : "source";
+        assert.deepEqual(
+          (yield* store.getThreadAttachmentIds(ThreadId.make(survivor))).toSorted(),
+          [...shared, `legacy_${survivor}_only`].toSorted(),
+        );
+      }).pipe(
+        Effect.provide(
+          storage === "sqlite"
+            ? ProjectionStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory))
+            : ProjectionStore.layerMemory,
+        ),
+      ),
+  );
+}

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

@@ -5212,6 +5212,16 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       FROM orchestration_v2_projection_messages AS message,
         json_each(message.payload_json, '$.attachments') AS attachment
       WHERE message.thread_id = ${threadId}
+        AND NOT EXISTS (
+          SELECT 1
+          FROM orchestration_v2_projection_threads AS remaining_thread
+          JOIN orchestration_v2_projection_messages AS remaining_message
+            ON remaining_message.thread_id = remaining_thread.thread_id,
+            json_each(remaining_message.payload_json, '$.attachments') AS remaining_attachment
+          WHERE remaining_thread.thread_id != ${threadId}
+            AND remaining_thread.deleted_at IS NULL
+            AND json_extract(remaining_attachment.value, '$.id') = json_extract(attachment.value, '$.id')
+        )
     `,
         sql<{ payload_json: string }>`
       SELECT payload_json
@@ -6810,12 +6820,26 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
               state.projections.get(threadId)?.turnItems.find((item) => item.id === itemId) ?? null,
           ),
         ),
-      getThreadAttachmentIds: (threadId) =>
-        service.getThreadProjection(threadId).pipe(
-          Effect.map((projection) => [
+      getThreadAttachmentIds: Effect.fn("ProjectionStore.getThreadAttachmentIds")(
+        function* (threadId) {
+          const projection = yield* service.getThreadProjection(threadId);
+          const state = yield* Ref.get(replayState);
+          // Relaunched threads reuse uploads, including legacy ids without an ownership prefix.
+          const referencedIds = new Set(
+            Array.from(state.projections.values()).flatMap((other) =>
+              other.thread.id !== threadId && other.thread.deletedAt === null
+                ? other.messages.flatMap((message) =>
+                    message.attachments.map((attachment) => attachment.id),
+                  )
+                : [],
+            ),
+          );
+          return [
             ...new Set([
               ...projection.messages.flatMap((message) =>
-                message.attachments.map((attachment) => attachment.id),
+                message.attachments
+                  .map((attachment) => attachment.id)
+                  .filter((id) => !referencedIds.has(id)),
               ),
               ...threadHtmlRenderAttachmentIds(
                 threadId,
@@ -6824,8 +6848,9 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
                 ),
               ),
             ]),
-          ]),
-        ),
+          ];
+        },
+      ),
       getThreadRecords: Effect.fn("ProjectionStore.getThreadRecords")(
         function* (threadId, fields, filter) {
           const projection = (yield* Ref.get(replayState)).projections.get(threadId);