fix(server): only the agent that staged an upload can send it

apps/server/src/attachmentStore.ts

@@ -27,7 +27,7 @@ const ATTACHMENT_ID_PATTERN = new RegExp(
 );
 
 export const PENDING_ATTACHMENT_THREAD_SEGMENT = "pending";
-const PENDING_ATTACHMENT_MAX_AGE_MS = 24 * 60 * 60 * 1000;
+export const PENDING_ATTACHMENT_MAX_AGE_MS = 24 * 60 * 60 * 1000;
 const PARTIAL_UPLOAD_MAX_AGE_MS = 60 * 60 * 1000;
 
 export function toSafeThreadAttachmentSegment(threadId: string): string | null {

apps/server/src/mcp/McpHttpServer.test.ts

@@ -58,6 +58,7 @@ import * as McpToolAccessTestkit from "./McpToolAccess.testkit.ts";
 import * as McpInvocationContext from "./McpInvocationContext.ts";
 import * as McpSessionRegistry from "./McpSessionRegistry.ts";
 import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
+import * as PendingUploadOwners from "./PendingUploadOwners.ts";
 import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
 import { OrchestratorToolkit } from "./toolkits/orchestrator/tools.ts";
 import { PreviewToolkit } from "./toolkits/preview/tools.ts";
@@ -342,6 +343,7 @@ const layerRecordingTools = (calls: Array<unknown>) => {
         Layer.mock(ServerSettings.ServerSettingsService)({}),
         Layer.mock(DocumentText.DocumentText)({}),
         Layer.mock(ProviderRegistry.ProviderRegistry)({}),
+        PendingUploadOwners.layer,
         ServerConfig.layerTest(process.cwd(), { prefix: "t3-mcp-http-server-test-" }),
         NodeCrypto.layer,
       ),

apps/server/src/mcp/McpHttpServer.ts

@@ -49,6 +49,7 @@ import * as McpToolAccess from "./McpToolAccess.ts";
 import * as McpToolExposure from "./McpToolExposure.ts";
 import { MCP_TOOL_EXPOSURE_HEADER } from "@t3tools/provider-core/server/mcpSession";
 import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
+import * as PendingUploadOwners from "./PendingUploadOwners.ts";
 import { PreviewControlsToolkit } from "./toolkits/previewControls/tools.ts";
 import * as PreviewControlsHandlers from "./toolkits/previewControls/handlers.ts";
 import { EnvironmentToolkit } from "./toolkits/environment/tools.ts";
@@ -1281,12 +1282,16 @@ const layerThemeToolkit = toolkitRegistration(ThemeToolkit, ThemeHandlers.layer)
   Layer.provide(ThreadCommandExecutor.layer),
 );
 
-export const layerProjectRegistration = toolkitRegistration(ProjectToolkit, ProjectHandlers.layer);
+// Both toolkits get the same memoized owners layer, so a launch sees the uploads prepare recorded.
+export const layerProjectRegistration = toolkitRegistration(
+  ProjectToolkit,
+  ProjectHandlers.layer,
+).pipe(Layer.provide(PendingUploadOwners.layer));
 
 export const layerAttachmentToolkit = toolkitRegistration(
   AttachmentToolkit,
   AttachmentHandlers.layer,
-);
+).pipe(Layer.provide(PendingUploadOwners.layer));
 
 export const layerPullRequestsToolkit = toolkitRegistration(
   PullRequestsToolkit,

apps/server/src/mcp/PendingUploadOwners.ts

@@ -0,0 +1,60 @@
+import { ATTACHMENT_UPLOAD_URL_TTL_MS, OrchestratorMcpFailure } from "@t3tools/contracts";
+import * as Clock from "effect/Clock";
+import * as Context from "effect/Context";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+
+import { PENDING_ATTACHMENT_MAX_AGE_MS } from "../attachmentStore.ts";
+import type { McpInvocationScope } from "./McpInvocationContext.ts";
+
+/** The sweep's clock starts at upload completion, possibly one upload-URL lifetime after issue. */
+const OWNER_TTL_MS = PENDING_ATTACHMENT_MAX_AGE_MS + ATTACHMENT_UPLOAD_URL_TTL_MS;
+
+const ownerKey = (scope: McpInvocationScope) => scope.thread?.threadId ?? scope.requestNamespace;
+
+/** The MCP caller that prepared each pending upload. Kept in memory, so a restart orphans them. */
+export class PendingUploadOwners extends Context.Service<
+  PendingUploadOwners,
+  {
+    readonly record: (scope: McpInvocationScope, attachmentId: string) => Effect.Effect<void>;
+    readonly requireOwner: (
+      scope: McpInvocationScope,
+      attachmentIds: ReadonlyArray<string>,
+      action: "send" | "discard",
+    ) => Effect.Effect<void, OrchestratorMcpFailure>;
+    readonly forget: (attachmentId: string) => Effect.Effect<void>;
+  }
+>()("t3/mcp/PendingUploadOwners") {}
+
+export const layer = Layer.sync(PendingUploadOwners, () => {
+  const owners = new Map<string, { readonly owner: string; readonly issuedAt: number }>();
+  return PendingUploadOwners.of({
+    record: Effect.fn("PendingUploadOwners.record")(function* (
+      scope: McpInvocationScope,
+      attachmentId: string,
+    ) {
+      const now = yield* Clock.currentTimeMillis;
+      for (const [id, entry] of owners) {
+        if (now - entry.issuedAt > OWNER_TTL_MS) owners.delete(id);
+      }
+      owners.set(attachmentId, { owner: ownerKey(scope), issuedAt: now });
+    }),
+    requireOwner: Effect.fn("PendingUploadOwners.requireOwner")(function* (
+      scope: McpInvocationScope,
+      attachmentIds: ReadonlyArray<string>,
+      action: "send" | "discard",
+    ) {
+      const owner = ownerKey(scope);
+      if (attachmentIds.every((id) => owners.get(id)?.owner === owner)) return;
+      return yield* new OrchestratorMcpFailure({
+        code: "invalid_request",
+        message: `Only the caller that prepared a pending upload can ${action} it.`,
+      });
+    }),
+    forget: Effect.fn("PendingUploadOwners.forget")((attachmentId: string) =>
+      Effect.sync(() => {
+        owners.delete(attachmentId);
+      }),
+    ),
+  });
+});

apps/server/src/mcp/toolkits/attachment/handlers.ts

@@ -1,17 +1,12 @@
-import {
-  ATTACHMENT_UPLOAD_URL_TTL_MS,
-  type ChatAttachment,
-  MessageId,
-  OrchestratorMcpFailure,
-} from "@t3tools/contracts";
-import * as Clock from "effect/Clock";
+import { type ChatAttachment, MessageId, OrchestratorMcpFailure } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 import * as AssetSigner from "../../../assets/AssetSigner.ts";
-import * as McpInvocationContext from "../../McpInvocationContext.ts";
 import * as Upload from "../../../assets/AttachmentUpload.ts";
 import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts";
 import * as ThreadMessageIntake from "../../../orchestration-v2/ThreadMessageIntake.ts";
+import * as McpInvocationContext from "../../McpInvocationContext.ts";
 import * as McpToolAccess from "../../McpToolAccess.ts";
+import * as PendingUploadOwners from "../../PendingUploadOwners.ts";
 import { newCommandId, readThread, unavailable } from "../../threadAccess.ts";
 import { AttachmentToolkit } from "./tools.ts";
 
@@ -35,18 +30,11 @@ export function resolveAttachmentReferences(
   });
 }
 
-/** The 24-hour sweep starts at upload completion, possibly one upload-URL lifetime after issue. */
-const UPLOAD_OWNER_TTL_MS = 24 * 60 * 60 * 1000 + ATTACHMENT_UPLOAD_URL_TTL_MS;
-
-const uploadOwner = McpInvocationContext.McpInvocationContext.pipe(
-  Effect.map((scope) => scope.thread?.threadId ?? scope.requestNamespace),
-);
-
 export const layer = McpToolAccess.toLayer(
   AttachmentToolkit,
   Effect.gen(function* () {
     const assetSigner = yield* AssetSigner.AssetSigner;
-    const uploadOwners = new Map<string, { readonly owner: string; readonly issuedAt: number }>();
+    const owners = yield* PendingUploadOwners.PendingUploadOwners;
     return {
       // A pending upload belongs to its caller until a send claims it, so
       // linked callers may stage the uploads their own sends need.
@@ -57,11 +45,10 @@ export const layer = McpToolAccess.toLayer(
               Effect.provideService(AssetSigner.AssetSigner, assetSigner),
               Effect.mapError(unavailable),
             );
-            const now = yield* Clock.currentTimeMillis;
-            for (const [id, entry] of uploadOwners) {
-              if (now - entry.issuedAt > UPLOAD_OWNER_TTL_MS) uploadOwners.delete(id);
-            }
-            uploadOwners.set(result.attachmentId, { owner: yield* uploadOwner, issuedAt: now });
+            yield* owners.record(
+              yield* McpInvocationContext.McpInvocationContext,
+              result.attachmentId,
+            );
             const { origin } = yield* McpInvocationContext.McpInvocationContext;
             return origin === undefined
               ? result
@@ -72,14 +59,13 @@ export const layer = McpToolAccess.toLayer(
       t3_attachment_discard: McpToolAccess.writes(
         (input) =>
           Effect.gen(function* () {
-            if (uploadOwners.get(input.attachmentId)?.owner !== (yield* uploadOwner)) {
-              return yield* new OrchestratorMcpFailure({
-                code: "invalid_request",
-                message: "Only the caller that prepared a pending upload can discard it.",
-              });
-            }
+            yield* owners.requireOwner(
+              yield* McpInvocationContext.McpInvocationContext,
+              [input.attachmentId],
+              "discard",
+            );
             yield* Upload.deletePendingAttachment(input.attachmentId);
-            uploadOwners.delete(input.attachmentId);
+            yield* owners.forget(input.attachmentId);
             return {};
           }),
         { linkedCallers: "allowed" },
@@ -94,6 +80,13 @@ export const layer = McpToolAccess.toLayer(
                 code: "invalid_request",
                 message: "Unarchive the target thread before sending attachments.",
               });
+            yield* owners.requireOwner(
+              yield* McpInvocationContext.McpInvocationContext,
+              input.attachments
+                .filter(Claims.attachmentIsPendingUpload)
+                .map((attachment) => attachment.id),
+              "send",
+            );
             const attachments = yield* resolveAttachmentReferences(
               input.attachments,
               projection.messages.flatMap((message) => message.attachments),