diff --git a/README.md b/README.md index a245dc3..12c2dc1 100644 --- a/README.md +++ b/README.md @@ -403,7 +403,7 @@ canvases execute inside the chat host's MCP Apps sandbox; use trusted local canv source, as for the other local views. Artifact source and saved state can contain private data; keep the history database private. -## Projects and remix +## Projects, remix, and scheduled runs Canvases and scripts support helper source files and pinned package dependencies. Use the gallery's source file picker and Dependencies editor, or pass @@ -416,4 +416,7 @@ Use **Remix** in the gallery, `canvas_remix` / `script_remix` through MCP, or `canvas remix SOURCE NEW_NAME` locally. Copies retain source provenance and start with fresh state, storage, secrets, and private hosted URLs. -See [project and remix contracts](docs/scripts.md) for details and limits. +Hosted and celld backends expose **Schedule** and **Run history** controls. Native +alarms support intervals and cron expressions with timezones, pause/resume, and +run-now. Run records live in host-owned SQLite. See +[project, remix, and scheduling contracts](docs/scripts.md) for details and limits. diff --git a/bun.lock b/bun.lock index 3bf5f07..9ecc01e 100644 --- a/bun.lock +++ b/bun.lock @@ -9,6 +9,7 @@ "@types/react": "^19.2.18", "@types/react-dom": "^19.2.4", "better-auth": "1.7.3", + "cron-parser": "5.10.0", "esbuild": "0.28.1", "react": "^19.2.8", "react-dom": "^19.2.8", @@ -266,6 +267,8 @@ "cors": ["cors@2.8.6", "", { "dependencies": { "object-assign": "^4", "vary": "^1" } }, "sha512-tJtZBBHA6vjIAaF6EnIaq6laBBP9aq/Y3ouVJjEfoHbRBcHBAHYcMh/w8LDrk2PvIMMq8gmopa5D4V8RmbrxGw=="], + "cron-parser": ["cron-parser@5.10.0", "", { "dependencies": { "luxon": "^3.7.2" } }, "sha512-izNAxJyRWUP8ljBoDSub5WyrVOUlT4SLGShswE7eoRBpp6QUsSycYxLBMJlbshgPBMcPT/nrfgjNY2918ayv2A=="], + "cross-spawn": ["cross-spawn@7.0.6", "", { "dependencies": { "path-key": "^3.1.0", "shebang-command": "^2.0.0", "which": "^2.0.1" } }, "sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA=="], "csstype": ["csstype@3.2.3", "", {}, "sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ=="], @@ -366,6 +369,8 @@ "lines-and-columns": ["lines-and-columns@1.2.4", "", {}, "sha512-7ylylesZQ/PV29jhEDl3Ufjo6ZX7gCqJr5F7PKrqc93v7fzSymt1BpwEU8nAUXs8qzzvqhbjhK5QZg6Mt/HkBg=="], + "luxon": ["luxon@3.7.2", "", {}, "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew=="], + "math-intrinsics": ["math-intrinsics@1.1.0", "", {}, "sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g=="], "media-typer": ["media-typer@1.1.1", "", {}, "sha512-yz3xRaG20c6/BOzvYoDaGtPmGscs7YivItZEEqe6GbwNfHuxu9YNmvnEkMzKldAGY4/80pRcQRZSEnhquk9XuQ=="], diff --git a/cloudflare/backend.ts b/cloudflare/backend.ts index cf28b6a..099b88d 100644 --- a/cloudflare/backend.ts +++ b/cloudflare/backend.ts @@ -1,6 +1,8 @@ import { DurableObject } from "cloudflare:workers"; import type { CanvasHttpRequest, CanvasHttpResponse } from "../src/httpTypes"; +import { CanvasExecution, type CanvasActive, type CanvasClaim, type CanvasScheduleInput } from "./canvas-execution"; + export type BackendEnv = { LOADER: WorkerLoader }; const MAX_BODY = 256 * 1024; @@ -43,24 +45,73 @@ async function serializeResponse(response: Response, method: string): Promise { private activeCode: string | undefined; + private execution: CanvasExecution; + constructor(state: DurableObjectState, env: BackendEnv) { + super(state, env); + this.execution = new CanvasExecution(state.storage); + } + activate(input: CanvasActive) { return this.execution.activate(input); } + async activeRevision() { + const active = await this.execution.active(); + return active ? { version_id: active.version_id, revision: active.revision, has_server: active.code !== null } : null; + } + runs(input: { limit?: number } = {}) { return this.execution.runs(input.limit); } + async schedule(input: CanvasScheduleInput = {}) { + if (input.request) { + nativeRequest(input.request); + input = { ...input, request: { ...input.request, method: input.request.method ?? "GET", headers: input.request.headers ?? [] } }; + } + if (input.action === "run_now") { + const schedule = await this.execution.get(); + if (!schedule) throw new Error("No schedule configured"); + await this.runScheduled(schedule.request, "manual"); + return this.execution.get(); + } + return this.execution.update(input); + } + async alarm() { + const claimed = await this.execution.claim(); + if (claimed) await this.runScheduled(claimed.schedule.request, "schedule", claimed); + } + private async runScheduled(request: CanvasHttpRequest, trigger: "manual" | "schedule", claimed?: CanvasClaim) { + const active = claimed ? claimed.active : await this.execution.active(); + if (!active?.code) { + const run = claimed?.run ?? this.execution.start(active?.version_id ?? "unavailable", trigger); + this.execution.finish(run, null); + return; + } + // Errors are recorded by request(). Never retry an invocation that may + // already have committed application writes. + try { await this.request({ ...active, code: active.code, request, trigger }, claimed?.run); } catch {} + } - async request(input: { code: string; hash: string; request: CanvasHttpRequest }): Promise { + async request(input: { code: string; hash: string; version_id?: string; request: CanvasHttpRequest; trigger?: "http" | "manual" | "schedule" }, claimedRun?: { id: string; started: number }): Promise { const request = nativeRequest(input.request); - if (this.activeCode !== input.hash) { - // Reload code while preserving the facet's database. Never delete/reset - // storage as a side effect of an edit, preview, or source restoration. - this.ctx.facets.abort("canvas", "Canvas server code updated"); - this.activeCode = input.hash; + const active = input.version_id ? undefined : await this.execution.active(); + const revision = input.version_id ?? (active?.hash === input.hash ? active.version_id : input.hash); + const run = claimedRun ?? this.execution.start(revision, input.trigger ?? "http"); + try { + if (this.activeCode !== input.hash) { + // Reload code while preserving the facet's database. Never delete/reset + // storage as a side effect of an edit, preview, or source restoration. + this.ctx.facets.abort("canvas", "Canvas server code updated"); + this.activeCode = input.hash; + } + const facet = this.ctx.facets.get("canvas", () => { + const worker = this.env.LOADER.get(input.hash, async () => ({ + compatibilityDate: "2026-09-06", + mainModule: "canvas-server.js", + modules: { "canvas-server.js": input.code }, + globalOutbound: null, + })); + return { class: worker.getDurableObjectClass("CanvasServer") }; + }); + const response = await serializeResponse(await facet.fetch(request), request.method); + this.execution.finish(run, response.status); + return response; + } catch (error) { + this.execution.finish(run, null); + throw error; } - const facet = this.ctx.facets.get("canvas", () => { - const worker = this.env.LOADER.get(input.hash, async () => ({ - compatibilityDate: "2026-09-06", - mainModule: "canvas-server.js", - modules: { "canvas-server.js": input.code }, - globalOutbound: null, - })); - return { class: worker.getDurableObjectClass("CanvasServer") }; - }); - return serializeResponse(await facet.fetch(request), request.method); } } diff --git a/cloudflare/canvas-execution.test.ts b/cloudflare/canvas-execution.test.ts new file mode 100644 index 0000000..3e35193 --- /dev/null +++ b/cloudflare/canvas-execution.test.ts @@ -0,0 +1,111 @@ +import { afterAll, afterEach, expect, spyOn, test } from "bun:test"; +import { Database } from "bun:sqlite"; +import { CanvasExecution } from "./canvas-execution"; + +// Real SQLite for host history; the KV/alarm adapter exposes observable state +// without requiring a Workers isolate for scheduling state transitions. +function fixture() { + const db = new Database(":memory:"); + const values = new Map(); + let alarm: number | null = null; + const storage = { + sql: { exec(query: string, ...params: (string | number | null)[]) { + const rows = db.prepare(query).all(...params); + return { toArray: () => rows }; + } }, + get: async (key: string) => values.get(key), + put: async (key: string, value: unknown) => { values.set(key, value); }, + setAlarm: async (time: number) => { alarm = time; }, + deleteAlarm: async () => { alarm = null; }, + transaction: async (callback: (storage: unknown) => Promise) => callback(storage), + }; + return { execution: new CanvasExecution(storage as unknown as DurableObjectStorage), storage, alarm: () => alarm, db }; +} +const now = spyOn(Date, "now"); +afterEach(() => now.mockReset()); +afterAll(() => now.mockRestore()); +const active = { code: "server", hash: "hash", version_id: "revision-one", revision: 1 }; +const request = { path: "/refresh", method: "POST", headers: [] as [string, string][] }; + +test("schedule claims once, pause cancels alarm, resume schedules a future occurrence", async () => { + now.mockReturnValue(1_800_000_000_000); + const { execution, alarm } = fixture(); + await execution.activate(active); + const saved = await execution.update({ action: "set", interval_seconds: 60, request }); + expect(saved).toMatchObject({ request, paused: false, next_run_at: Date.now() + 60_000 }); + expect(alarm()).toBe(saved.next_run_at); + expect(await execution.claim()).toBeNull(); + now.mockReturnValue(saved.next_run_at!); + expect(await execution.claim()).toMatchObject({ schedule: saved, active }); + expect(await execution.claim()).toBeNull(); + expect(alarm()).toBe(Date.now() + 60_000); + await execution.update({ action: "pause" }); + expect(alarm()).toBeNull(); + expect(await execution.claim()).toBeNull(); + expect(await execution.update({ action: "resume" })).toMatchObject({ paused: false, next_run_at: Date.now() + 60_000 }); +}); + +test("a claimed occurrence survives interruption without being replayed", async () => { + now.mockReturnValue(1_800_000_000_000); + const { execution, storage } = fixture(); + await execution.activate(active); + const schedule = await execution.update({ action: "set", interval_seconds: 60, request }); + now.mockReturnValue(schedule.next_run_at!); + const claimed = await execution.claim(); + expect(claimed).not.toBeNull(); + expect(execution.runs()).toMatchObject([{ id: claimed!.run.id, revision: active.version_id, status: "running", trigger: "schedule" }]); + const restarted = new CanvasExecution(storage as unknown as DurableObjectStorage); + expect(await restarted.claim()).toBeNull(); + expect(restarted.runs()).toMatchObject([{ id: claimed!.run.id, status: "interrupted" }]); +}); + +test("validated activation is monotonic and server removal pauses scheduling", async () => { + now.mockReturnValue(1_800_000_000_000); + const { execution, alarm } = fixture(); + await execution.activate({ ...active, revision: 3, version_id: "revision-three" }); + await execution.activate(active); + expect((await execution.active())?.version_id).toBe("revision-three"); + await execution.update({ action: "set", interval_seconds: 60, request }); + await execution.activate({ ...active, code: null, revision: 4 }); + expect(await execution.get()).toMatchObject({ paused: true, next_run_at: null }); + expect(alarm()).toBeNull(); + await expect(execution.update({ action: "resume" })).rejects.toThrow("no validated server"); +}); + +test("invalid schedule cannot replace existing state, cron uses configured timezone", async () => { + now.mockReturnValue(Date.parse("2026-09-07T10:00:00Z")); + const { execution } = fixture(); + await execution.activate(active); + const saved = await execution.update({ action: "set", cron: "0 9 * * *", timezone: "America/Los_Angeles", request }); + expect(saved.next_run_at).toBe(Date.parse("2026-09-07T16:00:00Z")); + await expect(execution.update({ action: "set", interval_seconds: 5, request })).rejects.toThrow(); + await expect(execution.update({ action: "set", interval_seconds: 60, cron: "* * * * *", request })).rejects.toThrow(); + expect(await execution.get()).toEqual(saved); +}); + +test("history records revision, trigger, duration and failures; restart retains uncertainty", () => { + now.mockReturnValue(1_800_000_000_000); + const { execution, storage } = fixture(); + const run = execution.start("version-1", "schedule"); + now.mockReturnValue(Date.now() + 123); + execution.finish(run, 204); + expect(execution.runs()).toMatchObject([{ revision: "version-1", trigger: "schedule", duration_ms: 123, status: "succeeded", http_status: 204 }]); + const failed = execution.start("version-2", "http"); + execution.finish(failed, 500); + const uncertain = execution.start("version-3", "manual"); + const recovered = new CanvasExecution(storage as unknown as DurableObjectStorage); + expect(recovered.runs().find(run => run.id === uncertain.id)).toMatchObject({ status: "interrupted", finished_at: null }); + expect(recovered.runs().find(run => run.id === failed.id)).toMatchObject({ status: "failed", http_status: 500 }); + expect(() => recovered.runs(101)).toThrow(); +}); + +test("compiled source and scheduled bodies larger than KV values persist in host SQLite", async () => { + now.mockReturnValue(1_800_000_000_000); + const { execution, storage } = fixture(); + const source = "x".repeat(200_000), body = btoa("y".repeat(200_000)); + await execution.activate({ ...active, code: source }); + await execution.update({ action: "set", interval_seconds: 3600, request: { ...request, body } }); + const restarted = new CanvasExecution(storage as unknown as DurableObjectStorage); + expect((await restarted.active())?.code).toBe(source); + expect((await restarted.get())?.request.body).toBe(body); +}); diff --git a/cloudflare/canvas-execution.ts b/cloudflare/canvas-execution.ts new file mode 100644 index 0000000..322ef3b --- /dev/null +++ b/cloudflare/canvas-execution.ts @@ -0,0 +1,87 @@ +import type { CanvasHttpRequest } from "../src/httpTypes"; +import { normalizeTiming, nextOccurrence, type ScheduleTiming } from "./schedule"; + +export type CanvasSchedule = ScheduleTiming & { paused: boolean; next_run_at: number | null; request: CanvasHttpRequest }; +export type CanvasScheduleInput = ScheduleTiming & { action?: "get" | "set" | "pause" | "resume" | "run_now"; request?: CanvasHttpRequest }; +export type CanvasRun = { id: string; revision: string; trigger: string; started_at: string; finished_at: string | null; duration_ms: number | null; status: string; http_status: number | null }; +export type CanvasActive = { code: string | null; hash: string; version_id: string; revision: number }; +export type CanvasClaim = { schedule: CanvasSchedule; active: CanvasActive | undefined; run: { id: string; started: number } }; + +/** Supervisor storage belongs to the host, separate from CanvasServer's facet. */ +export class CanvasExecution { + constructor(private storage: DurableObjectStorage) { + storage.sql.exec("create table if not exists host_schedule(id integer primary key check(id=1),payload text not null)"); + storage.sql.exec("create table if not exists active_server(id integer primary key check(id=1),payload text not null)"); + storage.sql.exec("create table if not exists execution_runs(id text primary key,revision text not null,trigger text not null,started_at text not null,finished_at text,duration_ms integer,status text not null,http_status integer)"); + storage.sql.exec("update execution_runs set status='interrupted' where status='running'"); + } + async activate(active: CanvasActive) { + await this.storage.transaction(async storage => { + const previous = this.readActive(); + if (previous && previous.revision > active.revision) return; + this.storage.sql.exec("insert into active_server(id,payload) values(1,?) on conflict(id) do update set payload=excluded.payload", JSON.stringify(active)); + if (active.code === null) { + const schedule = this.readSchedule(); + if (schedule) this.writeSchedule({ ...schedule, paused: true, next_run_at: null }); + await storage.deleteAlarm(); + } + }); + } + private readActive(): CanvasActive | undefined { + const row = this.storage.sql.exec<{ payload: string }>("select payload from active_server where id=1").toArray()[0]; + return row ? JSON.parse(row.payload) as CanvasActive : undefined; + } + async active() { return this.readActive(); } + start(revision: string, trigger: string) { + const id = crypto.randomUUID(), started = Date.now(); + this.storage.sql.exec("insert into execution_runs(id,revision,trigger,started_at,status) values(?,?,?,?,?)", id, revision, trigger, new Date(started).toISOString(), "running"); + return { id, started }; + } + finish(run: { id: string; started: number }, status: number | null) { + this.storage.sql.exec("update execution_runs set finished_at=?,duration_ms=?,status=?,http_status=? where id=?", new Date().toISOString(), Date.now() - run.started, status !== null && status < 400 ? "succeeded" : "failed", status, run.id); + this.storage.sql.exec("delete from execution_runs where status!='running' and id not in (select id from execution_runs order by started_at desc,rowid desc limit 1000)"); + } + runs(limit = 100) { + if (!Number.isSafeInteger(limit) || limit < 1 || limit > 100) throw new Error("limit must be between 1 and 100"); + return this.storage.sql.exec("select * from execution_runs order by started_at desc,rowid desc limit ?", limit).toArray(); + } + private readSchedule(): CanvasSchedule | null { + const row = this.storage.sql.exec<{ payload: string }>("select payload from host_schedule where id=1").toArray()[0]; + return row ? JSON.parse(row.payload) as CanvasSchedule : null; + } + private writeSchedule(schedule: CanvasSchedule) { + this.storage.sql.exec("insert into host_schedule(id,payload) values(1,?) on conflict(id) do update set payload=excluded.payload", JSON.stringify(schedule)); + } + async get() { return this.readSchedule(); } + async update(input: CanvasScheduleInput) { + const action = input.action ?? "get"; + if (action === "get") return this.get(); + if (!["set", "pause", "resume"].includes(action)) throw new Error("Invalid schedule action"); + return this.storage.transaction(async storage => { + const previous = this.readSchedule(); + if (action !== "set" && !previous) throw new Error("No schedule configured"); + const active = this.readActive(); + if (action !== "pause" && !active?.code) throw new Error("Canvas has no validated server"); + const timing = action === "set" ? normalizeTiming(input) : previous!; + const request = action === "set" ? input.request ?? { path: "/", method: "GET", headers: [] } : previous!.request; + const result: CanvasSchedule = { ...normalizeTiming(timing), request, paused: action === "pause", next_run_at: action === "pause" ? null : nextOccurrence(timing) }; + this.writeSchedule(result); + if (result.next_run_at === null) await storage.deleteAlarm(); else await storage.setAlarm(result.next_run_at); + return result; + }); + } + /** Claim before invoking; redelivered alarms cannot repeat side effects. */ + async claim(): Promise { + return this.storage.transaction(async storage => { + const schedule = this.readSchedule(); + if (!schedule || schedule.paused || schedule.next_run_at === null) return null; + if (schedule.next_run_at > Date.now()) { await storage.setAlarm(schedule.next_run_at); return null; } + const next = nextOccurrence(schedule); + this.writeSchedule({ ...schedule, next_run_at: next }); + await storage.setAlarm(next); + const active = this.readActive(); + const run = this.start(active?.version_id ?? "unavailable", "schedule"); + return { schedule, active, run }; + }); + } +} diff --git a/cloudflare/compiled-canvas-test-worker.ts b/cloudflare/compiled-canvas-test-worker.ts index 89ff96a..010377c 100644 --- a/cloudflare/compiled-canvas-test-worker.ts +++ b/cloudflare/compiled-canvas-test-worker.ts @@ -12,6 +12,8 @@ declare global { var compilerProbe: { allowed: boolean; client: number; server: number }; } +const activeRevisions = new Map(); + export default { async fetch(request: Request, env: Env) { try { @@ -24,9 +26,10 @@ export default { const workspace = url.searchParams.get("workspace") ?? "default"; const library = env.LIBRARY.getByName(libraryKey); const links = env.LINKS.getByName("deployment"); - const backends = { getByName: (key: string) => ({ request: async (input: { code: string; hash: string }) => ({ status: 200, headers: [], body: btoa(JSON.stringify({ key, ...input })) }) }) }; + const backends = { getByName: (key: string) => ({ activate: async (input: unknown) => { activeRevisions.set(key, input); }, request: async (input: { code: string; hash: string }) => ({ status: 200, headers: [], body: btoa(JSON.stringify({ key, ...input })) }) }) }; const service = new CloudCanvasService(library, workspace, backends as never, libraryKey, { links, origin: url.origin } as HostedArtifacts); const input = await request.json() as Record; + if (url.pathname === "/activated") return Response.json(activeRevisions.get(JSON.stringify([libraryKey, workspace, input.name])) ?? null); if (url.pathname === "/tool") return Response.json(await service.callTool(input.name, input.arguments)); if (url.pathname === "/preview") return Response.json(await service.preview(await service.snapshot(input))); if (url.pathname === "/request") return Response.json(await service.request(input, { path: "/", method: "GET", headers: [] })); diff --git a/cloudflare/compiled-canvas.test.ts b/cloudflare/compiled-canvas.test.ts index 01d2525..f63d479 100644 --- a/cloudflare/compiled-canvas.test.ts +++ b/cloudflare/compiled-canvas.test.ts @@ -52,6 +52,7 @@ test("edits persist compiled revisions; every read and restart works with the co expect(first.isError).not.toBe(true); const firstCanvas = first._meta.canvas; const firstVersion = firstCanvas.versionId; + expect(await call("/activated", { name: "example" })).toMatchObject({ version_id: firstVersion, code: expect.stringContaining("server-one") }); expect(await call("/compiler", { allowed: false })).toMatchObject({ client: 1, server: 1 }); for (let index = 0; index < 3; index++) { expect((await tool("canvas_open", { name: "example" }))._meta.canvas.js).toBe(firstCanvas.js); @@ -108,6 +109,7 @@ test("edits persist compiled revisions; every read and restart works with the co const projectEdit = await tool("canvas_edit", { name: "project", file: "helper.ts", edits: [{ old_text: "original-helper", new_text: "updated-helper" }] }); expect(projectEdit.isError).not.toBe(true); expect(projectEdit._meta.canvas.js).toContain("updated-helper"); + expect(await call("/activated", { name: "project" })).toMatchObject({ version_id: projectEdit._meta.canvas.versionId, code: expect.stringContaining("updated-helper") }); expect(projectEdit._meta.canvas.js).not.toBe(projectFirst.js); expect(projectEdit._meta.canvas.versionId).not.toBe(projectFirst.versionId); await call("/compiler", { allowed: false }); diff --git a/cloudflare/schedule.test.ts b/cloudflare/schedule.test.ts new file mode 100644 index 0000000..3163e40 --- /dev/null +++ b/cloudflare/schedule.test.ts @@ -0,0 +1,10 @@ +import {expect,test} from "bun:test"; +import {normalizeTiming,nextOccurrence} from "./schedule"; +test("interval and timezone cron next occurrences, including DST",()=>{ + expect(nextOccurrence(normalizeTiming({interval_seconds:60}),1000)).toBe(61000); + expect(new Date(nextOccurrence(normalizeTiming({cron:"0 9 * * *",timezone:"America/Los_Angeles"}),Date.parse("2026-03-07T18:00:00Z"))).toISOString()).toBe("2026-03-08T16:00:00.000Z"); + expect(new Date(nextOccurrence(normalizeTiming({cron:"0 9 * * *"}),Date.parse("2026-09-07T00:00:00Z"))).toISOString()).toBe("2026-09-07T09:00:00.000Z"); +}); +test("invalid and ambiguous schedules are rejected",()=>{ + for(const input of [{},{interval_seconds:60,cron:"* * * * *"},{interval_seconds:59},{interval_seconds:1.5},{interval_seconds:60,timezone:"UTC"},{cron:"* * * * * *"},{cron:"H * * * *"},{cron:"bad * * * *"},{cron:"* * * * *",timezone:"Unknown/Zone"}])expect(()=>normalizeTiming(input)).toThrow(); +}); diff --git a/cloudflare/schedule.ts b/cloudflare/schedule.ts new file mode 100644 index 0000000..701144c --- /dev/null +++ b/cloudflare/schedule.ts @@ -0,0 +1,22 @@ +import { CronExpressionParser } from "cron-parser"; + +export type ScheduleTiming = { interval_seconds?: number; cron?: string; timezone?: string }; +/** Five-field cron has minute precision; intervals are elapsed time independent of timezone. */ +export function normalizeTiming(input: ScheduleTiming): ScheduleTiming { + if ((input.interval_seconds !== undefined) === (input.cron !== undefined)) throw new Error("Provide exactly one of interval_seconds or cron"); + if (input.interval_seconds !== undefined) { + if (!Number.isSafeInteger(input.interval_seconds) || input.interval_seconds < 60 || input.interval_seconds > 31536000) throw new Error("interval_seconds must be between 60 and 31536000"); + if (input.timezone !== undefined) throw new Error("timezone applies only to cron schedules"); + return { interval_seconds: input.interval_seconds }; + } + if (typeof input.cron !== "string" || input.cron.length > 256 || input.cron.trim().split(/\s+/).length !== 5 || /[H?]/.test(input.cron)) throw new Error("cron must be a deterministic five-field expression"); + const timezone=input.timezone??"UTC"; + new Intl.DateTimeFormat("en",{timeZone:timezone}); + const timing={cron:input.cron.trim(),timezone}; + nextOccurrence(timing); + return timing; +} +export function nextOccurrence(timing: ScheduleTiming, now=Date.now()): number { + if (timing.interval_seconds !== undefined) return now+timing.interval_seconds*1000; + return CronExpressionParser.parse(timing.cron!, {currentDate:now,tz:timing.timezone??"UTC"}).next().getTime(); +} diff --git a/cloudflare/scheduling-celld.test.ts b/cloudflare/scheduling-celld.test.ts new file mode 100644 index 0000000..34ba5ce --- /dev/null +++ b/cloudflare/scheduling-celld.test.ts @@ -0,0 +1,54 @@ +import {afterAll,beforeAll,expect,test} from "bun:test"; +import {mkdtemp,rm} from "node:fs/promises"; +import {join} from "node:path"; +import {tmpdir} from "node:os"; +import {ensureCelldRuntime} from "../src/local/celld-runtime"; +const enabled=process.env.CELLD_SCHEDULING_INTEGRATION==="1"; +let directory="",url="",binary="",logs=""; +let child:ReturnType|undefined; +async function start(){ + child=Bun.spawn([binary,"dev",directory,"--host","127.0.0.1","--port",new URL(url).port,"--no-watch"],{cwd:directory,env:{...process.env,CELLD_WORKER_LOADER:"LOADER",CELLD_ESBUILD:join(import.meta.dir,"../node_modules/.bin/esbuild")},stdout:"pipe",stderr:"pipe"}); + for(const stream of [child.stdout,child.stderr]) if(typeof stream!=="number")void(async()=>{for await(const chunk of stream)logs+=new TextDecoder().decode(chunk);})(); + const deadline=Date.now()+45000; + while(Date.now(){ + if(!enabled)return; + directory=await mkdtemp(join(tmpdir(),"canvas-schedule-celld-")); + binary=process.env.CELLD_BIN??await ensureCelldRuntime({dataRoot:join(directory,"runtime")}); + const listener=Bun.listen({hostname:"127.0.0.1",port:0,socket:{data(){}}});url=`http://127.0.0.1:${listener.port}`;listener.stop(true); + const bundle=await Bun.build({entrypoints:[join(import.meta.dir,"scripts-test-worker.ts")],target:"browser",format:"esm",external:["cloudflare:workers","node:*","fs","fs/promises"]}); + if(!bundle.success)throw new Error(bundle.logs.join("\n")); + await Bun.write(join(directory,"worker.js"),await bundle.outputs[0]!.text()); + await Bun.write(join(directory,"wrangler.jsonc"),JSON.stringify({name:"schedule-test",main:"worker.js",compatibility_date:"2026-09-06",compatibility_flags:["nodejs_compat"],durable_objects:{bindings:[{name:"SCRIPTS",class_name:"ScriptLibrary"},{name:"SCRIPT_BACKENDS",class_name:"ScriptBackend"},{name:"LINKS",class_name:"ArtifactLinks"}]},migrations:[{tag:"v1",new_sqlite_classes:["ScriptLibrary","ScriptBackend","ArtifactLinks"]}]})); + await start(); +},180000); +afterAll(async()=>{await stop();if(directory)await rm(directory,{recursive:true,force:true});}); +async function call(path:string,input:unknown={}){const response=await fetch(`${url}/${path}?library=scheduled`,{method:"POST",body:JSON.stringify(input)});const value=await response.json() as any;if(!response.ok)throw new Error(JSON.stringify(value)+logs);return value.result;} +(enabled?test:test.skip)("celld persists schedules and run history across process restart and delivers native alarms",async()=>{ + const identity={workspace:"default",name:"handler"}; + const code=`export default {fetch(request,env){if(request.headers.get("x-canvas-internal-run")!=="caller-value")return new Response("header changed",{status:422});env.sql.exec("create table if not exists count(n integer)");env.sql.exec("insert into count values(1)");console.log("tick");return new Response("ok")}}`; + const draft=await call("writeDraft",{...identity,source:code}); + const active=await call("activate",{...identity,source_hash:draft.source_hash,code}); + await call("link",{hash:active.hash}); + await call("backend/schedule",{action:"set",identity:{...identity,libraryKey:"scheduled",origin:url},cron:"* * * * *",timezone:"UTC",request:{path:"/tick",method:"POST",headers:[["x-canvas-internal-run","caller-value"]]}}); + await call("backend/schedule",{action:"pause"}); + await call("backend/schedule",{action:"run_now"}); + expect((await call("backend/runs"))[0]).toMatchObject({status:"succeeded",trigger:"manual"}); + await stop();await start(); + expect(await call("backend/schedule")).toMatchObject({paused:true,cron:"* * * * *",timezone:"UTC"}); + expect(await call("backend/runs")).toHaveLength(1); + await call("backend/schedule",{action:"resume"}); + await call("backend/due"); + const deadline=Date.now()+20000;let runs:any[]=[]; + while(Date.now()r.trigger==="schedule"&&r.status!=="running"))break;await Bun.sleep(100);} + expect(runs).toHaveLength(2); + expect(runs[0]).toMatchObject({trigger:"schedule",status:"succeeded"}); + const runLogs=await call("backend/logs",{run_id:runs[0].id}); + expect(runLogs.some((log:any)=>log.message==="tick"&&log.run_id===runs[0].id)).toBe(true); + await call("backend/repeatAlarm"); + expect(await call("backend/runs")).toHaveLength(2); + await call("backend/schedule",{action:"pause"}); +},90000); diff --git a/cloudflare/script-backend.ts b/cloudflare/script-backend.ts index 03947c3..2b7aee9 100644 --- a/cloudflare/script-backend.ts +++ b/cloudflare/script-backend.ts @@ -1,15 +1,26 @@ +import { normalizeTiming, nextOccurrence, type ScheduleTiming } from "./schedule"; import { DurableObject } from "cloudflare:workers"; +import type { ScriptLibrary } from "./scripts"; +import type { ArtifactLinks } from "./links"; +import { nativeRequest } from "./backend"; +import type { CanvasHttpRequest } from "../src/httpTypes"; import { sha256 } from "./library"; -export type ScriptLog = { timestamp: string; level: string; message: string }; -export type ScriptRequest = {code: string; hash: string; secrets: Record; request: Request}; -type Env = { LOADER: WorkerLoader }; +export type ScriptLog = { timestamp: string; level: string; message: string; run_id?: string }; +export type ScriptRequest = {code: string; hash: string; secrets: Record; request: Request; trigger?: "http" | "manual" | "schedule"; run_id?: string}; +export type ScheduleIdentity = {libraryKey: string; workspace: string; name: string; origin: string}; +export type ScriptSchedule = ScheduleTiming & { paused: boolean; next_run_at: number | null; request: CanvasHttpRequest}; +export type ScriptRun = {id: string; revision: string; trigger: string; started_at: string; finished_at: string | null; duration_ms: number | null; status: string; http_status: number | null}; +type Env = { LOADER: WorkerLoader; SCRIPTS: DurableObjectNamespace; LINKS: DurableObjectNamespace }; const runtimeConfig = { compatibilityDate: "2026-09-06", compatibilityFlags: ["nodejs_compat"], limits: { cpuMs: 30000, subRequests: 50 } }; declare abstract class ScriptFacet extends DurableObject { takeLogs(): ScriptLog[]; } +const RUN_HEADER = "x-canvas-internal-run"; // Each script has its own isolate and native SQLite facet. Only the explicit // script secrets are supplied; the application's environment is never copied. -const logging = `import { env } from "cloudflare:workers"; +const logging = `import { AsyncLocalStorage } from "node:async_hooks"; +const secrets = __CANVAS_SCRIPT_SECRETS__; +export const execution = new AsyncLocalStorage(); let logs = []; const clean = (value, secrets) => { let message; @@ -19,16 +30,23 @@ const clean = (value, secrets) => { return message.slice(0, 2048); }; for (const level of ["log", "info", "warn", "error", "debug"]) console[level] = (...args) => { - logs.push({timestamp: new Date().toISOString(), level, message: args.slice(0, 16).map(value => clean(value, env.secrets)).join(" ").slice(0, 2048)}); + logs.push({run_id: execution.getStore(), timestamp: new Date().toISOString(), level, message: args.slice(0, 16).map(value => clean(value, secrets)).join(" ").slice(0, 2048)}); if (logs.length > 100) logs.shift(); }; export function takeLogs() { const result = logs; logs = []; return result; } `; const adapter = `import { DurableObject } from "cloudflare:workers"; -import { takeLogs } from "./logging.js"; +import { takeLogs, execution } from "./logging.js"; import handler from "./user.js"; export class ScriptRuntime extends DurableObject { - async fetch(request) { + fetch(incoming) { + const {id, original} = JSON.parse(incoming.headers.get("${RUN_HEADER}")); + const request = new Request(incoming); + if (original === null) request.headers.delete("${RUN_HEADER}"); + else request.headers.set("${RUN_HEADER}", original); + return execution.run(id, () => this.handle(request)); + } + async handle(request) { try { const response = await handler.fetch(request, {secrets: this.env.secrets, sql: this.ctx.storage.sql}, { waitUntil: promise => this.ctx.waitUntil(Promise.resolve(promise).catch(error => {console.error(error);})), @@ -49,6 +67,11 @@ export class ScriptBackend extends DurableObject { constructor(state: DurableObjectState, env: Env) { super(state,env); state.storage.sql.exec("create table if not exists execution_logs(id integer primary key autoincrement,timestamp text not null,level text not null,message text not null)"); + if (!state.storage.sql.exec<{name: string}>("pragma table_info(execution_logs)").toArray().some(column => column.name === "run_id")) state.storage.sql.exec("alter table execution_logs add column run_id text"); + state.storage.sql.exec("create table if not exists execution_runs(id text primary key,revision text not null,trigger text not null,started_at text not null,finished_at text,duration_ms integer,status text not null,http_status integer)"); + // A previous instance may have performed side effects before it stopped. + // Preserve uncertainty; never silently replay those invocations. + state.storage.sql.exec("update execution_runs set status='interrupted' where status='running'"); } private cacheKey(input: Omit) { return sha256(JSON.stringify([this.ctx.id.toString(), input.code, input.secrets, runtimeConfig, adapter, logging])); @@ -57,7 +80,7 @@ export class ScriptBackend extends DurableObject { const hash=this.cacheKey(input); return this.env.LOADER.get(hash,async()=>({ ...runtimeConfig, - mainModule:"adapter.js",modules:{"adapter.js":adapter,"logging.js":logging,"user.js":input.code}, + mainModule:"adapter.js",modules:{"adapter.js":adapter,"logging.js":logging.replace("__CANVAS_SCRIPT_SECRETS__", () => JSON.stringify(input.secrets)),"user.js":input.code}, env:{secrets:input.secrets}, })); } @@ -83,28 +106,106 @@ export class ScriptBackend extends DurableObject { return {class:worker.getDurableObjectClass("ScriptRuntime")}; }); this.facet=facet; - const started=Date.now(); + const started=Date.now(), runId=input.run_id ?? crypto.randomUUID(); + this.ctx.storage.sql.exec("insert or ignore into execution_runs(id,revision,trigger,started_at,status) values(?,?,?,?,?)",runId,input.hash,input.trigger??"http",new Date(started).toISOString(),"running"); try { - const response=await facet.fetch(input.request); - this.record({timestamp:new Date().toISOString(),level:"request",message:`${input.request.method} ${response.status} ${Date.now()-started}ms`}); + // Use the native fetch bridge: celld's generic RPC cannot clone Request. + // Restore the caller's header before user code sees the request. + const request = new Request(input.request); + request.headers.set(RUN_HEADER, JSON.stringify({ id: runId, original: request.headers.get(RUN_HEADER) })); + const response=await facet.fetch(request); + this.finish(runId,started,response.status); + this.record({run_id:runId,timestamp:new Date().toISOString(),level:"request",message:`${input.request.method} ${response.status} ${Date.now()-started}ms`}); await this.collect(facet); return response; } catch { - this.record({timestamp:new Date().toISOString(),level:"error",message:"Script execution failed"}); + this.finish(runId,started,500); + this.record({run_id:runId,timestamp:new Date().toISOString(),level:"error",message:"Script execution failed"}); return new Response("Script execution failed",{status:500}); } } + private finish(id: string, started: number, status: number) { + this.ctx.storage.sql.exec("update execution_runs set finished_at=?,duration_ms=?,status=?,http_status=? where id=?",new Date().toISOString(),Date.now()-started,status>=400?"failed":"succeeded",status,id); + this.ctx.storage.sql.exec("delete from execution_runs where status!='running' and id not in (select id from execution_runs order by started_at desc limit 1000)"); + } + runs(input: {limit?: number} = {}) { + const limit=input.limit??100; + if (!Number.isSafeInteger(limit)||limit<1||limit>100) throw new Error("limit must be between 1 and 100"); + return this.ctx.storage.sql.exec("select * from execution_runs order by started_at desc,rowid desc limit ?",limit).toArray(); + } + async schedule(input: {identity?: ScheduleIdentity; interval_seconds?: number; cron?: string; timezone?: string; request?: CanvasHttpRequest; action?: "get"|"set"|"pause"|"resume"|"run_now"} = {}): Promise { + const action=input.action??"get"; + if (!["get","set","pause","resume","run_now"].includes(action)) throw new Error("Unknown schedule action"); + if (action==="get") return await this.ctx.storage.get("schedule")??null; + if (action==="run_now") { await this.runScheduled("manual"); return await this.schedule(); } + let next: ScriptSchedule|null=null; + await this.ctx.storage.transaction(async storage => { + const previous=await storage.get("schedule"); + if (action!=="set" && !previous) throw new Error("No schedule configured"); + if (action==="set") { + const timing=normalizeTiming(input); + if (!input.identity) throw new Error("Schedule identity required"); + const request=input.request??{path:"/",method:"GET",headers:[]}; + nativeRequest(request); // Validate before persisting any state. + next={...timing,request,paused:false,next_run_at:nextOccurrence(timing)}; + } else next={...previous!,paused:action==="pause",next_run_at:action==="pause"?null:nextOccurrence(previous!)}; + if (input.identity) await storage.put("schedule_identity",input.identity); + await storage.put("schedule",next); + if (next.next_run_at===null) await storage.deleteAlarm(); else await storage.setAlarm(next.next_run_at); + }); + return next; + } + async alarm() { + // Commit the next occurrence before invoking user code. Alarm redelivery + // then sees a future timestamp and cannot replay this occurrence. + let due: {id:string;started:number;schedule:ScriptSchedule}|undefined; + await this.ctx.storage.transaction(async storage => { + due=undefined; + const schedule=await storage.get("schedule"); + if (!schedule || schedule.paused || schedule.next_run_at===null) return; + if (schedule.next_run_at>Date.now()) { await storage.setAlarm(schedule.next_run_at); return; } + const next=nextOccurrence(schedule); + await storage.put("schedule",{...schedule,next_run_at:next}); + await storage.setAlarm(next); + const id=crypto.randomUUID(),started=Date.now(); + this.ctx.storage.sql.exec("insert into execution_runs(id,revision,trigger,started_at,status) values(?,?,?,?,?)",id,"unavailable","schedule",new Date(started).toISOString(),"running"); + due={id,started,schedule}; + }); + if (due) await this.runScheduled("schedule",due); + } + private async runScheduled(trigger: "manual"|"schedule",claimed?:{id:string;started:number;schedule:ScriptSchedule}) { + const schedule=claimed?.schedule??await this.ctx.storage.get("schedule"); + const identity=await this.ctx.storage.get("schedule_identity"); + if (!schedule || !identity) throw new Error("No schedule configured"); + const id=claimed?.id??crypto.randomUUID(), started=claimed?.started??Date.now(); + this.ctx.storage.sql.exec("insert or ignore into execution_runs(id,revision,trigger,started_at,status) values(?,?,?,?,?)",id,"unavailable",trigger,new Date(started).toISOString(),"running"); + try { + const {libraryKey,workspace,name,origin}=identity; + const link=await this.env.LINKS.getByName("deployment").find({libraryKey,workspace,name,kind:"script"}); + if (!link?.script_hash) throw new Error("No active revision"); + const active=await this.env.SCRIPTS.getByName(libraryKey).active({workspace,name,hash:link.script_hash}); + this.ctx.storage.sql.exec("update execution_runs set revision=? where id=?",active.hash,id); + const incoming=nativeRequest(schedule.request); + const request=new Request(new URL(new URL(incoming.url).pathname+new URL(incoming.url).search,origin),incoming); + const response=await this.request({...active,request,trigger,run_id:id}); + await response.body?.cancel(); + } catch { + this.ctx.storage.sql.exec("insert or ignore into execution_runs(id,revision,trigger,started_at,status) values(?,?,?,?,?)",id,"unavailable",trigger,new Date(started).toISOString(),"running"); + this.finish(id,started,500); + this.record({run_id:id,timestamp:new Date().toISOString(),level:"error",message:"Scheduled execution failed; inspect the active revision and its configuration"}); + } + } private record(log: ScriptLog) { - this.ctx.storage.sql.exec("insert into execution_logs(timestamp,level,message) values(?,?,?)",log.timestamp,log.level,log.message.slice(0,2048)); + this.ctx.storage.sql.exec("insert into execution_logs(timestamp,level,message,run_id) values(?,?,?,?)",log.timestamp,log.level,log.message.slice(0,2048),log.run_id??null); this.ctx.storage.sql.exec("delete from execution_logs where id not in (select id from execution_logs order by id desc limit 100)"); } private async collect(facet: Fetcher) { try { for (const log of (await facet.takeLogs()).slice(-100)) this.record(log); } catch { /* An aborted isolate cannot supply pending logs. */ } } - async logs(input: {limit?: number} = {}) { + async logs(input: {limit?: number; run_id?: string} = {}) { const limit=input.limit??100; if (!Number.isSafeInteger(limit) || limit<1 || limit>100) throw new Error("limit must be between 1 and 100"); if (this.facet) await this.collect(this.facet); - return this.ctx.storage.sql.exec("select timestamp,level,message from execution_logs order by id desc limit ?",limit).toArray(); + return this.ctx.storage.sql.exec("select timestamp,level,message,run_id from execution_logs where (? is null or run_id=?) order by id desc limit ?",input.run_id??null,input.run_id??null,limit).toArray(); } } diff --git a/cloudflare/script-guide.ts b/cloudflare/script-guide.ts index 1aec33e..936707e 100644 --- a/cloudflare/script-guide.ts +++ b/cloudflare/script-guide.ts @@ -8,10 +8,10 @@ Use script_secrets({name,secrets:{TOKEN:"value",OLD_TOKEN:null}}) to set/delete script_write and script_edit save source before validation. Inspect applied, ok, and diagnostics: failed drafts remain editable, while the URL keeps serving the last validated code. No separate publish operation is needed. script_read reads draft or historical source, script_history lists revisions, script_version reads one, and script_restore restores and validates a prior revision. script_list, script_read, and gallery views never execute handlers. -script_run({name,request:{path:"/chosen-slug?key=value",method:"POST",headers:[["content-type","text/plain"]],body:"aGVsbG8="}}) explicitly invokes the last validated handler. The request path and query are passed exactly as provided, using the deployment origin; include the slug in path to reproduce a direct URL request. Body is base64. MCP requests and responses are limited to 256 KiB; use direct URLs for streaming or larger responses. script_logs reads bounded recent execution logs and errors. Direct URLs preserve incoming paths, methods and body bytes and return the handler response. Management cookies and Cloudflare Access credentials are stripped; private routes also strip Authorization. Public Authorization is passed to the handler. Script HTML gets an opaque-origin sandbox and responses cannot set cookies. Public handlers can implement their own authentication, including provider signature checks. Cloudflare Access policies protecting the entire hostname need a bypass for intended public paths. No schedules or separate publishing workflow are provided. Execution has 30 seconds CPU and 50 subrequests; logs retain 100 entries of up to 2 KiB with literal secret redaction. +script_run({name,request:{path:"/chosen-slug?key=value",method:"POST",headers:[["content-type","text/plain"]],body:"aGVsbG8="}}) explicitly invokes the last validated handler. The request path and query are passed exactly as provided, using the deployment origin; include the slug in path to reproduce a direct URL request. Body is base64. MCP requests and responses are limited to 256 KiB; use direct URLs for streaming or larger responses. script_logs reads bounded recent execution logs and errors. Direct URLs preserve incoming paths, methods and body bytes and return the handler response. Management cookies and Cloudflare Access credentials are stripped; private routes also strip Authorization. Public Authorization is passed to the handler. Script HTML gets an opaque-origin sandbox and responses cannot set cookies. Public handlers can implement their own authentication, including provider signature checks. Cloudflare Access policies protecting the entire hostname need a bypass for intended public paths. No separate publishing workflow is required. Execution has 30 seconds CPU and 50 subrequests; logs retain 100 entries of up to 2 KiB with literal secret redaction. -Remix -Use canvas_remix/script_remix with new_name and exactly one source name or version_id. Source provenance is retained, runtime data and configured secrets start empty, and hosted links start private. +Schedules and run history +Use script_schedule or canvas_schedule with name, action:set, request, and exactly one of interval_seconds (at least 60) or a five-field cron expression with optional IANA timezone (UTC default). Native Durable Object alarms run the latest validated revision on Cloudflare and celld. get, pause, resume, and run_now control the saved schedule. Missed occurrences are skipped; uncertain or failed runs are not automatically replayed. script_runs/canvas_runs expose up to 100 of 1000 retained host-owned run records, separate from application SQLite. Filter script_logs by run_id. HTTP-handler success does not imply waitUntil or streamed response completion. Remix with canvas_remix/script_remix, new_name and exactly one source name or version_id; source provenance is retained, runtime data and configured secrets start empty, and hosted links start private. Third-party dependencies Pass project:{files:{"lib/message.ts":"export const message = 'Hello';"},dependencies:{hono:"4.13.7"}} to script_write or canvas_write. Import helper modules with relative paths and packages by name. Omit project on write to preserve it; supply files/dependencies as full replacements to change or clear them. Dependency declarations require exact versions; tags, ranges, arbitrary registry URLs and uploaded lock contents are rejected. Registry requests and compressed/uncompressed package sizes are bounded, tarball integrity is checked, install scripts never run, and the complete dependency source snapshot is restored with each revision. The compiler supports one resolved version of each package; conflicting transitive requirements produce an explicit error. Required peers must be declared, except supported browser packages such as React supplied by Canvas's own runtime. Native addons and packages requiring a Node process cannot run in Workers. diff --git a/cloudflare/script-service.ts b/cloudflare/script-service.ts index 3a8474f..8019050 100644 --- a/cloudflare/script-service.ts +++ b/cloudflare/script-service.ts @@ -1,3 +1,4 @@ +import type { ScriptSchedule } from "./script-backend"; import { resolveProject } from "./project"; import type { CallToolResult } from "@modelcontextprotocol/sdk/types.js"; import { createHash } from "node:crypto"; @@ -52,7 +53,17 @@ export class CloudScriptService { return text({ versions, next_offset: versions.length === 100 ? offset + 100 : null }); } case "script_version": return text(await scripts.version({ workspace: this.artifacts.workspace, id: args.version_id as string })); - case "script_logs": return text({ logs: await hosted.scriptBackends.getByName(JSON.stringify([this.artifacts.libraryKey, this.artifacts.workspace, this.artifacts.target("script", name).name])).logs({ limit: args.limit as number | undefined }) }); + case "script_schedule": { + const target=this.artifacts.target("script",name); + const backend=hosted.scriptBackends.getByName(JSON.stringify([this.artifacts.libraryKey,this.artifacts.workspace,target.name])); + if (args.action==="set") { + const link=await hosted.links.find(target); + if (!link?.script_hash) throw new Error("Script has no validated URL revision"); + } + return text({schedule:await backend.schedule({action:args.action as "get"|"set"|"pause"|"resume"|"run_now"|undefined,interval_seconds:args.interval_seconds as number|undefined,cron:args.cron as string|undefined,timezone:args.timezone as string|undefined,request:args.request as ScriptSchedule["request"]|undefined,...(args.action==="set"?{identity:{libraryKey:this.artifacts.libraryKey,workspace:this.artifacts.workspace,name:target.name,origin:hosted.origin}}:{})})}); + } + case "script_runs": return text({runs:await hosted.scriptBackends.getByName(JSON.stringify([this.artifacts.libraryKey,this.artifacts.workspace,this.artifacts.target("script",name).name])).runs({limit:args.limit as number|undefined})}); + case "script_logs": return text({ logs: await hosted.scriptBackends.getByName(JSON.stringify([this.artifacts.libraryKey, this.artifacts.workspace, this.artifacts.target("script", name).name])).logs({ limit: args.limit as number | undefined, run_id: args.run_id as string | undefined }) }); case "script_secrets": { if (args.secrets !== undefined) { if (!args.secrets || typeof args.secrets !== "object" || Array.isArray(args.secrets)) throw new Error("secrets must be an object"); @@ -66,7 +77,7 @@ export class CloudScriptService { const active = await scripts.active({ ...input, hash: link.script_hash }); const incoming = nativeRequest(args.request as CanvasHttpRequest); const request = new Request(new URL(new URL(incoming.url).pathname + new URL(incoming.url).search, hosted.origin), incoming); - const response = await hosted.scriptBackends.getByName(JSON.stringify([this.artifacts.libraryKey, this.artifacts.workspace, this.artifacts.target("script", name).name])).request({ ...active, request }); + const response = await hosted.scriptBackends.getByName(JSON.stringify([this.artifacts.libraryKey, this.artifacts.workspace, this.artifacts.target("script", name).name])).request({ ...active, request, trigger: "manual" }); const envelope = await scriptResponse(response, incoming.method); return { ...text({ status: envelope.status, ...(link ? { url: `${hosted.origin}/${link.slug}` } : {}) }), structuredContent: { response: envelope } }; } diff --git a/cloudflare/scripts-test-worker.ts b/cloudflare/scripts-test-worker.ts index 74ed45d..f48566d 100644 --- a/cloudflare/scripts-test-worker.ts +++ b/cloudflare/scripts-test-worker.ts @@ -1,14 +1,22 @@ export { ScriptLibrary } from "./scripts"; -export { ScriptBackend } from "./script-backend"; +import { ScriptBackend as BaseScriptBackend, type ScriptSchedule } from "./script-backend"; +export {ArtifactLinks} from "./links"; +import type {ArtifactLinks} from "./links"; +export class ScriptBackend extends BaseScriptBackend { + async repeatAlarm() { await this.alarm(); } + async due() { const schedule=await this.ctx.storage.get("schedule"); await this.ctx.storage.put("schedule",{...schedule,next_run_at:Date.now()}); await this.ctx.storage.setAlarm(Date.now()+10); } +} import type { ScriptLibrary } from "./scripts"; -import type { ScriptBackend } from "./script-backend"; -type Env = { SCRIPTS: DurableObjectNamespace; SCRIPT_BACKENDS: DurableObjectNamespace }; + +type Env = { LINKS: DurableObjectNamespace; SCRIPTS: DurableObjectNamespace; SCRIPT_BACKENDS: DurableObjectNamespace }; export default { async fetch(request: Request, env: Env) { const url = new URL(request.url), name=url.searchParams.get("library")??"alice"; if (url.pathname==="/run") { const input=await request.json() as {code: string; secrets: Record; body?: string; method?: string}; return env.SCRIPT_BACKENDS.getByName(name).request({code:input.code,hash:"unused",secrets:input.secrets,request:new Request("https://scripts.invalid/chosen/path?one=two",{method:input.method??"POST",headers:{"x-original":"yes"},body:input.body??"original body"})}); } + if (url.pathname.startsWith("/backend/")) { try {const stub=env.SCRIPT_BACKENDS.getByName(name) as unknown as RecordPromise>;return Response.json({result:await stub[url.pathname.slice(9)]!(await request.json())});}catch(e){return Response.json({error:String(e)},{status:400});} } + if (url.pathname==="/link") {const input=await request.json() as {hash:string};const links=env.LINKS.getByName("deployment");const target={libraryKey:name,workspace:"default",name:"handler",kind:"script" as const};const generation=await links.begin(target);return Response.json({result:await links.commit(target,generation,{slug:name,access:"private",script_hash:input.hash})});} if (url.pathname==="/logs") return Response.json(await env.SCRIPT_BACKENDS.getByName(name).logs({})); if (url.pathname==="/validate") return Response.json(await env.SCRIPT_BACKENDS.getByName(name).validate(await request.json())); try { diff --git a/cloudflare/scripts.test.ts b/cloudflare/scripts.test.ts index 6625d5c..d26ec3b 100644 --- a/cloudflare/scripts.test.ts +++ b/cloudflare/scripts.test.ts @@ -4,7 +4,7 @@ let runtime: Miniflare; beforeAll(async()=>{ const build=await Bun.build({entrypoints:[new URL("./scripts-test-worker.ts",import.meta.url).pathname],target:"browser",format:"esm",external:["cloudflare:workers","node:*","fs","fs/promises"]}); if (!build.success) throw new Error(build.logs.join("\n")); - runtime=new Miniflare({unsafeInspectDurableObjects:true,cf:false,port:0,workers:[{config:{name:"scripts-test",type:"worker",compatibilityDate:"2026-09-06",compatibilityFlags:["nodejs_compat"],manifest:{mainModule:"test.js",modulesRoot:import.meta.dir,modules:{"test.js":{type:"esm",contents:await build.outputs[0]!.text()}}},env:{SCRIPTS:{type:"durable-object",worker:"scripts-test",exportName:"ScriptLibrary"},SCRIPT_BACKENDS:{type:"durable-object",worker:"scripts-test",exportName:"ScriptBackend"},LOADER:{type:"worker-loader"}},exports:{ScriptLibrary:{type:"durable-object",storage:"sqlite"},ScriptBackend:{type:"durable-object",storage:"sqlite"}}},dev:{outboundService:{type:"fetcher",handler:()=>new MFResponse("upstream success")}}}]}); + runtime=new Miniflare({unsafeInspectDurableObjects:true,cf:false,port:0,workers:[{config:{name:"scripts-test",type:"worker",compatibilityDate:"2026-09-06",compatibilityFlags:["nodejs_compat"],manifest:{mainModule:"test.js",modulesRoot:import.meta.dir,modules:{"test.js":{type:"esm",contents:await build.outputs[0]!.text()}}},env:{LINKS:{type:"durable-object",worker:"scripts-test",exportName:"ArtifactLinks"},SCRIPTS:{type:"durable-object",worker:"scripts-test",exportName:"ScriptLibrary"},SCRIPT_BACKENDS:{type:"durable-object",worker:"scripts-test",exportName:"ScriptBackend"},LOADER:{type:"worker-loader"}},exports:{ArtifactLinks:{type:"durable-object",storage:"sqlite"},ScriptLibrary:{type:"durable-object",storage:"sqlite"},ScriptBackend:{type:"durable-object",storage:"sqlite"}}},dev:{outboundService:{type:"fetcher",handler:()=>new MFResponse("upstream success")}}}]}); await runtime.ready; }); afterAll(async()=>{await runtime?.dispose();}); @@ -104,6 +104,63 @@ test("remix copies an immutable source revision, never secrets or activation, an await expect(call("remix",{workspace,name,version_id:version.id,new_name:"ambiguous"})).rejects.toThrow("not both"); }); +test("run history records HTTP outcomes and correlates concurrent logs without exposing host tables",async()=>{ + const code=`export default {async fetch(request,env){const tag=await request.text(); console.log('before',tag); await fetch('https://upstream.example'); console.log('after',tag); const host=env.sql.exec("select name from sqlite_master where name='execution_runs'").toArray();return Response.json({host,internal:request.headers.get("x-canvas-internal-run")});}}`; + const responses=await Promise.all([run(code,{},"history","first"),run(code,{},"history","second")]); + for(const response of responses) expect(await response.json()).toEqual({host:[],internal:null}); + const runs=await call("backend/runs",{},"history"); + expect(runs).toHaveLength(2); + for(const entry of runs) { + expect(entry).toMatchObject({status:"succeeded",trigger:"http",http_status:200,revision:"unused"}); + expect(entry.duration_ms).toBeGreaterThanOrEqual(0); + const logs=await call("backend/logs",{run_id:entry.id},"history"); + expect(logs.every((log:any)=>log.run_id===entry.id)).toBe(true); + const messages=logs.filter((log:any)=>log.level==="log").map((log:any)=>log.message); + expect(messages).toHaveLength(2); + expect(messages[0].split(" ")[1]).toBe(messages[1].split(" ")[1]); + } + await run('export default {fetch(){return new Response("bad",{status:503})}}',{},"history"); + expect((await call("backend/runs",{limit:1},"history"))[0]).toMatchObject({status:"failed",http_status:503}); + await expect(call("backend/runs",{limit:101},"history")).rejects.toThrow("limit"); +}); + +test("native alarms execute current revision and secrets, pause/resume and run-now preserve configuration, redelivery is skipped",async()=>{ + const library="scheduled"; + const code=`export default {fetch(request,env){console.log(env.secrets.TOKEN);env.sql.exec("create table if not exists ticks(n integer)");env.sql.exec("insert into ticks values(1)");return new Response("ok");}}`; + const draft=await call("writeDraft",{...identity,source:code},library); + const active=await call("activate",{...identity,source_hash:draft.source_hash,code},library); + await call("link",{hash:active.hash},library); + await call("setSecret",{...identity,key:"TOKEN",value:"schedule-secret"},library); + const config={action:"set",identity:{libraryKey:library,...identity,origin:"https://scripts.invalid"},interval_seconds:60,request:{path:"/tick",method:"POST",headers:[]}}; + const schedule=await call("backend/schedule",config,library); + expect(schedule).toMatchObject({interval_seconds:60,paused:false}); + expect(schedule.next_run_at).toBeGreaterThan(Date.now()); + expect(await call("backend/runs",{},library)).toEqual([]); + await expect(call("backend/schedule",{...config,interval_seconds:0},library)).rejects.toThrow("interval_seconds"); + expect(await call("backend/schedule",{},library)).toEqual(schedule); + await call("backend/schedule",{action:"pause"},library); + expect(await call("backend/schedule",{},library)).toMatchObject({paused:true,next_run_at:null}); + await call("backend/schedule",{action:"run_now"},library); + expect((await call("backend/runs",{},library))[0]).toMatchObject({trigger:"manual",status:"succeeded",revision:active.hash}); + await call("backend/schedule",{action:"resume"},library); + await call("backend/due",{},library); + const deadline=Date.now()+5000; + let runs:any[]=[]; + while(Date.now()r.trigger==="schedule"&&r.status!=="running"))break;await Bun.sleep(25);} + expect(runs).toHaveLength(2); + expect(runs[0]).toMatchObject({trigger:"schedule",status:"succeeded"}); + await call("backend/repeatAlarm",{},library); // duplicate alarm delivery + expect(await call("backend/runs",{},library)).toHaveLength(2); + expect(JSON.stringify(await call("backend/logs",{},library))).not.toContain("schedule-secret"); + const newer=code.replace('new Response("ok")','new Response("failed",{status:503})'); + const revision=await call("writeDraft",{...identity,source:newer},library); + const activated=await call("activate",{...identity,source_hash:revision.source_hash,code:newer},library); + await call("link",{hash:activated.hash},library); + await call("backend/schedule",{action:"run_now"},library); + expect((await call("backend/runs",{limit:1},library))[0]).toMatchObject({status:"failed",revision:activated.hash,http_status:503}); + await call("backend/schedule",{action:"pause"},library); +}); + test("project changes invalidate script validation and restore complete module snapshots", async () => { const identity = { workspace: "projects", name: "handler" }; const project = { files: { "helper.ts": "export const answer = 1;", "other.ts": "unchanged" }, dependencies: { "lodash-es": "4.17.21" }, lock: { "node_modules/lodash-es/index.js": "export const version = 1;" } }; diff --git a/cloudflare/service.ts b/cloudflare/service.ts index d747ffa..a77a90d 100644 --- a/cloudflare/service.ts +++ b/cloudflare/service.ts @@ -4,6 +4,7 @@ import type { DurableObjectStub, DurableObjectNamespace } from "@cloudflare/work import type { CallToolResult } from "@modelcontextprotocol/sdk/types.js"; import type { CanvasLibrary, CanvasEdit, CompiledCanvas } from "./library"; import { compileCanvasSource, compileCanvasServerSource, typecheckCanvasSource, typecheckCanvasServerSource } from "./compiler"; +import type { CanvasScheduleInput } from "./canvas-execution"; import type { CanvasBackend } from "./backend"; import type { LinkUpdate } from "./links"; import { ArtifactService, type HostedArtifacts } from "./artifact-service"; @@ -120,8 +121,31 @@ export class CloudCanvasService { const { _meta, ...payload } = preview; return { ...text(payload, !preview.ok), structuredContent: payload, ...(_meta ? { _meta } : {}) }; } + case "canvas_schedule": + case "canvas_runs": { + const snapshot = await this.snapshot({ name: args.name as string }); + const backend = this.backends.getByName(JSON.stringify([this.libraryKey, this.workspace, snapshot.name])); + if (name === "canvas_runs") { + const payload = { runs: await backend.runs({ limit: args.limit as number | undefined }) }; + return { ...text(payload), structuredContent: payload }; + } + // Existing canvases gain a validated active revision when first scheduled. + // Reads and pause remain usable while a draft is invalid. + if (args.action === "set" || args.action === "resume") { + const current = await backend.activeRevision(); + if (!current) { + const compiled = await this.compile(snapshot); + if (!compiled.ok) throw new Error(formatCanvasCheck(compiled.diagnostics)); + const preview = await this.preview(snapshot, compiled); + if (preview.ok && "canvas" in preview) await this.activate(snapshot, preview.canvas.versionId, preview.canvas.revision, compiled); + } + } + const { name: _name, ...input } = args; + const payload = { schedule: await backend.schedule(input as CanvasScheduleInput) }; + return { ...text(payload), structuredContent: payload }; + } case "canvas_request": { - const response = await this.request({ name: args.name as string | undefined, version_id: args.version_id as string | undefined }, args.request as CanvasHttpRequest); + const response = await this.request({ name: args.name as string | undefined, version_id: args.version_id as string | undefined }, args.request as CanvasHttpRequest, "manual"); return { ...text({ status: response.status }), structuredContent: { response } }; } default: throw new Error("Unknown tool"); @@ -144,7 +168,7 @@ export class CloudCanvasService { runtime: compiled.artifact.runtime, compiled_id: compiled.artifact.id, initial_state: snapshot.state, mode: "preview", ...(snapshot.version_id ? { version_id: snapshot.version_id } : {}), }); - const canvas = { name: version.name, versionId: version.id, eventId: event.id, sourceHash: version.source_hash }; + const canvas = { name: version.name, versionId: version.id, revision: version.revision, eventId: event.id, sourceHash: version.source_hash }; return { ...base, ok: true, canvas, _meta: { canvas: { ...canvas, js: compiled.js, state: snapshot.state, server: snapshot.server_source !== null, ...(this.plugins ? { plugins: true } : {}), ...(this.fileStorage ? { files: true } : {}) } satisfies CanvasAppPayload } }; } @@ -173,6 +197,7 @@ export class CloudCanvasService { return { ...text(superseded, true), structuredContent: superseded }; } } + if (details.ok && "canvas" in details) await this.activate(mutation, details.canvas.versionId, details.canvas.revision, compiled); const result = { ...payload, ...await this.artifacts.linkDetails(target) }; return { ...text(result, !details.ok), structuredContent: { ...result, preview: { ...details, ...await this.artifacts.linkDetails(target) } }, ...(_meta ? { _meta } : {}) }; } catch (error) { @@ -211,7 +236,14 @@ export class CloudCanvasService { }] }; } - async request(selection: { name?: string; version_id?: string }, request: CanvasHttpRequest) { + private async activate(snapshot: Snapshot, version_id: string, revision: number, compiled: Compiled) { + if (!compiled.ok || !compiled.artifact) throw new Error(formatCanvasCheck(compiled.diagnostics)); + const code = compiled.artifact.server_js; + const hash = createHash("sha256").update(code ?? "").digest("hex"); + await this.backends.getByName(JSON.stringify([this.libraryKey, this.workspace, snapshot.name])).activate({ code, hash, version_id, revision }); + } + + async request(selection: { name?: string; version_id?: string }, request: CanvasHttpRequest, trigger: "http" | "manual" = "http") { const snapshot = await this.snapshot(selection.version_id ? { version_id: selection.version_id } : { name: selection.name }); if (selection.name && snapshot.name !== selection.name.trim().replace(/\.canvas\.tsx$/, "")) throw new Error("Server version does not belong to this canvas"); if (snapshot.server_source === null) throw new Error("Canvas has no server"); @@ -220,7 +252,7 @@ export class CloudCanvasService { const code = compiled.artifact.server_js; const hash = createHash("sha256").update(code).digest("hex"); const backend = this.backends.getByName(JSON.stringify([this.libraryKey, this.workspace, snapshot.name])); - return backend.request({ code, hash, request }); + return backend.request({ code, hash, version_id: snapshot.version_id ?? undefined, request, trigger }); } async fileRequest(selection: { name?: string; version_id?: string }, input: CanvasFileRequest, writable = true): Promise { diff --git a/cloudflare/tool-contract.ts b/cloudflare/tool-contract.ts index 29307ee..25d7171 100644 --- a/cloudflare/tool-contract.ts +++ b/cloudflare/tool-contract.ts @@ -102,7 +102,9 @@ addScriptTool("script_run", "Execute a script's last validated handler. request. addScriptTool("script_history", "List script source history, 100 entries per page.", { ...nameProperty, ...pageProperty }); addScriptTool("script_version", "Read an immutable script source revision. Never returns secrets.", { version_id: { type: "string" } }, ["version_id"]); addScriptTool("script_restore", "Restore a script revision as a new draft and activate it if validation succeeds. Persistent storage remains live.", { ...nameProperty, version_id: { type: "string" } }, ["name", "version_id"]); -addScriptTool("script_logs", "Read bounded recent execution logs and errors without executing the script.", { ...nameProperty, limit: { type: "integer", minimum: 1, maximum: 100 } }, ["name"]); +addScriptTool("script_schedule", "Inspect or configure recurring execution of the latest validated script. Use action set with exactly one of interval_seconds or five-field cron (timezone defaults UTC) and an optional request. Pause/resume preserve settings; run_now explicitly invokes the saved request, including while paused. Failed or interrupted runs are never automatically replayed; missed occurrences are skipped.", {...nameProperty,action:{type:"string",enum:["get","set","pause","resume","run_now"]},interval_seconds:{type:"integer",minimum:60,maximum:31536000},cron:{type:"string",maxLength:256},timezone:{type:"string",maxLength:100},request:CANVAS_REQUEST_SCHEMA},["name"]); +addScriptTool("script_runs", "Inspect recent invocation outcomes, revision hashes, triggers, and HTTP-handler durations. Retains 1000 runs, returns the latest 100. Interrupted means completion is unknown; background waitUntil work and streamed response completion are not part of HTTP-handler success.", {...nameProperty,limit:{type:"integer",minimum:1,maximum:100}},["name"]); +addScriptTool("script_logs", "Read bounded recent execution logs and errors without executing the script. Optionally filter by run_id from script_runs.", { ...nameProperty, run_id:{type:"string"}, limit: { type: "integer", minimum: 1, maximum: 100 } }, ["name"]); addScriptTool("script_secrets", "Set or delete per-script secrets (null deletes). Omit secrets to list names. Values are never returned or added to source history.", { ...nameProperty, secrets: { type: "object", maxProperties: 32, additionalProperties: { type: ["string", "null"], maxLength: 4096 } } }, ["name"]); addScriptTool("plugins_list", "List deployment-installed plugins, browser availability, and operation schemas and read-only hints. Does not execute operations.", {}); @@ -114,6 +116,12 @@ addScriptTool("canvas_plugin_call", "Call a deployment-installed server function }, ["plugin", "operation", "input"]); CLOUD_MCP_TOOLS[CLOUD_MCP_TOOLS.length - 1]!.annotations = { readOnlyHint: false, destructiveHint: true, openWorldHint: true }; +addScriptTool("canvas_schedule", "Configure recurring requests to a canvas's latest validated native server. Uses Durable Object alarms, with no automatic retry after uncertain side effects. Exactly one of interval_seconds or cron is required for set; timezone defaults to UTC. run_now executes the configured request, including while paused. Historical previews do not change the scheduled revision.", { + name: { type: "string", minLength: 1 }, action: { type: "string", enum: ["get", "set", "pause", "resume", "run_now"], default: "get" }, + interval_seconds: { type: "integer", minimum: 60, maximum: 31536000 }, cron: { type: "string", maxLength: 256 }, timezone: { type: "string", maxLength: 100 }, request: CANVAS_REQUEST_SCHEMA, +}, ["name"]); +addScriptTool("canvas_runs", "Read recent native canvas server invocations, their source revision, trigger, duration and outcome. Host-owned history is separate from application SQLite. An interrupted run may already have performed writes and is never automatically retried.", { name: { type: "string", minLength: 1 }, limit: { type: "integer", minimum: 1, maximum: 100 } }, ["name"]); +CLOUD_MCP_TOOLS[CLOUD_MCP_TOOLS.length - 1]!.annotations = { readOnlyHint: true }; // Project source is additive to legacy entrypoints; dependency locks are generated by the service. const projectProperty = { type: "object", properties: { files: { type: "object", maxProperties: 64, additionalProperties: { type: "string" }, description: "Relative helper modules (.ts/.tsx/.js/.json). Import from ./path. Full replacement; omit project to preserve existing project." }, diff --git a/cloudflare/worker.test.ts b/cloudflare/worker.test.ts index a850662..536c011 100644 --- a/cloudflare/worker.test.ts +++ b/cloudflare/worker.test.ts @@ -685,3 +685,34 @@ test("hosted remix tools create private independent artifacts and reject overwri expect(await run("remix-script")).toEqual({count:2,secret:"original-only"}); expect((await client.callTool({name:"script_remix",arguments:{name:"remix-script",new_name:"remix-script-copy"}})).isError).toBe(true); }, 120000); + +test("canvas schedules retain validated revisions, keep host history isolated, and pause on server removal", async () => { + const name = "scheduled-counter"; + const write = await client.callTool({ name: "canvas_write", arguments: { name, contents: counterClient, server: counterServer } }); + expect(write.isError).not.toBe(true); + const version = (write._meta as { canvas: { versionId: string } }).canvas.versionId; + const tool = async (nameOfTool: string, args: Record = {}) => { + const result = await client.callTool({ name: nameOfTool, arguments: { name, ...args } }); + expect(result.isError).not.toBe(true); + return result.structuredContent as any; + }; + const saved = await tool("canvas_schedule", { action: "set", interval_seconds: 3600, request: { path: "/counter", method: "POST" } }); + expect(saved.schedule).toMatchObject({ paused: false, interval_seconds: 3600 }); + await tool("canvas_schedule", { action: "pause" }); + const invalid = await client.callTool({ name: "canvas_write", arguments: { name, contents: "export default function {" } }); + expect(invalid.isError).toBe(true); + await tool("canvas_schedule", { action: "run_now" }); + const response = await tool("canvas_request", { version_id: version, request: { path: "/counter" } }); + expect(JSON.parse(atob(response.response.body)).value).toBe(1); + const runs = (await tool("canvas_runs")).runs; + expect(runs).toEqual(expect.arrayContaining([expect.objectContaining({ revision: version, trigger: "manual", status: "succeeded", http_status: 200 })])); + // A canvas's own SQL database cannot inspect supervisor execution history. + const inspectServer = `import { DurableObject } from "cloudflare:workers"; +export class CanvasServer extends DurableObject { fetch() { return Response.json(this.ctx.storage.sql.exec("select name from sqlite_master where name='execution_runs'").toArray()); } }`; + expect((await client.callTool({ name: "canvas_write", arguments: { name, contents: counterClient, server: inspectServer } })).isError).not.toBe(true); + const inspected = await tool("canvas_request", { request: { path: "/" } }); + expect(JSON.parse(atob(inspected.response.body))).toEqual([]); + expect((await client.callTool({ name: "canvas_write", arguments: { name, contents: counterClient, server: null } })).isError).not.toBe(true); + expect((await tool("canvas_schedule")).schedule).toMatchObject({ paused: true, next_run_at: null }); + expect((await client.callTool({ name: "canvas_schedule", arguments: { name, action: "resume" } })).isError).toBe(true); +}, 120000); diff --git a/docs/scripts.md b/docs/scripts.md index 69f2906..069ac29 100644 --- a/docs/scripts.md +++ b/docs/scripts.md @@ -110,3 +110,15 @@ Agents can call `script_guide` to retrieve the authoring contract through MCP. Select a working copy or historical revision in the gallery, then **Remix** and choose a new name. MCP offers `canvas_remix` and `script_remix`, with `new_name` and exactly one of `name` or `version_id`. Hosted calls optionally accept a new `slug`; otherwise it defaults to the destination name. Sources and destinations belong to the authenticated library and selected workspace. A remix creates a separate artifact and records its immutable source revision as provenance. Canvas browser and server sources stay paired. The new artifact starts with empty runtime data and no copied secret values; its hosted URL starts private. Destination names cannot overwrite existing drafts or archived artifacts, and URL collisions are rejected. Invalid source remains saved as a remix draft, with validation diagnostics; it receives a working URL only after successful validation. Remixing does not execute the request handler. + +## Schedules and execution history + +Hosted Cloudflare and celld servers can schedule scripts and canvas backends with native Durable Object alarms. Configure **Schedule** in the gallery, or call `script_schedule` / `canvas_schedule` with `name` and `action: "set"`. Supply exactly one of `interval_seconds` (at least 60) or a five-field `cron` expression. Cron accepts an IANA `timezone`, defaulting to `UTC`. Supply a relative HTTP `request` (path, method, headers, optional base64 body); the default is GET `/`. + +Use `action: "get"`, `"pause"`, `"resume"`, or `"run_now"` to inspect or control the schedule. Run-now executes the saved request, including while paused. Each artifact has one schedule. Removing a canvas server pauses its schedule. The local filesystem CLI/stdio runtime does not execute backends or run a scheduler; use `canvas server` for local scheduled execution. + +Schedules execute the latest validated revision, not an invalid draft or a historical preview. An occurrence is durably claimed before calling user code; missed intervals are skipped rather than accumulated. Failures and uncertain interrupted invocations are not automatically replayed. Application handlers should make repeated writes idempotent when that matters. A local celld server must be running for alarms to execute. + +`script_runs` and `canvas_runs` expose the latest 100 of up to 1,000 retained executions: revision, trigger, start/end times, duration, HTTP status and outcome. The gallery's **Run history** shows the same records. These records live in host-owned SQLite, separate from the user handler's database, and survive code updates and source restores. An `interrupted` record means completion is unknown and side effects may already have occurred. + +Script logs can be filtered by `run_id` from `script_runs`; the existing 100-entry log retention still applies. Script success describes the HTTP handler returning a response, not completion of `waitUntil` tasks or streamed response bodies. Canvas history includes response serialization; canvas console capture is not provided. diff --git a/e2e/execution-controls.test.ts b/e2e/execution-controls.test.ts new file mode 100644 index 0000000..703c275 --- /dev/null +++ b/e2e/execution-controls.test.ts @@ -0,0 +1,42 @@ +import {mkdtemp,rm} from "node:fs/promises"; +import {tmpdir} from "node:os"; +import {join} from "node:path"; +import {expect,test} from "bun:test"; +import {chromium} from "playwright"; + +test("schedule controls ignore unfinished request fields except when saving",async()=>{ + const directory=await mkdtemp(join(tmpdir(),"execution-controls-")); + const entry=join(directory,"fixture.tsx"); + await Bun.write(entry,`import {createElement} from ${JSON.stringify(Bun.resolveSync("react",import.meta.dir))};import {createRoot} from ${JSON.stringify(Bun.resolveSync("react-dom/client",import.meta.dir))};import {ExecutionControls} from ${JSON.stringify(new URL("../src/gallery/execution-controls.tsx",import.meta.url).pathname)};createRoot(document.getElementById("root")).render(createElement(ExecutionControls,{workspace:"default",name:"example"}));`); + let build; + try { build=await Bun.build({entrypoints:[entry],target:"browser"}); } finally { await rm(directory,{recursive:true,force:true}); } + expect(build.success).toBe(true); + const js=await build.outputs[0]!.text(); + const actions:string[]=[]; + let schedule={interval_seconds:3600,paused:false,next_run_at:Date.now()+3600000,request:{path:"/",method:"GET",headers:[]}}; + const server=Bun.serve({hostname:"127.0.0.1",port:0,async fetch(request){ + if(new URL(request.url).pathname==="/client.js")return new Response(js,{headers:{"content-type":"text/javascript"}}); + if(new URL(request.url).pathname==="/api/tools"){ + const call=await request.json() as {arguments:{action:string}};actions.push(call.arguments.action); + if(call.arguments.action==="pause")schedule={...schedule,paused:true}; + if(call.arguments.action==="resume")schedule={...schedule,paused:false}; + return Response.json({structuredContent:{schedule}}); + } + return new Response('
',{headers:{"content-type":"text/html"}}); + }}); + const browser=await chromium.launch({headless:true}); + try{ + const page=await browser.newPage();await page.goto(server.url.href); + await page.getByText("Schedule",{exact:true}).click(); + await page.getByLabel("Schedule headers").fill("{"); + await page.getByRole("button",{name:"Load schedule",exact:true}).click(); + await page.getByText("Next run:",{exact:false}).waitFor(); + await page.getByLabel("Schedule headers").fill("{"); + await page.getByRole("button",{name:"Pause schedule",exact:true}).click(); + await page.getByRole("button",{name:"Resume schedule",exact:true}).click(); + await page.getByRole("button",{name:"Run schedule now",exact:true}).click(); + await page.getByRole("button",{name:"Save schedule",exact:true}).click(); + await page.getByRole("alert").waitFor(); + expect(actions).toEqual(["get","pause","resume","run_now"]); + }finally{await browser.close();server.stop(true);} +},30000); diff --git a/package.json b/package.json index 8289728..7544249 100644 --- a/package.json +++ b/package.json @@ -91,6 +91,7 @@ "src/viewer.ts", "src/localProject.ts", "cloudflare/project.ts", + "src/gallery/execution-controls.tsx", "src/gallery/project-editor.tsx" ], "publishConfig": { @@ -101,7 +102,7 @@ "mcp": "bun run src/cli.ts mcp", "test": "bun test src scripts/prepare-plugins.test.ts examples/runner-status --timeout 30000", "test:e2e": "bun test e2e/herdr.e2e.test.ts", - "test:mcp-ui": "bun test e2e/mcp-app.test.ts e2e/gallery-scripts.test.ts e2e/browser-plugins.test.ts e2e/plugin-bridge.test.ts e2e/routing.test.ts e2e/runner-status.test.ts --timeout 30000", + "test:mcp-ui": "bun test e2e/mcp-app.test.ts e2e/gallery-scripts.test.ts e2e/execution-controls.test.ts e2e/browser-plugins.test.ts e2e/plugin-bridge.test.ts e2e/routing.test.ts e2e/runner-status.test.ts --timeout 30000", "typecheck": "tsc --noEmit && tsc -p examples/runner-status/tsconfig.json --noEmit && tsc -p examples/runner-status/tsconfig.local.json --noEmit", "dev:cloudflare": "wrangler dev --local --var ENVIRONMENT:local", "prepare:celld": "bun run scripts/prepare-celld.ts", @@ -125,6 +126,7 @@ "@types/react": "^19.2.18", "@types/react-dom": "^19.2.4", "better-auth": "1.7.3", + "cron-parser": "5.10.0", "esbuild": "0.28.1", "react": "^19.2.8", "react-dom": "^19.2.8", diff --git a/src/gallery/client.tsx b/src/gallery/client.tsx index a844457..f6ae467 100644 --- a/src/gallery/client.tsx +++ b/src/gallery/client.tsx @@ -2,6 +2,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { createRoot } from "react-dom/client"; import { authClient, signInUrl } from "../auth/client-api"; import type { GalleryArtifact, GalleryData } from "./types"; +import { ExecutionControls } from "./execution-controls"; import { CanvasSourcePanel, LinkSettings, ScriptPanel } from "./hosted"; import { canvasFileTransferUrl } from "../sdk/files"; import { RemixPanel } from "./remix"; @@ -59,6 +60,9 @@ const styles = ` .script-panel details { margin-top: 16px; border-top: 1px solid var(--line); padding-top: 12px; } .script-panel summary { cursor: pointer; } .script-panel pre { white-space: pre-wrap; overflow-wrap: anywhere; } + .canvas-execution { padding: 8px; border-bottom: 1px solid var(--line); max-height: 45vh; overflow: auto; } + .canvas-execution summary { cursor: pointer; } + .canvas-execution td, .canvas-execution th { padding: 4px 8px; text-align: left; } .muted { color: var(--muted); } .canvas-stage { position: relative; display: flex; min-width: 0; min-height: 0; overflow: hidden; } .preview-frame { display: block; width: 100%; height: 100%; border: 0; } @@ -495,6 +499,7 @@ function App() {
{remixing && !creatingScript && selectedArtifact && resolvedVersion ? setRemixing(false)} onSaved={async name => { createdRemix.current = {name,workspace:selectedArtifact.workspace,kind:selectedArtifact.kind ?? "canvas"}; await loadGallery(); setSelectedVersion("working"); setQuery(""); setRemixing(false); }} /> : null} + {!creatingScript && selectedArtifact && selectedArtifact.kind !== "script" && gallery?.capabilities?.links ?
: null} {!creatingScript && selectedArtifact && gallery?.capabilities?.links ? : null}
{creatingScript ? ( diff --git a/src/gallery/execution-controls.tsx b/src/gallery/execution-controls.tsx new file mode 100644 index 0000000..964f413 --- /dev/null +++ b/src/gallery/execution-controls.tsx @@ -0,0 +1,43 @@ +import {useState} from "react"; +import {galleryTool} from "./hosted"; + +type Schedule = {interval_seconds?:number;cron?:string;timezone?:string;paused:boolean;next_run_at:number|null;request:{path:string;method:string;headers:[string,string][];body?:string}}; +type Run = {id:string;revision:string;trigger:string;started_at:string;duration_ms:number|null;status:string;http_status:number|null}; +const unpack=(value:unknown):any=>typeof value==="string"?JSON.parse(value):value; +export function ExecutionControls({workspace,name,kind="script"}:{workspace:string;name:string;kind?:"script"|"canvas"}) { + const [schedule,setSchedule]=useState(null),[loaded,setLoaded]=useState(false); + const [mode,setMode]=useState("interval"),[interval,setInterval]=useState("3600"),[cron,setCron]=useState("0 * * * *"),[timezone,setTimezone]=useState("UTC"); + const [path,setPath]=useState("/"),[method,setMethod]=useState("GET"),[headers,setHeaders]=useState("[]"),[body,setBody]=useState(""); + const [runs,setRuns]=useState([]),[logs,setLogs]=useState(""),[error,setError]=useState(""),[busy,setBusy]=useState(false); + async function perform(work:()=>Promise) {setBusy(true);setError("");try{await work();}catch(e){setError(e instanceof Error?e.message:String(e));}finally{setBusy(false);}} + async function control(action:string) { + const request=action==="set"?{path,method,headers:JSON.parse(headers),...(body?{body:btoa(Array.from(new TextEncoder().encode(body),b=>String.fromCharCode(b)).join(""))}:{})}:undefined; + const result=unpack(await galleryTool(workspace,`${kind}_schedule`,{name,action,...(action==="set"?{...(mode==="interval"?{interval_seconds:Number(interval)}:{cron,timezone}),request}:{})})); + setSchedule(result.schedule);setLoaded(true); + if(action==="get"&&result.schedule){const s=result.schedule as Schedule;setMode(s.cron?"cron":"interval");setInterval(String(s.interval_seconds??3600));setCron(s.cron??"0 * * * *");setTimezone(s.timezone??"UTC");setPath(s.request.path);setMethod(s.request.method);setHeaders(JSON.stringify(s.request.headers));setBody(s.request.body?new TextDecoder().decode(Uint8Array.from(atob(s.request.body),c=>c.charCodeAt(0))):"");} + } + return <> + {error?

{error}

:null} +
Schedule + + {loaded?

{schedule?(schedule.paused?"Paused":`Next run: ${new Date(schedule.next_run_at!).toISOString()} (UTC)`):"No schedule configured"}

:null} +
{event.preventDefault();void perform(()=>control("set"));}}> + + {mode==="interval"?:<>} + + +