diff --git a/apps/server/src/attachmentStore.ts b/apps/server/src/attachmentStore.ts index a2b531e3fa5b0..8d36b97763103 100644 --- a/apps/server/src/attachmentStore.ts +++ b/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 { diff --git a/apps/server/src/mcp/McpHttpServer.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts index 3c80cdf55fe1e..ba089d3752001 100644 --- a/apps/server/src/mcp/McpHttpServer.test.ts +++ b/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) => { 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, ), diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 752702e390a2f..015fd143815f9 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/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, diff --git a/apps/server/src/mcp/PendingUploadOwners.ts b/apps/server/src/mcp/PendingUploadOwners.ts new file mode 100644 index 0000000000000..f7b335835045c --- /dev/null +++ b/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; + readonly requireOwner: ( + scope: McpInvocationScope, + attachmentIds: ReadonlyArray, + action: "send" | "discard", + ) => Effect.Effect; + readonly forget: (attachmentId: string) => Effect.Effect; + } +>()("t3/mcp/PendingUploadOwners") {} + +export const layer = Layer.sync(PendingUploadOwners, () => { + const owners = new Map(); + 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, + 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); + }), + ), + }); +}); diff --git a/apps/server/src/mcp/toolkits/attachment/handlers.ts b/apps/server/src/mcp/toolkits/attachment/handlers.ts index 032ab3a1d0ed6..26a1030e61b24 100644 --- a/apps/server/src/mcp/toolkits/attachment/handlers.ts +++ b/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(); + 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), diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 35a18f5f0b460..17b538097ade9 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -1,4 +1,5 @@ import * as AssetSigner from "../../assets/AssetSigner.ts"; +import * as Upload from "../../assets/AttachmentUpload.ts"; import * as DocumentText from "../../provider/DocumentText.ts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -49,6 +50,9 @@ import type * as ProjectionStore from "../../orchestration-v2/ProjectionStore.ts import * as ThreadSearch from "../../orchestration-v2/ThreadSearch.ts"; import * as PreviewBrowser from "../../preview/PreviewBrowser.ts"; import * as ProjectService from "../../project/ProjectService.ts"; +import * as ProjectCloneTracker from "../../project/ProjectCloneTracker.ts"; +import * as ManagedProjectFolders from "../../project/ManagedProjectFolders.ts"; +import * as Repositories from "../../sourceControl/SourceControlRepositoryService.ts"; import * as ProviderRegistry from "../../provider/ProviderRegistry.ts"; import * as SecretRequests from "../../secrets/SecretRequests.ts"; import * as RemoteDelegation from "../../peer/RemoteDelegation.ts"; @@ -1233,64 +1237,161 @@ it.effect("discovers workflow stop targets and requires orchestration capability ); }); -it.effect("only the caller that prepared a pending upload can discard it", () => - Effect.gen(function* () { - const server = yield* McpServer.McpServer; - const call = ( - name: string, - args: Record, - invocation: McpInvocationContext.McpInvocationScope, - ) => - server - .callTool({ name, arguments: args }) - .pipe( - Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), - Effect.provideService(McpSchema.McpServerClient, client), - ); - const prepared = yield* call( - "t3_attachment_prepare_upload", - { upload: { name: "shot.png", mimeType: "image/png", sizeBytes: 4 } }, - scope, - ); - const { attachmentId } = prepared.structuredContent as { readonly attachmentId: string }; - const otherThread = { - ...scope, - requestNamespace: "other-session", - thread: { ...scope.thread!, threadId: ThreadId.make("other-thread") }, - }; +it.effect( + "only the caller that prepared a pending upload can send, launch with or discard it", + () => { + const sent: Array = []; + const launchedBy: Array = []; + return Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const call = ( + name: string, + args: Record, + invocation: McpInvocationContext.McpInvocationScope, + ) => + server + .callTool({ name, arguments: args }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), + Effect.provideService(McpSchema.McpServerClient, client), + ); + const upload = { type: "file", name: "notes.txt", mimeType: "text/plain", sizeBytes: 4 }; + const prepared = yield* call("t3_attachment_prepare_upload", { upload }, scope); + const { attachmentId, relativeUrl } = prepared.structuredContent as { + readonly attachmentId: string; + readonly relativeUrl: string; + }; + const claims = yield* Upload.validateAttachmentUploadToken( + relativeUrl.slice(`${Upload.ATTACHMENT_UPLOAD_ROUTE_PREFIX}/`.length), + ); + expect(yield* Upload.storeAttachmentUpload(claims!, new Uint8Array(4))).toEqual({ ok: true }); + const otherThread = { + ...scope, + requestNamespace: "other-session", + thread: { ...scope.thread!, threadId: ThreadId.make("other-thread") }, + }; + const attachments = [{ ...upload, id: attachmentId }]; - const refused = yield* call("t3_attachment_discard", { attachmentId }, otherThread); - expect(declaredFailure(refused)).toMatchObject({ code: "invalid_request" }); + const refused = yield* call("t3_attachment_discard", { attachmentId }, otherThread); + expect(declaredFailure(refused)).toMatchObject({ code: "invalid_request" }); + const stolen = yield* call("t3_thread_send_attachments", { attachments }, otherThread); + expect(declaredFailure(stolen)).toMatchObject({ code: "invalid_request" }); + const launchedWithStolen = yield* call( + "t3_thread_launch", + { title: "Stolen", attachments }, + otherThread, + ); + expect(declaredFailure(launchedWithStolen)).toMatchObject({ code: "invalid_request" }); + expect(sent).toEqual([]); + expect(launchedBy).toEqual([]); + const own = yield* call("t3_thread_send_attachments", { attachments }, scope); + expect(own.isError).toBe(false); + expect(sent).toEqual([threadId]); + const ownLaunch = yield* call("t3_thread_launch", { title: "Own", attachments }, scope); + expect(ownLaunch.isError).toBe(false); + expect(launchedBy).toEqual([threadId]); - yield* TestClock.adjust(24 * 60 * 60 * 1000 + 5 * 60_000); - yield* call( - "t3_attachment_prepare_upload", - { upload: { name: "later.png", mimeType: "image/png", sizeBytes: 4 } }, - scope, - ); + yield* TestClock.adjust(24 * 60 * 60 * 1000 + 5 * 60_000); + yield* call( + "t3_attachment_prepare_upload", + { upload: { name: "later.png", mimeType: "image/png", sizeBytes: 4 } }, + scope, + ); - const discarded = yield* call( - "t3_attachment_discard", - { attachmentId }, - { ...scope, requestNamespace: "mcp-core-session-2" }, - ); - expect(discarded.isError).toBe(false); - }).pipe( - Effect.provide( - McpHttpServer.layerAttachmentToolkit.pipe( - Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), - Layer.provide(McpToolAccessTestkit.liveThreadsLayer), - Layer.provide(AssetSigner.layer.pipe(Layer.provideMerge(ServerSecretStore.layer))), - Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-mcp-attachment-" })), - Layer.provide( - DocumentText.layer.pipe( - Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + const discarded = yield* call( + "t3_attachment_discard", + { attachmentId }, + { ...scope, requestNamespace: "mcp-core-session-2" }, + ); + expect(discarded.isError).toBe(false); + }).pipe( + Effect.provide( + Layer.mergeAll( + McpHttpServer.layerAttachmentToolkit, + McpHttpServer.layerProjectRegistration, + ).pipe( + Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), + Layer.provide( + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: (input) => + Effect.sync(() => { + launchedBy.push(input.initialMessage?.senderThreadId); + return { + threadId: input.threadId, + projection: { + thread: { + id: input.threadId, + projectId: input.projectId, + modelSelection: input.modelSelection, + }, + runs: [], + messages: [{ attachments: input.initialMessage?.attachments ?? [] }], + }, + resumed: false, + } as unknown as ThreadLaunchService.ThreadLaunchResult; + }), + }), + ), + Layer.provide( + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), + Layer.provide( + Layer.mock(ProjectService.ProjectService)({ + getById: (projectId) => + Effect.succeedSome({ + id: projectId, + title: "MCP test", + workspaceRoot: "/repo", + defaultModelSelection: null, + scripts: [], + createdAt: "2026-10-18T00:00:00.000Z", + updatedAt: "2026-10-18T00:00:00.000Z", + deletedAt: null, + }), + }), + ), + Layer.provide(Layer.mock(ProjectCloneTracker.ProjectCloneTracker)({})), + Layer.provide( + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + }), + ), + Layer.provide(Layer.mock(Repositories.SourceControlRepositoryService)({})), + Layer.provide( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (id) => Effect.succeed(McpToolAccessTestkit.liveThreadShell(id)), + getProjectThreadRecords: (input) => + Effect.succeed( + McpToolAccessTestkit.idleThreadProjection( + McpToolAccessTestkit.liveThreadShell(input.threadId), + ), + ), + sendToThread: (input) => + Effect.sync(() => { + sent.push(input.threadId); + return { + run: { id: RunId.make("run:sent"), status: "running" }, + message: { attachments: input.attachments }, + } as unknown as ThreadManagement.ThreadManagementSendResult; + }), + }), + ), + Layer.provideMerge(AssetSigner.layer.pipe(Layer.provideMerge(ServerSecretStore.layer))), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-mcp-attachment-" }), + ), + Layer.provideMerge( + DocumentText.layer.pipe( + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ), ), + Layer.provide(NodeCrypto.layer), ), - Layer.provide(NodeCrypto.layer), ), - ), - ), + ); + }, ); it.effect( diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index c95fadd9096e2..e67e0c8e0f0d2 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -44,6 +44,7 @@ import * as GitVcsDriver from "../../../vcs/GitVcsDriver.ts"; import * as VcsProcess from "../../../vcs/VcsProcess.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as McpToolAccess from "../../McpToolAccess.ts"; +import * as PendingUploadOwners from "../../PendingUploadOwners.ts"; import * as ProjectHandlers from "./handlers.ts"; import { ProjectToolkit } from "./tools.ts"; @@ -170,6 +171,7 @@ const clientLaunchHarness = (input: { ), }), Layer.mock(PeerForwarding.PeerForwarding)({}), + PendingUploadOwners.layer, Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects", isScratchProject: () => Effect.succeed(false), diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index 7a2fb5381227e..9e0dc654376ab 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -29,6 +29,7 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { resolveRuntimeMode } from "../../delegatedTaskModes.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as McpToolAccess from "../../McpToolAccess.ts"; +import * as PendingUploadOwners from "../../PendingUploadOwners.ts"; import * as OrchestrationMcp from "../../OrchestratorMcpService.ts"; import { newCommandId, readCaller, resolveProjectId, unavailable } from "../../threadAccess.ts"; import { ProjectToolkit } from "./tools.ts"; @@ -129,6 +130,11 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { code: "invalid_request", message: "A new thread accepts only pending attachment uploads.", }); + yield* (yield* PendingUploadOwners.PendingUploadOwners).requireOwner( + scope, + attachments.map((attachment) => attachment.id), + "send", + ); if ( input.scratch === true && (input.projectId !== undefined || input.workspaceStrategy !== undefined) diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts index 7de832eea28ad..209dac268c744 100644 --- a/apps/server/src/mcp/toolkits/project/tools.ts +++ b/apps/server/src/mcp/toolkits/project/tools.ts @@ -38,6 +38,7 @@ import * as ThreadManagementService from "../../../orchestration-v2/ThreadManage import * as SourceControlRepositoryService from "../../../sourceControl/SourceControlRepositoryService.ts"; import * as GitVcsDriver from "../../../vcs/GitVcsDriver.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as PendingUploadOwners from "../../PendingUploadOwners.ts"; const shared = { success: Project, @@ -178,6 +179,7 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", { ProviderRegistry.ProviderRegistry, DocumentText.DocumentText, Digest.Digest, + PendingUploadOwners.PendingUploadOwners, ], }) .annotate(Tool.Destructive, true)