diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index c35b605dca4fb..e9576c33e1ca3 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -527,17 +527,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 681f5ae3368ae..417031d6c742b 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -2556,6 +2556,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]! })), }), @@ -2717,7 +2718,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 })), }), @@ -2806,7 +2807,7 @@ describe("OrchestratorMcpService provider resolution", () => { Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(ScheduledTaskService.ScheduledTaskService)({ - list: () => Effect.succeed({ tasks: [task] }), + find: () => Effect.succeed(task), upsert: (input) => Ref.update(saved, (strategies) => [...strategies, input.workspaceStrategy]).pipe( Effect.andThen( diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 074b19fad3fec..304ab20e822b4 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1964,15 +1964,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 10a6f3c30c602..29371976fe865 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -179,6 +179,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 602031290763d..d8395b923916c 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -700,13 +700,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 }, @@ -724,34 +719,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, @@ -759,9 +801,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( @@ -775,8 +818,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 9dd04b41c9275..2d8070e2cb935 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -254,6 +254,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, @@ -2476,6 +2477,7 @@ export const layer = Layer.effect( return ScheduledTaskService.of({ list, + find: findTask, subscribeList, upsert, setEnabled,