From 37d0b46b2c863db2c835f2e1db471abb4d427a69 Mon Sep 17 00:00:00 2001 From: pallaoro Date: Wed, 16 Sep 2026 18:02:03 +0200 Subject: [PATCH] Expose trigger CRUD on the flow server so the dashboard can schedule flows Until now a cron trigger could only be created by the agent through the flow_trigger tool. The dashboard's Schedule page had no way to make one, so the two schedulers on a box looked like one, and "New task" could only mean a message to the agent. The flow server gains /flows/triggers: list, create, update (pause via enabled:false), delete, and run-now. Every write goes through the same validation the tool uses (assertValidSchedule) plus a flow-existence check, and arms the scheduler immediately. The server reaches the scheduler through an accessor rather than a captured instance, because register() re-runs on plugin reloads and replaces the scheduler while the first server instance survives; a captured reference would point at a stopped scheduler and re-arm its timers. Trigger records carry an origin field (agent or dashboard) so the two kinds stay tellable apart once both exist. --- package-lock.json | 4 +- package.json | 2 +- src/core/serve.ts | 187 +++++++++++++++++++++++++++++++++++ src/core/triggers.ts | 4 + src/plugin/index.ts | 35 ++++--- tests/serve-triggers.test.ts | 156 +++++++++++++++++++++++++++++ 6 files changed, 372 insertions(+), 16 deletions(-) create mode 100644 tests/serve-triggers.test.ts diff --git a/package-lock.json b/package-lock.json index 4eebffb..83d2d62 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@clawnify/clawflow", - "version": "1.6.2", + "version": "1.6.3", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@clawnify/clawflow", - "version": "1.6.2", + "version": "1.6.3", "license": "MIT", "dependencies": { "croner": "^10.0.1" diff --git a/package.json b/package.json index 82312c1..dbdbb18 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@clawnify/clawflow", - "version": "1.6.2", + "version": "1.6.3", "description": "The n8n for agents. A declarative, AI-native workflow format that agents can read, write, and run.", "type": "module", "main": "./dist/index.js", diff --git a/src/core/serve.ts b/src/core/serve.ts index 65809af..e204f5d 100644 --- a/src/core/serve.ts +++ b/src/core/serve.ts @@ -3,6 +3,8 @@ import * as path from "path"; import type { FlowDefinition, ServeConfig } from "./types.js"; import type { FlowRunner } from "./runner.js"; import { validateFlow } from "./validate.js"; +import type { CreateTriggerInput, TriggerRecord, TriggerStore } from "./triggers.js"; +import { assertValidSchedule, type TriggerScheduler } from "./scheduler.js"; import { resolveRunnableFlow } from "./manage.js"; // ---- Flow Server ---------------------------------------------------------------- @@ -28,6 +30,14 @@ export interface FlowServerOpts { warn: (msg: string) => void; error: (msg: string) => void; }; + /** + * Trigger CRUD for the dashboard (/:basePath/triggers). An accessor, not a + * captured instance: register() re-runs on plugin reloads and replaces the + * scheduler while this server instance survives (activeServer guard), so + * the routes must reach the scheduler that is armed right now, never the + * one that was armed when the server first started. + */ + triggers?: () => { store: TriggerStore; scheduler: TriggerScheduler } | null; } const MAX_BODY_BYTES = 1_048_576; // 1 MB @@ -128,6 +138,183 @@ export function startFlowServer(opts: FlowServerOpts): http.Server { return; } + // Trigger CRUD — the HTTP twin of the flow_trigger tool, so a schedule + // made from the dashboard is validated exactly like one the agent makes: + // GET /:basePath/triggers[?flow=] list (with live nextRunAt) + // POST /:basePath/triggers create { flow, cron, tz?, inputs?, description?, enabled?, version? } + // PATCH /:basePath/triggers/:id update any of those, enabled:false pauses + // DELETE /:basePath/triggers/:id + // POST /:basePath/triggers/:id/run fire now (202; the run is async) + const triggersBase = `${basePath}/triggers`; + if (pathname === triggersBase || pathname.startsWith(`${triggersBase}/`)) { + const wired = opts.triggers?.() ?? null; + if (!wired) { + json(res, 501, { error: "Trigger scheduler not available on this server" }); + return; + } + const { store, scheduler } = wired; + const rest = pathname.slice(triggersBase.length).replace(/^\//, ""); + const [id, action, extra] = rest ? rest.split("/") : []; + const withNext = (r: TriggerRecord) => ({ + ...r, + nextRunAt: scheduler.nextRun(r.id)?.toISOString() ?? r.nextRunAt ?? null, + }); + const readJson = async (): Promise | null> => { + const raw = await readBody(req); + if (!raw) return {}; + try { + const parsed = JSON.parse(raw); + return parsed && typeof parsed === "object" && !Array.isArray(parsed) + ? (parsed as Record) + : null; + } catch { + return null; + } + }; + const validVersion = (v: unknown): v is number | "@published" => + v === "@published" || (typeof v === "number" && Number.isInteger(v) && v > 0); + try { + if (extra !== undefined) { + json(res, 404, { error: "Not found" }); + return; + } + if (!id && req.method === "GET") { + const flow = url.searchParams.get("flow") ?? undefined; + const records = store.list(flow ? { flowName: flow } : undefined); + json(res, 200, { ok: true, triggers: records.map(withNext) }); + return; + } + if (!id && req.method === "POST") { + const body = await readJson(); + if (!body) { + json(res, 400, { error: "Invalid JSON body" }); + return; + } + const flowName = + typeof body.flow === "string" ? body.flow + : typeof body.flowName === "string" ? body.flowName + : null; + if (!flowName || typeof body.cron !== "string") { + json(res, 400, { error: '"flow" and "cron" are required' }); + return; + } + if (body.tz !== undefined && typeof body.tz !== "string") { + json(res, 400, { error: '"tz" must be an IANA timezone string' }); + return; + } + if (body.version !== undefined && !validVersion(body.version)) { + json(res, 400, { error: '"version" must be a positive integer or "@published"' }); + return; + } + if (!resolveRunnableFlow(workspace, flowsDir, flowName)) { + json(res, 404, { error: `Flow not found: ${flowName}` }); + return; + } + assertValidSchedule(body.cron, body.tz as string | undefined); + const input: CreateTriggerInput = { + flowName, + cron: body.cron, + ...(typeof body.tz === "string" ? { tz: body.tz } : {}), + ...(body.inputs && typeof body.inputs === "object" && !Array.isArray(body.inputs) + ? { inputs: body.inputs as Record } + : {}), + ...(body.version !== undefined ? { version: body.version as number | "@published" } : {}), + ...(typeof body.description === "string" ? { description: body.description } : {}), + ...(typeof body.enabled === "boolean" ? { enabled: body.enabled } : {}), + origin: body.origin === "agent" ? "agent" : "dashboard", + }; + const record = store.create(input); + scheduler.sync(); + log.info(`[clawflow] trigger ${record.id} created via http → ${flowName} (${record.cron})`); + json(res, 201, { ok: true, trigger: withNext(store.get(record.id) ?? record) }); + return; + } + if (id && !action && req.method === "PATCH") { + const current = store.get(id); + if (!current) { + json(res, 404, { error: `Trigger not found: ${id}` }); + return; + } + const body = await readJson(); + if (!body) { + json(res, 400, { error: "Invalid JSON body" }); + return; + } + if (body.cron !== undefined && typeof body.cron !== "string") { + json(res, 400, { error: '"cron" must be a string' }); + return; + } + if (body.tz !== undefined && typeof body.tz !== "string") { + json(res, 400, { error: '"tz" must be an IANA timezone string' }); + return; + } + if (body.version !== undefined && !validVersion(body.version)) { + json(res, 400, { error: '"version" must be a positive integer or "@published"' }); + return; + } + const cron = typeof body.cron === "string" ? body.cron : current.cron; + const tz = typeof body.tz === "string" ? body.tz : current.tz; + // Re-validate the resulting schedule, not just the field that + // changed — a new tz can invalidate an expression that was fine. + assertValidSchedule(cron, tz); + const patch: Partial = { cron }; + if (typeof body.tz === "string") patch.tz = body.tz; + if (body.inputs !== undefined) { + if (body.inputs === null) patch.inputs = undefined; + else if (typeof body.inputs === "object" && !Array.isArray(body.inputs)) { + patch.inputs = body.inputs as Record; + } else { + json(res, 400, { error: '"inputs" must be an object' }); + return; + } + } + if (body.version !== undefined) patch.version = body.version as number | "@published"; + if (typeof body.description === "string") patch.description = body.description; + if (typeof body.enabled === "boolean") { + patch.enabled = body.enabled; + if (!body.enabled) patch.nextRunAt = undefined; + } + store.update(id, patch); + scheduler.sync(); + json(res, 200, { ok: true, trigger: withNext(store.get(id)!) }); + return; + } + if (id && !action && req.method === "DELETE") { + if (!store.remove(id)) { + json(res, 404, { error: `Trigger not found: ${id}` }); + return; + } + scheduler.sync(); + log.info(`[clawflow] trigger ${id} deleted via http`); + json(res, 200, { ok: true, id }); + return; + } + if (id && action === "run" && req.method === "POST") { + if (!store.get(id)) { + json(res, 404, { error: `Trigger not found: ${id}` }); + return; + } + // Same contract as /:flowName/run: acknowledge, then run detached. + // The outcome lands on the record (lastInstanceId / lastStatus). + json(res, 202, { ok: true, id }); + scheduler.runNow(id).then((result) => { + if (result && !result.ok) { + log.error(`[clawflow] trigger ${id} run-now failed: ${result.error ?? "unknown error"}`); + } + }).catch((err) => { + log.error(`[clawflow] trigger ${id} run-now crashed: ${err instanceof Error ? err.message : String(err)}`); + }); + return; + } + json(res, 404, { error: "Not found" }); + } catch (err) { + // Everything that throws here is a client-side problem: a schedule the + // validator rejects, a duplicate id, a record that vanished mid-request. + json(res, 400, { error: err instanceof Error ? err.message : String(err) }); + } + return; + } + // Run a flow: POST /:basePath/:flowName/run const runPattern = new RegExp( `^${basePath.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}/([a-zA-Z0-9_-]+)/run$`, diff --git a/src/core/triggers.ts b/src/core/triggers.ts index a4b177f..2c6fb64 100644 --- a/src/core/triggers.ts +++ b/src/core/triggers.ts @@ -34,6 +34,8 @@ export interface TriggerRecord { lastInstanceId?: string; /** Informational: when the scheduler expects to fire next. */ nextRunAt?: string; + /** Who created it: the agent (flow_trigger tool) or a person (dashboard). */ + origin?: "agent" | "dashboard"; } export interface CreateTriggerInput { @@ -46,6 +48,7 @@ export interface CreateTriggerInput { description?: string; /** Explicit id (used when restoring); generated when omitted. */ id?: string; + origin?: "agent" | "dashboard"; } export class TriggerStore { @@ -74,6 +77,7 @@ export class TriggerStore { ...(input.inputs ? { inputs: input.inputs } : {}), enabled: input.enabled !== false, ...(input.description ? { description: input.description } : {}), + ...(input.origin ? { origin: input.origin } : {}), createdAt: now, updatedAt: now, }; diff --git a/src/plugin/index.ts b/src/plugin/index.ts index abeedf7..3aab8ea 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -23,6 +23,7 @@ import type { FlowDefinition, FlowNode, PluginConfig, BranchNode, ConditionNode, // scheduler at module scope, like the flow server's activeServer, and retire // it before arming its replacement. let activeScheduler: TriggerScheduler | null = null; +let activeTriggerStore: TriggerStore | null = null; /** The scheduler currently armed in this process (tests + diagnostics). */ export function activeTriggerScheduler(): TriggerScheduler | null { @@ -112,19 +113,6 @@ function register(api: PluginApi) { const runner = new FlowRunner(pluginCfg); const store = runner.getStore(); - // ---- Flow server (optional) --------------------------------------------------- - // Skip when spawned as a child agent (CLAWFLOW_NO_SERVE) to avoid port conflicts. - if (pluginCfg.serve && !process.env.CLAWFLOW_NO_SERVE) { - startFlowServer({ - runner, - serve: pluginCfg.serve, - // Same workspace the flow_* tools use, so a triggered run resolves the - // published version exactly like flow_run does. - workspace, - logger: api.logger, - }); - } - // ---- Trigger scheduler -------------------------------------------------------- // Fires scheduled flows on this box. Skipped for child agents alongside the // flow server so a spawned agent never double-fires its parent's triggers. @@ -140,9 +128,29 @@ function register(api: PluginApi) { if (!process.env.CLAWFLOW_NO_SERVE) { activeScheduler?.stop(); activeScheduler = scheduler; + activeTriggerStore = triggerStore; scheduler.start(); } + // ---- Flow server (optional) --------------------------------------------------- + // Skip when spawned as a child agent (CLAWFLOW_NO_SERVE) to avoid port conflicts. + // Started after the scheduler so its trigger routes can reach it; the + // accessor keeps them on the currently armed scheduler across reloads. + if (pluginCfg.serve && !process.env.CLAWFLOW_NO_SERVE) { + startFlowServer({ + runner, + serve: pluginCfg.serve, + // Same workspace the flow_* tools use, so a triggered run resolves the + // published version exactly like flow_run does. + workspace, + logger: api.logger, + triggers: () => + activeScheduler && activeTriggerStore + ? { store: activeTriggerStore, scheduler: activeScheduler } + : null, + }); + } + // ---- Approval gate for flow_run ----------------------------------------------- // Flows can call HTTP, exec, and agent tools, so by default we prompt the user // before each run. Disable entirely (`approval.enabled: false`) or skip for @@ -2079,6 +2087,7 @@ By default a trigger runs the latest PUBLISHED version; pin one with "version".` inputs: params.inputs, version: params.version, description: params.description, + origin: "agent", }); scheduler.sync(); const stored = triggerStore.get(record.id) ?? record; diff --git a/tests/serve-triggers.test.ts b/tests/serve-triggers.test.ts new file mode 100644 index 0000000..0825a24 --- /dev/null +++ b/tests/serve-triggers.test.ts @@ -0,0 +1,156 @@ +import { describe, it, before, after } from "node:test"; +import assert from "node:assert/strict"; +import * as fs from "fs"; +import * as path from "path"; +import * as os from "os"; + +import { FlowRunner, startFlowServer } from "../src/index.js"; +import { TriggerStore } from "../src/core/triggers.js"; +import { TriggerScheduler } from "../src/core/scheduler.js"; +import type { FlowDefinition, PluginConfig } from "../src/index.js"; + +// The HTTP twin of the flow_trigger tool. A schedule made from the dashboard +// goes through these routes, so they must validate like the tool does and +// arm the scheduler immediately. + +const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "ocf-serve-triggers-")); +const workspace = path.join(tmpDir, "workspace"); +const flowsDir = path.join(workspace, "flows"); +const cfg: PluginConfig = { + stateDir: path.join(tmpDir, "state"), + memoryDir: path.join(tmpDir, "memory"), +}; +const silent = { info: () => {}, warn: () => {}, error: () => {} }; + +let server: ReturnType; +let base = ""; +let store: TriggerStore; +let scheduler: TriggerScheduler; + +function writeFlow(name: string): void { + const def: FlowDefinition = { + flow: name, + nodes: [{ name: "step1", do: "code", run: "1", output: "result" }], + }; + fs.mkdirSync(flowsDir, { recursive: true }); + fs.writeFileSync(path.join(flowsDir, `${name}.json`), JSON.stringify(def, null, 2)); +} + +async function call(method: string, p: string, body?: unknown) { + const res = await fetch(`${base}${p}`, { + method, + headers: body !== undefined ? { "Content-Type": "application/json" } : {}, + body: body !== undefined ? JSON.stringify(body) : undefined, + }); + return { status: res.status, body: (await res.json()) as Record }; +} + +before(async () => { + writeFlow("digest"); + const runner = new FlowRunner(cfg); + store = new TriggerStore(workspace); + scheduler = new TriggerScheduler({ runner, store, workspace, flowsDir, logger: silent }); + server = startFlowServer({ + runner, + serve: { port: 0, flowsDir }, + workspace, + logger: silent, + triggers: () => ({ store, scheduler }), + }); + await new Promise((resolve) => server.on("listening", resolve)); + const addr = server.address() as { port: number }; + base = `http://127.0.0.1:${addr.port}/flows`; +}); + +after(async () => { + scheduler.stop(); + await new Promise((resolve) => server.close(() => resolve())); + fs.rmSync(tmpDir, { recursive: true, force: true }); +}); + +describe("trigger routes", () => { + let id = ""; + + it("rejects a schedule for a flow that does not exist", async () => { + const r = await call("POST", "/triggers", { flow: "nope", cron: "0 8 * * *" }); + assert.equal(r.status, 404); + assert.match(r.body.error, /Flow not found/); + }); + + it("rejects an invalid schedule with the validator's message", async () => { + const r = await call("POST", "/triggers", { flow: "digest", cron: "* * * * * *" }); + assert.equal(r.status, 400); + assert.match(r.body.error, /once a minute/); + }); + + it("requires flow and cron", async () => { + const r = await call("POST", "/triggers", { flow: "digest" }); + assert.equal(r.status, 400); + }); + + it("creates a trigger, arms it, and tags it as dashboard-made", async () => { + const r = await call("POST", "/triggers", { + flow: "digest", + cron: "0 8 * * *", + tz: "Europe/Amsterdam", + description: "morning digest", + }); + assert.equal(r.status, 201, JSON.stringify(r.body)); + const t = r.body.trigger; + id = t.id; + assert.equal(t.flowName, "digest"); + assert.equal(t.enabled, true); + assert.equal(t.origin, "dashboard"); + assert.ok(t.nextRunAt, "not armed: no nextRunAt"); + assert.ok(scheduler.nextRun(id), "scheduler did not arm the new trigger"); + }); + + it("lists triggers, optionally by flow", async () => { + const all = await call("GET", "/triggers"); + assert.equal(all.status, 200); + assert.equal(all.body.triggers.length, 1); + const none = await call("GET", "/triggers?flow=other"); + assert.equal(none.body.triggers.length, 0); + }); + + it("pauses through PATCH and disarms", async () => { + const r = await call("PATCH", `/triggers/${id}`, { enabled: false }); + assert.equal(r.status, 200, JSON.stringify(r.body)); + assert.equal(r.body.trigger.enabled, false); + assert.equal(r.body.trigger.nextRunAt, null); + assert.equal(scheduler.nextRun(id), null, "paused trigger still armed"); + }); + + it("resumes and retimes, re-validating the schedule", async () => { + const bad = await call("PATCH", `/triggers/${id}`, { cron: "not a cron" }); + assert.equal(bad.status, 400); + const r = await call("PATCH", `/triggers/${id}`, { enabled: true, cron: "30 9 * * 1-5" }); + assert.equal(r.status, 200, JSON.stringify(r.body)); + assert.equal(r.body.trigger.cron, "30 9 * * 1-5"); + assert.ok(r.body.trigger.nextRunAt); + assert.ok(scheduler.nextRun(id), "resumed trigger not armed"); + }); + + it("runs now, detached, and records the run on the trigger", async () => { + const r = await call("POST", `/triggers/${id}/run`); + assert.equal(r.status, 202); + const deadline = Date.now() + 5000; + let ran = store.get(id); + while ((!ran?.lastRunAt) && Date.now() < deadline) { + await new Promise((res) => setTimeout(res, 50)); + ran = store.get(id); + } + assert.ok(ran?.lastRunAt, "run-now did not record a run"); + assert.equal(ran?.lastStatus, "ok", `run failed: ${ran?.lastError}`); + }); + + it("404s on unknown ids and deletes", async () => { + assert.equal((await call("PATCH", "/triggers/missing", { enabled: false })).status, 404); + assert.equal((await call("POST", "/triggers/missing/run")).status, 404); + const del = await call("DELETE", `/triggers/${id}`); + assert.equal(del.status, 200); + assert.equal(scheduler.nextRun(id), null); + assert.equal((await call("DELETE", `/triggers/${id}`)).status, 404); + assert.equal((await call("GET", "/triggers")).body.triggers.length, 0); + }); +});