From d1e19662ba7f1b2bf4d51066a93bfba237d79570 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 10:28:31 -0600 Subject: [PATCH] perf(server): scheduled-task MCP tools look up one task instead of listing all run_scheduled_task_now, and schedule_task updates and deletes, loaded and decoded every scheduled task to find one id. Expose the service's existing single-row lookup as `find` and use it, so these calls no longer scale with the task count or fail when an unrelated row cannot be decoded. Co-Authored-By: Claude Opus 5.5 --- .../OrchestratorMcpService.activity.test.ts | 16 ++- .../src/mcp/OrchestratorMcpService.test.ts | 5 +- apps/server/src/mcp/OrchestratorMcpService.ts | 7 +- ...OrchestratorMcpToolkit.integration.test.ts | 1 + apps/server/src/mcp/toolkits/core.test.ts | 102 +++++++++++++----- .../src/mcp/toolkits/thread/handlers.ts | 4 +- .../scheduledTasks/ScheduledTaskService.ts | 2 + 7 files changed, 90 insertions(+), 47 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index b290ad124c0f2..c129ff0a61a6c 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -430,17 +430,13 @@ it("readThread and sendToThread reach threads in other projects", async () => { Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(ScheduledTaskService.ScheduledTaskService)({ - list: () => + find: () => Effect.succeed({ - tasks: [ - { - id: "task-foreign", - projectId: foreignProjectId, - runtimeMode: "approval-required", - interactionMode: "default", - } as never, - ], - }), + id: "task-foreign", + projectId: foreignProjectId, + runtimeMode: "approval-required", + interactionMode: "default", + } as never), } satisfies Partial), Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ list: () => Effect.succeed([]), diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 47a6bda166ce1..8ef1b6b85d8d7 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -2400,6 +2400,7 @@ describe("OrchestratorMcpService provider resolution", () => { Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(ScheduledTaskService.ScheduledTaskService)({ list: () => Effect.succeed({ tasks }), + find: (id) => Effect.succeed(tasks.find((task) => task.id === id) ?? null), upsert: () => Ref.update(upserted, (count) => count + 1).pipe(Effect.as({ task: tasks[0]! })), }), @@ -2559,7 +2560,7 @@ describe("OrchestratorMcpService provider resolution", () => { Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(ScheduledTaskService.ScheduledTaskService)({ - list: () => Effect.succeed({ tasks: [bound] }), + find: () => Effect.succeed(bound), upsert: () => Ref.update(upserted, (count) => count + 1).pipe(Effect.as({ task: bound })), }), @@ -2643,7 +2644,7 @@ describe("OrchestratorMcpService provider resolution", () => { Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(ScheduledTaskService.ScheduledTaskService)({ - list: () => Effect.succeed({ tasks: [existing] }), + find: () => Effect.succeed(existing), upsert: (input) => Ref.update(saved, (strategies) => [...strategies, input.workspaceStrategy]).pipe( Effect.as({ task: existing }), diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index b124a03ea2c40..27201ea292aaa 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1934,15 +1934,14 @@ const make = Effect.gen(function* () { readonly interactionMode: ProviderInteractionMode; }, ): Effect.fn.Return { - const { tasks } = yield* scheduledTasks - .list() + const task = yield* scheduledTasks + .find(scheduledTaskId) .pipe( Effect.mapError((error) => failure("orchestration_error", `Could not load scheduled task: ${error.message}`), ), ); - const task = tasks.find((candidate) => candidate.id === scheduledTaskId); - if (task === undefined) { + if (task === null) { return yield* failure("task_not_found", `Scheduled task ${scheduledTaskId} was not found.`); } for (const modes of yield* scheduledTaskRunModes(task)) { diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index bedf2ce0111d3..5fe0c9e810998 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -177,6 +177,7 @@ const layerUnusedScheduledTaskStub = Layer.succeed( ScheduledTaskService.ScheduledTaskService, ScheduledTaskService.ScheduledTaskService.of({ list: () => Effect.succeed({ tasks: [] }), + find: () => Effect.die("ScheduledTaskService.find is unused in this test"), subscribeList: () => Stream.succeed({ tasks: [] }), upsert: () => Effect.die("ScheduledTaskService.upsert is unused in this test"), setEnabled: () => Effect.die("ScheduledTaskService.setEnabled is unused in this test"), diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 5774f0e4c4f65..6985db01592fc 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -695,13 +695,8 @@ it.effect("a caller cannot rewrite a scheduled task that runs above its own mode Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})), Layer.provide( Layer.mock(ScheduledTaskService.ScheduledTaskService)({ - list: () => - Effect.succeed({ - tasks: [ - scheduledTask("task-full-access", "full-access"), - scheduledTask("task-auto", "auto"), - ], - }), + find: (id) => + Effect.succeed(scheduledTask(id, id === "task-full-access" ? "full-access" : "auto")), upsert: (input) => Effect.succeed({ task: { ...(scheduledTask(input.id ?? "task-auto", "auto") as object), ...input }, @@ -719,34 +714,81 @@ it.effect("a caller cannot rewrite a scheduled task that runs above its own mode ), ); +const runnableTask = Schema.decodeSync(ScheduledTask)(scheduledTask("task-a", "full-access")); + it.effect.each([ { - scenario: "explains why a webhook task cannot run manually", - tasks: [ - { - ...Schema.decodeSync(ScheduledTask)(scheduledTask("task-webhook", "full-access")), - schedule: { type: "webhook" as const, signature: null }, + scenario: "runs an existing task by its id", + find: Effect.succeed(runnableTask), + runNow: Effect.succeed({ task: { ...runnableTask, runCount: 1 }, threadId: null }), + expected: { + structuredContent: { + taskId: "task-a", + threadId: null, + lastRunStatus: "never", + runCount: 1, + nextRunAt: null, }, - ], - error: new ScheduledTaskError({ - message: "Webhook tasks run when their URL receives a request.", - cause: new Error("private-webhook-cause"), + failure: undefined, + }, + }, + { + scenario: "explains why a webhook task cannot run manually", + find: Effect.succeed({ + ...runnableTask, + schedule: { type: "webhook" as const, signature: null }, }), - code: "orchestration_error", - message: "Webhook tasks run when their URL receives a request.", + runNow: Effect.fail( + new ScheduledTaskError({ + message: "Webhook tasks run when their URL receives a request.", + cause: new Error("private-webhook-cause"), + }), + ), + expected: { + structuredContent: undefined, + failure: { + _tag: "OrchestratorMcpFailure", + code: "orchestration_error", + message: "Webhook tasks run when their URL receives a request.", + }, + }, }, { scenario: "reports a missing task before trying to run it", - tasks: [], - error: new ScheduledTaskError({ message: "must not run a missing task" }), - code: "invalid_request", - message: "The scheduled task was not found.", + find: Effect.succeed(null), + runNow: Effect.die("must not run a missing task"), + expected: { + structuredContent: undefined, + failure: { + _tag: "OrchestratorMcpFailure", + code: "invalid_request", + message: "The scheduled task was not found.", + }, + }, }, -])("run_scheduled_task_now $scenario", ({ tasks, error, code, message }) => + { + scenario: "hides a storage failure while looking up the task", + find: Effect.fail( + new ScheduledTaskError({ + message: "Could not load schedule task.", + cause: new Error("private-storage-cause"), + }), + ), + runNow: Effect.die("must not run after a failed lookup"), + expected: { + structuredContent: undefined, + failure: { + _tag: "OrchestratorMcpFailure", + code: "orchestration_error", + message: "The operation could not be completed.", + }, + }, + }, +])("run_scheduled_task_now $scenario", ({ find, runNow, expected }) => Effect.gen(function* () { const server = yield* McpServer.McpServer; const result = yield* server - .callTool({ name: "run_scheduled_task_now", arguments: { taskId: "task-webhook" } }) + .callTool({ name: "run_scheduled_task_now", arguments: { taskId: "task-a" } }) .pipe( Effect.provideService( McpInvocationContext.McpInvocationContext, @@ -754,9 +796,10 @@ it.effect.each([ ), Effect.provideService(McpSchema.McpServerClient, client), ); - expect(result.isError).toBe(true); - expect(result.structuredContent).toBeUndefined(); - expect(declaredFailure(result)).toEqual({ _tag: "OrchestratorMcpFailure", code, message }); + expect({ + structuredContent: result.structuredContent, + failure: declaredFailure(result), + }).toEqual(expected); expect(JSON.stringify(result)).not.toContain("private-"); }).pipe( Effect.provide( @@ -770,8 +813,9 @@ it.effect.each([ Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), Layer.provide( Layer.mock(ScheduledTaskService.ScheduledTaskService)({ - list: () => Effect.succeed({ tasks }), - runNow: () => Effect.fail(error), + list: () => Effect.die("must not list every task to run one"), + find: () => find, + runNow: () => runNow, }), ), ), diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index f968312437518..7aaee1571f0e9 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -94,8 +94,8 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { run_scheduled_task_now: McpToolAccess.writesEnvironment((input) => Effect.gen(function* () { const scheduler = yield* ScheduledTasks.ScheduledTaskService; - const { tasks } = yield* scheduler.list().pipe(Effect.mapError(unavailable)); - if (!tasks.some((task) => task.id === input.taskId)) + const found = yield* scheduler.find(input.taskId).pipe(Effect.mapError(unavailable)); + if (found === null) return yield* new OrchestratorMcpFailure({ code: "invalid_request", message: "The scheduled task was not found.", diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index d69254009b04c..d91d355f72c6f 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -230,6 +230,7 @@ export class ScheduledTaskService extends Context.Service< ScheduledTaskService, { readonly list: () => Effect.Effect; + readonly find: (id: ScheduledTaskId) => Effect.Effect; readonly subscribeList: () => Stream.Stream; readonly upsert: ( input: ScheduledTaskUpsertInput, @@ -2335,6 +2336,7 @@ export const layer = Layer.effect( return ScheduledTaskService.of({ list, + find: findTask, subscribeList, upsert, setEnabled,