1import {
2 ATTACHMENT_UPLOAD_URL_TTL_MS,
3 type ChatAttachment,
4 MessageId,
5 OrchestratorMcpFailure,
6} from "@t3tools/contracts";
7import * as Clock from "effect/Clock";
8import * as Effect from "effect/Effect";
9import * as McpInvocationContext from "../../McpInvocationContext.ts";
10import * as Upload from "../../../assets/AttachmentUpload.ts";
11import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts";
12import * as ThreadMessageIntake from "../../../orchestration-v2/ThreadMessageIntake.ts";
13import * as McpToolAccess from "../../McpToolAccess.ts";
14import { newCommandId, readThread, unavailable } from "../../threadAccess.ts";
15import { AttachmentToolkit } from "./tools.ts";
16
17export function resolveAttachmentReferences(
18 requested: ReadonlyArray<ChatAttachment>,
19 stored: ReadonlyArray<ChatAttachment>,
20) {
21 const owned = new Map(stored.map((attachment) => [attachment.id, attachment]));
22 return Effect.forEach(requested, (attachment) => {
23 const canonical = Claims.attachmentIsPendingUpload(attachment)
24 ? attachment
25 : owned.get(attachment.id);
26 return canonical === undefined
27 ? Effect.fail(
28 new OrchestratorMcpFailure({
29 code: "invalid_request",
30 message: "Attachments must be pending uploads or already belong to the target thread.",
31 }),
32 )
33 : Effect.succeed(canonical);
34 });
35}
36
37/**
38 * As long as the pending file can live. The sweep counts its 24 hours from when
39 * the upload finishes, which can be up to the upload URL's lifetime after issue.
40 */
41const UPLOAD_OWNER_TTL_MS = 24 * 60 * 60 * 1000 + ATTACHMENT_UPLOAD_URL_TTL_MS;
42
43/** A thread owns its uploads across provider sessions; an outside client per MCP session. */
44const uploadOwner = McpInvocationContext.McpInvocationContext.pipe(
45 Effect.map((scope) => scope.thread?.threadId ?? scope.requestNamespace),
46);
47
48export const layer = McpToolAccess.toLayer(
49 AttachmentToolkit,
50 Effect.sync(() => {
51 // Which caller prepared each pending upload, so only that caller can discard it.
52 const uploadOwners = new Map<string, { readonly owner: string; readonly issuedAt: number }>();
53 return {
54 t3_attachment_prepare_upload: McpToolAccess.writes((input) =>
55 Effect.gen(function* () {
56 const result = yield* Upload.issueAttachmentUploadUrl(input.upload).pipe(
57 Effect.mapError(unavailable),
58 );
59 const now = yield* Clock.currentTimeMillis;
60 for (const [id, entry] of uploadOwners) {
61 if (now - entry.issuedAt > UPLOAD_OWNER_TTL_MS) uploadOwners.delete(id);
62 }
63 uploadOwners.set(result.attachmentId, { owner: yield* uploadOwner, issuedAt: now });
64 return result;
65 }),
66 ),
67 t3_attachment_discard: McpToolAccess.writes((input) =>
68 Effect.gen(function* () {
69 if (uploadOwners.get(input.attachmentId)?.owner !== (yield* uploadOwner)) {
70 return yield* new OrchestratorMcpFailure({
71 code: "invalid_request",
72 message: "Only the caller that prepared a pending upload can discard it.",
73 });
74 }
75 yield* Upload.deletePendingAttachment(input.attachmentId);
76 uploadOwners.delete(input.attachmentId);
77 return {};
78 }),
79 ),
80 t3_thread_send_attachments: McpToolAccess.writesThreads(
81 (input) => [input.threadId],
82 (input) =>
83 Effect.gen(function* () {
84 const { caller, projection } = yield* readThread(input.threadId, ["messages"]);
85 if (projection.thread.archivedAt !== null)
86 return yield* new OrchestratorMcpFailure({
87 code: "invalid_request",
88 message: "Unarchive the target thread before sending attachments.",
89 });
90 const attachments = yield* resolveAttachmentReferences(
91 input.attachments,
92 projection.messages.flatMap((message) => message.attachments),
93 );
94 const commandId = yield* newCommandId();
95 const messageId = MessageId.make(commandId);
96 const result = yield* ThreadMessageIntake.sendToThread({
97 projectId: projection.thread.projectId,
98 threadId: projection.thread.id,
99 commandId,
100 messageId,
101 ...(caller === undefined ? {} : { senderThreadId: caller.id }),
102 text: input.message ?? "",
103 attachments,
104 mode: "auto",
105 createdBy: "agent",
106 creationSource: "mcp",
107 }).pipe(
108 Effect.mapError((error) =>
109 error._tag === "AttachmentClaimError"
110 ? new OrchestratorMcpFailure({
111 code: "orchestration_error",
112 message: error.message,
113 })
114 : unavailable(),
115 ),
116 );
117 return {
118 threadId: projection.thread.id,
119 messageId,
120 runId: result.run.id,
121 status: result.run.status,
122 };
123 }),
124 ),
125 };
126 }),
127);