diff --git a/docs/use/execute.md b/docs/use/execute.md index 3c7e82fb98..73a9810420 100644 --- a/docs/use/execute.md +++ b/docs/use/execute.md @@ -32,6 +32,23 @@ 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_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_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/build-capability-registry.workers.test.ts b/packages/worker/src/mcp/capabilities/build-capability-registry.workers.test.ts index d803dbb22f..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 @@ -74,6 +74,23 @@ 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_upsert', + 'scheduler_list', + 'scheduler_get', + '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/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..e8fe58ac42 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/domain.ts @@ -0,0 +1,21 @@ +import { defineDomain } from '../define-domain.ts' +import { capabilityDomainNames } from '../domain-metadata.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 { schedulerUpsertCapability } from './scheduler-upsert.ts' + +export const schedulerDomain = defineDomain({ + name: capabilityDomainNames.scheduler, + description: + '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: [ + schedulerUpsertCapability, + schedulerListCapability, + schedulerGetCapability, + schedulerDeleteCapability, + schedulerRunNowCapability, + ], +}) 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-upsert.ts b/packages/worker/src/mcp/capabilities/scheduler/scheduler-upsert.ts new file mode 100644 index 0000000000..37896b66a2 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/scheduler-upsert.ts @@ -0,0 +1,56 @@ +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 { schedulerCreate, schedulerUpdate } from '#worker/scheduler/client.ts' +import { + requireSchedulerUser, + schedulerJobViewSchema, + schedulerUpsertInputSchema, +} from './shared.ts' + +export const schedulerUpsertCapability = defineDomainCapability( + capabilityDomainNames.scheduler, + { + name: 'scheduler_upsert', + description: + '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: 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: { + 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..2239ddfdcc --- /dev/null +++ b/packages/worker/src/mcp/capabilities/scheduler/shared.ts @@ -0,0 +1,154 @@ +import { z } from 'zod' +import { + type ScheduledJobView, + type SchedulerExecutionResult, + type SchedulerUpsertInput, +} 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 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({ + 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/mcp-server.mcp-e2e.test.ts b/packages/worker/src/mcp/mcp-server.mcp-e2e.test.ts index 8addc332a4..134f351295 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,141 @@ 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_upsert({ + 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 (params) => { + return await codemode.scheduler_run_now({ id: params.jobId }) + }`, + params: { 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 futureRunAt = new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString() + const updateResult = await mcpClient.client.callTool({ + name: 'execute', + arguments: { + code: `async (params) => { + return await codemode.scheduler_upsert({ + id: params.jobId, + enabled: false, + schedule: { type: 'once', runAt: params.futureRunAt }, + }) + }`, + params: { jobId, futureRunAt }, + }, + }) + 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: futureRunAt, + }) + expect(updateStructured?.result?.nextRunAt).toBe(futureRunAt) + + const getResult = await mcpClient.client.callTool({ + name: 'execute', + arguments: { + code: `async (params) => { + return await codemode.scheduler_get({ id: params.jobId }) + }`, + params: { 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 (params) => { + return await codemode.scheduler_delete({ id: params.jobId }) + }`, + params: { 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) @@ -295,13 +430,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) @@ -313,11 +449,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) diff --git a/packages/worker/src/mcp/server-instructions.ts b/packages/worker/src/mcp/server-instructions.ts index c5bcd651d9..9732d9e778 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_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 cad59e980b..0dc886c2a7 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_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: - \`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..d1402e7bd6 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_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/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 } 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.node.test.ts b/packages/worker/src/scheduler/process-due-jobs.node.test.ts new file mode 100644 index 0000000000..94a08b7b41 --- /dev/null +++ b/packages/worker/src/scheduler/process-due-jobs.node.test.ts @@ -0,0 +1,85 @@ +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).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 () => { + 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/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.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 new file mode 100644 index 0000000000..f32dc24c64 --- /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 fields.join(' ') +} + +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..507467030c --- /dev/null +++ b/packages/worker/src/scheduler/scheduler-do.ts @@ -0,0 +1,400 @@ +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, + isJobDue, + normalizeScheduledJobSchedule, + normalizeSchedulerTimezone, + toScheduledJobView, +} from './schedule.ts' +import { + type PersistedSchedulerCallerContext, + type ScheduledJob, + type SchedulerCreateInput, + type SchedulerExecutionResult, + type SchedulerUpdateInput, +} from './types.ts' + +const jobStorageKeyPrefix = 'job:' + +function getJobStorageKey(jobId: string) { + return `${jobStorageKeyPrefix}${jobId}` +} + +function getJobCallerContextStorageKey(jobId: string) { + return `job-context:${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) => + isJobDue(job, now), + ) + if (dueJobs.length === 0) { + await this.syncAlarm() + return + } + const result = await processDueJobs({ + jobs: dueJobs, + now, + 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() + } + + private async handleCreateJob(request: Request) { + const payload = + await this.parseMutationRequest(request) + 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.persistCallerContext(job.id, payload.callerContext) + 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.', + ) + } + const existing = await this.requireStoredJob(jobId) + 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 = + nextTimezone === null + ? normalizeSchedulerTimezone(null) + : normalizeSchedulerTimezone(nextTimezone ?? existing.timezone) + const shouldRecomputeNextRunAt = + JSON.stringify(schedule) !== JSON.stringify(existing.schedule) || + timezone !== existing.timezone || + (existing.enabled === false && nextEnabled === true) + const nextRunAt = + shouldRecomputeNextRunAt + ? computeNextRunAt({ + schedule, + timezone, + }) + : existing.nextRunAt + 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: nextEnabled, + nextRunAt, + } + await this.ctx.storage.put(getJobStorageKey(jobId), updated) + await this.persistCallerContext(jobId, payload.callerContext) + await this.syncAlarm() + return toScheduledJobView(updated) + } + + 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, + 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.', + ) + } + 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, + } + 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), + 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: [], + } + } + 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, + 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( + jobId: string, + callerContext: PersistedSchedulerCallerContext, + ) { + await this.ctx.storage.put( + getJobCallerContextStorageKey(jobId), + callerContext, + ) + } + + private async getPersistedCallerContext(jobId: string) { + return ( + (await this.ctx.storage.get( + getJobCallerContextStorageKey(jobId), + )) ?? 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..467f92d7c0 --- /dev/null +++ b/packages/worker/src/scheduler/types.ts @@ -0,0 +1,80 @@ +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 +} + +export type SchedulerUpsertInput = { + 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": [