feat(mcp): expose attachment upload and send#10564

@@ -0,0 +1,89 @@
+import { type ChatAttachment, MessageId, OrchestratorMcpFailure } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Upload from "../../../assets/AttachmentUpload.ts";
+import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts";
+import {
+  newCommandId,
+  readMutationCaller,
+  readWritableThread,
+  unavailable,
+} from "../../threadAccess.ts";
+import { AttachmentToolkit } from "./tools.ts";
+
+export function resolveAttachmentReferences(
+  requested: ReadonlyArray<ChatAttachment>,
+  stored: ReadonlyArray<ChatAttachment>,
+) {
+  const owned = new Map(stored.map((attachment) => [attachment.id, attachment]));
+  return Effect.forEach(requested, (attachment) => {
+    const canonical = Claims.attachmentIsPendingUpload(attachment)
+      ? attachment
+      : owned.get(attachment.id);
+    return canonical === undefined
+      ? Effect.fail(
+          new OrchestratorMcpFailure({
+            code: "invalid_request",
+            message: "Attachments must be pending uploads or already belong to the target thread.",
+          }),
+        )
+      : Effect.succeed(canonical);
+  });
+}
+
+export const AttachmentHandlersLive = AttachmentToolkit.toLayer({
+  t3_attachment_prepare_upload: (input) =>
+    Effect.gen(function* () {
+      yield* readMutationCaller();
+      return yield* Upload.issueAttachmentUploadUrl(input.upload).pipe(
+        Effect.mapError(unavailable),
+      );
+    }),
+  t3_attachment_discard: (input) =>
+    Effect.gen(function* () {
+      yield* readMutationCaller();
+      yield* Upload.deletePendingAttachment(input.attachmentId);
+      return {};
+    }),
+  t3_thread_send_attachments: (input) =>
+    Effect.gen(function* () {
+      const { caller, threads, projection } = yield* readWritableThread(input.threadId);
+      const attachments = yield* resolveAttachmentReferences(
+        input.attachments,
+        projection.messages.flatMap((message) => message.attachments),
+      );
+      const commandId = yield* newCommandId();
+      const messageId = MessageId.make(commandId);
+      const claimed = yield* Claims.claimPendingAttachments({

@@ -0,0 +1,80 @@
+import { MessageId, OrchestratorMcpFailure } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Upload from "../../../assets/AttachmentUpload.ts";
+import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts";
+import {
+  newCommandId,
+  readMutationCaller,
+  readWritableThread,
+  unavailable,
+} from "../../threadAccess.ts";
+import { AttachmentToolkit } from "./tools.ts";
+
+export const AttachmentHandlersLive = AttachmentToolkit.toLayer({
+  t3_attachment_prepare_upload: (input) =>
+    Effect.gen(function* () {
+      yield* readMutationCaller();
+      return yield* Upload.issueAttachmentUploadUrl(input.upload).pipe(
+        Effect.mapError(unavailable),
+      );
+    }),
+  t3_attachment_discard: (input) =>
+    Effect.gen(function* () {
+      yield* readMutationCaller();
+      yield* Upload.deletePendingAttachment(input.attachmentId);
+      return {};
+    }),
+  t3_thread_send_attachments: (input) =>
+    Effect.gen(function* () {
+      const { caller, threads, projection } = yield* readWritableThread(input.threadId);
+      const owned = new Set(
+        projection.messages.flatMap((message) =>
+          message.attachments.map((attachment) => attachment.id),
+        ),
+      );
+      if (
+        input.attachments.some(
+          (attachment) =>
+            !Claims.attachmentIsPendingUpload(attachment) && !owned.has(attachment.id),

@@ -0,0 +1,86 @@
+import { type ChatAttachment, MessageId, OrchestratorMcpFailure } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Upload from "../../../assets/AttachmentUpload.ts";
+import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts";
+import * as ThreadMessageIntake from "../../../orchestration-v2/ThreadMessageIntake.ts";
+import {
+  newCommandId,
+  readMutationCaller,
+  readWritableThread,
+  unavailable,
+} from "../../threadAccess.ts";
+import { AttachmentToolkit } from "./tools.ts";
+
+export function resolveAttachmentReferences(
+  requested: ReadonlyArray<ChatAttachment>,
+  stored: ReadonlyArray<ChatAttachment>,
+) {
+  const owned = new Map(stored.map((attachment) => [attachment.id, attachment]));
+  return Effect.forEach(requested, (attachment) => {
+    const canonical = Claims.attachmentIsPendingUpload(attachment)

@@ -0,0 +1,73 @@
+import {
+  AttachmentCreateUploadUrlInput,
+  AttachmentCreateUploadUrlResult,
+  AttachmentDeleteInput,
+  ChatImageAttachment,
+  ChatFileAttachment,
+  MessageId,
+  RunId,
+  ThreadId,
+  OrchestrationV2RunStatus,
+  OrchestratorMcpFailure,
+} from "@t3tools/contracts";
+import * as Crypto from "effect/Crypto";
+import * as FileSystem from "effect/FileSystem";
+import * as Schema from "effect/Schema";
+import { Tool, Toolkit } from "effect/unstable/ai";
+import { ServerSecretStore } from "../../../auth/ServerSecretStore.ts";
+import { ServerConfig } from "../../../config.ts";
+import { ThreadManagementService } from "../../../orchestration-v2/ThreadManagementService.ts";
+import { McpInvocationContext } from "../../McpInvocationContext.ts";
+
+const shared = {
+  failure: OrchestratorMcpFailure,
+  failureMode: "return" as const,
+  dependencies: [
+    McpInvocationContext,
+    ThreadManagementService,
+    ServerConfig,
+    ServerSecretStore,
+    FileSystem.FileSystem,
+    Crypto.Crypto,
+  ],
+};
+export const AttachmentUploadTool = Tool.make("t3_attachment_prepare_upload", {
+  ...shared,
+  description:
+    "Create the app's signed upload URL. PUT the exact bytes to relativeUrl on this MCP server's HTTP origin, then pass the attachment metadata and returned ID to t3_thread_send_attachments. Upload is separate from sending; provider attachment support is decided by its adapter.",

@@ -0,0 +1,80 @@
+import { MessageId, OrchestratorMcpFailure } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Upload from "../../../assets/AttachmentUpload.ts";
+import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts";
+import {
+  newCommandId,
+  readMutationCaller,
+  readWritableThread,
+  unavailable,
+} from "../../threadAccess.ts";
+import { AttachmentToolkit } from "./tools.ts";
+
+export const AttachmentHandlersLive = AttachmentToolkit.toLayer({
+  t3_attachment_prepare_upload: (input) =>
+    Effect.gen(function* () {
+      yield* readMutationCaller();
+      return yield* Upload.issueAttachmentUploadUrl(input.upload).pipe(
+        Effect.mapError(unavailable),
+      );
+    }),
+  t3_attachment_discard: (input) =>
+    Effect.gen(function* () {
+      yield* readMutationCaller();
+      yield* Upload.deletePendingAttachment(input.attachmentId);
+      return {};
+    }),
+  t3_thread_send_attachments: (input) =>
+    Effect.gen(function* () {
+      const { caller, threads, projection } = yield* readWritableThread(input.threadId);
+      const owned = new Set(
+        projection.messages.flatMap((message) =>
+          message.attachments.map((attachment) => attachment.id),
+        ),
+      );
+      if (
+        input.attachments.some(
+          (attachment) =>
+            !Claims.attachmentIsPendingUpload(attachment) && !owned.has(attachment.id),
+        )
+      )
+        return yield* new OrchestratorMcpFailure({
+          code: "invalid_request",
+          message: "Attachments must be pending uploads or already belong to the target thread.",
+        });

@@ -0,0 +1,72 @@
+import { MessageId, OrchestratorMcpFailure } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Upload from "../../../assets/AttachmentUpload.ts";
+import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts";
+import {
+  newCommandId,
+  readMutationCaller,
+  readWritableThread,
+  unavailable,
+} from "../../threadAccess.ts";
+import { AttachmentToolkit } from "./tools.ts";
+
+export const AttachmentHandlersLive = AttachmentToolkit.toLayer({
+  t3_attachment_prepare_upload: (input) =>
+    Effect.gen(function* () {
+      yield* readMutationCaller();
+      return yield* Upload.issueAttachmentUploadUrl(input.upload).pipe(
+        Effect.mapError(unavailable),
+      );
+    }),
+  t3_attachment_discard: (input) =>
+    Effect.gen(function* () {
+      yield* readMutationCaller();
+      yield* Upload.deletePendingAttachment(input.attachmentId);
+      return {};
+    }),
+  t3_thread_send_attachments: (input) =>
+    Effect.gen(function* () {
+      const { caller, threads, projection } = yield* readWritableThread(input.threadId);
+      const owned = new Set(
+        projection.messages.flatMap((message) =>
+          message.attachments.map((attachment) => attachment.id),
+        ),
+      );
+      if (
+        input.attachments.some(
+          (attachment) =>
+            !Claims.attachmentIsPendingUpload(attachment) && !owned.has(attachment.id),
+        )
+      )
+        return yield* new OrchestratorMcpFailure({
+          code: "invalid_request",
+          message: "Attachments must be pending uploads or already belong to the target thread.",
+        });
+      const commandId = yield* newCommandId();
+      const messageId = MessageId.make(commandId);
+      const claimed = yield* Claims.claimPendingAttachments({
+        threadId: projection.thread.id,
+        attachments: input.attachments,
+      }).pipe(Effect.mapError(unavailable));