refactor(server): share attachment message intake#10580

@@ -0,0 +1,50 @@
+import type { OrchestrationV2Command } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+
+import * as AttachmentClaims from "./AttachmentClaims.ts";
+import * as ThreadLaunch from "./ThreadLaunchService.ts";
+import * as ThreadManagement from "./ThreadManagementService.ts";
+
+// Claim failures clean up their partial batch. Once dispatch/launch starts, its
+// failure can follow a durable commit, so callers must not delete claimed files.
+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 });
+});
+
+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 });
+});
+
+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({

@@ -0,0 +1,50 @@
+import type { OrchestrationV2Command } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+
+import * as AttachmentClaims from "./AttachmentClaims.ts";
+import * as ThreadLaunch from "./ThreadLaunchService.ts";
+import * as ThreadManagement from "./ThreadManagementService.ts";
+
+// Claim failures clean up their partial batch. Once dispatch/launch starts, its
+// failure can follow a durable commit, so callers must not delete claimed files.
+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 });

@@ -0,0 +1,50 @@
+import type { OrchestrationV2Command } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+
+import * as AttachmentClaims from "./AttachmentClaims.ts";
+import * as ThreadLaunch from "./ThreadLaunchService.ts";
+import * as ThreadManagement from "./ThreadManagementService.ts";
+
+// Claim failures clean up their partial batch. Once dispatch/launch starts, its
+// failure can follow a durable commit, so callers must not delete claimed files.
+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 });
+});
+
+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 });
+});
+
+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 },
+  });

@@ -1454,104 +1432,68 @@ const makeWsRpcLayer = (
         [ORCHESTRATION_V2_WS_METHODS.launchThread]: (input) =>
           observeRpcEffect(
             ORCHESTRATION_V2_WS_METHODS.launchThread,
-            Effect.gen(function* () {
-              const pendingUploads =
-                input.initialMessage?.attachments.some(attachmentIsPendingUpload) ?? false;
-              if (pendingUploads && input.threadId === undefined) {
-                return yield* new OrchestrationV2ThreadLaunchError({
+            startup
+              .enqueueCommand(
+                ThreadMessageIntake.launchThread({
                   commandId: input.commandId,
+                  ...(input.threadId === undefined ? {} : { threadId: input.threadId }),
+                  ...(input.reuseExistingThread === undefined
+                    ? {}
+                    : { reuseExistingThread: input.reuseExistingThread }),
                   projectId: input.projectId,
-                  message: "Uploaded attachments need a thread id at launch.",
-                });
-              }
-              const claimed =
-                pendingUploads && input.threadId !== undefined && input.initialMessage !== undefined
-                  ? yield* claimPendingAttachments({
-                      threadId: input.threadId,
-                      attachments: input.initialMessage.attachments,
-                    }).pipe(
-                      Effect.mapError(
-                        (cause) =>
-                          new OrchestrationV2ThreadLaunchError({
-                            commandId: input.commandId,
-                            projectId: input.projectId,
-                            message: cause.message,
-                            cause,
-                          }),
-                      ),
-                    )
-                  : null;
-              const initialMessage =
-                input.initialMessage === undefined
-                  ? undefined
-                  : claimed === null
-                    ? input.initialMessage
-                    : { ...input.initialMessage, attachments: claimed.attachments };
-              return yield* startup
-                .enqueueCommand(
-                  threadLaunch.launch({
-                    commandId: input.commandId,
-                    ...(input.threadId === undefined ? {} : { threadId: input.threadId }),
-                    ...(input.reuseExistingThread === undefined
-                      ? {}
-                      : { reuseExistingThread: input.reuseExistingThread }),
-                    projectId: input.projectId,
-                    title: input.title,
-                    ...(input.generateTitle === undefined
-                      ? {}
-                      : { generateTitle: input.generateTitle }),
-                    modelSelection: input.modelSelection,
-                    runtimeMode: input.runtimeMode,
-                    interactionMode: input.interactionMode,
-                    workspaceStrategy: input.workspaceStrategy,
-                    ...(initialMessage === undefined
-                      ? {}
-                      : {
-                          initialMessage: {
-                            ...(initialMessage.messageId === undefined
-                              ? {}
-                              : { messageId: initialMessage.messageId }),
-                            text: initialMessage.text,
-                            attachments: initialMessage.attachments,
-                          },
-                        }),
-                    createdBy: "user",
-                    creationSource: input.creationSource ?? "web",
-                  }),
-                )
-                .pipe(
-                  Effect.tapError(() =>
-                    claimed === null
-                      ? Effect.void
-                      : releaseClaimedAttachments(claimed.claimedPaths),
-                  ),
-                  Effect.tap(() =>
-                    analytics
-                      .record("client.thread.started", originProps)
-                      .pipe(
-                        Effect.andThen(
-                          input.initialMessage === undefined
-                            ? Effect.void
-                            : analytics.record("client.turn.requested", originProps),
-                        ),
-                        Effect.ignore,
-                      ),
-                  ),
-                  Effect.map((result) => ({
-                    ...result,
-                    projection: projectThreadProjectionForWire(result.projection),
-                  })),
-                  Effect.mapError(
-                    (cause) =>
-                      new OrchestrationV2ThreadLaunchError({
-                        commandId: input.commandId,
-                        projectId: input.projectId,
-                        message: "Failed to launch thread",
-                        cause,
+                  title: input.title,
+                  ...(input.generateTitle === undefined
+                    ? {}
+                    : { generateTitle: input.generateTitle }),
+                  modelSelection: input.modelSelection,
+                  runtimeMode: input.runtimeMode,
+                  interactionMode: input.interactionMode,
+                  workspaceStrategy: input.workspaceStrategy,
+                  ...(input.initialMessage === undefined
+                    ? {}
+                    : {
+                        initialMessage: {
+                          ...(input.initialMessage.messageId === undefined
+                            ? {}
+                            : { messageId: input.initialMessage.messageId }),
+                          text: input.initialMessage.text,
+                          attachments: input.initialMessage.attachments,
+                        },
                       }),
-                  ),
-                );
-            }),
+                  createdBy: "user",
+                  creationSource: input.creationSource ?? "web",
+                }).pipe(Effect.provide(intakeContext)),
+              )
+              .pipe(
+                Effect.tap(() =>
+                  analytics
+                    .record("client.thread.started", originProps)
+                    .pipe(
+                      Effect.andThen(
+                        input.initialMessage === undefined
+                          ? Effect.void
+                          : analytics.record("client.turn.requested", originProps),
+                      ),
+                      Effect.ignore,
+                    ),
+                ),
+                Effect.map((result) => ({
+                  ...result,
+                  projection: projectThreadProjectionForWire(result.projection),
+                })),
+                Effect.mapError(
+                  (cause) =>
+                    new OrchestrationV2ThreadLaunchError({
+                      commandId: input.commandId,
+                      projectId: input.projectId,
+                      message:
+                        cause._tag === "AttachmentClaimError"
+                          ? cause.message
+                          : "Failed to launch thread",
+                      cause,
+                    }),
+                ),