diff --git a/README.md b/README.md
index a751129..8584576 100644
--- a/README.md
+++ b/README.md
@@ -65,6 +65,7 @@ Workflows today are written **for** agents, not **by** them. Visual canvas tools
 - **Approval gates** — `do: wait` pauses for human review, resumes with a token
 - **External events** — `waitForEvent` blocks until an external system pushes data
 - **Per-node retry** — exponential, linear, or constant backoff on any node
+- **Scheduled triggers** — cron schedules as first-class records: many per flow, each with its own inputs, paused without touching the flow
 
 ### Portability
 - **OpenClaw plugin** — run flows as agent tools today
@@ -478,8 +479,60 @@ Any string field supports `{{ path.to.value }}` interpolation resolved against f
 **Important:** templates reference the **`output` key**, not the node name. If a node has `"name": "get_data", "output": "api"`, reference it as `{{ api }}` — not `{{ get_data }}`.
 
 Flow state starts as `{ inputs: <payload> }` and grows as nodes complete. The
-caller (CLI, webhook server, parent flow, dashboard) is responsible for
-producing that payload — the flow itself is trigger-agnostic.
+caller (CLI, webhook server, scheduled trigger, parent flow, dashboard) is
+responsible for producing that payload — the flow definition itself stays
+trigger-agnostic. Schedules live in their own records, not in the flow (see
+[Scheduling](#scheduling)).
+
+---
+
+## Scheduling
+
+A flow runs when something invokes it: an agent, an HTTP call, or a **trigger**.
+
+A trigger is a first-class record — not a field on the flow definition:
+
+```bash
+flow_trigger action: "create" flow: "daily-digest" cron: "0 9 * * *" tz: "Europe/Rome"
+```
+
+```
+● daily-digest-ab12cd34
+    flow: daily-digest (latest published)
+    cron: 0 9 * * * [Europe/Rome]
+    next: 2026-08-22T07:00:00.000Z
+```
+
+That separation is deliberate. A schedule is mutable operational state — paused
+at 2am, retimed, pointed at a different version — while a published flow version
+is an immutable artifact. Embedding one in the other would make "pause" mean
+"publish a new version", and would let an unrelated publish silently re-arm a
+schedule the draft happened to carry. Temporal made the same move from
+cron-in-workflow to a separate Schedules object, for the same reason.
+
+Because triggers are their own records, one flow can have many, each with its
+own payload:
+
+```bash
+flow_trigger action: "create" flow: "digest" cron: "0 * * * *" inputs: '{"customer":"acme"}'
+flow_trigger action: "create" flow: "digest" cron: "0 9 * * *" inputs: '{"customer":"globex"}'
+```
+
+| Behavior | Rule |
+|---|---|
+| Expression | Standard 5-field cron; minimum interval 60s |
+| Timezone | IANA name (`Europe/Rome`); host local time when omitted |
+| Version | Latest published by default; pin with `version: 2` |
+| Missed runs | **Not replayed** — a host that was down at 09:00 waits for the next occurrence |
+| Overlap | A tick is skipped if the previous run is still going |
+| Flow deleted | Triggers are **paused**, not removed, so a restore keeps them |
+| Approval | Arming/pausing/deleting a schedule is gated like flow authoring |
+
+Records live in `.clawflow/triggers/<id>.json`, alongside `.clawflow/versions/`.
+Out-of-process callers — the Clawnify hook server and dashboard — read and write
+them by importing `TriggerStore` from `dist`, the same way they read published
+versions. The running scheduler re-reads the directory every 30s, so a schedule
+created or edited in the dashboard goes live without a restart.
 
 ---
 
@@ -542,6 +595,7 @@ Eleven tools registered in OpenClaw:
 | `flow_read` | Read a flow definition (draft or specific version), inspect single nodes |
 | `flow_publish` | Publish current draft as a new numbered version |
 | `flow_edit` | Edit nodes in a flow definition (set, update, add, remove, move, wrap, revert, list) |
+| `flow_trigger` | Schedule a flow on cron (create, update, list, pause, resume, delete, run_now) |
 
 **Config:**
 ```json
@@ -557,7 +611,7 @@ Eleven tools registered in OpenClaw:
   "agents": {
     "list": [{
       "id": "main",
-      "tools": { "alsoAllow": ["flow_create", "flow_delete", "flow_restore_from_bin", "flow_run", "flow_resume", "flow_send_event", "flow_status", "flow_list", "flow_read", "flow_publish", "flow_edit"] }
+      "tools": { "alsoAllow": ["flow_create", "flow_delete", "flow_restore_from_bin", "flow_run", "flow_resume", "flow_send_event", "flow_status", "flow_list", "flow_read", "flow_publish", "flow_edit", "flow_trigger"] }
     }]
   }
 }
diff --git a/openclaw.plugin.json b/openclaw.plugin.json
index fdbc553..331a95b 100644
--- a/openclaw.plugin.json
+++ b/openclaw.plugin.json
@@ -1,8 +1,8 @@
 {
   "id": "clawflow",
   "name": "ClawFlow",
-  "description": "The n8n for agents. Declarative, AI-native workflow engine — LLM-writable, Cloudflare-portable.",
-  "version": "1.4.1",
+  "description": "The n8n for agents. Declarative, AI-native workflow engine \u2014 LLM-writable, Cloudflare-portable.",
+  "version": "1.6.0",
   "skills": [
     "./skills/clawflow"
   ],
@@ -21,7 +21,8 @@
       "flow_list",
       "flow_read",
       "flow_publish",
-      "flow_edit"
+      "flow_edit",
+      "flow_trigger"
     ]
   },
   "configSchema": {
@@ -129,7 +130,7 @@
           "gateMutations": {
             "type": "boolean",
             "default": true,
-            "description": "Gate flow authoring/publishing (flow_create/edit/publish/delete) behind approval on every call, independently of `enabled` (which governs flow_run). Set false to disable."
+            "description": "Gate flow authoring/publishing (flow_create/edit/publish/delete) and schedule changes (flow_trigger create/delete/pause/resume/run_now) behind approval on every call, independently of `enabled` (which governs flow_run). Set false to disable."
           }
         }
       }
diff --git a/package-lock.json b/package-lock.json
index 27cfb74..2bbeb46 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,13 +1,16 @@
 {
   "name": "@clawnify/clawflow",
-  "version": "1.3.1",
+  "version": "1.5.1",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@clawnify/clawflow",
-      "version": "1.3.1",
+      "version": "1.5.1",
       "license": "MIT",
+      "dependencies": {
+        "croner": "^10.0.1"
+      },
       "devDependencies": {
         "@types/node": "^25.5.0",
         "tsx": "^4.21.0",
@@ -477,6 +480,24 @@
         "undici-types": "~7.18.0"
       }
     },
+    "node_modules/croner": {
+      "version": "10.0.1",
+      "resolved": "https://registry.npmjs.org/croner/-/croner-10.0.1.tgz",
+      "integrity": "sha512-ixNtAJndqh173VQ4KodSdJEI6nuioBWI0V1ITNKhZZsO0pEMoDxz539T4FTTbSZ/xIOSuDnzxLVRqBVSvPNE2g==",
+      "funding": [
+        {
+          "type": "other",
+          "url": "https://paypal.me/hexagonpp"
+        },
+        {
+          "type": "github",
+          "url": "https://github.com/sponsors/hexagon"
+        }
+      ],
+      "engines": {
+        "node": ">=18.0"
+      }
+    },
     "node_modules/esbuild": {
       "version": "0.27.4",
       "resolved": "https://registry.npmjs.org/esbuild/-/esbuild-0.27.4.tgz",
diff --git a/package.json b/package.json
index a335ae0..798d09d 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@clawnify/clawflow",
-  "version": "1.5.1",
+  "version": "1.6.0",
   "description": "The n8n for agents. A declarative, AI-native workflow format that agents can read, write, and run.",
   "type": "module",
   "main": "./dist/index.js",
@@ -85,5 +85,8 @@
   "homepage": "https://github.com/clawnify/clawflow#readme",
   "bugs": {
     "url": "https://github.com/clawnify/clawflow/issues"
+  },
+  "dependencies": {
+    "croner": "^10.0.1"
   }
 }
diff --git a/skills/clawflow/SKILL.md b/skills/clawflow/SKILL.md
index 972a3f6..63a3b39 100644
--- a/skills/clawflow/SKILL.md
+++ b/skills/clawflow/SKILL.md
@@ -570,6 +570,8 @@ flows/
   triggers, dashboard runs) resolve the same way: latest published version, draft
   only when nothing is published. So a draft edit does not change what a live
   trigger executes once the flow has been published at least once.
+- Scheduled triggers (`flow_trigger`) resolve the same way unless pinned to a
+  version — see "Scheduling a flow" below.
 - `flow_run file: "my-flow" draft: true` — explicitly run the working copy
 - `flow_run file: "my-flow" version: 2` — run a specific version
 - `flow_read file: "my-flow" version: 1` — inspect a specific published version
@@ -577,6 +579,50 @@ flows/
 - Edits to the draft never affect published versions
 - **Do NOT create separate files for versions** (e.g. `my-flow-v2.json`). Use `flow_publish` instead.
 
+## Scheduling a flow
+
+`flow_trigger` runs a flow on a cron schedule on this box. A trigger is a
+**separate record**, not a field on the flow: one flow can carry several
+triggers with different cadences and different inputs, and pausing one never
+edits the flow or mints a new version.
+
+```
+flow_trigger action: "create" flow: "daily-digest" cron: "0 9 * * *" tz: "Europe/Rome"
+flow_trigger action: "update" id: "daily-digest-ab12cd34" cron: "0 7 * * *"
+flow_trigger action: "list"
+flow_trigger action: "pause"   id: "daily-digest-ab12cd34"
+flow_trigger action: "resume"  id: "daily-digest-ab12cd34"
+flow_trigger action: "run_now" id: "daily-digest-ab12cd34"
+flow_trigger action: "delete"  id: "daily-digest-ab12cd34"
+```
+
+**Per-trigger inputs** — the same flow, two customers, two cadences:
+
+```
+flow_trigger action: "create" flow: "digest" cron: "0 * * * *" inputs: { "customer": "acme" }
+flow_trigger action: "create" flow: "digest" cron: "0 9 * * *" inputs: { "customer": "globex" }
+```
+
+**Rules:**
+- Standard 5-field cron. The minimum interval is 60 seconds — sub-minute
+  schedules are rejected, not silently accepted.
+- `tz` is an IANA name (`Europe/Rome`). Host local time when omitted.
+- A trigger runs the **latest published version** by default. Pass `version: 2`
+  to pin one — useful when you want a schedule to keep running a known-good
+  definition while the draft moves on.
+- **Missed runs are not replayed.** If the host was down at 09:00 the trigger
+  does not fire at boot; it waits for the next occurrence.
+- Runs never overlap: if the previous run is still going, the tick is skipped.
+- Soft-deleting a flow (`flow_delete`) **pauses** its triggers rather than
+  removing them, so a restore keeps them — restore then requires an explicit
+  `resume`, so nothing silently re-arms.
+- Triggers are plain records on disk, so the Clawnify dashboard reads and edits
+  the same ones you see here. A change made there is picked up by the running
+  scheduler within ~30s — no restart.
+- Arming, editing, pausing, resuming or deleting a schedule requires approval by default
+  (same gate as flow authoring) — a schedule runs a flow unattended with tool
+  access, so it is treated as a mutation.
+
 ## Reading and discovering flows
 
 - `flow_list` — lists all flows with their description, declared `inputs:` block, and published version info
diff --git a/src/core/manage.ts b/src/core/manage.ts
index e93c7f2..6abb444 100644
--- a/src/core/manage.ts
+++ b/src/core/manage.ts
@@ -119,3 +119,46 @@ export function publishDraft(workspace: string, file: string): PublishResult {
     totalVersions: nextVersion,
   };
 }
+
+/**
+ * Resolve what an incoming trigger should execute: a pinned version when one is
+ * requested, else the latest PUBLISHED version, else the draft.
+ *
+ * This is the one place that answers "which definition does a trigger run".
+ * The flow server, the scheduler, and any off-box caller share it so a webhook,
+ * a cron trigger, and an agent run can never execute different definitions of
+ * the same flow — the invariant 1.5.1 established.
+ *
+ * Returns null when the flow (or the pinned version) does not exist.
+ */
+export function resolveRunnableFlow(
+  workspace: string,
+  flowsDir: string,
+  flowName: string,
+  version?: number | "@published",
+): { def: FlowDefinition; version: number | null; source: string } | null {
+  const safe = flowName.replace(/[^a-zA-Z0-9_-]/g, "");
+  if (!safe) return null;
+
+  if (typeof version === "number") {
+    const def = readVersion(workspace, safe, version);
+    return def ? { def, version, source: `v${version}` } : null;
+  }
+
+  const latest = readLatestVersion(workspace, safe);
+  if (latest) {
+    return { def: latest.def, version: latest.version, source: `v${latest.version}` };
+  }
+
+  const file = path.join(flowsDir, `${safe}.json`);
+  if (!fs.existsSync(file)) return null;
+  try {
+    return {
+      def: JSON.parse(fs.readFileSync(file, "utf8")) as FlowDefinition,
+      version: null,
+      source: "draft (no published versions)",
+    };
+  } catch {
+    return null;
+  }
+}
diff --git a/src/core/scheduler.ts b/src/core/scheduler.ts
new file mode 100644
index 0000000..8d1ddcb
--- /dev/null
+++ b/src/core/scheduler.ts
@@ -0,0 +1,302 @@
+import { randomUUID } from "crypto";
+import { Cron } from "croner";
+
+import { resolveRunnableFlow } from "./manage.js";
+import type { TriggerStore, TriggerRecord } from "./triggers.js";
+import type { FlowDefinition, FlowResult } from "./types.js";
+
+// ---- Trigger Scheduler ----------------------------------------------------------
+// Arms one croner timer per enabled trigger record and fires the flow directly
+// through the runner — the same path the HTTP flow server takes, so a cron fire,
+// a webhook and an agent run all resolve the same definition.
+//
+// Missed runs are NOT replayed. If the host was down at 09:00 the trigger does
+// not fire at boot; it waits for the next occurrence. Replaying side-effecting
+// flows after downtime is the worse failure, and croner gives us this for free
+// by always scheduling forward from now.
+
+/** Triggers may not fire more often than once a minute. */
+const MIN_INTERVAL_MS = 60_000;
+
+export interface SchedulerLogger {
+  info: (msg: string) => void;
+  warn: (msg: string) => void;
+  error: (msg: string) => void;
+}
+
+export interface FlowRunnerLike {
+  run(
+    def: FlowDefinition,
+    inputs: unknown,
+    instanceId: string,
+  ): Promise<FlowResult>;
+}
+
+export interface TriggerSchedulerOpts {
+  runner: FlowRunnerLike;
+  store: TriggerStore;
+  workspace: string;
+  flowsDir: string;
+  logger?: SchedulerLogger;
+  /**
+   * How often to re-read the trigger records from disk, in ms.
+   * Records are also written by out-of-process callers — the Clawnify hook
+   * server and dashboard import TriggerStore from dist the same way they read
+   * published versions — so the scheduler cannot rely on its own tools being
+   * the only writer. Set 0 to disable (tests).
+   */
+  resyncMs?: number;
+}
+
+/**
+ * Validate a schedule before it is stored.
+ *
+ * Triggers are agent-writable, so this is a trust boundary: croner accepts
+ * 6-field patterns, and `* * * * * *` on a flow with AI nodes is a runaway cost
+ * incident rather than a schedule. Rejects anything firing more than once a
+ * minute, and anything croner cannot parse.
+ */
+export function assertValidSchedule(cron: string, tz?: string): void {
+  let first: Date | null;
+  let second: Date | null;
+
+  // croner constructs lazily: an invalid timezone surfaces from nextRun(), not
+  // from the constructor, so both must be inside the guard.
+  try {
+    const job = new Cron(cron, { timezone: tz });
+    first = job.nextRun();
+    second = first ? job.nextRun(first) : null;
+  } catch (err) {
+    throw new Error(
+      `Invalid schedule "${cron}"${tz ? ` (tz ${tz})` : ""}: ${
+        err instanceof Error ? err.message : String(err)
+      }`,
+    );
+  }
+
+  if (!first) throw new Error(`Schedule "${cron}" never fires`);
+
+  if (second && second.getTime() - first.getTime() < MIN_INTERVAL_MS) {
+    throw new Error(
+      `Schedule "${cron}" fires more than once a minute — the minimum interval is 60s`,
+    );
+  }
+}
+
+const DEFAULT_RESYNC_MS = 30_000;
+
+export class TriggerScheduler {
+  private jobs = new Map<string, Cron>();
+  private armedSpecs = new Map<string, string>();
+  private resyncTimer: ReturnType<typeof setInterval> | null = null;
+  private readonly opts: TriggerSchedulerOpts;
+  private readonly log: SchedulerLogger;
+
+  constructor(opts: TriggerSchedulerOpts) {
+    this.opts = opts;
+    this.log = opts.logger ?? {
+      info: console.log,
+      warn: console.warn,
+      error: console.error,
+    };
+  }
+
+  /** Arm every enabled trigger. Safe to call repeatedly. */
+  sync(): void {
+    const records = this.opts.store.list({ enabledOnly: true });
+    const live = new Set(records.map((r) => r.id));
+
+    for (const [id, job] of this.jobs) {
+      if (!live.has(id)) {
+        job.stop();
+        this.jobs.delete(id);
+      }
+    }
+
+    for (const record of records) {
+      // Re-arm only when the schedule itself changed. Rebuilding an unchanged
+      // timer on every resync would reset its countdown and could starve a
+      // trigger that fires less often than the resync interval.
+      const armed = this.armedSpecs.get(record.id);
+      const spec = `${record.cron}|${record.tz ?? ""}`;
+      if (armed === spec && this.jobs.has(record.id)) continue;
+
+      this.jobs.get(record.id)?.stop();
+      this.jobs.delete(record.id);
+      this.arm(record);
+      this.armedSpecs.set(record.id, spec);
+    }
+
+    for (const id of [...this.armedSpecs.keys()]) {
+      if (!live.has(id)) this.armedSpecs.delete(id);
+    }
+  }
+
+  start(): void {
+    this.sync();
+    this.log.info(`[clawflow] scheduler armed ${this.jobs.size} trigger(s)`);
+
+    const every = this.opts.resyncMs ?? DEFAULT_RESYNC_MS;
+    if (every > 0 && !this.resyncTimer) {
+      this.resyncTimer = setInterval(() => {
+        try {
+          this.sync();
+        } catch (err) {
+          this.log.error(
+            `[clawflow] scheduler resync failed: ${
+              err instanceof Error ? err.message : String(err)
+            }`,
+          );
+        }
+      }, every);
+      this.resyncTimer.unref?.();
+    }
+  }
+
+  stop(): void {
+    if (this.resyncTimer) {
+      clearInterval(this.resyncTimer);
+      this.resyncTimer = null;
+    }
+    for (const job of this.jobs.values()) job.stop();
+    this.jobs.clear();
+    this.armedSpecs.clear();
+  }
+
+  /** Informational: when an armed trigger next fires. */
+  nextRun(id: string): Date | null {
+    return this.jobs.get(id)?.nextRun() ?? null;
+  }
+
+  /** Fire a trigger immediately, ignoring its schedule (but not its existence). */
+  async runNow(id: string): Promise<FlowResult | null> {
+    const record = this.opts.store.get(id);
+    if (!record) throw new Error(`Trigger not found: ${id}`);
+    return this.fire(record, { manual: true });
+  }
+
+  // ---- Internals ---------------------------------------------------------------
+
+  private arm(record: TriggerRecord): void {
+    let job: Cron;
+    try {
+      job = new Cron(
+        record.cron,
+        {
+          timezone: record.tz,
+          // Never start a run while the previous one is still going.
+          protect: true,
+          // Don't hold the process open on our account.
+          unref: true,
+          catch: true,
+        },
+        () => {
+          void this.fire(record);
+        },
+      );
+    } catch (err) {
+      this.log.error(
+        `[clawflow] trigger ${record.id} has an unusable schedule "${record.cron}": ${
+          err instanceof Error ? err.message : String(err)
+        }`,
+      );
+      return;
+    }
+
+    let next: Date | null = null;
+    try {
+      next = job.nextRun();
+    } catch (err) {
+      job.stop();
+      this.log.error(
+        `[clawflow] trigger ${record.id} has an unusable schedule "${record.cron}"${
+          record.tz ? ` (tz ${record.tz})` : ""
+        }: ${err instanceof Error ? err.message : String(err)}`,
+      );
+      return;
+    }
+
+    this.jobs.set(record.id, job);
+    if (next) {
+      this.opts.store.update(record.id, { nextRunAt: next.toISOString() });
+    }
+  }
+
+  private async fire(
+    armed: TriggerRecord,
+    opts?: { manual?: boolean },
+  ): Promise<FlowResult | null> {
+    // Re-read: the armed copy is a snapshot, and the record may have been
+    // paused, retimed or repointed since it was armed.
+    const record = this.opts.store.get(armed.id);
+    if (!record) return null;
+    if (!record.enabled && !opts?.manual) return null;
+
+    const loaded = resolveRunnableFlow(
+      this.opts.workspace,
+      this.opts.flowsDir,
+      record.flowName,
+      record.version,
+    );
+
+    if (!loaded) {
+      const error =
+        typeof record.version === "number"
+          ? `Flow "${record.flowName}" v${record.version} not found`
+          : `Flow "${record.flowName}" not found`;
+      this.log.error(`[clawflow] trigger ${record.id} skipped: ${error}`);
+      this.opts.store.update(record.id, {
+        lastRunAt: new Date().toISOString(),
+        lastStatus: "skipped",
+        lastError: error,
+      });
+      return null;
+    }
+
+    const instanceId = randomUUID();
+    this.log.info(
+      `[clawflow] trigger ${record.id} → ${record.flowName} ${loaded.source} (${instanceId})`,
+    );
+
+    try {
+      const result = await this.opts.runner.run(
+        loaded.def,
+        record.inputs ?? {},
+        instanceId,
+      );
+      this.opts.store.update(record.id, {
+        lastRunAt: new Date().toISOString(),
+        lastStatus: result.ok ? "ok" : "error",
+        lastInstanceId: instanceId,
+        ...(result.ok ? { lastError: undefined } : { lastError: result.error }),
+        ...(this.nextRunIso(record.id) ?? {}),
+      });
+      if (!result.ok) {
+        this.log.error(
+          `[clawflow] trigger ${record.id} run failed: ${result.error ?? "unknown error"}`,
+        );
+      }
+      return result;
+    } catch (err) {
+      const message = err instanceof Error ? err.message : String(err);
+      this.log.error(`[clawflow] trigger ${record.id} crashed: ${message}`);
+      this.opts.store.update(record.id, {
+        lastRunAt: new Date().toISOString(),
+        lastStatus: "error",
+        lastInstanceId: instanceId,
+        lastError: message,
+        ...(this.nextRunIso(record.id) ?? {}),
+      });
+      return null;
+    }
+  }
+
+  private nextRunIso(id: string): { nextRunAt: string } | null {
+    try {
+      const next = this.jobs.get(id)?.nextRun();
+      return next ? { nextRunAt: next.toISOString() } : null;
+    } catch {
+      return null;
+    }
+  }
+}
diff --git a/src/core/serve.ts b/src/core/serve.ts
index a86aa83..65809af 100644
--- a/src/core/serve.ts
+++ b/src/core/serve.ts
@@ -1,10 +1,9 @@
 import * as http from "http";
-import * as fs from "fs";
 import * as path from "path";
 import type { FlowDefinition, ServeConfig } from "./types.js";
 import type { FlowRunner } from "./runner.js";
 import { validateFlow } from "./validate.js";
-import { readLatestVersion } from "./manage.js";
+import { resolveRunnableFlow } from "./manage.js";
 
 // ---- Flow Server ----------------------------------------------------------------
 // Lightweight HTTP server that runs flows on POST. Trigger semantics (webhooks,
@@ -45,39 +44,6 @@ function resolveFlowsDir(serve: ServeConfig, workspace: string): string {
   return serve.flowsDir ?? path.join(workspace, "flows");
 }
 
-/**
- * Resolve what an incoming trigger should execute: the latest PUBLISHED
- * version when the flow has one, else the draft.
- *
- * Same precedence as the flow_run tool — a webhook and an agent run must never
- * execute different definitions of the same flow, or "publish" means nothing
- * for every off-box caller (webhooks, HTTP triggers, the dashboard).
- * Unpublished flows still run from the draft so a flow works the moment it's
- * written.
- */
-function loadFlow(
-  workspace: string,
-  flowsDir: string,
-  flowName: string,
-): { def: FlowDefinition; source: string } | null {
-  const safe = flowName.replace(/[^a-zA-Z0-9_-]/g, "");
-  if (!safe) return null;
-
-  const latest = readLatestVersion(workspace, safe);
-  if (latest) return { def: latest.def, source: `v${latest.version}` };
-
-  const file = path.join(flowsDir, `${safe}.json`);
-  if (!fs.existsSync(file)) return null;
-  try {
-    return {
-      def: JSON.parse(fs.readFileSync(file, "utf8")) as FlowDefinition,
-      source: "draft (no published versions)",
-    };
-  } catch {
-    return null;
-  }
-}
-
 function json(
   res: http.ServerResponse,
   status: number,
@@ -176,7 +142,7 @@ export function startFlowServer(opts: FlowServerOpts): http.Server {
     const flowName = match[1];
 
     try {
-      const loaded = loadFlow(workspace, flowsDir, flowName);
+      const loaded = resolveRunnableFlow(workspace, flowsDir, flowName);
       if (!loaded) {
         json(res, 404, { error: `Flow not found: ${flowName}` });
         return;
diff --git a/src/core/triggers.ts b/src/core/triggers.ts
new file mode 100644
index 0000000..a4b177f
--- /dev/null
+++ b/src/core/triggers.ts
@@ -0,0 +1,169 @@
+import * as fs from "fs";
+import * as path from "path";
+import { randomUUID } from "crypto";
+
+// ---- Trigger Records ------------------------------------------------------------
+// A trigger is a schedule that fires a flow. It is a first-class record, NOT a
+// field on the flow definition: a schedule is mutable operational state (pause it
+// at 2am, retime it, point it at a different version) while a published flow
+// version is an immutable artifact. Embedding one in the other would make "pause"
+// mean "publish a new version", and would let an unrelated publish silently
+// re-arm a schedule the draft happened to carry.
+//
+// Layout mirrors versions: .clawflow/triggers/<triggerId>.json — one file per
+// record, so N triggers can target one flow with different cadences and inputs.
+
+export interface TriggerRecord {
+  id: string;
+  flowName: string;
+  /** Which definition to run: a pinned version, or "@published" (latest). */
+  version: number | "@published";
+  /** Standard 5-field cron expression. */
+  cron: string;
+  /** IANA timezone (e.g. "Europe/Rome"). Host local time when unset. */
+  tz?: string;
+  /** Payload passed as the flow's inputs on every fire. */
+  inputs?: Record<string, unknown>;
+  enabled: boolean;
+  description?: string;
+  createdAt: string;
+  updatedAt: string;
+  lastRunAt?: string;
+  lastStatus?: "ok" | "error" | "skipped";
+  lastError?: string;
+  lastInstanceId?: string;
+  /** Informational: when the scheduler expects to fire next. */
+  nextRunAt?: string;
+}
+
+export interface CreateTriggerInput {
+  flowName: string;
+  cron: string;
+  version?: number | "@published";
+  tz?: string;
+  inputs?: Record<string, unknown>;
+  enabled?: boolean;
+  description?: string;
+  /** Explicit id (used when restoring); generated when omitted. */
+  id?: string;
+}
+
+export class TriggerStore {
+  private dir: string;
+
+  constructor(workspace?: string, triggersDir?: string) {
+    const root =
+      workspace ??
+      process.env.OPENCLAW_WORKSPACE ??
+      process.env.HOME ??
+      ".";
+    this.dir = triggersDir ?? path.join(root, ".clawflow", "triggers");
+    fs.mkdirSync(this.dir, { recursive: true });
+  }
+
+  create(input: CreateTriggerInput): TriggerRecord {
+    const now = new Date().toISOString();
+    const id = input.id ?? generateId(input.flowName);
+    if (this.get(id)) throw new Error(`Trigger already exists: ${id}`);
+    const record: TriggerRecord = {
+      id,
+      flowName: input.flowName,
+      version: input.version ?? "@published",
+      cron: input.cron,
+      ...(input.tz ? { tz: input.tz } : {}),
+      ...(input.inputs ? { inputs: input.inputs } : {}),
+      enabled: input.enabled !== false,
+      ...(input.description ? { description: input.description } : {}),
+      createdAt: now,
+      updatedAt: now,
+    };
+    this.write(record);
+    return record;
+  }
+
+  get(id: string): TriggerRecord | null {
+    const file = this.filePath(id);
+    if (!fs.existsSync(file)) return null;
+    try {
+      return JSON.parse(fs.readFileSync(file, "utf8")) as TriggerRecord;
+    } catch {
+      return null;
+    }
+  }
+
+  update(id: string, patch: Partial<TriggerRecord>): TriggerRecord {
+    const existing = this.get(id);
+    if (!existing) throw new Error(`Trigger not found: ${id}`);
+    const updated: TriggerRecord = {
+      ...existing,
+      ...patch,
+      id: existing.id,
+      updatedAt: new Date().toISOString(),
+    };
+    this.write(updated);
+    return updated;
+  }
+
+  remove(id: string): boolean {
+    const file = this.filePath(id);
+    if (!fs.existsSync(file)) return false;
+    fs.unlinkSync(file);
+    return true;
+  }
+
+  list(opts?: { flowName?: string; enabledOnly?: boolean }): TriggerRecord[] {
+    if (!fs.existsSync(this.dir)) return [];
+    const records = fs
+      .readdirSync(this.dir)
+      .filter((f) => f.endsWith(".json"))
+      .map((f) => {
+        try {
+          return JSON.parse(
+            fs.readFileSync(path.join(this.dir, f), "utf8"),
+          ) as TriggerRecord;
+        } catch {
+          return null;
+        }
+      })
+      .filter((r): r is TriggerRecord => r !== null);
+
+    return records
+      .filter((r) => (opts?.flowName ? r.flowName === opts.flowName : true))
+      .filter((r) => (opts?.enabledOnly ? r.enabled : true))
+      .sort((a, b) => a.id.localeCompare(b.id));
+  }
+
+  /**
+   * Pause every trigger targeting a flow, returning the ids paused.
+   *
+   * Called when a flow is soft-deleted. Paused rather than removed so a restore
+   * keeps its schedules — and so restoring never silently re-arms unattended
+   * runs; that takes an explicit resume.
+   */
+  pauseForFlow(flowName: string): string[] {
+    const affected = this.list({ flowName }).filter((r) => r.enabled);
+    for (const record of affected) {
+      this.update(record.id, { enabled: false, nextRunAt: undefined });
+    }
+    return affected.map((r) => r.id);
+  }
+
+  // ---- Internals ---------------------------------------------------------------
+
+  private filePath(id: string): string {
+    const safe = id.replace(/[^a-zA-Z0-9_-]/g, "_");
+    return path.join(this.dir, `${safe}.json`);
+  }
+
+  private write(record: TriggerRecord): void {
+    fs.writeFileSync(
+      this.filePath(record.id),
+      JSON.stringify(record, null, 2),
+    );
+  }
+}
+
+function generateId(flowName: string): string {
+  const safe = flowName.replace(/[^a-zA-Z0-9_-]/g, "-").slice(0, 32);
+  return `${safe}-${randomUUID().slice(0, 8)}`;
+}
diff --git a/src/index.ts b/src/index.ts
index 5a19428..1bd0e3f 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -56,6 +56,10 @@ export type {
 } from "./core/types.js";
 export { parseDuration, MODEL_MAP, DEFAULT_MODEL, NODE_KEYS } from "./core/types.js";
 export { startFlowServer } from "./core/serve.js";
+export { TriggerStore } from "./core/triggers.js";
+export type { TriggerRecord, CreateTriggerInput } from "./core/triggers.js";
+export { TriggerScheduler, assertValidSchedule } from "./core/scheduler.js";
+export type { TriggerSchedulerOpts } from "./core/scheduler.js";
 export type { FlowServerOpts } from "./core/serve.js";
 export type { ServeConfig } from "./core/types.js";
 export {
@@ -65,5 +69,6 @@ export {
   listVersions,
   readVersion,
   readLatestVersion,
+  resolveRunnableFlow,
 } from "./core/manage.js";
 export type { PublishResult } from "./core/manage.js";
diff --git a/src/plugin/index.ts b/src/plugin/index.ts
index ff0a34e..95ff337 100644
--- a/src/plugin/index.ts
+++ b/src/plugin/index.ts
@@ -4,6 +4,8 @@ import * as path from "node:path";
 import { FlowRunner, sendEvent } from "../core/runner.js";
 
 import { startFlowServer } from "../core/serve.js";
+import { TriggerStore } from "../core/triggers.js";
+import { TriggerScheduler, assertValidSchedule } from "../core/scheduler.js";
 import { validateFlow } from "../core/validate.js";
 import {
   publishDraft,
@@ -110,6 +112,22 @@ function register(api: PluginApi) {
     });
   }
 
+  // ---- 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.
+  const flowsDir = pluginCfg.serve?.flowsDir ?? path.join(workspace, "flows");
+  const triggerStore = new TriggerStore(workspace);
+  const scheduler = new TriggerScheduler({
+    runner,
+    store: triggerStore,
+    workspace,
+    flowsDir,
+    logger: api.logger,
+  });
+  if (!process.env.CLAWFLOW_NO_SERVE) {
+    scheduler.start();
+  }
+
   // ---- 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
@@ -132,6 +150,18 @@ function register(api: PluginApi) {
   // independently of `enabled` (which governs flow_run) and does NOT honor
   // skipSessionPatterns. Kill-switch: approval.gateMutations=false.
   const gateMutations = approvalCfg.gateMutations !== false;
+  // Arming a schedule commits the box to running a flow unattended, on a timer,
+  // with tool access — at least as consequential as editing one, so it is gated
+  // the same way. Reads (list) are not gated.
+  const TRIGGER_MUTATION_VERBS: Record<string, string> = {
+    create: "Schedule",
+    update: "Reschedule",
+    delete: "Unschedule",
+    pause: "Pause schedule for",
+    resume: "Resume schedule for",
+    run_now: "Run now",
+  };
+
   const MUTATION_VERBS: Record<string, string> = {
     flow_create: "Create",
     flow_edit: "Edit",
@@ -147,14 +177,22 @@ function register(api: PluginApi) {
 
         // Flow-authoring tools — always gate (no skipSessionPatterns). No
         // allow-always persist path, so every call re-prompts.
-        const mutationVerb = toolName ? MUTATION_VERBS[toolName] : undefined;
+        // flow_trigger is action-shaped: gate the mutating actions, let reads through.
+        let mutationVerb = toolName ? MUTATION_VERBS[toolName] : undefined;
+        if (toolName === "flow_trigger") {
+          const action = (event.params as { action?: string } | undefined)?.action;
+          mutationVerb = TRIGGER_MUTATION_VERBS[action ?? ""] ?? undefined;
+        }
         if (gateMutations && mutationVerb) {
-          const mp = (event.params ?? {}) as { file?: string; flow?: string };
-          const name = mp.flow ?? mp.file ?? "inline flow";
+          const mp = (event.params ?? {}) as { file?: string; flow?: string; id?: string };
+          const name = mp.flow ?? mp.file ?? mp.id ?? "inline flow";
           return {
             requireApproval: {
               title: `${mutationVerb} clawflow "${name}"?`.slice(0, 80),
-              description: "Creates, edits, publishes, or deletes a flow definition.",
+              description:
+                toolName === "flow_trigger"
+                  ? "Changes an unattended schedule that runs a flow on a timer."
+                  : "Creates, edits, publishes, or deletes a flow definition.",
               severity: "warning",
               timeoutMs: approvalTimeoutMs,
               timeoutBehavior: approvalTimeoutBehavior,
@@ -418,11 +456,19 @@ Safe for agents to call without fear of data loss.`,
 
         fs.renameSync(abs, binPath);
 
+        // Schedules outlive a soft delete, but must not keep firing at a flow
+        // that is no longer there. Paused, not removed, so a restore keeps them.
+        const pausedIds = triggerStore.pauseForFlow(path.basename(abs, ".json"));
+        const pausedNote =
+          pausedIds.length > 0
+            ? `\nPaused ${pausedIds.length} trigger(s) for this flow: ${pausedIds.join(", ")}`
+            : "";
+
         return {
           content: [
             {
               type: "text",
-              text: `Flow moved to bin: ${binPath}`,
+              text: `Flow moved to bin: ${binPath}${pausedNote}`,
             },
           ],
         };
@@ -1941,6 +1987,212 @@ re-send the whole flow: a single large edit is slow and can stall mid-generation
     },
     { optional: true },
   );
+
+  // ---- flow_trigger -------------------------------------------------------------
+
+  api.registerTool(
+    {
+      name: "flow_trigger",
+      description: `Schedule a flow to run on a recurring cron schedule on this box.
+
+A trigger is a separate record, not a field on the flow — so one flow can have
+several triggers with different cadences and different inputs, and pausing one
+never touches the flow definition or mints a new version.
+
+Actions:
+  create   — arm a new schedule (flow + cron required)
+  update   — edit an existing schedule (cron, tz, inputs, version, description)
+  list     — list triggers, optionally filtered by flow
+  pause    — stop firing, keep the record
+  resume   — start firing again
+  delete   — remove the record
+  run_now  — fire immediately, ignoring the schedule
+
+Schedules are standard 5-field cron ("0 9 * * *"). Minimum interval is 60s.
+Runs missed while the host was down are NOT replayed.
+By default a trigger runs the latest PUBLISHED version; pin one with "version".`,
+
+      parameters: {
+        type: "object",
+        required: ["action"],
+        properties: {
+          action: {
+            type: "string",
+            enum: ["create", "update", "list", "pause", "resume", "delete", "run_now"],
+          },
+          id: {
+            type: "string",
+            description: "Trigger id. Required for update/pause/resume/delete/run_now.",
+          },
+          flow: {
+            type: "string",
+            description: "Flow name. Required for create; optional filter for list.",
+          },
+          cron: {
+            type: "string",
+            description:
+              'Cron expression, e.g. "0 9 * * *". Required for create; optional for update.',
+          },
+          tz: {
+            type: "string",
+            description: 'IANA timezone, e.g. "Europe/Rome". Host local time when omitted.',
+          },
+          inputs: {
+            type: "object",
+            additionalProperties: true,
+            description: "Payload passed as the flow's inputs on every fire.",
+          },
+          version: {
+            type: "number",
+            description: "Pin a published version. Omit to always run the latest published.",
+          },
+          description: { type: "string" },
+        },
+      },
+
+      async execute(
+        _id: string,
+        params: {
+          action: string;
+          id?: string;
+          flow?: string;
+          cron?: string;
+          tz?: string;
+          inputs?: Record<string, unknown>;
+          version?: number;
+          description?: string;
+        },
+      ) {
+        const text = (t: string) => ({ content: [{ type: "text", text: t }] });
+
+        const requireId = (): string => {
+          if (!params.id) throw new Error(`"id" is required for action "${params.action}"`);
+          if (!triggerStore.get(params.id)) throw new Error(`Trigger not found: ${params.id}`);
+          return params.id;
+        };
+
+        const describe = (r: {
+          id: string;
+          flowName: string;
+          cron: string;
+          tz?: string;
+          version: number | "@published";
+          enabled: boolean;
+          nextRunAt?: string;
+          lastRunAt?: string;
+          lastStatus?: string;
+        }) =>
+          `${r.enabled ? "●" : "○"} ${r.id}\n` +
+          `    flow: ${r.flowName} (${r.version === "@published" ? "latest published" : `v${r.version}`})\n` +
+          `    cron: ${r.cron}${r.tz ? ` [${r.tz}]` : ""}\n` +
+          `    next: ${r.enabled ? (r.nextRunAt ?? "—") : "paused"}` +
+          (r.lastRunAt ? `\n    last: ${r.lastRunAt} (${r.lastStatus ?? "?"})` : "");
+
+        try {
+          switch (params.action) {
+            case "create": {
+              if (!params.flow) throw new Error('"flow" is required for action "create"');
+              if (!params.cron) throw new Error('"cron" is required for action "create"');
+              // Validate before storing: a bad schedule must fail here, not at
+              // fire time on a box nobody is watching.
+              assertValidSchedule(params.cron, params.tz);
+              const record = triggerStore.create({
+                flowName: params.flow,
+                cron: params.cron,
+                tz: params.tz,
+                inputs: params.inputs,
+                version: params.version,
+                description: params.description,
+              });
+              scheduler.sync();
+              const stored = triggerStore.get(record.id) ?? record;
+              return text(`Scheduled.\n\n${describe(stored)}`);
+            }
+
+            case "update": {
+              const id = requireId();
+              const current = triggerStore.get(id)!;
+              const cron = params.cron ?? current.cron;
+              const tz = params.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);
+              triggerStore.update(id, {
+                cron,
+                ...(params.tz !== undefined ? { tz: params.tz } : {}),
+                ...(params.inputs !== undefined ? { inputs: params.inputs } : {}),
+                ...(params.version !== undefined ? { version: params.version } : {}),
+                ...(params.description !== undefined
+                  ? { description: params.description }
+                  : {}),
+              });
+              scheduler.sync();
+              return text(`Updated.\n\n${describe(triggerStore.get(id)!)}`);
+            }
+
+            case "list": {
+              const records = triggerStore.list(
+                params.flow ? { flowName: params.flow } : undefined,
+              );
+              if (records.length === 0) {
+                return text(
+                  params.flow
+                    ? `No triggers for flow "${params.flow}".`
+                    : "No triggers on this box.",
+                );
+              }
+              return text(records.map(describe).join("\n\n"));
+            }
+
+            case "pause": {
+              const id = requireId();
+              triggerStore.update(id, { enabled: false, nextRunAt: undefined });
+              scheduler.sync();
+              return text(`Paused ${id}. The flow definition is unchanged.`);
+            }
+
+            case "resume": {
+              const id = requireId();
+              triggerStore.update(id, { enabled: true });
+              scheduler.sync();
+              return text(`Resumed.\n\n${describe(triggerStore.get(id)!)}`);
+            }
+
+            case "delete": {
+              const id = requireId();
+              triggerStore.remove(id);
+              scheduler.sync();
+              return text(`Deleted trigger ${id}.`);
+            }
+
+            case "run_now": {
+              const id = requireId();
+              const result = await scheduler.runNow(id);
+              if (!result) {
+                const stored = triggerStore.get(id);
+                return text(
+                  `Trigger ${id} did not run: ${stored?.lastError ?? "flow could not be resolved"}`,
+                );
+              }
+              return text(
+                `Ran ${id} → instance ${result.instanceId} (${result.status}).`,
+              );
+            }
+
+            default:
+              return text(`Unknown action: "${params.action}"`);
+          }
+        } catch (err) {
+          return text(
+            `flow_trigger ${params.action} failed: ${
+              err instanceof Error ? err.message : String(err)
+            }`,
+          );
+        }
+      },
+    },
+    { optional: true },
+  );
 }
 
 export default {
diff --git a/tests/approval-gate.test.ts b/tests/approval-gate.test.ts
index 7a8dd02..639306f 100644
--- a/tests/approval-gate.test.ts
+++ b/tests/approval-gate.test.ts
@@ -87,3 +87,58 @@ describe("clawflow approval gate — flow mutation tools", () => {
     assert.equal(await hook({ toolName: "flow_read", params: { file: "f" } }), undefined);
   });
 });
+
+describe("clawflow approval gate — flow_trigger", () => {
+  it("gates the mutating actions", async () => {
+    const hook = captureHook({});
+    for (const [action, verb] of [
+      ["create", "Schedule"],
+      ["update", "Reschedule"],
+      ["delete", "Unschedule"],
+      ["pause", "Pause schedule for"],
+      ["resume", "Resume schedule for"],
+      ["run_now", "Run now"],
+    ] as const) {
+      const res = await hook({
+        toolName: "flow_trigger",
+        params: { action, flow: "my-flow" },
+      });
+      assert.ok(res && res.requireApproval, `flow_trigger ${action} should require approval`);
+      assert.match(res.requireApproval!.title, new RegExp(`^${verb} clawflow "my-flow"`));
+      assert.match(res.requireApproval!.description, /unattended schedule/);
+    }
+  });
+
+  it("does not gate list", async () => {
+    const hook = captureHook({});
+    const res = await hook({ toolName: "flow_trigger", params: { action: "list" } });
+    assert.ok(!res || !res.requireApproval, "listing triggers is a read, not a mutation");
+  });
+
+  it("names the trigger id when there is no flow name in params", async () => {
+    const hook = captureHook({});
+    const res = await hook({
+      toolName: "flow_trigger",
+      params: { action: "pause", id: "digest-ab12cd34" },
+    });
+    assert.match(res!.requireApproval!.title, /"digest-ab12cd34"/);
+  });
+
+  it("gates trigger mutations even when the flow_run gate is disabled", async () => {
+    const hook = captureHook({ approval: { enabled: false } });
+    const res = await hook({
+      toolName: "flow_trigger",
+      params: { action: "create", flow: "x" },
+    });
+    assert.ok(res && res.requireApproval, "arming a schedule must still gate");
+  });
+
+  it("respects the gateMutations kill-switch", async () => {
+    const hook = captureHook({ approval: { gateMutations: false } });
+    const res = await hook({
+      toolName: "flow_trigger",
+      params: { action: "create", flow: "x" },
+    });
+    assert.ok(!res || !res.requireApproval);
+  });
+});
diff --git a/tests/flow-trigger-tool.test.ts b/tests/flow-trigger-tool.test.ts
new file mode 100644
index 0000000..fc06325
--- /dev/null
+++ b/tests/flow-trigger-tool.test.ts
@@ -0,0 +1,245 @@
+import { describe, it, 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 plugin from "../src/plugin/index.js";
+import type { FlowDefinition } from "../src/index.js";
+
+const roots: string[] = [];
+
+type ToolResult = { content: Array<{ type: string; text: string }> };
+type Tool = {
+  name: string;
+  execute: (id: string, params: Record<string, unknown>) => Promise<ToolResult>;
+};
+
+/** Register the plugin against a mock api and return the flow_trigger tool. */
+function harness(): { tool: Tool; workspace: string } {
+  const workspace = fs.mkdtempSync(path.join(os.tmpdir(), "ocf-trigtool-"));
+  roots.push(workspace);
+  const tools = new Map<string, Tool>();
+  const api = {
+    registerTool: (def: Tool) => tools.set(def.name, def),
+    registerHook: () => {},
+    config: {
+      workspace,
+      plugins: {
+        entries: {
+          clawflow: {
+            config: {
+              stateDir: path.join(workspace, "state"),
+              memoryDir: path.join(workspace, "memory"),
+            },
+          },
+        },
+      },
+    },
+    logger: { info: () => {}, warn: () => {}, error: () => {} },
+  };
+  plugin.register(api as never);
+  const tool = tools.get("flow_trigger");
+  assert.ok(tool, "flow_trigger tool was not registered");
+  return { tool: tool!, workspace };
+}
+
+function writeFlow(workspace: string, name: string, run: string): void {
+  const def: FlowDefinition = {
+    flow: name,
+    nodes: [{ name: "step1", do: "code", run, output: "result" }],
+  };
+  fs.mkdirSync(path.join(workspace, "flows"), { recursive: true });
+  fs.writeFileSync(
+    path.join(workspace, "flows", `${name}.json`),
+    JSON.stringify(def, null, 2),
+  );
+}
+
+const say = (r: ToolResult) => r.content[0].text;
+const idFrom = (text: string) => {
+  const m = text.match(/[●○] (\S+)/);
+  assert.ok(m, `no trigger id in output:\n${text}`);
+  return m![1];
+};
+
+after(() => {
+  for (const dir of roots) fs.rmSync(dir, { recursive: true, force: true });
+});
+
+describe("flow_trigger tool", () => {
+  it("creates a trigger and reports its schedule", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "digest", "1");
+
+    const out = say(
+      await tool.execute("t", {
+        action: "create",
+        flow: "digest",
+        cron: "0 9 * * *",
+        tz: "Europe/Rome",
+      }),
+    );
+    assert.match(out, /Scheduled\./);
+    assert.match(out, /flow: digest \(latest published\)/);
+    assert.match(out, /cron: 0 9 \* \* \* \[Europe\/Rome\]/);
+  });
+
+  it("rejects a sub-minute schedule instead of storing it", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "spammy", "1");
+
+    const out = say(
+      await tool.execute("t", { action: "create", flow: "spammy", cron: "* * * * * *" }),
+    );
+    assert.match(out, /minimum interval is 60s/);
+    const list = say(await tool.execute("t", { action: "list" }));
+    assert.match(list, /No triggers/, "nothing was stored");
+  });
+
+  it("rejects an unparseable schedule", async () => {
+    const { tool } = harness();
+    const out = say(
+      await tool.execute("t", { action: "create", flow: "f", cron: "every tuesday" }),
+    );
+    assert.match(out, /Invalid schedule/);
+  });
+
+  it("requires flow and cron on create", async () => {
+    const { tool } = harness();
+    assert.match(say(await tool.execute("t", { action: "create", cron: "0 9 * * *" })), /"flow" is required/);
+    assert.match(say(await tool.execute("t", { action: "create", flow: "f" })), /"cron" is required/);
+  });
+
+  it("lists several triggers for one flow and filters by flow", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "multi", "1");
+    writeFlow(workspace, "other", "1");
+    await tool.execute("t", { action: "create", flow: "multi", cron: "0 * * * *", inputs: { c: "a" } });
+    await tool.execute("t", { action: "create", flow: "multi", cron: "0 9 * * *", inputs: { c: "b" } });
+    await tool.execute("t", { action: "create", flow: "other", cron: "0 9 * * *" });
+
+    const all = say(await tool.execute("t", { action: "list" }));
+    assert.equal((all.match(/[●○] /g) ?? []).length, 3);
+
+    const filtered = say(await tool.execute("t", { action: "list", flow: "multi" }));
+    assert.equal((filtered.match(/[●○] /g) ?? []).length, 2, "two schedules on one flow");
+  });
+
+  it("pauses and resumes without touching the flow definition", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "pauseme", "1");
+    const before = fs.readFileSync(path.join(workspace, "flows", "pauseme.json"), "utf8");
+    const id = idFrom(
+      say(await tool.execute("t", { action: "create", flow: "pauseme", cron: "0 9 * * *" })),
+    );
+
+    assert.match(say(await tool.execute("t", { action: "pause", id })), /Paused/);
+    assert.match(say(await tool.execute("t", { action: "list" })), /^○/m);
+
+    assert.match(say(await tool.execute("t", { action: "resume", id })), /Resumed/);
+    assert.match(say(await tool.execute("t", { action: "list" })), /^●/m);
+
+    const after = fs.readFileSync(path.join(workspace, "flows", "pauseme.json"), "utf8");
+    assert.equal(after, before, "flow definition is byte-identical");
+    assert.equal(
+      fs.existsSync(path.join(workspace, ".clawflow", "versions", "pauseme")),
+      false,
+      "pausing did not mint a version",
+    );
+  });
+
+  it("edits an existing schedule in place", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "editme", "1");
+    const id = idFrom(
+      say(await tool.execute("t", { action: "create", flow: "editme", cron: "0 9 * * *" })),
+    );
+
+    const out = say(
+      await tool.execute("t", {
+        action: "update",
+        id,
+        cron: "30 6 * * 1",
+        tz: "Europe/Rome",
+        inputs: { customer: "acme" },
+      }),
+    );
+    assert.match(out, /Updated\./);
+    assert.match(out, /cron: 30 6 \* \* 1 \[Europe\/Rome\]/);
+
+    const listed = say(await tool.execute("t", { action: "list" }));
+    assert.match(listed, /30 6 \* \* 1/, "the change persisted");
+    assert.doesNotMatch(listed, /0 9 \* \* \*/, "old cadence is gone");
+  });
+
+  it("keeps untouched fields when updating one", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "partial", "1");
+    const id = idFrom(
+      say(await tool.execute("t", {
+        action: "create", flow: "partial", cron: "0 9 * * *", tz: "Europe/Rome",
+      })),
+    );
+    const out = say(await tool.execute("t", { action: "update", id, cron: "0 10 * * *" }));
+    assert.match(out, /cron: 0 10 \* \* \* \[Europe\/Rome\]/, "timezone survived");
+  });
+
+  it("rejects an edit that would produce an invalid schedule", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "badedit", "1");
+    const id = idFrom(
+      say(await tool.execute("t", { action: "create", flow: "badedit", cron: "0 9 * * *" })),
+    );
+
+    assert.match(
+      say(await tool.execute("t", { action: "update", id, cron: "* * * * * *" })),
+      /minimum interval is 60s/,
+    );
+    assert.match(
+      say(await tool.execute("t", { action: "update", id, tz: "Mars/Olympus" })),
+      /Invalid schedule/,
+    );
+    assert.match(
+      say(await tool.execute("t", { action: "list" })),
+      /0 9 \* \* \*/,
+      "the original schedule is intact",
+    );
+  });
+
+  it("deletes a trigger", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "gone", "1");
+    const id = idFrom(say(await tool.execute("t", { action: "create", flow: "gone", cron: "0 9 * * *" })));
+
+    assert.match(say(await tool.execute("t", { action: "delete", id })), /Deleted/);
+    assert.match(say(await tool.execute("t", { action: "list" })), /No triggers/);
+  });
+
+  it("errors clearly on a missing id", async () => {
+    const { tool } = harness();
+    assert.match(say(await tool.execute("t", { action: "pause" })), /"id" is required/);
+    assert.match(say(await tool.execute("t", { action: "pause", id: "nope" })), /Trigger not found/);
+  });
+
+  it("run_now fires the flow immediately", async () => {
+    const { tool, workspace } = harness();
+    writeFlow(workspace, "runnow", "'fired'");
+    const id = idFrom(say(await tool.execute("t", { action: "create", flow: "runnow", cron: "0 9 * * *" })));
+
+    const out = say(await tool.execute("t", { action: "run_now", id }));
+    assert.match(out, /Ran .* → instance .* \(completed\)/);
+  });
+
+  it("run_now reports a missing flow instead of pretending it ran", async () => {
+    const { tool } = harness();
+    const id = idFrom(say(await tool.execute("t", { action: "create", flow: "ghost", cron: "0 9 * * *" })));
+    const out = say(await tool.execute("t", { action: "run_now", id }));
+    assert.match(out, /did not run/);
+  });
+
+  it("rejects an unknown action", async () => {
+    const { tool } = harness();
+    assert.match(say(await tool.execute("t", { action: "explode" })), /Unknown action/);
+  });
+});
diff --git a/tests/scheduler.test.ts b/tests/scheduler.test.ts
new file mode 100644
index 0000000..2fc2391
--- /dev/null
+++ b/tests/scheduler.test.ts
@@ -0,0 +1,296 @@
+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 } from "../src/index.js";
+import { TriggerStore } from "../src/core/triggers.js";
+import { TriggerScheduler, assertValidSchedule } from "../src/core/scheduler.js";
+import { publishDraft } from "../src/core/manage.js";
+import type { FlowDefinition, PluginConfig } from "../src/index.js";
+
+const tmpDir = path.join(os.tmpdir(), `ocf-scheduler-test-${Date.now()}`);
+const workspace = path.join(tmpDir, "workspace");
+const flowsDir = path.join(workspace, "flows");
+const baseCfg: PluginConfig = {
+  stateDir: path.join(tmpDir, "state"),
+  memoryDir: path.join(tmpDir, "memory"),
+};
+
+const silent = { info: () => {}, warn: () => {}, error: () => {} };
+
+function cleanup() {
+  fs.rmSync(tmpDir, { recursive: true, force: true });
+}
+
+function writeDraft(name: string, run: string): void {
+  const def: FlowDefinition = {
+    flow: name,
+    nodes: [{ name: "step1", do: "code", run, output: "result" }],
+  };
+  fs.mkdirSync(flowsDir, { recursive: true });
+  fs.writeFileSync(path.join(flowsDir, `${name}.json`), JSON.stringify(def, null, 2));
+}
+
+function harness() {
+  const store = new TriggerStore(workspace);
+  const runner = new FlowRunner(baseCfg);
+  const scheduler = new TriggerScheduler({
+    runner,
+    store,
+    workspace,
+    flowsDir,
+    logger: silent,
+  });
+  return { store, scheduler };
+}
+
+describe("assertValidSchedule", () => {
+  it("accepts a standard 5-field expression with a timezone", () => {
+    assert.doesNotThrow(() => assertValidSchedule("0 9 * * *", "Europe/Rome"));
+  });
+
+  it("rejects an unparseable expression", () => {
+    assert.throws(() => assertValidSchedule("not a cron"), /Invalid schedule/);
+  });
+
+  it("rejects an invalid timezone", () => {
+    assert.throws(() => assertValidSchedule("0 9 * * *", "Mars/Olympus"), /Invalid schedule/);
+  });
+
+  it("rejects sub-minute schedules — agents can write these", () => {
+    assert.throws(() => assertValidSchedule("* * * * * *"), /minimum interval is 60s/);
+    assert.throws(() => assertValidSchedule("*/5 * * * * *"), /minimum interval is 60s/);
+  });
+
+  it("accepts every-minute, the fastest allowed cadence", () => {
+    assert.doesNotThrow(() => assertValidSchedule("* * * * *"));
+  });
+});
+
+describe("TriggerScheduler", () => {
+  before(cleanup);
+  after(cleanup);
+
+  it("arms only enabled triggers and disarms paused ones", () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("armed", "1");
+
+    const on = store.create({ flowName: "armed", cron: "0 9 * * *" });
+    const off = store.create({ flowName: "armed", cron: "0 10 * * *", enabled: false });
+
+    scheduler.start();
+    assert.ok(scheduler.nextRun(on.id), "enabled trigger is armed");
+    assert.equal(scheduler.nextRun(off.id), null, "disabled trigger is not armed");
+
+    store.update(on.id, { enabled: false });
+    scheduler.sync();
+    assert.equal(scheduler.nextRun(on.id), null, "pausing disarms on next sync");
+    scheduler.stop();
+  });
+
+  it("records nextRunAt when arming", () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("nextrun", "1");
+    const rec = store.create({ flowName: "nextrun", cron: "0 9 * * *", tz: "Europe/Rome" });
+
+    scheduler.start();
+    const stored = store.get(rec.id);
+    assert.ok(stored?.nextRunAt, "nextRunAt persisted");
+    assert.equal(
+      new Date(stored!.nextRunAt!).toISOString(),
+      scheduler.nextRun(rec.id)!.toISOString(),
+      "persisted value matches the armed timer",
+    );
+    scheduler.stop();
+  });
+
+  it("runs the published version, not the draft", async () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("versioned", "'v1'");
+    publishDraft(workspace, "versioned");
+    // Draft moves on; the published version must still be what fires.
+    writeDraft("versioned", "'draft-edit'");
+
+    const rec = store.create({ flowName: "versioned", cron: "0 9 * * *" });
+    const result = await scheduler.runNow(rec.id);
+
+    assert.equal(result?.ok, true);
+    assert.equal(result?.state.result, "v1", "fired the published version");
+    scheduler.stop();
+  });
+
+  it("honors a pinned version", async () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("pinned", "'one'");
+    publishDraft(workspace, "pinned");
+    writeDraft("pinned", "'two'");
+    publishDraft(workspace, "pinned");
+
+    const rec = store.create({ flowName: "pinned", cron: "0 9 * * *", version: 1 });
+    const result = await scheduler.runNow(rec.id);
+    assert.equal(result?.state.result, "one", "ran v1, not latest");
+    scheduler.stop();
+  });
+
+  it("passes the record's inputs to the flow", async () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("with-inputs", "state.inputs.customer");
+    const rec = store.create({
+      flowName: "with-inputs",
+      cron: "0 9 * * *",
+      inputs: { customer: "acme" },
+    });
+
+    const result = await scheduler.runNow(rec.id);
+    assert.equal(result?.state.result, "acme");
+    scheduler.stop();
+  });
+
+  it("records a skip when the flow is missing instead of throwing", async () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    const rec = store.create({ flowName: "ghost", cron: "0 9 * * *" });
+
+    const result = await scheduler.runNow(rec.id);
+    assert.equal(result, null);
+    const stored = store.get(rec.id);
+    assert.equal(stored?.lastStatus, "skipped");
+    assert.match(stored?.lastError ?? "", /not found/);
+    scheduler.stop();
+  });
+
+  it("records run bookkeeping after a successful fire", async () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("bookkeeping", "1");
+    const rec = store.create({ flowName: "bookkeeping", cron: "0 9 * * *" });
+
+    await scheduler.runNow(rec.id);
+    const stored = store.get(rec.id);
+    assert.equal(stored?.lastStatus, "ok");
+    assert.ok(stored?.lastRunAt, "lastRunAt stamped");
+    assert.ok(stored?.lastInstanceId, "instance id recorded");
+    scheduler.stop();
+  });
+
+  it("does not fire a paused trigger even if its timer survives", async () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("paused", "1");
+    const rec = store.create({ flowName: "paused", cron: "0 9 * * *" });
+    store.update(rec.id, { enabled: false });
+
+    // Simulate the armed timer firing after the record was paused.
+    const fire = (scheduler as unknown as {
+      fire: (r: unknown) => Promise<unknown>;
+    }).fire.bind(scheduler);
+    const result = await fire(rec);
+
+    assert.equal(result, null, "re-read the record and declined to run");
+    assert.equal(store.get(rec.id)?.lastStatus, undefined, "no run recorded");
+    scheduler.stop();
+  });
+
+  it("picks up a trigger written out-of-process", () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("offbox", "1");
+    scheduler.start();
+
+    // The Clawnify hook server / dashboard writes records through TriggerStore
+    // imported from dist — a different process, same directory.
+    const offBox = new TriggerStore(workspace);
+    const rec = offBox.create({ flowName: "offbox", cron: "0 9 * * *" });
+    assert.equal(scheduler.nextRun(rec.id), null, "not armed until the next resync");
+
+    scheduler.sync();
+    assert.ok(scheduler.nextRun(rec.id), "armed after resync");
+    scheduler.stop();
+  });
+
+  it("picks up an out-of-process edit and re-arms to the new cadence", () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("retimed", "1");
+    const rec = store.create({ flowName: "retimed", cron: "0 9 * * *", tz: "UTC" });
+    scheduler.start();
+    const before = scheduler.nextRun(rec.id)!;
+
+    new TriggerStore(workspace).update(rec.id, { cron: "0 10 * * *" });
+    scheduler.sync();
+
+    const after = scheduler.nextRun(rec.id)!;
+    assert.notEqual(before.toISOString(), after.toISOString(), "re-armed to the new time");
+    assert.equal(after.getUTCHours(), 10);
+    scheduler.stop();
+  });
+
+  it("does not rebuild an unchanged timer on resync", () => {
+    cleanup();
+    const { store, scheduler } = harness();
+    writeDraft("stable", "1");
+    const rec = store.create({ flowName: "stable", cron: "0 9 * * *" });
+    scheduler.start();
+
+    const jobs = (scheduler as unknown as { jobs: Map<string, unknown> }).jobs;
+    const first = jobs.get(rec.id);
+    scheduler.sync();
+    assert.equal(jobs.get(rec.id), first, "same timer survives a no-op resync");
+
+    store.update(rec.id, { cron: "0 11 * * *" });
+    scheduler.sync();
+    assert.notEqual(jobs.get(rec.id), first, "a real change rebuilds it");
+    scheduler.stop();
+  });
+
+  it("fires repeatedly on its own timer, end to end", async () => {
+    cleanup();
+    const store = new TriggerStore(workspace);
+    writeDraft("ticker", "'tick'");
+
+    // Count real fires through the runner so this proves the timer RE-ARMS,
+    // not just that it fired once.
+    const inner = new FlowRunner(baseCfg);
+    const fired: string[] = [];
+    const counting = {
+      run: (def: FlowDefinition, inputs: unknown, instanceId: string) => {
+        fired.push(instanceId);
+        return inner.run(def, inputs, instanceId);
+      },
+    };
+    const scheduler = new TriggerScheduler({
+      runner: counting,
+      store,
+      workspace,
+      flowsDir,
+      logger: silent,
+    });
+
+    // Sub-minute is rejected at the tool boundary; written straight to the
+    // store here so the timer path itself can be exercised in-test.
+    const rec = store.create({ flowName: "ticker", cron: "* * * * * *" });
+
+    scheduler.start();
+    await new Promise((r) => setTimeout(r, 3200));
+    scheduler.stop();
+
+    assert.ok(
+      fired.length >= 2,
+      `expected the timer to re-arm and fire at least twice, got ${fired.length}`,
+    );
+    assert.equal(new Set(fired).size, fired.length, "each fire gets its own instance id");
+
+    const stored = store.get(rec.id);
+    assert.equal(stored?.lastStatus, "ok", "the timer fired the flow unattended");
+    assert.ok(stored?.lastRunAt, "run was recorded");
+    assert.ok(stored?.nextRunAt, "next occurrence recorded after firing");
+  });
+
+});
diff --git a/tests/triggers.test.ts b/tests/triggers.test.ts
new file mode 100644
index 0000000..438ee50
--- /dev/null
+++ b/tests/triggers.test.ts
@@ -0,0 +1,157 @@
+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 { TriggerStore } from "../src/core/triggers.js";
+
+const tmpDir = path.join(os.tmpdir(), `ocf-triggers-test-${Date.now()}`);
+const workspace = path.join(tmpDir, "workspace");
+
+function cleanup() {
+  fs.rmSync(tmpDir, { recursive: true, force: true });
+}
+
+function freshStore(): TriggerStore {
+  cleanup();
+  return new TriggerStore(workspace);
+}
+
+describe("TriggerStore", () => {
+  before(cleanup);
+  after(cleanup);
+
+  it("creates a record with defaults and round-trips it", () => {
+    const store = freshStore();
+    const created = store.create({ flowName: "daily-digest", cron: "0 9 * * *" });
+
+    assert.equal(created.flowName, "daily-digest");
+    assert.equal(created.version, "@published", "defaults to latest published");
+    assert.equal(created.enabled, true, "defaults to enabled");
+    assert.ok(created.id.startsWith("daily-digest-"));
+
+    const read = store.get(created.id);
+    assert.deepEqual(read, created, "survives a write/read round-trip");
+  });
+
+  it("persists to .clawflow/triggers alongside versions", () => {
+    const store = freshStore();
+    const created = store.create({ flowName: "f", cron: "* * * * *" });
+    const file = path.join(workspace, ".clawflow", "triggers", `${created.id}.json`);
+    assert.ok(fs.existsSync(file), "record written to the expected path");
+  });
+
+  it("holds N triggers for one flow with different cadences and inputs", () => {
+    const store = freshStore();
+    const hourly = store.create({
+      flowName: "digest",
+      cron: "0 * * * *",
+      inputs: { customer: "acme" },
+    });
+    const daily = store.create({
+      flowName: "digest",
+      cron: "0 9 * * *",
+      tz: "Europe/Rome",
+      inputs: { customer: "globex" },
+    });
+
+    assert.notEqual(hourly.id, daily.id, "ids are distinct");
+    const forFlow = store.list({ flowName: "digest" });
+    assert.equal(forFlow.length, 2);
+    assert.deepEqual(
+      forFlow.map((r) => r.inputs?.customer).sort(),
+      ["acme", "globex"],
+      "each carries its own inputs",
+    );
+  });
+
+  it("pauses without touching the flow definition", () => {
+    const store = freshStore();
+    const created = store.create({ flowName: "f", cron: "0 9 * * *" });
+    const paused = store.update(created.id, { enabled: false });
+
+    assert.equal(paused.enabled, false);
+    assert.equal(paused.cron, created.cron, "cadence untouched");
+    assert.notEqual(paused.updatedAt, undefined);
+    assert.equal(store.list({ enabledOnly: true }).length, 0);
+    assert.equal(store.list().length, 1, "paused records are still listed");
+  });
+
+  it("pins a version and keeps it across updates", () => {
+    const store = freshStore();
+    const created = store.create({ flowName: "f", cron: "0 9 * * *", version: 3 });
+    assert.equal(created.version, 3);
+    const touched = store.update(created.id, { lastStatus: "ok" });
+    assert.equal(touched.version, 3, "run bookkeeping does not drift the pin");
+  });
+
+  it("refuses to overwrite an existing id", () => {
+    const store = freshStore();
+    const created = store.create({ flowName: "f", cron: "* * * * *" });
+    assert.throws(
+      () => store.create({ id: created.id, flowName: "f", cron: "* * * * *" }),
+      /already exists/,
+    );
+  });
+
+  it("throws when updating a missing record", () => {
+    const store = freshStore();
+    assert.throws(() => store.update("nope", { enabled: false }), /not found/);
+  });
+
+  it("pauses a flow's triggers when it is soft-deleted, keeping the records", () => {
+    const store = freshStore();
+    store.create({ flowName: "doomed", cron: "0 9 * * *" });
+    store.create({ flowName: "doomed", cron: "0 10 * * *" });
+    store.create({ flowName: "survivor", cron: "0 9 * * *" });
+
+    const paused = store.pauseForFlow("doomed");
+    assert.equal(paused.length, 2);
+    assert.equal(store.list({ flowName: "doomed" }).length, 2, "records survive for restore");
+    assert.equal(
+      store.list({ flowName: "doomed", enabledOnly: true }).length,
+      0,
+      "none of them still fire",
+    );
+    assert.equal(
+      store.list({ flowName: "survivor", enabledOnly: true }).length,
+      1,
+      "unrelated flow untouched",
+    );
+  });
+
+  it("pauseForFlow reports only what it actually changed", () => {
+    const store = freshStore();
+    const already = store.create({ flowName: "f", cron: "0 9 * * *", enabled: false });
+    const live = store.create({ flowName: "f", cron: "0 10 * * *" });
+    const paused = store.pauseForFlow("f");
+    assert.deepEqual(paused, [live.id], "already-paused record not reported");
+    assert.ok(already.id);
+  });
+
+  it("remove() reports whether anything was deleted", () => {
+    const store = freshStore();
+    const created = store.create({ flowName: "f", cron: "* * * * *" });
+    assert.equal(store.remove(created.id), true);
+    assert.equal(store.remove(created.id), false, "second delete is a no-op");
+  });
+
+  it("skips corrupt files instead of failing the whole listing", () => {
+    const store = freshStore();
+    store.create({ flowName: "good", cron: "* * * * *" });
+    fs.writeFileSync(
+      path.join(workspace, ".clawflow", "triggers", "corrupt.json"),
+      "{ not json",
+    );
+    assert.equal(store.list().length, 1, "one good record still lists");
+  });
+
+  it("sanitizes ids so a record cannot escape the triggers dir", () => {
+    const store = freshStore();
+    const created = store.create({ id: "../../escape", flowName: "f", cron: "* * * * *" });
+    assert.ok(store.get(created.id), "readable through the store");
+    const escaped = path.join(workspace, ".clawflow", "escape.json");
+    assert.ok(!fs.existsSync(escaped), "did not write outside the triggers dir");
+  });
+});