From 37ad460edb609473e2523d0bf73ca6c20660821b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 19:34:50 +0000 Subject: [PATCH 1/9] test(worker): stub SCHEDULER_DO in oauth-handlers workers-unit env createEnv omitted SCHEDULER_DO after EnvSchema required it, so getEnv() threw and resolveSessionEmail swallowed the error as a null session. Co-authored-by: Kent C. Dodds --- packages/worker/src/oauth-handlers.workers.test.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/packages/worker/src/oauth-handlers.workers.test.ts b/packages/worker/src/oauth-handlers.workers.test.ts index 79c453680e..d4cd00d518 100644 --- a/packages/worker/src/oauth-handlers.workers.test.ts +++ b/packages/worker/src/oauth-handlers.workers.test.ts @@ -79,6 +79,17 @@ async function createDatabase(password: string) { } as unknown as D1Database } +function mockSchedulerDoNamespace(): DurableObjectNamespace { + return { + idFromName() { + return { toString: () => 'scheduler-do-test-id' } as DurableObjectId + }, + get() { + return {} as DurableObjectStub + }, + } as unknown as DurableObjectNamespace +} + function createEnv( helpers: OAuthHelpers, appDb?: D1Database, @@ -89,6 +100,7 @@ function createEnv( OAUTH_PROVIDER: helpers, APP_DB: resolvedDb, COOKIE_SECRET: cookieSecretValue, + SCHEDULER_DO: mockSchedulerDoNamespace(), } as unknown as Env } From 21d6a7db66b42a644ffc73287de303ee355e53d1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 19:36:07 +0000 Subject: [PATCH 2/9] Add codemode scheduler durable object Co-authored-by: Kent C. Dodds --- docs/use/execute.md | 18 + package-lock.json | 20 + package.json | 1 + packages/worker/src/env-schema.ts | 6 + packages/worker/src/index.ts | 10 +- .../src/mcp/capabilities/builtin-domains.ts | 2 + .../src/mcp/capabilities/domain-metadata.ts | 1 + .../src/mcp/capabilities/scheduler/domain.ts | 23 ++ .../src/mcp/capabilities/scheduler/index.ts | 7 + .../scheduler/scheduler-create.ts | 28 ++ .../scheduler/scheduler-delete.ts | 28 ++ .../capabilities/scheduler/scheduler-get.ts | 24 ++ .../capabilities/scheduler/scheduler-list.ts | 29 ++ .../scheduler/scheduler-run-now.ts | 31 ++ .../scheduler/scheduler-update.ts | 39 ++ .../src/mcp/capabilities/scheduler/shared.ts | 133 +++++++ .../worker/src/mcp/server-instructions.ts | 1 + packages/worker/src/mcp/tools/execute.ts | 4 + packages/worker/src/mcp/tools/search.ts | 2 + packages/worker/src/scheduler/client.ts | 112 ++++++ .../worker/src/scheduler/process-due-jobs.ts | 64 ++++ packages/worker/src/scheduler/schedule.ts | 123 +++++++ packages/worker/src/scheduler/scheduler-do.ts | 344 ++++++++++++++++++ packages/worker/src/scheduler/types.ts | 70 ++++ packages/worker/wrangler.jsonc | 16 + 25 files changed, 1135 insertions(+), 1 deletion(-) create mode 100644 packages/worker/src/mcp/capabilities/scheduler/domain.ts create mode 100644 packages/worker/src/mcp/capabilities/scheduler/index.ts create mode 100644 packages/worker/src/mcp/capabilities/scheduler/scheduler-create.ts create mode 100644 packages/worker/src/mcp/capabilities/scheduler/scheduler-delete.ts create mode 100644 packages/worker/src/mcp/capabilities/scheduler/scheduler-get.ts create mode 100644 packages/worker/src/mcp/capabilities/scheduler/scheduler-list.ts create mode 100644 packages/worker/src/mcp/capabilities/scheduler/scheduler-run-now.ts create mode 100644 packages/worker/src/mcp/capabilities/scheduler/scheduler-update.ts create mode 100644 packages/worker/src/mcp/capabilities/scheduler/shared.ts create mode 100644 packages/worker/src/scheduler/client.ts create mode 100644 packages/worker/src/scheduler/process-due-jobs.ts create mode 100644 packages/worker/src/scheduler/schedule.ts create mode 100644 packages/worker/src/scheduler/scheduler-do.ts create mode 100644 packages/worker/src/scheduler/types.ts diff --git a/docs/use/execute.md b/docs/use/execute.md index 3c7e82fb98..ab915277aa 100644 --- a/docs/use/execute.md +++ b/docs/use/execute.md @@ -32,6 +32,24 @@ To run persisted user code by name, use **`meta_run_skill`** with **`name`** and optional **`params`**. To inspect source, use **`meta_get_skill`**. You can also inline saved skill code into **execute** when that fits the workflow. +## Scheduled jobs + +Kody can persist scheduled codemode jobs per user: + +- **`scheduler_create`** — create a one-shot or recurring cron job +- **`scheduler_list`** — list jobs with next run time, last run status, and a + human-readable schedule summary +- **`scheduler_get`** — inspect one job +- **`scheduler_update`** — change code, params, schedule, timezone, name, or + enabled state +- **`scheduler_delete`** — remove a job +- **`scheduler_run_now`** — trigger a job immediately without changing its normal + schedule + +Cron schedules use standard **5-field cron syntax** and may include an IANA +timezone such as **`America/New_York`**. One-shot schedules use an ISO 8601 UTC +timestamp such as **`2026-04-17T15:00:00Z`**. + ## Long-term memory Kody can surface a small number of relevant long-term memories when you pass a diff --git a/package-lock.json b/package-lock.json index 76b853ade7..9d69a699d7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -27,6 +27,7 @@ "acorn": "^8.16.0", "agents": "^0.7.6", "bonjour": "^3.5.1", + "croner": "^10.0.1", "get-port": "^7.1.0", "remix": "3.0.0-alpha.3", "workers-ai-provider": "^3.1.2", @@ -11391,6 +11392,25 @@ "node": ">=20" } }, + "node_modules/croner": { + "version": "10.0.1", + "resolved": "https://registry.npmjs.org/croner/-/croner-10.0.1.tgz", + "integrity": "sha512-ixNtAJndqh173VQ4KodSdJEI6nuioBWI0V1ITNKhZZsO0pEMoDxz539T4FTTbSZ/xIOSuDnzxLVRqBVSvPNE2g==", + "funding": [ + { + "type": "other", + "url": "https://paypal.me/hexagonpp" + }, + { + "type": "github", + "url": "https://github.com/sponsors/hexagon" + } + ], + "license": "MIT", + "engines": { + "node": ">=18.0" + } + }, "node_modules/cross-spawn": { "version": "7.0.6", "license": "MIT", diff --git a/package.json b/package.json index c74b772758..f19fffce7b 100644 --- a/package.json +++ b/package.json @@ -94,6 +94,7 @@ "acorn": "^8.16.0", "agents": "^0.7.6", "bonjour": "^3.5.1", + "croner": "^10.0.1", "get-port": "^7.1.0", "remix": "3.0.0-alpha.3", "workers-ai-provider": "^3.1.2", diff --git a/packages/worker/src/env-schema.ts b/packages/worker/src/env-schema.ts index b460642d89..fbf808c48d 100644 --- a/packages/worker/src/env-schema.ts +++ b/packages/worker/src/env-schema.ts @@ -165,6 +165,12 @@ export const EnvSchema = object({ 'COOKIE_SECRET must be at least 32 characters for session signing.', ), APP_DB: d1DatabaseSchema, + SCHEDULER_DO: createSchema((value, context) => { + if (value) { + return { value: value as DurableObjectNamespace } + } + return fail('Missing SCHEDULER_DO binding for scheduled codemode jobs.', context.path) + }), APP_BASE_URL: optionalUrlStringSchema, APP_COMMIT_SHA: optionalCommitShaSchema, CLOUDFLARE_EMAIL_FROM: optionalNonEmptyStringSchema, diff --git a/packages/worker/src/index.ts b/packages/worker/src/index.ts index cd779f64dc..3fba49569d 100644 --- a/packages/worker/src/index.ts +++ b/packages/worker/src/index.ts @@ -4,6 +4,7 @@ import { ChatAgent } from './chat-agent.ts' import { HomeConnectorSession } from './home/session.ts' import { HomeMCP } from './home/mcp.ts' import { MCP } from './mcp/index.ts' +import { SchedulerDO } from './scheduler/scheduler-do.ts' import { chatAgentBasePath } from '@kody-internal/shared/chat-routes.ts' import { getWorkerSentryOptions } from './sentry-options.ts' import { handleRequest } from '#app/handler.ts' @@ -38,7 +39,14 @@ import { parseConnectorRoutePath, } from './remote-connector/connector-session-key.ts' -export { ChatAgent, CodemodeFetchGateway, HomeConnectorSession, HomeMCP, MCP } +export { + ChatAgent, + CodemodeFetchGateway, + HomeConnectorSession, + HomeMCP, + MCP, + SchedulerDO, +} const claudeWidgetDomainSuffix = '.claudemcpcontent.com' diff --git a/packages/worker/src/mcp/capabilities/builtin-domains.ts b/packages/worker/src/mcp/capabilities/builtin-domains.ts index aefe636b3e..b6c062bf08 100644 --- a/packages/worker/src/mcp/capabilities/builtin-domains.ts +++ b/packages/worker/src/mcp/capabilities/builtin-domains.ts @@ -1,6 +1,7 @@ import { appsDomain } from './apps/domain.ts' import { codingDomain } from './coding/domain.ts' import { metaDomain } from './meta/domain.ts' +import { schedulerDomain } from './scheduler/domain.ts' import { secretsDomain } from './secrets/domain.ts' import { valuesDomain } from './values/domain.ts' @@ -16,6 +17,7 @@ export const builtinDomains = [ appsDomain, codingDomain, metaDomain, + schedulerDomain, secretsDomain, valuesDomain, ] as const diff --git a/packages/worker/src/mcp/capabilities/domain-metadata.ts b/packages/worker/src/mcp/capabilities/domain-metadata.ts index 21707d73dc..96b05b30fd 100644 --- a/packages/worker/src/mcp/capabilities/domain-metadata.ts +++ b/packages/worker/src/mcp/capabilities/domain-metadata.ts @@ -5,6 +5,7 @@ export const capabilityDomainNames = { home: 'home', math: 'math', meta: 'meta', + scheduler: 'scheduler', secrets: 'secrets', values: 'values', } as const diff --git a/packages/worker/src/mcp/capabilities/scheduler/domain.ts b/packages/worker/src/mcp/capabilities/scheduler/domain.ts new file mode 100644 index 0000000000..523d31d35b --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/domain.ts @@ -0,0 +1,23 @@ +import { defineDomain } from '../define-domain.ts' +import { capabilityDomainNames } from '../domain-metadata.ts' +import { schedulerCreateCapability } from './scheduler-create.ts' +import { schedulerDeleteCapability } from './scheduler-delete.ts' +import { schedulerGetCapability } from './scheduler-get.ts' +import { schedulerListCapability } from './scheduler-list.ts' +import { schedulerRunNowCapability } from './scheduler-run-now.ts' +import { schedulerUpdateCapability } from './scheduler-update.ts' + +export const schedulerDomain = defineDomain({ + name: capabilityDomainNames.scheduler, + description: + 'Create, inspect, update, delete, and trigger scheduled codemode jobs that run once at a UTC timestamp or recur on a timezone-aware cron schedule.', + keywords: ['schedule', 'scheduler', 'cron', 'datetime', 'run later', 'job'], + capabilities: [ + schedulerCreateCapability, + schedulerListCapability, + schedulerGetCapability, + schedulerUpdateCapability, + schedulerDeleteCapability, + schedulerRunNowCapability, + ], +}) diff --git a/packages/worker/src/mcp/capabilities/scheduler/index.ts b/packages/worker/src/mcp/capabilities/scheduler/index.ts new file mode 100644 index 0000000000..468847188d --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/index.ts @@ -0,0 +1,7 @@ +export { schedulerDomain } from './domain.ts' +export { schedulerCreateCapability } from './scheduler-create.ts' +export { schedulerDeleteCapability } from './scheduler-delete.ts' +export { schedulerGetCapability } from './scheduler-get.ts' +export { schedulerListCapability } from './scheduler-list.ts' +export { schedulerRunNowCapability } from './scheduler-run-now.ts' +export { schedulerUpdateCapability } from './scheduler-update.ts' diff --git a/packages/worker/src/mcp/capabilities/scheduler/scheduler-create.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-create.ts new file mode 100644 index 0000000000..ca299e4fa5 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/scheduler-create.ts @@ -0,0 +1,28 @@ +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { type CapabilityContext } from '#mcp/capabilities/types.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { schedulerCreate } from '#worker/scheduler/client.ts' +import { schedulerCreateInputSchema, schedulerJobViewSchema } from './shared.ts' + +export const schedulerCreateCapability = defineDomainCapability( + capabilityDomainNames.scheduler, + { + name: 'scheduler_create', + description: + 'Create a scheduled codemode job for the signed-in user. Supports one-shot UTC timestamps or recurring 5-field cron expressions with IANA timezone interpretation.', + keywords: ['schedule', 'cron', 'datetime', 'job', 'automation'], + readOnly: false, + idempotent: false, + destructive: false, + inputSchema: schedulerCreateInputSchema, + outputSchema: schedulerJobViewSchema, + async handler(args, ctx: CapabilityContext) { + const user = requireMcpUser(ctx.callerContext) + return schedulerCreate(ctx.env, user.userId, { + callerContext: ctx.callerContext, + body: args, + }) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/scheduler/scheduler-delete.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-delete.ts new file mode 100644 index 0000000000..d5ae46f3ec --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/scheduler-delete.ts @@ -0,0 +1,28 @@ +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { type CapabilityContext } from '#mcp/capabilities/types.ts' +import { schedulerDelete } from '#worker/scheduler/client.ts' +import { + scheduledJobIdInputSchema, + schedulerDeleteOutputSchema, +} from './shared.ts' +import { requireMcpUser } from '../meta/require-user.ts' + +export const schedulerDeleteCapability = defineDomainCapability( + capabilityDomainNames.scheduler, + { + name: 'scheduler_delete', + description: + 'Delete a scheduled job for the signed-in user by id and update the Durable Object alarm to the next pending run.', + keywords: ['scheduler', 'delete', 'job', 'remove'], + readOnly: false, + idempotent: false, + destructive: true, + inputSchema: scheduledJobIdInputSchema, + outputSchema: schedulerDeleteOutputSchema, + async handler(args, ctx: CapabilityContext) { + const user = requireMcpUser(ctx.callerContext) + return schedulerDelete(ctx.env, user.userId, args.id) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/scheduler/scheduler-get.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-get.ts new file mode 100644 index 0000000000..bed6693f91 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/scheduler-get.ts @@ -0,0 +1,24 @@ +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { schedulerGet } from '#worker/scheduler/client.ts' +import { scheduledJobIdInputSchema, schedulerJobViewSchema } from './shared.ts' + +export const schedulerGetCapability = defineDomainCapability( + capabilityDomainNames.scheduler, + { + name: 'scheduler_get', + description: + 'Get one scheduled codemode job by id for the signed-in user, including next run time and last execution state.', + keywords: ['schedule', 'scheduler', 'job', 'get'], + readOnly: true, + idempotent: true, + destructive: false, + inputSchema: scheduledJobIdInputSchema, + outputSchema: schedulerJobViewSchema, + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + return schedulerGet(ctx.env, user.userId, args.id) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/scheduler/scheduler-list.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-list.ts new file mode 100644 index 0000000000..53ce4aeba5 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/scheduler-list.ts @@ -0,0 +1,29 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { type CapabilityContext } from '#mcp/capabilities/types.ts' +import { schedulerList } from '#worker/scheduler/client.ts' +import { + requireSchedulerUser, + schedulerCapabilityKeywords, + schedulerJobViewSchema, +} from './shared.ts' + +export const schedulerListCapability = defineDomainCapability( + capabilityDomainNames.scheduler, + { + name: 'scheduler_list', + description: + 'List all scheduled jobs for the signed-in user, including next run time, last run status, and a human-readable schedule summary.', + keywords: [...schedulerCapabilityKeywords], + readOnly: true, + idempotent: true, + destructive: false, + inputSchema: z.object({}), + outputSchema: z.array(schedulerJobViewSchema), + async handler(_args, ctx: CapabilityContext) { + const user = requireSchedulerUser(ctx) + return schedulerList(ctx.env, user.userId) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/scheduler/scheduler-run-now.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-run-now.ts new file mode 100644 index 0000000000..c2d8a29a75 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/scheduler-run-now.ts @@ -0,0 +1,31 @@ +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { type CapabilityContext } from '#mcp/capabilities/types.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { schedulerRunNow } from '#worker/scheduler/client.ts' +import { + scheduledJobIdInputSchema, + schedulerRunNowOutputSchema, +} from './shared.ts' + +export const schedulerRunNowCapability = defineDomainCapability( + capabilityDomainNames.scheduler, + { + name: 'scheduler_run_now', + description: + 'Trigger a scheduled job immediately outside its normal schedule and return the execution result plus the updated job state.', + keywords: ['schedule', 'run now', 'trigger', 'job'], + readOnly: false, + idempotent: false, + destructive: false, + inputSchema: scheduledJobIdInputSchema, + outputSchema: schedulerRunNowOutputSchema, + async handler(args, ctx: CapabilityContext) { + const user = requireMcpUser(ctx.callerContext) + return schedulerRunNow(ctx.env, user.userId, { + callerContext: ctx.callerContext, + body: { id: args.id }, + }) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/scheduler/scheduler-update.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-update.ts new file mode 100644 index 0000000000..a570cf408d --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/scheduler-update.ts @@ -0,0 +1,39 @@ +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { type CapabilityContext } from '#mcp/capabilities/types.ts' +import { schedulerUpdate } from '#worker/scheduler/client.ts' +import { + requireSchedulerUser, + schedulerJobViewSchema, + schedulerUpdateInputSchema, +} from './shared.ts' + +export const schedulerUpdateCapability = defineDomainCapability( + capabilityDomainNames.scheduler, + { + name: 'scheduler_update', + description: + 'Update a scheduled codemode job by id. You can change the name, code, params, schedule, timezone, or enabled flag. Changing the schedule recomputes nextRunAt immediately.', + keywords: ['scheduler', 'update', 'cron', 'datetime', 'job'], + readOnly: false, + idempotent: false, + destructive: false, + inputSchema: schedulerUpdateInputSchema, + outputSchema: schedulerJobViewSchema, + async handler(args, ctx: CapabilityContext) { + const user = requireSchedulerUser(ctx) + return schedulerUpdate(ctx.env, user.userId, { + callerContext: ctx.callerContext, + body: { + id: args.id, + ...(args.name !== undefined ? { name: args.name } : {}), + ...(args.code !== undefined ? { code: args.code } : {}), + ...(args.params !== undefined ? { params: args.params } : {}), + ...(args.schedule !== undefined ? { schedule: args.schedule } : {}), + ...(args.timezone !== undefined ? { timezone: args.timezone } : {}), + ...(args.enabled !== undefined ? { enabled: args.enabled } : {}), + }, + }) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/scheduler/shared.ts b/packages/worker/src/mcp/capabilities/scheduler/shared.ts new file mode 100644 index 0000000000..38e7fc3018 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/shared.ts @@ -0,0 +1,133 @@ +import { z } from 'zod' +import { + type ScheduledJobView, + type SchedulerCreateInput, + type SchedulerExecutionResult, + type SchedulerUpdateInput, +} from '#worker/scheduler/types.ts' +import { type CapabilityContext } from '#mcp/capabilities/types.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' + +export const schedulerScheduleSchema = z.discriminatedUnion('type', [ + z.object({ + type: z.literal('cron'), + expression: z + .string() + .min(1) + .describe( + 'Standard 5-field cron expression: minute hour day-of-month month day-of-week.', + ), + }), + z.object({ + type: z.literal('once'), + runAt: z + .string() + .min(1) + .describe('ISO 8601 UTC timestamp for a one-shot run.'), + }), +]) + +export const schedulerJobViewSchema = z.object({ + id: z.string(), + name: z.string(), + code: z.string(), + params: z.record(z.string(), z.unknown()).optional(), + schedule: schedulerScheduleSchema, + timezone: z.string(), + enabled: z.boolean(), + createdAt: z.string(), + lastRunAt: z.string().optional(), + lastRunStatus: z.enum(['success', 'error']).optional(), + lastRunError: z.string().optional(), + nextRunAt: z.string(), + scheduleSummary: z.string(), +}) satisfies z.ZodType + +export const schedulerCapabilityKeywords = [ + 'schedule', + 'scheduler', + 'cron', + 'datetime', + 'job', +] as const + +export const scheduledJobIdInputSchema = z.object({ + id: z.string().min(1).describe('Identifier of the scheduled job.'), +}) + +export const schedulerCreateInputSchema = z.object({ + name: z.string().min(1).describe('Human-friendly job name.'), + code: z + .string() + .min(1) + .describe('Codemode async arrow function source to execute on schedule.'), + params: z + .record(z.string(), z.unknown()) + .optional() + .describe('Optional params injected into the scheduled codemode execution.'), + schedule: schedulerScheduleSchema, + timezone: z + .string() + .optional() + .describe( + 'Optional IANA timezone used when interpreting cron expressions. Defaults to UTC.', + ), + enabled: z + .boolean() + .optional() + .describe('Whether the new job should be enabled immediately. Defaults to true.'), +}) satisfies z.ZodType + +export const schedulerUpdateInputSchema = z.object({ + id: z.string().min(1).describe('Existing scheduled job id.'), + name: z.string().min(1).optional().describe('Optional new job name.'), + code: z + .string() + .min(1) + .optional() + .describe('Optional replacement codemode source.'), + params: z + .record(z.string(), z.unknown()) + .nullable() + .optional() + .describe( + 'Optional replacement params. Pass null to clear previously stored params.', + ), + schedule: schedulerScheduleSchema.optional(), + timezone: z + .string() + .nullable() + .optional() + .describe('Optional replacement IANA timezone. Pass null to reset to UTC.'), + enabled: z + .boolean() + .optional() + .describe('Optional replacement enabled flag.'), +}) satisfies z.ZodType + +export const schedulerExecutionResultSchema = z.discriminatedUnion('ok', [ + z.object({ + ok: z.literal(true), + result: z.unknown().optional(), + logs: z.array(z.string()), + }), + z.object({ + ok: z.literal(false), + error: z.string(), + logs: z.array(z.string()), + }), +]) satisfies z.ZodType + +export const schedulerRunNowOutputSchema = z.object({ + job: schedulerJobViewSchema, + execution: schedulerExecutionResultSchema, +}) + +export const schedulerDeleteOutputSchema = z.object({ + id: z.string(), + deleted: z.literal(true), +}) + +export function requireSchedulerUser(ctx: CapabilityContext) { + return requireMcpUser(ctx.callerContext) +} diff --git a/packages/worker/src/mcp/server-instructions.ts b/packages/worker/src/mcp/server-instructions.ts index c5bcd651d9..29d6a02172 100644 --- a/packages/worker/src/mcp/server-instructions.ts +++ b/packages/worker/src/mcp/server-instructions.ts @@ -22,6 +22,7 @@ Conventions - \`memoryContext\`: short and task-focused. Kody may use it to surface a few relevant long-term memories and suppress repeats within the same \`conversationId\`. - Do not ask the user to paste secrets in chat; use saved secrets or \`open_generated_ui\`. - \`meta_save_skill\`: repeatable workflows only; optional \`collection\`; same name replaces an existing skill. One-off work: \`execute\`. Skill params: pass via \`meta_run_skill\` → \`params\` in codemode. +- \`scheduler_create\` / \`scheduler_update\`: schedule codemode to run later or on a recurring cron in an IANA timezone. Use \`scheduler_list\` / \`scheduler_get\` to inspect status and next run times, and \`scheduler_run_now\` to trigger the stored job immediately without changing its schedule. - \`ui_save_app\` / \`app_id\`: persisted UI artifacts (hidden from search unless \`hidden: false\`). \`codemode.secret_list\` / \`secret_set\`: metadata-only list; set only for values already in trusted execution (see \`execute\` tool description). - Memory writes are verify-first: always run \`meta_memory_verify\` before \`meta_memory_upsert\` or \`meta_memory_delete\`. Kody retrieves related memories; the consuming agent decides whether to upsert, delete, both, or do nothing. \`meta_memory_upsert\` creates a new memory when \`memory_id\` is omitted and updates an existing memory when \`memory_id\` is provided. - User-specific MCP instructions: \`meta_get_mcp_server_instructions\` / \`meta_set_mcp_server_instructions\` (signed-in users). Updates apply to **new** MCP sessions (reconnect to refresh what the host shows). diff --git a/packages/worker/src/mcp/tools/execute.ts b/packages/worker/src/mcp/tools/execute.ts index cad59e980b..84b29680c0 100644 --- a/packages/worker/src/mcp/tools/execute.ts +++ b/packages/worker/src/mcp/tools/execute.ts @@ -38,6 +38,10 @@ builtin capability). Discover names with \`search\`; for one capability’s Saved skills: prefer \`meta_run_skill({ name, params })\`, or \`meta_get_skill\` then paste code here. +Scheduled jobs: use \`scheduler_create\`, \`scheduler_list\`, \`scheduler_get\`, +\`scheduler_update\`, \`scheduler_delete\`, and \`scheduler_run_now\` to manage +one-shot or recurring codemode executions for the signed-in user. + Sandbox surface: - \`codemode\`: \`(args) => Promise\` per capability. - \`refreshAccessToken(providerName)\`, \`createAuthenticatedFetch(providerName)\` for OAuth connectors. diff --git a/packages/worker/src/mcp/tools/search.ts b/packages/worker/src/mcp/tools/search.ts index 4b03f25b21..8ecd7ecae2 100644 --- a/packages/worker/src/mcp/tools/search.ts +++ b/packages/worker/src/mcp/tools/search.ts @@ -136,6 +136,8 @@ empty ranked list. capabilities. Types and fields: see response. Run a skill: \`meta_run_skill({ name, params })\`; source: \`meta_get_skill\`. +Manage schedules: \`scheduler_create\`, \`scheduler_list\`, \`scheduler_get\`, +\`scheduler_update\`, \`scheduler_delete\`, \`scheduler_run_now\`. Apps: \`open_generated_ui({ app_id })\`. Secrets: never raw in results; use \`codemode.secret_list\` during execute and UI for missing values. Persisted values use \`codemode.value_get\` / \`codemode.value_list\`. Connectors diff --git a/packages/worker/src/scheduler/client.ts b/packages/worker/src/scheduler/client.ts new file mode 100644 index 0000000000..20a783f629 --- /dev/null +++ b/packages/worker/src/scheduler/client.ts @@ -0,0 +1,112 @@ +import { type McpCallerContext } from '@kody-internal/shared/chat.ts' +import { + type ScheduledJobView, + type SchedulerCreateInput, + type SchedulerExecutionResult, + type SchedulerUpdateInput, +} from './types.ts' + +type SchedulerMutationRequest = { + callerContext: McpCallerContext + body?: TBody +} + +type SchedulerRunNowResponse = { + job: ScheduledJobView + execution: SchedulerExecutionResult +} + +function getSchedulerStub(env: Env, userId: string) { + return env.SCHEDULER_DO.get(env.SCHEDULER_DO.idFromName(userId)) +} + +async function schedulerJsonRequest( + stub: DurableObjectStub, + input: { + method: string + pathname: string + body?: unknown + } +) { + const response = await stub.fetch(`https://scheduler.internal${input.pathname}`, { + method: input.method, + headers: input.body + ? { + 'Content-Type': 'application/json', + } + : undefined, + body: input.body === undefined ? undefined : JSON.stringify(input.body), + }) + if (!response.ok) { + throw new Error(await response.text()) + } + return (await response.json()) as TResult +} + +export async function schedulerCreate( + env: Env, + userId: string, + request: SchedulerMutationRequest, +) { + return schedulerJsonRequest(getSchedulerStub(env, userId), { + method: 'POST', + pathname: '/jobs', + body: request, + }) +} + +export async function schedulerList(env: Env, userId: string) { + return schedulerJsonRequest>(getSchedulerStub(env, userId), { + method: 'GET', + pathname: '/jobs', + }) +} + +export async function schedulerGet(env: Env, userId: string, jobId: string) { + return schedulerJsonRequest(getSchedulerStub(env, userId), { + method: 'GET', + pathname: `/jobs/${encodeURIComponent(jobId)}`, + }) +} + +export async function schedulerUpdate( + env: Env, + userId: string, + request: SchedulerMutationRequest, +) { + const id = request.body?.id + if (!id) { + throw new Error('Scheduler update requires a job id.') + } + return schedulerJsonRequest(getSchedulerStub(env, userId), { + method: 'PATCH', + pathname: `/jobs/${encodeURIComponent(id)}`, + body: request, + }) +} + +export async function schedulerDelete(env: Env, userId: string, jobId: string) { + return schedulerJsonRequest<{ id: string; deleted: true }>( + getSchedulerStub(env, userId), + { + method: 'DELETE', + pathname: `/jobs/${encodeURIComponent(jobId)}`, + }, + ) +} + +export async function schedulerRunNow( + env: Env, + userId: string, + request: SchedulerMutationRequest<{ id: string }>, +) { + const id = request.body?.id + if (!id) { + throw new Error('Scheduler run_now requires a job id.') + } + return schedulerJsonRequest(getSchedulerStub(env, userId), { + method: 'POST', + pathname: `/jobs/${encodeURIComponent(id)}/run-now`, + body: request, + }) +} diff --git a/packages/worker/src/scheduler/process-due-jobs.ts b/packages/worker/src/scheduler/process-due-jobs.ts new file mode 100644 index 0000000000..deea97b6c0 --- /dev/null +++ b/packages/worker/src/scheduler/process-due-jobs.ts @@ -0,0 +1,64 @@ +import { + computeNextRunAt, + formatSchedulerError, +} from './schedule.ts' +import { + type ScheduledJob, + type SchedulerExecutionResult, +} from './types.ts' + +type ProcessDueJobsResult = { + deleteJobIds: Array + saveJobs: Array +} + +export async function processDueJobs(input: { + jobs: Array + now?: Date + executeJob: (job: ScheduledJob) => Promise +}): Promise { + const now = input.now ?? new Date() + const lastRunAt = now.toISOString() + const deleteJobIds: Array = [] + const saveJobs: Array = [] + + for (const job of input.jobs) { + const execution = await input.executeJob(job).catch((error) => ({ + ok: false as const, + error: formatSchedulerError(error), + logs: [], + })) + + if (job.schedule.type === 'once') { + deleteJobIds.push(job.id) + continue + } + + try { + saveJobs.push({ + ...job, + lastRunAt, + lastRunStatus: execution.ok ? 'success' : 'error', + lastRunError: execution.ok ? undefined : execution.error, + nextRunAt: computeNextRunAt({ + schedule: job.schedule, + timezone: job.timezone, + from: now, + }), + }) + } catch (error) { + saveJobs.push({ + ...job, + enabled: false, + lastRunAt, + lastRunStatus: 'error', + lastRunError: `Failed to reschedule job: ${formatSchedulerError(error)}`, + }) + } + } + + return { + deleteJobIds, + saveJobs, + } +} diff --git a/packages/worker/src/scheduler/schedule.ts b/packages/worker/src/scheduler/schedule.ts new file mode 100644 index 0000000000..cb6a07a99d --- /dev/null +++ b/packages/worker/src/scheduler/schedule.ts @@ -0,0 +1,123 @@ +import { Cron } from 'croner' +import { type ScheduledJob, type ScheduledJobSchedule, type ScheduledJobView } from './types.ts' + +export const defaultSchedulerTimezone = 'UTC' + +const cronFieldPattern = /\s+/ + +export function formatSchedulerError(error: unknown): string { + if (typeof error === 'string') return error + if (error instanceof Error) return error.message + return String(error) +} + +export function normalizeSchedulerTimezone(timezone?: string | null) { + return timezone?.trim() || defaultSchedulerTimezone +} + +export function assertValidSchedulerTimezone(timezone: string) { + try { + new Intl.DateTimeFormat('en-US', { timeZone: timezone }).format(new Date()) + } catch { + throw new Error(`Invalid IANA timezone "${timezone}".`) + } +} + +export function assertValidCronExpression(expression: string) { + const trimmed = expression.trim() + const fields = trimmed.split(cronFieldPattern).filter(Boolean) + if (fields.length !== 5) { + throw new Error( + 'Cron expressions must use standard 5-field syntax: minute hour day-of-month month day-of-week.', + ) + } + return trimmed +} + +export function parseOnceRunAt(runAt: string) { + const trimmed = runAt.trim() + if (!trimmed) { + throw new Error('Once schedules require a non-empty runAt timestamp.') + } + if (!/(?:Z|[+-]00:00)$/i.test(trimmed)) { + throw new Error( + 'Once schedules must use an ISO 8601 UTC timestamp (for example 2026-04-17T15:00:00Z).', + ) + } + const parsed = new Date(trimmed) + if (Number.isNaN(parsed.valueOf())) { + throw new Error('Once schedules require a valid ISO 8601 timestamp.') + } + return parsed +} + +export function computeNextRunAt(input: { + schedule: ScheduledJobSchedule + timezone?: string | null + from?: Date | string | null +}) { + const timezone = normalizeSchedulerTimezone(input.timezone) + const from = + input.from instanceof Date + ? input.from + : input.from + ? new Date(input.from) + : new Date() + if (Number.isNaN(from.valueOf())) { + throw new Error('Cannot compute the next run time from an invalid reference date.') + } + if (input.schedule.type === 'once') { + return parseOnceRunAt(input.schedule.runAt).toISOString() + } + assertValidSchedulerTimezone(timezone) + const expression = assertValidCronExpression(input.schedule.expression) + const cron = new Cron(expression, { + paused: true, + timezone, + }) + const nextRun = cron.nextRun(from) + if (!nextRun) { + throw new Error('Cron expression does not produce a future run time.') + } + return nextRun.toISOString() +} + +export function normalizeScheduledJobSchedule(schedule: ScheduledJobSchedule) { + if (schedule.type === 'once') { + return { + type: 'once' as const, + runAt: parseOnceRunAt(schedule.runAt).toISOString(), + } + } + return { + type: 'cron' as const, + expression: assertValidCronExpression(schedule.expression), + } +} + +export function formatScheduleSummary(input: { + schedule: ScheduledJobSchedule + timezone?: string | null +}) { + const timezone = normalizeSchedulerTimezone(input.timezone) + if (input.schedule.type === 'once') { + return `Runs once at ${parseOnceRunAt(input.schedule.runAt).toISOString()}` + } + return `Runs on cron "${assertValidCronExpression(input.schedule.expression)}" in ${timezone}` +} + +export function isJobDue(job: ScheduledJob, now = new Date()) { + if (!job.enabled) return false + return new Date(job.nextRunAt).valueOf() <= now.valueOf() +} + +export function toScheduledJobView(job: ScheduledJob): ScheduledJobView { + return { + ...job, + timezone: normalizeSchedulerTimezone(job.timezone), + scheduleSummary: formatScheduleSummary({ + schedule: job.schedule, + timezone: job.timezone, + }), + } +} diff --git a/packages/worker/src/scheduler/scheduler-do.ts b/packages/worker/src/scheduler/scheduler-do.ts new file mode 100644 index 0000000000..000b004769 --- /dev/null +++ b/packages/worker/src/scheduler/scheduler-do.ts @@ -0,0 +1,344 @@ +import * as Sentry from '@sentry/cloudflare' +import { DurableObject, exports as workerExports } from 'cloudflare:workers' +import { type McpCallerContext } from '@kody-internal/shared/chat.ts' +import { buildSentryOptions } from '#worker/sentry-options.ts' +import { createMcpCallerContext, parseMcpCallerContext } from '#mcp/context.ts' +import { runCodemodeWithRegistry } from '#mcp/run-codemode-registry.ts' +import { processDueJobs } from './process-due-jobs.ts' +import { + computeNextRunAt, + formatSchedulerError, + normalizeScheduledJobSchedule, + normalizeSchedulerTimezone, + toScheduledJobView, +} from './schedule.ts' +import { + type PersistedSchedulerCallerContext, + type ScheduledJob, + type SchedulerCreateInput, + type SchedulerExecutionResult, + type SchedulerUpdateInput, +} from './types.ts' + +const callerContextStorageKey = 'scheduler:caller-context' +const jobStorageKeyPrefix = 'job:' + +function getJobStorageKey(jobId: string) { + return `${jobStorageKeyPrefix}${jobId}` +} + +function requirePersistableSchedulerCallerContext( + callerContext: McpCallerContext, +): PersistedSchedulerCallerContext { + if (!callerContext.user) { + throw new Error('Authenticated MCP user is required for scheduler operations.') + } + return createMcpCallerContext({ + baseUrl: callerContext.baseUrl, + user: callerContext.user, + homeConnectorId: callerContext.homeConnectorId ?? null, + remoteConnectors: callerContext.remoteConnectors ?? null, + storageContext: callerContext.storageContext ?? null, + }) as PersistedSchedulerCallerContext +} + +function normalizeJobName(name: string) { + const trimmed = name.trim() + if (!trimmed) { + throw new Error('Scheduled jobs require a non-empty name.') + } + return trimmed +} + +function normalizeJobCode(code: string) { + if (!code.trim()) { + throw new Error('Scheduled jobs require non-empty code.') + } + return code +} + +function normalizeOptionalParams( + params: Record | null | undefined, +): Record | undefined { + return params === null || params === undefined ? undefined : params +} + +class SchedulerDOBase extends DurableObject { + async fetch(request: Request): Promise { + try { + const url = new URL(request.url) + const segments = url.pathname.split('/').filter(Boolean) + if (segments.length === 1 && segments[0] === 'jobs') { + if (request.method === 'GET') { + return Response.json(await this.handleListJobs()) + } + if (request.method === 'POST') { + return Response.json(await this.handleCreateJob(request)) + } + } + if (segments.length === 2 && segments[0] === 'jobs') { + const jobId = decodeURIComponent(segments[1] ?? '') + if (request.method === 'GET') { + return Response.json(await this.handleGetJob(jobId)) + } + if (request.method === 'PATCH') { + return Response.json(await this.handleUpdateJob(jobId, request)) + } + if (request.method === 'DELETE') { + return Response.json(await this.handleDeleteJob(jobId)) + } + } + if ( + segments.length === 3 && + segments[0] === 'jobs' && + segments[2] === 'run-now' && + request.method === 'POST' + ) { + return Response.json( + await this.handleRunNow(decodeURIComponent(segments[1] ?? ''), request), + ) + } + return new Response('Not found', { status: 404 }) + } catch (error) { + return new Response(formatSchedulerError(error), { status: 400 }) + } + } + + async alarm(): Promise { + const now = new Date() + const dueJobs = (await this.listStoredJobs()).filter( + (job) => job.enabled && new Date(job.nextRunAt).valueOf() <= now.valueOf(), + ) + if (dueJobs.length === 0) { + await this.syncAlarm() + return + } + const callerContext = await this.getPersistedCallerContext() + const result = await processDueJobs({ + jobs: dueJobs, + now, + executeJob: async (job) => this.executeJob(job, callerContext), + }) + for (const job of result.saveJobs) { + await this.ctx.storage.put(getJobStorageKey(job.id), job) + } + for (const jobId of result.deleteJobIds) { + await this.ctx.storage.delete(getJobStorageKey(jobId)) + } + await this.syncAlarm() + } + + private async handleCreateJob(request: Request) { + const payload = await this.parseMutationRequest(request) + await this.persistCallerContext(payload.callerContext) + const timezone = normalizeSchedulerTimezone(payload.body.timezone) + const schedule = normalizeScheduledJobSchedule(payload.body.schedule) + const now = new Date().toISOString() + const job: ScheduledJob = { + id: crypto.randomUUID(), + name: normalizeJobName(payload.body.name), + code: normalizeJobCode(payload.body.code), + params: normalizeOptionalParams(payload.body.params), + schedule, + timezone, + enabled: payload.body.enabled ?? true, + createdAt: now, + nextRunAt: computeNextRunAt({ + schedule, + timezone, + }), + } + await this.ctx.storage.put(getJobStorageKey(job.id), job) + await this.syncAlarm() + return toScheduledJobView(job) + } + + private async handleListJobs() { + return (await this.listStoredJobs()).map((job) => toScheduledJobView(job)) + } + + private async handleGetJob(jobId: string) { + const job = await this.requireStoredJob(jobId) + return toScheduledJobView(job) + } + + private async handleUpdateJob(jobId: string, request: Request) { + const payload = await this.parseMutationRequest(request) + if (payload.body.id && payload.body.id !== jobId) { + throw new Error('Scheduler job id in the body must match the request path.') + } + await this.persistCallerContext(payload.callerContext) + const existing = await this.requireStoredJob(jobId) + const schedule = payload.body.schedule + ? normalizeScheduledJobSchedule(payload.body.schedule) + : existing.schedule + const timezone = + payload.body.timezone === null + ? normalizeSchedulerTimezone(null) + : normalizeSchedulerTimezone(payload.body.timezone ?? existing.timezone) + const updated: ScheduledJob = { + ...existing, + name: + payload.body.name === undefined + ? existing.name + : normalizeJobName(payload.body.name), + code: + payload.body.code === undefined + ? existing.code + : normalizeJobCode(payload.body.code), + params: + payload.body.params === undefined + ? existing.params + : normalizeOptionalParams(payload.body.params), + schedule, + timezone, + enabled: payload.body.enabled ?? existing.enabled, + nextRunAt: computeNextRunAt({ + schedule, + timezone, + }), + } + await this.ctx.storage.put(getJobStorageKey(jobId), updated) + await this.syncAlarm() + return toScheduledJobView(updated) + } + + private async handleDeleteJob(jobId: string) { + await this.requireStoredJob(jobId) + await this.ctx.storage.delete(getJobStorageKey(jobId)) + await this.syncAlarm() + return { + id: jobId, + deleted: true as const, + } + } + + private async handleRunNow(jobId: string, request: Request) { + const payload = await this.parseMutationRequest<{ id?: string }>(request) + if (payload.body.id && payload.body.id !== jobId) { + throw new Error('Scheduler job id in the body must match the request path.') + } + await this.persistCallerContext(payload.callerContext) + const existing = await this.requireStoredJob(jobId) + const execution = await this.executeJob(existing, payload.callerContext) + const updated: ScheduledJob = { + ...existing, + lastRunAt: new Date().toISOString(), + lastRunStatus: execution.ok ? 'success' : 'error', + lastRunError: execution.ok ? undefined : execution.error, + } + await this.ctx.storage.put(getJobStorageKey(jobId), updated) + return { + job: toScheduledJobView(updated), + execution, + } + } + + private async executeJob( + job: ScheduledJob, + callerContext: PersistedSchedulerCallerContext | null, + ): Promise { + if (!callerContext) { + return { + ok: false, + error: + 'Scheduler caller context is missing. Update or recreate a scheduled job to refresh its execution context.', + logs: [], + } + } + const execution = await runCodemodeWithRegistry( + this.env, + callerContext, + job.code, + job.params, + workerExports, + ) + if (execution.error) { + return { + ok: false, + error: formatSchedulerError(execution.error), + logs: execution.logs ?? [], + } + } + return { + ok: true, + result: execution.result, + logs: execution.logs ?? [], + } + } + + private async parseMutationRequest(request: Request) { + const raw = (await request.json()) as { + callerContext?: unknown + body?: TBody + } + const callerContext = requirePersistableSchedulerCallerContext( + parseMcpCallerContext(raw.callerContext), + ) + if (raw.body === undefined) { + throw new Error('Scheduler mutation requests require a JSON body.') + } + return { + callerContext, + body: raw.body, + } + } + + private async persistCallerContext(callerContext: PersistedSchedulerCallerContext) { + await this.ctx.storage.put(callerContextStorageKey, callerContext) + } + + private async getPersistedCallerContext() { + return ( + (await this.ctx.storage.get( + callerContextStorageKey, + )) ?? null + ) + } + + private async requireStoredJob(jobId: string) { + const job = await this.ctx.storage.get(getJobStorageKey(jobId)) + if (!job) { + throw new Error(`Scheduled job "${jobId}" was not found.`) + } + return { + ...job, + timezone: normalizeSchedulerTimezone(job.timezone), + } + } + + private async listStoredJobs() { + const jobs = [ + ...(await this.ctx.storage.list({ prefix: jobStorageKeyPrefix })) + .values(), + ].map((job) => ({ + ...job, + timezone: normalizeSchedulerTimezone(job.timezone), + })) + return jobs.sort((left, right) => { + const leftTime = new Date(left.nextRunAt).valueOf() + const rightTime = new Date(right.nextRunAt).valueOf() + if (leftTime !== rightTime) return leftTime - rightTime + return left.name.localeCompare(right.name) + }) + } + + private async syncAlarm() { + const enabledJobs = (await this.listStoredJobs()).filter((job) => job.enabled) + if (enabledJobs.length === 0) { + await this.ctx.storage.deleteAlarm() + return + } + const nextRunJob = enabledJobs.reduce((earliest, job) => + new Date(job.nextRunAt).valueOf() < new Date(earliest.nextRunAt).valueOf() + ? job + : earliest, + ) + await this.ctx.storage.setAlarm(new Date(nextRunJob.nextRunAt)) + } +} + +export const SchedulerDO = Sentry.instrumentDurableObjectWithSentry( + (env: Env) => buildSentryOptions(env), + SchedulerDOBase, +) diff --git a/packages/worker/src/scheduler/types.ts b/packages/worker/src/scheduler/types.ts new file mode 100644 index 0000000000..c27053d46c --- /dev/null +++ b/packages/worker/src/scheduler/types.ts @@ -0,0 +1,70 @@ +import { type McpCallerContext, type McpUserContext } from '@kody-internal/shared/chat.ts' + +export type ScheduledJobSchedule = + | { + type: 'cron' + expression: string + } + | { + type: 'once' + runAt: string + } + +export type ScheduledJobRunStatus = 'success' | 'error' + +export type ScheduledJob = { + id: string + name: string + code: string + params?: Record + schedule: ScheduledJobSchedule + timezone: string + enabled: boolean + createdAt: string + lastRunAt?: string + lastRunStatus?: ScheduledJobRunStatus + lastRunError?: string + nextRunAt: string +} + +export type ScheduledJobView = ScheduledJob & { + scheduleSummary: string +} + +export type SchedulerExecutionResult = + | { + ok: true + result?: unknown + logs: Array + } + | { + ok: false + error: string + logs: Array + } + +export type PersistedSchedulerCallerContext = Pick< + McpCallerContext, + 'baseUrl' | 'homeConnectorId' | 'remoteConnectors' | 'storageContext' +> & { + user: McpUserContext +} + +export type SchedulerCreateInput = { + name: string + code: string + params?: Record + schedule: ScheduledJobSchedule + timezone?: string | null + enabled?: boolean +} + +export type SchedulerUpdateInput = { + id: string + name?: string + code?: string + params?: Record | null + schedule?: ScheduledJobSchedule + timezone?: string | null + enabled?: boolean +} diff --git a/packages/worker/wrangler.jsonc b/packages/worker/wrangler.jsonc index 957c74981f..7b2b7c58ee 100644 --- a/packages/worker/wrangler.jsonc +++ b/packages/worker/wrangler.jsonc @@ -18,6 +18,10 @@ "new_sqlite_classes": ["HomeConnectorSession", "HomeMCP"], "tag": "v3", }, + { + "new_sqlite_classes": ["SchedulerDO"], + "tag": "v4", + }, ], "observability": { @@ -57,6 +61,10 @@ "class_name": "HomeMCP", "name": "HOME_MCP_OBJECT", }, + { + "class_name": "SchedulerDO", + "name": "SCHEDULER_DO", + }, ], }, "d1_databases": [ @@ -116,6 +124,10 @@ "class_name": "HomeMCP", "name": "HOME_MCP_OBJECT", }, + { + "class_name": "SchedulerDO", + "name": "SCHEDULER_DO", + }, ], }, "d1_databases": [ @@ -175,6 +187,10 @@ "class_name": "HomeMCP", "name": "HOME_MCP_OBJECT", }, + { + "class_name": "SchedulerDO", + "name": "SCHEDULER_DO", + }, ], }, "d1_databases": [ From db297331a215b0adc012b2c5924d55671edfc2f1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 19:42:27 +0000 Subject: [PATCH 3/9] Add scheduler coverage and test stubs Co-authored-by: Kent C. Dodds --- .../build-capability-registry.workers.test.ts | 18 +++ .../worker/src/mcp/mcp-server.mcp-e2e.test.ts | 130 ++++++++++++++++++ .../scheduler/process-due-jobs.node.test.ts | 81 +++++++++++ .../src/scheduler/schedule.node.test.ts | 54 ++++++++ packages/worker/src/scheduler/schedule.ts | 2 +- 5 files changed, 284 insertions(+), 1 deletion(-) create mode 100644 packages/worker/src/scheduler/process-due-jobs.node.test.ts create mode 100644 packages/worker/src/scheduler/schedule.node.test.ts diff --git a/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts b/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts index d803dbb22f..493b93e15a 100644 --- a/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts +++ b/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts @@ -74,6 +74,24 @@ test('buildCapabilityRegistry rejects duplicate domain registration', () => { ) }) +test('builtin capability domains include scheduler', async () => { + const { builtinDomains } = await import('./builtin-domains.ts') + expect(builtinDomains.some((domain) => domain.name === 'scheduler')).toBe(true) + const schedulerDomain = builtinDomains.find( + (domain) => domain.name === 'scheduler', + ) + expect( + schedulerDomain?.capabilities.map((capability) => capability.name), + ).toEqual([ + 'scheduler_create', + 'scheduler_list', + 'scheduler_get', + 'scheduler_update', + 'scheduler_delete', + 'scheduler_run_now', + ]) +}) + test('defineDomain rejects duplicate capability names within one domain', () => { const one = defineDomainCapability(capabilityDomainNames.apps, { name: 'dup', diff --git a/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts index 8addc332a4..c6d61749f5 100644 --- a/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts +++ b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts @@ -160,6 +160,136 @@ test('mcp server saves skills and run_skill works', async () => { expect(runStructured?.result?.result).toEqual({ ok: true }) }) +test('mcp server manages scheduled codemode jobs', async () => { + await using database = await createTestDatabase() + await using server = await startDevServer(database.persistDir) + await using mcpClient = await createMcpClient(server.origin, database.user) + + const createResult = await mcpClient.client.callTool({ + name: 'execute', + arguments: { + code: `async () => { + return await codemode.scheduler_create({ + name: 'daily-summary', + code: 'async () => ({ ran: true })', + schedule: { type: 'cron', expression: '0 7 * * *' }, + timezone: 'America/New_York', + }) + }`, + }, + }) + const createStructured = (createResult as CallToolResult).structuredContent as + | { result?: { id?: string; timezone?: string; scheduleSummary?: string } } + | undefined + const jobId = createStructured?.result?.id + expect(typeof jobId).toBe('string') + expect(createStructured?.result?.timezone).toBe('America/New_York') + expect(createStructured?.result?.scheduleSummary).toContain( + 'America/New_York', + ) + + const listResult = await mcpClient.client.callTool({ + name: 'execute', + arguments: { + code: `async () => { + return await codemode.scheduler_list({}) + }`, + }, + }) + const listStructured = (listResult as CallToolResult).structuredContent as + | { + result?: Array<{ + id?: string + nextRunAt?: string + scheduleSummary?: string + }> + } + | undefined + const listedJob = listStructured?.result?.find((job) => job.id === jobId) + expect(listedJob?.nextRunAt).toMatch(/Z$/) + expect(listedJob?.scheduleSummary).toContain('cron') + + const runNowResult = await mcpClient.client.callTool({ + name: 'execute', + arguments: { + code: `async () => { + return await codemode.scheduler_run_now({ id: ${JSON.stringify(jobId)} }) + }`, + }, + }) + const runNowStructured = (runNowResult as CallToolResult).structuredContent as + | { + result?: { + job?: { lastRunStatus?: string; lastRunAt?: string } + execution?: { ok?: boolean; result?: { ran?: boolean } } + } + } + | undefined + expect(runNowStructured?.result?.execution?.ok).toBe(true) + expect(runNowStructured?.result?.execution?.result).toEqual({ ran: true }) + expect(runNowStructured?.result?.job?.lastRunStatus).toBe('success') + expect(runNowStructured?.result?.job?.lastRunAt).toMatch(/Z$/) + + const updateResult = await mcpClient.client.callTool({ + name: 'execute', + arguments: { + code: `async () => { + return await codemode.scheduler_update({ + id: ${JSON.stringify(jobId)}, + enabled: false, + schedule: { type: 'once', runAt: '2026-04-18T15:00:00Z' }, + }) + }`, + }, + }) + const updateStructured = (updateResult as CallToolResult).structuredContent as + | { + result?: { + enabled?: boolean + nextRunAt?: string + schedule?: { type?: string; runAt?: string } + } + } + | undefined + expect(updateStructured?.result?.enabled).toBe(false) + expect(updateStructured?.result?.schedule).toEqual({ + type: 'once', + runAt: '2026-04-18T15:00:00.000Z', + }) + expect(updateStructured?.result?.nextRunAt).toBe('2026-04-18T15:00:00.000Z') + + const getResult = await mcpClient.client.callTool({ + name: 'execute', + arguments: { + code: `async () => { + return await codemode.scheduler_get({ id: ${JSON.stringify(jobId)} }) + }`, + }, + }) + const getStructured = (getResult as CallToolResult).structuredContent as + | { result?: { id?: string; enabled?: boolean; scheduleSummary?: string } } + | undefined + expect(getStructured?.result?.id).toBe(jobId) + expect(getStructured?.result?.enabled).toBe(false) + expect(getStructured?.result?.scheduleSummary).toContain('Runs once at') + + const deleteResult = await mcpClient.client.callTool({ + name: 'execute', + arguments: { + code: `async () => { + return await codemode.scheduler_delete({ id: ${JSON.stringify(jobId)} }) + }`, + }, + }) + const deleteStructured = (deleteResult as CallToolResult).structuredContent as + | { result?: { id?: string; deleted?: boolean } } + | undefined + expect(deleteStructured?.result).toEqual({ + id: jobId, + deleted: true, + }) +}) + test('mcp server ui_save_app hidden flag and search listing', async () => { await using database = await createTestDatabase() await using server = await startDevServer(database.persistDir) diff --git a/packages/worker/src/scheduler/process-due-jobs.node.test.ts b/packages/worker/src/scheduler/process-due-jobs.node.test.ts new file mode 100644 index 0000000000..6634e1b701 --- /dev/null +++ b/packages/worker/src/scheduler/process-due-jobs.node.test.ts @@ -0,0 +1,81 @@ +import { expect, test } from 'vitest' +import { processDueJobs } from './process-due-jobs.ts' +import { type ScheduledJob } from './types.ts' + +function createCronJob(overrides: Partial = {}): ScheduledJob { + return { + id: 'job-1', + name: 'Morning job', + code: 'async () => ({ ok: true })', + schedule: { + type: 'cron', + expression: '0 7 * * *', + }, + timezone: 'UTC', + enabled: true, + createdAt: '2026-04-12T00:00:00.000Z', + nextRunAt: '2026-04-12T07:00:00.000Z', + ...overrides, + } +} + +test('processDueJobs records failures without aborting later jobs', async () => { + const now = new Date('2026-04-12T07:00:00.000Z') + const first = createCronJob({ id: 'job-1' }) + const second = createCronJob({ id: 'job-2', name: 'Second job' }) + + const result = await processDueJobs({ + jobs: [first, second], + now, + async executeJob(job) { + if (job.id === 'job-1') { + throw new Error('boom') + } + return { + ok: true, + logs: ['ok'], + result: { ok: true }, + } + }, + }) + + expect(result.deleteJobIds).toEqual([]) + expect(result.saveJobs).toHaveLength(2) + expect(result.saveJobs[0]).toMatchObject({ + id: 'job-1', + lastRunStatus: 'error', + lastRunError: 'boom', + lastRunAt: now.toISOString(), + }) + expect(result.saveJobs[1]).toMatchObject({ + id: 'job-2', + lastRunStatus: 'success', + lastRunAt: now.toISOString(), + }) +}) + +test('processDueJobs deletes one-shot jobs after execution', async () => { + const onceJob = createCronJob({ + id: 'job-once', + schedule: { + type: 'once', + runAt: '2026-04-12T07:00:00.000Z', + }, + nextRunAt: '2026-04-12T07:00:00.000Z', + }) + + const result = await processDueJobs({ + jobs: [onceJob], + now: new Date('2026-04-12T07:00:00.000Z'), + async executeJob() { + return { + ok: false, + error: 'expected failure', + logs: [], + } + }, + }) + + expect(result.deleteJobIds).toEqual(['job-once']) + expect(result.saveJobs).toEqual([]) +}) diff --git a/packages/worker/src/scheduler/schedule.node.test.ts b/packages/worker/src/scheduler/schedule.node.test.ts new file mode 100644 index 0000000000..94483a2623 --- /dev/null +++ b/packages/worker/src/scheduler/schedule.node.test.ts @@ -0,0 +1,54 @@ +import { expect, test } from 'vitest' +import { + computeNextRunAt, + formatScheduleSummary, + normalizeScheduledJobSchedule, +} from './schedule.ts' + +test('computeNextRunAt handles timezone-aware cron schedules', () => { + const nextRunAt = computeNextRunAt({ + schedule: { + type: 'cron', + expression: '0 7 * * *', + }, + timezone: 'America/New_York', + from: '2026-03-08T10:30:00.000Z', + }) + + expect(nextRunAt).toBe('2026-03-08T11:00:00.000Z') +}) + +test('computeNextRunAt preserves UTC once schedules', () => { + const nextRunAt = computeNextRunAt({ + schedule: { + type: 'once', + runAt: '2026-04-17T15:00:00Z', + }, + }) + + expect(nextRunAt).toBe('2026-04-17T15:00:00.000Z') +}) + +test('normalizeScheduledJobSchedule trims cron expressions', () => { + expect( + normalizeScheduledJobSchedule({ + type: 'cron', + expression: ' */15 7 * * 1-5 ', + }), + ).toEqual({ + type: 'cron', + expression: '*/15 7 * * 1-5', + }) +}) + +test('formatScheduleSummary includes timezone for cron jobs', () => { + expect( + formatScheduleSummary({ + schedule: { + type: 'cron', + expression: '0 7 * * *', + }, + timezone: 'America/Los_Angeles', + }), + ).toBe('Runs on cron "0 7 * * *" in America/Los_Angeles') +}) diff --git a/packages/worker/src/scheduler/schedule.ts b/packages/worker/src/scheduler/schedule.ts index cb6a07a99d..f32dc24c64 100644 --- a/packages/worker/src/scheduler/schedule.ts +++ b/packages/worker/src/scheduler/schedule.ts @@ -31,7 +31,7 @@ export function assertValidCronExpression(expression: string) { 'Cron expressions must use standard 5-field syntax: minute hour day-of-month month day-of-week.', ) } - return trimmed + return fields.join(' ') } export function parseOnceRunAt(runAt: string) { From 61a7542b0ca2e24c235a6ef24c3e7a003a3eecd6 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 19:50:36 +0000 Subject: [PATCH 4/9] Use isJobDue in scheduler alarm Co-authored-by: Kent C. Dodds --- packages/worker/src/scheduler/scheduler-do.ts | 47 ++++++++++++++----- 1 file changed, 34 insertions(+), 13 deletions(-) diff --git a/packages/worker/src/scheduler/scheduler-do.ts b/packages/worker/src/scheduler/scheduler-do.ts index 000b004769..79dc124a7b 100644 --- a/packages/worker/src/scheduler/scheduler-do.ts +++ b/packages/worker/src/scheduler/scheduler-do.ts @@ -8,6 +8,7 @@ import { processDueJobs } from './process-due-jobs.ts' import { computeNextRunAt, formatSchedulerError, + isJobDue, normalizeScheduledJobSchedule, normalizeSchedulerTimezone, toScheduledJobView, @@ -31,7 +32,9 @@ function requirePersistableSchedulerCallerContext( callerContext: McpCallerContext, ): PersistedSchedulerCallerContext { if (!callerContext.user) { - throw new Error('Authenticated MCP user is required for scheduler operations.') + throw new Error( + 'Authenticated MCP user is required for scheduler operations.', + ) } return createMcpCallerContext({ baseUrl: callerContext.baseUrl, @@ -95,7 +98,10 @@ class SchedulerDOBase extends DurableObject { request.method === 'POST' ) { return Response.json( - await this.handleRunNow(decodeURIComponent(segments[1] ?? ''), request), + await this.handleRunNow( + decodeURIComponent(segments[1] ?? ''), + request, + ), ) } return new Response('Not found', { status: 404 }) @@ -106,8 +112,8 @@ class SchedulerDOBase extends DurableObject { async alarm(): Promise { const now = new Date() - const dueJobs = (await this.listStoredJobs()).filter( - (job) => job.enabled && new Date(job.nextRunAt).valueOf() <= now.valueOf(), + const dueJobs = (await this.listStoredJobs()).filter((job) => + isJobDue(job, now), ) if (dueJobs.length === 0) { await this.syncAlarm() @@ -129,7 +135,8 @@ class SchedulerDOBase extends DurableObject { } private async handleCreateJob(request: Request) { - const payload = await this.parseMutationRequest(request) + const payload = + await this.parseMutationRequest(request) await this.persistCallerContext(payload.callerContext) const timezone = normalizeSchedulerTimezone(payload.body.timezone) const schedule = normalizeScheduledJobSchedule(payload.body.schedule) @@ -163,9 +170,12 @@ class SchedulerDOBase extends DurableObject { } private async handleUpdateJob(jobId: string, request: Request) { - const payload = await this.parseMutationRequest(request) + const payload = + await this.parseMutationRequest(request) if (payload.body.id && payload.body.id !== jobId) { - throw new Error('Scheduler job id in the body must match the request path.') + throw new Error( + 'Scheduler job id in the body must match the request path.', + ) } await this.persistCallerContext(payload.callerContext) const existing = await this.requireStoredJob(jobId) @@ -216,7 +226,9 @@ class SchedulerDOBase extends DurableObject { private async handleRunNow(jobId: string, request: Request) { const payload = await this.parseMutationRequest<{ id?: string }>(request) if (payload.body.id && payload.body.id !== jobId) { - throw new Error('Scheduler job id in the body must match the request path.') + throw new Error( + 'Scheduler job id in the body must match the request path.', + ) } await this.persistCallerContext(payload.callerContext) const existing = await this.requireStoredJob(jobId) @@ -284,7 +296,9 @@ class SchedulerDOBase extends DurableObject { } } - private async persistCallerContext(callerContext: PersistedSchedulerCallerContext) { + private async persistCallerContext( + callerContext: PersistedSchedulerCallerContext, + ) { await this.ctx.storage.put(callerContextStorageKey, callerContext) } @@ -297,7 +311,9 @@ class SchedulerDOBase extends DurableObject { } private async requireStoredJob(jobId: string) { - const job = await this.ctx.storage.get(getJobStorageKey(jobId)) + const job = await this.ctx.storage.get( + getJobStorageKey(jobId), + ) if (!job) { throw new Error(`Scheduled job "${jobId}" was not found.`) } @@ -309,8 +325,11 @@ class SchedulerDOBase extends DurableObject { private async listStoredJobs() { const jobs = [ - ...(await this.ctx.storage.list({ prefix: jobStorageKeyPrefix })) - .values(), + ...( + await this.ctx.storage.list({ + prefix: jobStorageKeyPrefix, + }) + ).values(), ].map((job) => ({ ...job, timezone: normalizeSchedulerTimezone(job.timezone), @@ -324,7 +343,9 @@ class SchedulerDOBase extends DurableObject { } private async syncAlarm() { - const enabledJobs = (await this.listStoredJobs()).filter((job) => job.enabled) + const enabledJobs = (await this.listStoredJobs()).filter( + (job) => job.enabled, + ) if (enabledJobs.length === 0) { await this.ctx.storage.deleteAlarm() return From b924feb863e7c5ee74c037e1c37f9e5c4d111480 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 20:00:37 +0000 Subject: [PATCH 5/9] Fix scheduler update next run logic Co-authored-by: Kent C. Dodds --- .../src/mcp/capabilities/scheduler/index.ts | 7 ------- packages/worker/src/scheduler/scheduler-do.ts | 16 +++++++++++----- 2 files changed, 11 insertions(+), 12 deletions(-) delete mode 100644 packages/worker/src/mcp/capabilities/scheduler/index.ts diff --git a/packages/worker/src/mcp/capabilities/scheduler/index.ts b/packages/worker/src/mcp/capabilities/scheduler/index.ts deleted file mode 100644 index 468847188d..0000000000 --- a/packages/worker/src/mcp/capabilities/scheduler/index.ts +++ /dev/null @@ -1,7 +0,0 @@ -export { schedulerDomain } from './domain.ts' -export { schedulerCreateCapability } from './scheduler-create.ts' -export { schedulerDeleteCapability } from './scheduler-delete.ts' -export { schedulerGetCapability } from './scheduler-get.ts' -export { schedulerListCapability } from './scheduler-list.ts' -export { schedulerRunNowCapability } from './scheduler-run-now.ts' -export { schedulerUpdateCapability } from './scheduler-update.ts' diff --git a/packages/worker/src/scheduler/scheduler-do.ts b/packages/worker/src/scheduler/scheduler-do.ts index 79dc124a7b..2a75ead21f 100644 --- a/packages/worker/src/scheduler/scheduler-do.ts +++ b/packages/worker/src/scheduler/scheduler-do.ts @@ -179,13 +179,22 @@ class SchedulerDOBase extends DurableObject { } await this.persistCallerContext(payload.callerContext) const existing = await this.requireStoredJob(jobId) - const schedule = payload.body.schedule + const hasScheduleUpdate = payload.body.schedule !== undefined + const hasTimezoneUpdate = payload.body.timezone !== undefined + const schedule = hasScheduleUpdate ? normalizeScheduledJobSchedule(payload.body.schedule) : existing.schedule const timezone = payload.body.timezone === null ? normalizeSchedulerTimezone(null) : normalizeSchedulerTimezone(payload.body.timezone ?? existing.timezone) + const nextRunAt = + hasScheduleUpdate || hasTimezoneUpdate + ? computeNextRunAt({ + schedule, + timezone, + }) + : existing.nextRunAt const updated: ScheduledJob = { ...existing, name: @@ -203,10 +212,7 @@ class SchedulerDOBase extends DurableObject { schedule, timezone, enabled: payload.body.enabled ?? existing.enabled, - nextRunAt: computeNextRunAt({ - schedule, - timezone, - }), + nextRunAt, } await this.ctx.storage.put(getJobStorageKey(jobId), updated) await this.syncAlarm() From 393e02b34435883aefc26c3f0ae40d123dff3c17 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 20:09:43 +0000 Subject: [PATCH 6/9] Fix run-now for one-shot jobs Co-authored-by: Kent C. Dodds --- packages/worker/src/scheduler/scheduler-do.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/packages/worker/src/scheduler/scheduler-do.ts b/packages/worker/src/scheduler/scheduler-do.ts index 2a75ead21f..b3d6335b8f 100644 --- a/packages/worker/src/scheduler/scheduler-do.ts +++ b/packages/worker/src/scheduler/scheduler-do.ts @@ -245,7 +245,12 @@ class SchedulerDOBase extends DurableObject { lastRunStatus: execution.ok ? 'success' : 'error', lastRunError: execution.ok ? undefined : execution.error, } - await this.ctx.storage.put(getJobStorageKey(jobId), updated) + if (existing.schedule.type === 'once') { + await this.ctx.storage.delete(getJobStorageKey(jobId)) + await this.syncAlarm() + } else { + await this.ctx.storage.put(getJobStorageKey(jobId), updated) + } return { job: toScheduledJobView(updated), execution, From 6b5faf3d425d3f4b3d12405010ad2b1262de6096 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 20:34:55 +0000 Subject: [PATCH 7/9] test(mcp-e2e): pass memory id via execute params for delete/get Interpolating memory_id into generated codemode source triggered a workerd SWC internal error; using execute's params keeps user code stable. Co-authored-by: Kent C. Dodds --- packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts index c6d61749f5..d94ec326e5 100644 --- a/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts +++ b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts @@ -425,13 +425,14 @@ test('mcp memory verify upsert and delete', async () => { const softDeleteResult = await mcpClient.client.callTool({ name: 'execute', arguments: { - code: `async () => { + code: `async (params) => { return await codemode.meta_memory_delete({ - memory_id: ${JSON.stringify(memoryId)}, + memory_id: params.memoryId, verified_by_agent: true, verification_reference: 'verify-delete-1', }) }`, + params: { memoryId }, }, }) const softDeleteStructured = (softDeleteResult as CallToolResult) @@ -443,11 +444,12 @@ test('mcp memory verify upsert and delete', async () => { const getDeletedResult = await mcpClient.client.callTool({ name: 'execute', arguments: { - code: `async () => { + code: `async (params) => { return await codemode.meta_memory_get({ - memory_id: ${JSON.stringify(memoryId)}, + memory_id: params.memoryId, }) }`, + params: { memoryId }, }, }) const getDeletedStructured = (getDeletedResult as CallToolResult) From 73552e418b417d29077ac01b413501c2e1eeb276 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 21:09:53 +0000 Subject: [PATCH 8/9] Fix scheduler DO review feedback Co-authored-by: Kent C. Dodds --- .../worker/src/mcp/mcp-server.mcp-e2e.test.ts | 7 +- .../scheduler/process-due-jobs.node.test.ts | 26 ++++--- packages/worker/src/scheduler/scheduler-do.ts | 72 ++++++++++++------- 3 files changed, 67 insertions(+), 38 deletions(-) diff --git a/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts index d94ec326e5..3afcb9b364 100644 --- a/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts +++ b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts @@ -230,6 +230,7 @@ test('mcp server manages scheduled codemode jobs', async () => { expect(runNowStructured?.result?.job?.lastRunStatus).toBe('success') expect(runNowStructured?.result?.job?.lastRunAt).toMatch(/Z$/) + const futureRunAt = new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString() const updateResult = await mcpClient.client.callTool({ name: 'execute', arguments: { @@ -237,7 +238,7 @@ test('mcp server manages scheduled codemode jobs', async () => { return await codemode.scheduler_update({ id: ${JSON.stringify(jobId)}, enabled: false, - schedule: { type: 'once', runAt: '2026-04-18T15:00:00Z' }, + schedule: { type: 'once', runAt: ${JSON.stringify(futureRunAt)} }, }) }`, }, @@ -254,9 +255,9 @@ test('mcp server manages scheduled codemode jobs', async () => { expect(updateStructured?.result?.enabled).toBe(false) expect(updateStructured?.result?.schedule).toEqual({ type: 'once', - runAt: '2026-04-18T15:00:00.000Z', + runAt: futureRunAt, }) - expect(updateStructured?.result?.nextRunAt).toBe('2026-04-18T15:00:00.000Z') + expect(updateStructured?.result?.nextRunAt).toBe(futureRunAt) const getResult = await mcpClient.client.callTool({ name: 'execute', diff --git a/packages/worker/src/scheduler/process-due-jobs.node.test.ts b/packages/worker/src/scheduler/process-due-jobs.node.test.ts index 6634e1b701..94a08b7b41 100644 --- a/packages/worker/src/scheduler/process-due-jobs.node.test.ts +++ b/packages/worker/src/scheduler/process-due-jobs.node.test.ts @@ -41,17 +41,21 @@ test('processDueJobs records failures without aborting later jobs', async () => expect(result.deleteJobIds).toEqual([]) expect(result.saveJobs).toHaveLength(2) - expect(result.saveJobs[0]).toMatchObject({ - id: 'job-1', - lastRunStatus: 'error', - lastRunError: 'boom', - lastRunAt: now.toISOString(), - }) - expect(result.saveJobs[1]).toMatchObject({ - id: 'job-2', - lastRunStatus: 'success', - lastRunAt: now.toISOString(), - }) + expect(result.saveJobs).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + id: 'job-1', + lastRunStatus: 'error', + lastRunError: 'boom', + lastRunAt: now.toISOString(), + }), + expect.objectContaining({ + id: 'job-2', + lastRunStatus: 'success', + lastRunAt: now.toISOString(), + }), + ]), + ) }) test('processDueJobs deletes one-shot jobs after execution', async () => { diff --git a/packages/worker/src/scheduler/scheduler-do.ts b/packages/worker/src/scheduler/scheduler-do.ts index b3d6335b8f..507467030c 100644 --- a/packages/worker/src/scheduler/scheduler-do.ts +++ b/packages/worker/src/scheduler/scheduler-do.ts @@ -21,13 +21,16 @@ import { type SchedulerUpdateInput, } from './types.ts' -const callerContextStorageKey = 'scheduler:caller-context' const jobStorageKeyPrefix = 'job:' function getJobStorageKey(jobId: string) { return `${jobStorageKeyPrefix}${jobId}` } +function getJobCallerContextStorageKey(jobId: string) { + return `job-context:${jobId}` +} + function requirePersistableSchedulerCallerContext( callerContext: McpCallerContext, ): PersistedSchedulerCallerContext { @@ -119,17 +122,18 @@ class SchedulerDOBase extends DurableObject { await this.syncAlarm() return } - const callerContext = await this.getPersistedCallerContext() const result = await processDueJobs({ jobs: dueJobs, now, - executeJob: async (job) => this.executeJob(job, callerContext), + executeJob: async (job) => + this.executeJob(job, await this.getPersistedCallerContext(job.id)), }) for (const job of result.saveJobs) { await this.ctx.storage.put(getJobStorageKey(job.id), job) } for (const jobId of result.deleteJobIds) { await this.ctx.storage.delete(getJobStorageKey(jobId)) + await this.ctx.storage.delete(getJobCallerContextStorageKey(jobId)) } await this.syncAlarm() } @@ -137,7 +141,6 @@ class SchedulerDOBase extends DurableObject { private async handleCreateJob(request: Request) { const payload = await this.parseMutationRequest(request) - await this.persistCallerContext(payload.callerContext) const timezone = normalizeSchedulerTimezone(payload.body.timezone) const schedule = normalizeScheduledJobSchedule(payload.body.schedule) const now = new Date().toISOString() @@ -156,6 +159,7 @@ class SchedulerDOBase extends DurableObject { }), } await this.ctx.storage.put(getJobStorageKey(job.id), job) + await this.persistCallerContext(job.id, payload.callerContext) await this.syncAlarm() return toScheduledJobView(job) } @@ -177,19 +181,23 @@ class SchedulerDOBase extends DurableObject { 'Scheduler job id in the body must match the request path.', ) } - await this.persistCallerContext(payload.callerContext) const existing = await this.requireStoredJob(jobId) - const hasScheduleUpdate = payload.body.schedule !== undefined - const hasTimezoneUpdate = payload.body.timezone !== undefined - const schedule = hasScheduleUpdate - ? normalizeScheduledJobSchedule(payload.body.schedule) + const nextSchedule = payload.body.schedule + const nextTimezone = payload.body.timezone + const nextEnabled = payload.body.enabled ?? existing.enabled + const schedule = nextSchedule !== undefined + ? normalizeScheduledJobSchedule(nextSchedule) : existing.schedule const timezone = - payload.body.timezone === null + nextTimezone === null ? normalizeSchedulerTimezone(null) - : normalizeSchedulerTimezone(payload.body.timezone ?? existing.timezone) + : normalizeSchedulerTimezone(nextTimezone ?? existing.timezone) + const shouldRecomputeNextRunAt = + JSON.stringify(schedule) !== JSON.stringify(existing.schedule) || + timezone !== existing.timezone || + (existing.enabled === false && nextEnabled === true) const nextRunAt = - hasScheduleUpdate || hasTimezoneUpdate + shouldRecomputeNextRunAt ? computeNextRunAt({ schedule, timezone, @@ -211,10 +219,11 @@ class SchedulerDOBase extends DurableObject { : normalizeOptionalParams(payload.body.params), schedule, timezone, - enabled: payload.body.enabled ?? existing.enabled, + enabled: nextEnabled, nextRunAt, } await this.ctx.storage.put(getJobStorageKey(jobId), updated) + await this.persistCallerContext(jobId, payload.callerContext) await this.syncAlarm() return toScheduledJobView(updated) } @@ -222,6 +231,7 @@ class SchedulerDOBase extends DurableObject { private async handleDeleteJob(jobId: string) { await this.requireStoredJob(jobId) await this.ctx.storage.delete(getJobStorageKey(jobId)) + await this.ctx.storage.delete(getJobCallerContextStorageKey(jobId)) await this.syncAlarm() return { id: jobId, @@ -236,7 +246,6 @@ class SchedulerDOBase extends DurableObject { 'Scheduler job id in the body must match the request path.', ) } - await this.persistCallerContext(payload.callerContext) const existing = await this.requireStoredJob(jobId) const execution = await this.executeJob(existing, payload.callerContext) const updated: ScheduledJob = { @@ -247,9 +256,11 @@ class SchedulerDOBase extends DurableObject { } if (existing.schedule.type === 'once') { await this.ctx.storage.delete(getJobStorageKey(jobId)) + await this.ctx.storage.delete(getJobCallerContextStorageKey(jobId)) await this.syncAlarm() } else { await this.ctx.storage.put(getJobStorageKey(jobId), updated) + await this.persistCallerContext(jobId, payload.callerContext) } return { job: toScheduledJobView(updated), @@ -269,13 +280,22 @@ class SchedulerDOBase extends DurableObject { logs: [], } } - const execution = await runCodemodeWithRegistry( - this.env, - callerContext, - job.code, - job.params, - workerExports, - ) + let execution + try { + execution = await runCodemodeWithRegistry( + this.env, + callerContext, + job.code, + job.params, + workerExports, + ) + } catch (error) { + return { + ok: false, + error: formatSchedulerError(error), + logs: [], + } + } if (execution.error) { return { ok: false, @@ -308,15 +328,19 @@ class SchedulerDOBase extends DurableObject { } private async persistCallerContext( + jobId: string, callerContext: PersistedSchedulerCallerContext, ) { - await this.ctx.storage.put(callerContextStorageKey, callerContext) + await this.ctx.storage.put( + getJobCallerContextStorageKey(jobId), + callerContext, + ) } - private async getPersistedCallerContext() { + private async getPersistedCallerContext(jobId: string) { return ( (await this.ctx.storage.get( - callerContextStorageKey, + getJobCallerContextStorageKey(jobId), )) ?? null ) } From b43e83307d5b1f054e2a0e0170025f4fdfda0ebe Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 12 Apr 2026 21:42:36 +0000 Subject: [PATCH 9/9] Switch scheduler create/update to upsert Co-authored-by: Kent C. Dodds --- docs/use/execute.md | 5 +- .../build-capability-registry.workers.test.ts | 3 +- .../src/mcp/capabilities/scheduler/domain.ts | 8 +- .../scheduler/scheduler-create.ts | 28 ---- ...cheduler-update.ts => scheduler-upsert.ts} | 33 +++-- .../src/mcp/capabilities/scheduler/shared.ts | 123 ++++++++++-------- .../worker/src/mcp/mcp-server.mcp-e2e.test.ts | 26 ++-- .../worker/src/mcp/server-instructions.ts | 2 +- packages/worker/src/mcp/tools/execute.ts | 4 +- packages/worker/src/mcp/tools/search.ts | 4 +- packages/worker/src/scheduler/types.ts | 10 ++ 11 files changed, 133 insertions(+), 113 deletions(-) delete mode 100644 packages/worker/src/mcp/capabilities/scheduler/scheduler-create.ts rename packages/worker/src/mcp/capabilities/scheduler/{scheduler-update.ts => scheduler-upsert.ts} (50%) diff --git a/docs/use/execute.md b/docs/use/execute.md index ab915277aa..73a9810420 100644 --- a/docs/use/execute.md +++ b/docs/use/execute.md @@ -36,12 +36,11 @@ inline saved skill code into **execute** when that fits the workflow. Kody can persist scheduled codemode jobs per user: -- **`scheduler_create`** — create a one-shot or recurring cron job +- **`scheduler_upsert`** — create a new one-shot or recurring cron job, or + update an existing one when you pass **`id`** - **`scheduler_list`** — list jobs with next run time, last run status, and a human-readable schedule summary - **`scheduler_get`** — inspect one job -- **`scheduler_update`** — change code, params, schedule, timezone, name, or - enabled state - **`scheduler_delete`** — remove a job - **`scheduler_run_now`** — trigger a job immediately without changing its normal schedule diff --git a/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts b/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts index 493b93e15a..1f979f64eb 100644 --- a/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts +++ b/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts @@ -83,10 +83,9 @@ test('builtin capability domains include scheduler', async () => { expect( schedulerDomain?.capabilities.map((capability) => capability.name), ).toEqual([ - 'scheduler_create', + 'scheduler_upsert', 'scheduler_list', 'scheduler_get', - 'scheduler_update', 'scheduler_delete', 'scheduler_run_now', ]) diff --git a/packages/worker/src/mcp/capabilities/scheduler/domain.ts b/packages/worker/src/mcp/capabilities/scheduler/domain.ts index 523d31d35b..e8fe58ac42 100644 --- a/packages/worker/src/mcp/capabilities/scheduler/domain.ts +++ b/packages/worker/src/mcp/capabilities/scheduler/domain.ts @@ -1,22 +1,20 @@ import { defineDomain } from '../define-domain.ts' import { capabilityDomainNames } from '../domain-metadata.ts' -import { schedulerCreateCapability } from './scheduler-create.ts' import { schedulerDeleteCapability } from './scheduler-delete.ts' import { schedulerGetCapability } from './scheduler-get.ts' import { schedulerListCapability } from './scheduler-list.ts' import { schedulerRunNowCapability } from './scheduler-run-now.ts' -import { schedulerUpdateCapability } from './scheduler-update.ts' +import { schedulerUpsertCapability } from './scheduler-upsert.ts' export const schedulerDomain = defineDomain({ name: capabilityDomainNames.scheduler, description: - 'Create, inspect, update, delete, and trigger scheduled codemode jobs that run once at a UTC timestamp or recur on a timezone-aware cron schedule.', + 'Create, upsert, inspect, delete, and trigger scheduled codemode jobs that run once at a UTC timestamp or recur on a timezone-aware cron schedule.', keywords: ['schedule', 'scheduler', 'cron', 'datetime', 'run later', 'job'], capabilities: [ - schedulerCreateCapability, + schedulerUpsertCapability, schedulerListCapability, schedulerGetCapability, - schedulerUpdateCapability, schedulerDeleteCapability, schedulerRunNowCapability, ], diff --git a/packages/worker/src/mcp/capabilities/scheduler/scheduler-create.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-create.ts deleted file mode 100644 index ca299e4fa5..0000000000 --- a/packages/worker/src/mcp/capabilities/scheduler/scheduler-create.ts +++ /dev/null @@ -1,28 +0,0 @@ -import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' -import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' -import { type CapabilityContext } from '#mcp/capabilities/types.ts' -import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' -import { schedulerCreate } from '#worker/scheduler/client.ts' -import { schedulerCreateInputSchema, schedulerJobViewSchema } from './shared.ts' - -export const schedulerCreateCapability = defineDomainCapability( - capabilityDomainNames.scheduler, - { - name: 'scheduler_create', - description: - 'Create a scheduled codemode job for the signed-in user. Supports one-shot UTC timestamps or recurring 5-field cron expressions with IANA timezone interpretation.', - keywords: ['schedule', 'cron', 'datetime', 'job', 'automation'], - readOnly: false, - idempotent: false, - destructive: false, - inputSchema: schedulerCreateInputSchema, - outputSchema: schedulerJobViewSchema, - async handler(args, ctx: CapabilityContext) { - const user = requireMcpUser(ctx.callerContext) - return schedulerCreate(ctx.env, user.userId, { - callerContext: ctx.callerContext, - body: args, - }) - }, - }, -) diff --git a/packages/worker/src/mcp/capabilities/scheduler/scheduler-update.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-upsert.ts similarity index 50% rename from packages/worker/src/mcp/capabilities/scheduler/scheduler-update.ts rename to packages/worker/src/mcp/capabilities/scheduler/scheduler-upsert.ts index a570cf408d..37896b66a2 100644 --- a/packages/worker/src/mcp/capabilities/scheduler/scheduler-update.ts +++ b/packages/worker/src/mcp/capabilities/scheduler/scheduler-upsert.ts @@ -1,27 +1,44 @@ +import { type CapabilityContext } from '#mcp/capabilities/types.ts' import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' -import { type CapabilityContext } from '#mcp/capabilities/types.ts' -import { schedulerUpdate } from '#worker/scheduler/client.ts' +import { schedulerCreate, schedulerUpdate } from '#worker/scheduler/client.ts' import { requireSchedulerUser, schedulerJobViewSchema, - schedulerUpdateInputSchema, + schedulerUpsertInputSchema, } from './shared.ts' -export const schedulerUpdateCapability = defineDomainCapability( +export const schedulerUpsertCapability = defineDomainCapability( capabilityDomainNames.scheduler, { - name: 'scheduler_update', + name: 'scheduler_upsert', description: - 'Update a scheduled codemode job by id. You can change the name, code, params, schedule, timezone, or enabled flag. Changing the schedule recomputes nextRunAt immediately.', - keywords: ['scheduler', 'update', 'cron', 'datetime', 'job'], + 'Create a new scheduled codemode job when id is omitted, or update an existing scheduled job when id is provided. Supports one-shot UTC timestamps and recurring 5-field cron expressions with IANA timezone interpretation.', + keywords: ['scheduler', 'upsert', 'create', 'update', 'cron', 'datetime'], readOnly: false, idempotent: false, destructive: false, - inputSchema: schedulerUpdateInputSchema, + inputSchema: schedulerUpsertInputSchema, outputSchema: schedulerJobViewSchema, async handler(args, ctx: CapabilityContext) { const user = requireSchedulerUser(ctx) + if (args.id === undefined) { + return schedulerCreate(ctx.env, user.userId, { + callerContext: ctx.callerContext, + body: { + name: args.name ?? '', + code: args.code ?? '', + ...(args.params !== undefined && args.params !== null + ? { params: args.params } + : {}), + schedule: args.schedule!, + ...(args.timezone !== undefined && args.timezone !== null + ? { timezone: args.timezone } + : {}), + ...(args.enabled !== undefined ? { enabled: args.enabled } : {}), + }, + }) + } return schedulerUpdate(ctx.env, user.userId, { callerContext: ctx.callerContext, body: { diff --git a/packages/worker/src/mcp/capabilities/scheduler/shared.ts b/packages/worker/src/mcp/capabilities/scheduler/shared.ts index 38e7fc3018..2239ddfdcc 100644 --- a/packages/worker/src/mcp/capabilities/scheduler/shared.ts +++ b/packages/worker/src/mcp/capabilities/scheduler/shared.ts @@ -1,9 +1,8 @@ import { z } from 'zod' import { type ScheduledJobView, - type SchedulerCreateInput, type SchedulerExecutionResult, - type SchedulerUpdateInput, + type SchedulerUpsertInput, } from '#worker/scheduler/types.ts' import { type CapabilityContext } from '#mcp/capabilities/types.ts' import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' @@ -55,55 +54,77 @@ export const scheduledJobIdInputSchema = z.object({ id: z.string().min(1).describe('Identifier of the scheduled job.'), }) -export const schedulerCreateInputSchema = z.object({ - name: z.string().min(1).describe('Human-friendly job name.'), - code: z - .string() - .min(1) - .describe('Codemode async arrow function source to execute on schedule.'), - params: z - .record(z.string(), z.unknown()) - .optional() - .describe('Optional params injected into the scheduled codemode execution.'), - schedule: schedulerScheduleSchema, - timezone: z - .string() - .optional() - .describe( - 'Optional IANA timezone used when interpreting cron expressions. Defaults to UTC.', - ), - enabled: z - .boolean() - .optional() - .describe('Whether the new job should be enabled immediately. Defaults to true.'), -}) satisfies z.ZodType - -export const schedulerUpdateInputSchema = z.object({ - id: z.string().min(1).describe('Existing scheduled job id.'), - name: z.string().min(1).optional().describe('Optional new job name.'), - code: z - .string() - .min(1) - .optional() - .describe('Optional replacement codemode source.'), - params: z - .record(z.string(), z.unknown()) - .nullable() - .optional() - .describe( - 'Optional replacement params. Pass null to clear previously stored params.', - ), - schedule: schedulerScheduleSchema.optional(), - timezone: z - .string() - .nullable() - .optional() - .describe('Optional replacement IANA timezone. Pass null to reset to UTC.'), - enabled: z - .boolean() - .optional() - .describe('Optional replacement enabled flag.'), -}) satisfies z.ZodType +export const schedulerUpsertInputSchema = z + .object({ + id: z + .string() + .min(1) + .optional() + .describe( + 'Existing scheduled job id to update in place. Omit to create a new scheduled job.', + ), + name: z + .string() + .min(1) + .optional() + .describe('Job name. Required when creating a new scheduled job.'), + code: z + .string() + .min(1) + .optional() + .describe( + 'Codemode async arrow function source. Required when creating a new scheduled job.', + ), + params: z + .record(z.string(), z.unknown()) + .nullable() + .optional() + .describe( + 'Optional params injected into the scheduled codemode execution. Pass null to clear previously stored params during an update.', + ), + schedule: schedulerScheduleSchema + .optional() + .describe( + 'Schedule definition. Required when creating a new scheduled job.', + ), + timezone: z + .string() + .nullable() + .optional() + .describe( + 'Optional IANA timezone used when interpreting cron expressions. Pass null to reset to UTC during an update.', + ), + enabled: z + .boolean() + .optional() + .describe( + 'Whether the job should be enabled. Defaults to true when creating a new scheduled job.', + ), + }) + .superRefine((value, ctx) => { + if (value.id !== undefined) return + if (value.name === undefined) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['name'], + message: 'name is required when creating a scheduled job.', + }) + } + if (value.code === undefined) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['code'], + message: 'code is required when creating a scheduled job.', + }) + } + if (value.schedule === undefined) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['schedule'], + message: 'schedule is required when creating a scheduled job.', + }) + } + }) satisfies z.ZodType export const schedulerExecutionResultSchema = z.discriminatedUnion('ok', [ z.object({ diff --git a/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts index 3afcb9b364..134f351295 100644 --- a/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts +++ b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts @@ -169,7 +169,7 @@ test('mcp server manages scheduled codemode jobs', async () => { name: 'execute', arguments: { code: `async () => { - return await codemode.scheduler_create({ + return await codemode.scheduler_upsert({ name: 'daily-summary', code: 'async () => ({ ran: true })', schedule: { type: 'cron', expression: '0 7 * * *' }, @@ -212,9 +212,10 @@ test('mcp server manages scheduled codemode jobs', async () => { const runNowResult = await mcpClient.client.callTool({ name: 'execute', arguments: { - code: `async () => { - return await codemode.scheduler_run_now({ id: ${JSON.stringify(jobId)} }) + code: `async (params) => { + return await codemode.scheduler_run_now({ id: params.jobId }) }`, + params: { jobId }, }, }) const runNowStructured = (runNowResult as CallToolResult).structuredContent as @@ -234,13 +235,14 @@ test('mcp server manages scheduled codemode jobs', async () => { const updateResult = await mcpClient.client.callTool({ name: 'execute', arguments: { - code: `async () => { - return await codemode.scheduler_update({ - id: ${JSON.stringify(jobId)}, + code: `async (params) => { + return await codemode.scheduler_upsert({ + id: params.jobId, enabled: false, - schedule: { type: 'once', runAt: ${JSON.stringify(futureRunAt)} }, + schedule: { type: 'once', runAt: params.futureRunAt }, }) }`, + params: { jobId, futureRunAt }, }, }) const updateStructured = (updateResult as CallToolResult).structuredContent as @@ -262,9 +264,10 @@ test('mcp server manages scheduled codemode jobs', async () => { const getResult = await mcpClient.client.callTool({ name: 'execute', arguments: { - code: `async () => { - return await codemode.scheduler_get({ id: ${JSON.stringify(jobId)} }) + code: `async (params) => { + return await codemode.scheduler_get({ id: params.jobId }) }`, + params: { jobId }, }, }) const getStructured = (getResult as CallToolResult).structuredContent as @@ -277,9 +280,10 @@ test('mcp server manages scheduled codemode jobs', async () => { const deleteResult = await mcpClient.client.callTool({ name: 'execute', arguments: { - code: `async () => { - return await codemode.scheduler_delete({ id: ${JSON.stringify(jobId)} }) + code: `async (params) => { + return await codemode.scheduler_delete({ id: params.jobId }) }`, + params: { jobId }, }, }) const deleteStructured = (deleteResult as CallToolResult).structuredContent as diff --git a/packages/worker/src/mcp/server-instructions.ts b/packages/worker/src/mcp/server-instructions.ts index 29d6a02172..9732d9e778 100644 --- a/packages/worker/src/mcp/server-instructions.ts +++ b/packages/worker/src/mcp/server-instructions.ts @@ -22,7 +22,7 @@ Conventions - \`memoryContext\`: short and task-focused. Kody may use it to surface a few relevant long-term memories and suppress repeats within the same \`conversationId\`. - Do not ask the user to paste secrets in chat; use saved secrets or \`open_generated_ui\`. - \`meta_save_skill\`: repeatable workflows only; optional \`collection\`; same name replaces an existing skill. One-off work: \`execute\`. Skill params: pass via \`meta_run_skill\` → \`params\` in codemode. -- \`scheduler_create\` / \`scheduler_update\`: schedule codemode to run later or on a recurring cron in an IANA timezone. Use \`scheduler_list\` / \`scheduler_get\` to inspect status and next run times, and \`scheduler_run_now\` to trigger the stored job immediately without changing its schedule. +- \`scheduler_upsert\`: create a new scheduled codemode job or update an existing one in place. Use \`scheduler_list\` / \`scheduler_get\` to inspect status and next run times, and \`scheduler_run_now\` to trigger the stored job immediately without changing its schedule. - \`ui_save_app\` / \`app_id\`: persisted UI artifacts (hidden from search unless \`hidden: false\`). \`codemode.secret_list\` / \`secret_set\`: metadata-only list; set only for values already in trusted execution (see \`execute\` tool description). - Memory writes are verify-first: always run \`meta_memory_verify\` before \`meta_memory_upsert\` or \`meta_memory_delete\`. Kody retrieves related memories; the consuming agent decides whether to upsert, delete, both, or do nothing. \`meta_memory_upsert\` creates a new memory when \`memory_id\` is omitted and updates an existing memory when \`memory_id\` is provided. - User-specific MCP instructions: \`meta_get_mcp_server_instructions\` / \`meta_set_mcp_server_instructions\` (signed-in users). Updates apply to **new** MCP sessions (reconnect to refresh what the host shows). diff --git a/packages/worker/src/mcp/tools/execute.ts b/packages/worker/src/mcp/tools/execute.ts index 84b29680c0..0dc886c2a7 100644 --- a/packages/worker/src/mcp/tools/execute.ts +++ b/packages/worker/src/mcp/tools/execute.ts @@ -38,8 +38,8 @@ builtin capability). Discover names with \`search\`; for one capability’s Saved skills: prefer \`meta_run_skill({ name, params })\`, or \`meta_get_skill\` then paste code here. -Scheduled jobs: use \`scheduler_create\`, \`scheduler_list\`, \`scheduler_get\`, -\`scheduler_update\`, \`scheduler_delete\`, and \`scheduler_run_now\` to manage +Scheduled jobs: use \`scheduler_upsert\`, \`scheduler_list\`, \`scheduler_get\`, +\`scheduler_delete\`, and \`scheduler_run_now\` to manage one-shot or recurring codemode executions for the signed-in user. Sandbox surface: diff --git a/packages/worker/src/mcp/tools/search.ts b/packages/worker/src/mcp/tools/search.ts index 8ecd7ecae2..d1402e7bd6 100644 --- a/packages/worker/src/mcp/tools/search.ts +++ b/packages/worker/src/mcp/tools/search.ts @@ -136,8 +136,8 @@ empty ranked list. capabilities. Types and fields: see response. Run a skill: \`meta_run_skill({ name, params })\`; source: \`meta_get_skill\`. -Manage schedules: \`scheduler_create\`, \`scheduler_list\`, \`scheduler_get\`, -\`scheduler_update\`, \`scheduler_delete\`, \`scheduler_run_now\`. +Manage schedules: \`scheduler_upsert\`, \`scheduler_list\`, \`scheduler_get\`, +\`scheduler_delete\`, \`scheduler_run_now\`. Apps: \`open_generated_ui({ app_id })\`. Secrets: never raw in results; use \`codemode.secret_list\` during execute and UI for missing values. Persisted values use \`codemode.value_get\` / \`codemode.value_list\`. Connectors diff --git a/packages/worker/src/scheduler/types.ts b/packages/worker/src/scheduler/types.ts index c27053d46c..467f92d7c0 100644 --- a/packages/worker/src/scheduler/types.ts +++ b/packages/worker/src/scheduler/types.ts @@ -68,3 +68,13 @@ export type SchedulerUpdateInput = { timezone?: string | null enabled?: boolean } + +export type SchedulerUpsertInput = { + id?: string + name?: string + code?: string + params?: Record | null + schedule?: ScheduledJobSchedule + timezone?: string | null + enabled?: boolean +}