juliusmarminge · Sep 7, 2026 · Sep 7, 2026 · Sep 7, 2026 · Sep 7, 2026
diff --git a/apps/server/src/orchestration-v2/AttachmentClaims.ts b/apps/server/src/orchestration-v2/AttachmentClaims.ts
@@ -28,8 +28,7 @@ export function attachmentIsPendingUpload(attachment: ChatAttachment): boolean {
   return parseThreadSegmentFromAttachmentId(attachment.id) === PENDING_ATTACHMENT_THREAD_SEGMENT;
 }
 
-/** Best-effort removal of claimed copies after a failed dispatch: the pending
- *  upload remains the retry source, so only the thread-scoped copies go. */
+/** Remove partial claims only before dispatch, or after proving they were not accepted. */
 export const releaseClaimedAttachments = Effect.fn("AttachmentClaims.releaseClaimedAttachments")(
   function* (claimedPaths: ReadonlyArray<string>) {
     if (claimedPaths.length === 0) return;

diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts
@@ -1,5 +1,12 @@
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import * as FileSystem from "effect/FileSystem";
+import * as ServerConfig from "../config.ts";
+import { createPendingAttachmentId, resolveAttachmentPath } from "../attachmentStore.ts";
+import * as ThreadMessageIntake from "./ThreadMessageIntake.ts";
 import { assert, it, vi } from "@effect/vitest";
 import {
+  ChatAttachmentId,
+  type ChatAttachment,
   CommandId,
   DEFAULT_SERVER_SETTINGS,
   GitCommandError,
@@ -1286,3 +1293,139 @@ it.effect("does not depend on the legacy launch workflow table", () => {
     assert.equal(launched.projection.messages[0]?.text, "No private workflow state");
   }).pipe(Effect.provide(harness.layer));
 });
+
+it.effect("shared intake preserves durable attachment bytes after a lost launch result", () => {
+  const harness = makeHarness();
+  const files = ServerConfig.layerTest(process.cwd(), { prefix: "t3-message-intake-" }).pipe(
+    Layer.provideMerge(NodeServices.layer),
+  );
+  return Effect.gen(function* () {
+    const config = yield* ServerConfig.ServerConfig;
+    const fs = yield* FileSystem.FileSystem;
+    const launches = yield* ThreadLaunch.ThreadLaunchService;
+    const threads = yield* ThreadManagement.ThreadManagementService;
+    const pendingId = createPendingAttachmentId();
+    assert.isNotNull(pendingId);
+    const attachment: ChatAttachment = {
+      type: "image",
+      id: ChatAttachmentId.make(pendingId),
+      name: "image.png",
+      mimeType: "image/png",
+      sizeBytes: 4,
+    };
+    const pendingPath = resolveAttachmentPath({
+      attachmentsDir: config.attachmentsDir,
+      attachment,
+    });
+    assert.isNotNull(pendingPath);
+    yield* fs.makeDirectory(config.attachmentsDir, { recursive: true });
+    yield* fs.writeFile(pendingPath, new Uint8Array([1, 2, 3, 4]));
+    const input = {
+      ...launchInput({ command: "intake-launch", thread: "intake-thread" }),
+      initialMessage: {
+        messageId: MessageId.make("intake-first"),
+        text: "First",
+        attachments: [attachment],
+      },
+    };
+    const failed = yield* ThreadMessageIntake.launchThread(input).pipe(
+      Effect.provideService(ThreadLaunch.ThreadLaunchService, {
+        launch: (request) =>
+          launches.launch(request).pipe(
+            Effect.andThen(
+              new ThreadLaunch.ThreadLaunchError({
+                operation: "create-thread",
+                commandId: request.commandId,
+                projectId,
+                cause: "lost result after acceptance",
+              }),
+            ),
+          ),
+      }),
+      Effect.flip,
+    );
+    assert.equal(failed._tag, "ThreadLaunchError");
+    // The observer failed, but the real V2 message and its bytes were accepted.
+    const accepted = yield* threads.getThreadProjection(input.threadId);
+    const stored = accepted.messages.find(
+      (message) => message.id === input.initialMessage.messageId,
+    );
+    assert.isDefined(stored);
+    assert.notEqual(stored.attachments[0]?.id, attachment.id);
+    const storedPath = resolveAttachmentPath({
+      attachmentsDir: config.attachmentsDir,
+      attachment: stored.attachments[0]!,
+    });
+    assert.isNotNull(storedPath);
+    assert.deepEqual(yield* fs.readFile(storedPath), new Uint8Array([1, 2, 3, 4]));
+    assert.deepEqual(yield* fs.readFile(pendingPath), new Uint8Array([1, 2, 3, 4]));
+
+    const replayed = yield* ThreadMessageIntake.launchThread(input);
+    assert.equal(replayed.projection.messages[0]?.id, stored.id);
+    assert.deepEqual(replayed.projection.messages[0]?.attachments, stored.attachments);
+    const claimedFiles = Effect.map(fs.readDirectory(config.attachmentsDir), (files) =>
+      files.filter((name) => !name.startsWith("pending-")),
+    );
+    assert.equal((yield* claimedFiles).length, 1);
+
+    const missingProject = yield* ThreadMessageIntake.launchThread({
+      ...input,
+      commandId: CommandId.make("intake-no-project"),
+      projectId: ProjectId.make("missing-project"),
+    }).pipe(Effect.flip);
+    assert.equal(missingProject._tag, "ThreadLaunchError");
+    assert.equal((yield* claimedFiles).length, 1);
+    const missingThread = yield* ThreadMessageIntake.dispatchCommand({
+      type: "message.dispatch",
+      commandId: CommandId.make("intake-no-thread"),
+      threadId: ThreadId.make("missing-thread"),
+      messageId: MessageId.make("intake-missing"),
+      text: "Missing",
+      attachments: [attachment],
+      dispatchMode: { type: "start_immediately" },
+      createdBy: "user",
+      creationSource: "web",
+    }).pipe(Effect.flip);
+    assert.equal(missingThread._tag, "OrchestratorProjectionError");
+    assert.equal((yield* claimedFiles).length, 1);
+
+    // Both ordinary command intake (RPC) and send intake (MCP) use the same store.
+    const dispatch = ThreadMessageIntake.dispatchCommand({
+      type: "message.dispatch",
+      commandId: CommandId.make("intake-dispatch"),
+      threadId: input.threadId,
+      messageId: MessageId.make("intake-second"),
+      text: "Second",
+      attachments: [attachment],
+      dispatchMode: { type: "queue_after_active" },
+      createdBy: "user",
+      creationSource: "web",
+    });
+    yield* dispatch;
+    yield* dispatch;
+    const send = ThreadMessageIntake.sendToThread({
+      commandId: CommandId.make("intake-send"),
+      projectId,
+      threadId: input.threadId,
+      messageId: MessageId.make("intake-third"),
+      text: "Third",
+      attachments: [attachment],
+      mode: "queue",
+      createdBy: "agent",
+      creationSource: "mcp",
+    });
+    yield* send;
+    yield* send;
+    assert.equal((yield* claimedFiles).length, 3);
+    const final = yield* threads.getThreadProjection(input.threadId);
+    assert.equal(final.messages.length, 3);
+    for (const message of final.messages) {
+      const path = resolveAttachmentPath({
+        attachmentsDir: config.attachmentsDir,
+        attachment: message.attachments[0]!,
+      });
+      assert.isNotNull(path);
+      assert.deepEqual(yield* fs.readFile(path), new Uint8Array([1, 2, 3, 4]));
+    }
+  }).pipe(Effect.provide(Layer.mergeAll(harness.layer, files)));
+});
diff --git a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts
@@ -0,0 +1,134 @@
+import type { ChatAttachment, OrchestrationV2Command } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Schema from "effect/Schema";
+import { resolveAttachmentPath } from "../attachmentStore.ts";
+import * as ServerConfig from "../config.ts";
+import * as Orchestrator from "./Orchestrator.ts";
+
+import * as AttachmentClaims from "./AttachmentClaims.ts";
+import * as ThreadLaunch from "./ThreadLaunchService.ts";
+import * as ThreadManagement from "./ThreadManagementService.ts";
+
+// These dispatcher failures occur in receipt validation or planning, before
+// commitCommand. Generic dispatch errors can follow a commit and remain uncertain.
+function dispatchWasNotAccepted(
+  error: Orchestrator.OrchestratorV2Error | ThreadManagement.ThreadManagementError,
+) {
+  switch (error._tag) {
+    case "OrchestratorProjectionError":
+    case "OrchestratorProviderAdapterError":
+    case "OrchestratorCommandPreviouslyRejectedError":
+    case "OrchestratorCommandIdConflictError":
+      return true;
+    default:
+      return false;
+  }
+}
+const isOrchestratorError = Schema.is(Orchestrator.OrchestratorV2Error);
+
+const releaseUnusedClaims = Effect.fn("ThreadMessageIntake.releaseUnusedClaims")(function* (
+  claimed: AttachmentClaims.ClaimedAttachments,
+  accepted: ReadonlyArray<ChatAttachment>,
+) {
+  if (claimed.claimedPaths.length === 0) return;
+  const config = yield* ServerConfig.ServerConfig;
+  const retained = new Set(
+    accepted.map((attachment) =>
+      resolveAttachmentPath({
+        attachmentsDir: config.attachmentsDir,
+        attachment,
+      }),
+    ),
+  );
+  yield* AttachmentClaims.releaseClaimedAttachments(
+    claimed.claimedPaths.filter((path) => !retained.has(path)),
+  );
+});
+
+export const dispatchCommand = Effect.fn("ThreadMessageIntake.dispatchCommand")(function* (
+  command: OrchestrationV2Command,
+) {
+  const threads = yield* ThreadManagement.ThreadManagementService;
+  if (command.type !== "message.dispatch") return yield* threads.dispatch(command);
+  const claimed = yield* AttachmentClaims.claimPendingAttachments({
+    threadId: command.threadId,
+    attachments: command.attachments,
+  });
+  return yield* threads.dispatch({ ...command, attachments: claimed.attachments }).pipe(
+    Effect.tap((result) =>
+      releaseUnusedClaims(
+        claimed,
+        result.storedEvents.flatMap(({ event }) =>
+          event.type === "message.updated" ? event.payload.attachments : [],
+        ),
+      ),
+    ),
+    Effect.tapError((error) =>
+      dispatchWasNotAccepted(error)
+        ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
+        : Effect.void,
+    ),
+  );
+});
+
+export const sendToThread = Effect.fn("ThreadMessageIntake.sendToThread")(function* (
+  input: ThreadManagement.ThreadManagementSendInput,
+) {
+  const threads = yield* ThreadManagement.ThreadManagementService;
+  const claimed = yield* AttachmentClaims.claimPendingAttachments(input);
+  return yield* threads.sendToThread({ ...input, attachments: claimed.attachments }).pipe(
+    Effect.tap((result) => releaseUnusedClaims(claimed, result.message.attachments)),
+    Effect.tapError((error) =>
+      dispatchWasNotAccepted(error)
+        ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
+        : Effect.void,
+    ),
+  );
+});
+
+export const launchThread = Effect.fn("ThreadMessageIntake.launchThread")(function* (
+  input: ThreadLaunch.ThreadLaunchInput,
+) {
+  const launches = yield* ThreadLaunch.ThreadLaunchService;
+  if (!input.initialMessage?.attachments.some(AttachmentClaims.attachmentIsPendingUpload)) {
+    return yield* launches.launch(input);
+  }
+  if (input.threadId === undefined) {
+    return yield* new AttachmentClaims.AttachmentClaimError({
+      message: "Uploaded attachments need a thread id at launch.",
+    });
+  }
+  const claimed = yield* AttachmentClaims.claimPendingAttachments({
+    threadId: input.threadId,
+    attachments: input.initialMessage.attachments,
+  });
+  return yield* launches
+    .launch({
+      ...input,
+      initialMessage: { ...input.initialMessage, attachments: claimed.attachments },
+    })
+    .pipe(
+      Effect.tap((result) =>
+        releaseUnusedClaims(
+          claimed,
+          result.projection.messages.flatMap((message) => message.attachments),
+        ),
+      ),
+      Effect.tapError((error) => {
+        // Project/receipt reads precede message dispatch. The create-thread error
+        // also wraps post-message projection reads, so its tag alone is not proof.
+        const notAccepted =
+          error.operation === "resolve-project" ||
+          error.operation === "read-receipt" ||
+          ((error.operation === "create-thread" || error.operation === "dispatch-message") &&
+            isOrchestratorError(error.cause) &&
+            // Projection errors under create-thread can occur after the message commit.
+            (error.operation !== "create-thread" ||
+              error.cause._tag !== "OrchestratorProjectionError") &&
+            dispatchWasNotAccepted(error.cause));
+        return notAccepted
+          ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
+          : Effect.void;
+      }),
+    );
+});