From f8aaf683d293676d5b1c8d54f006226053d51d94 Mon Sep 17 00:00:00 2001 From: pallaoro Date: Sun, 8 Nov 2026 14:13:13 +0100 Subject: [PATCH 1/4] Add native SQLite and PostgreSQL execution storage --- .github/workflows/ci.yml | 14 ++ README.md | 130 ++++++++++++++++- openclaw.plugin.json | 58 ++++++++ package-lock.json | 147 ++++++++++++++++++- package.json | 10 +- scripts/migrate-storage.ts | 17 +++ src/core/active-runs.ts | 2 +- src/core/migrate-storage.ts | 47 +++++++ src/core/runner.ts | 212 +++++++++++++--------------- src/core/runs.ts | 9 +- src/core/sql-driver.ts | 188 +++++++++++++++++++++++++ src/core/sql-store.ts | 199 ++++++++++++++++++++++++++ src/core/storage.ts | 51 +++++++ src/core/store.ts | 23 +++ src/core/types.ts | 2 + src/index.ts | 4 + src/plugin/index.ts | 19 ++- tests/storage.test.ts | 274 ++++++++++++++++++++++++++++++++++++ 18 files changed, 1269 insertions(+), 137 deletions(-) create mode 100644 scripts/migrate-storage.ts create mode 100644 src/core/migrate-storage.ts create mode 100644 src/core/sql-driver.ts create mode 100644 src/core/sql-store.ts create mode 100644 src/core/storage.ts create mode 100644 tests/storage.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 41a5844..6828e82 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -9,6 +9,20 @@ on: jobs: check: runs-on: ubuntu-latest + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: clawflow + POSTGRES_PASSWORD: clawflow-test + POSTGRES_DB: clawflow_test + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U clawflow -d clawflow_test" + --health-interval 5s --health-timeout 5s --health-retries 10 + env: + CLAWFLOW_TEST_POSTGRES_URL: postgresql://clawflow:clawflow-test@localhost:5432/clawflow_test steps: - uses: actions/checkout@v4 diff --git a/README.md b/README.md index d65ccb5..0ff6a33 100644 --- a/README.md +++ b/README.md @@ -562,11 +562,11 @@ Strongly recommended on `do: http` and `do: ai` nodes. Do not use on `do: wait` ## Flow State and Durability -Every flow run gets a unique `instanceId`. The runner persists state after every completed node to `~/.openclaw/flow-state/.json`. +Every flow run gets a unique `instanceId`. The runner persists state after every completed node. JSON files in `stateDir` remain the default; native SQLite and PostgreSQL are explicit alternatives. All three store checkpoints, memoized outputs, traces, the running step and pending approvals. **What this means in practice:** -- Gateway restarts mid-flow? Already-completed nodes are not re-run. The flow resumes from the last checkpoint. +- A paused flow can be explicitly resumed after a gateway restart, using its stored checkpoint. - A node that took 30 seconds to run won't run again on resume — its memoized output is loaded from disk. - An approval-gated flow can stay paused for days. The state survives indefinitely. @@ -574,9 +574,131 @@ Every flow run gets a unique `instanceId`. The runner persists state after every **Cancelling a run.** `flow_cancel { instanceId }`, or `POST /flows/runs/:instanceId/cancel` on the flow server (optional body `{ "reason": "..." }`), stops a run that has not finished. A run executing in the process stops on the step it is on: an agent step's `openclaw agent` process gets SIGTERM, on which the CLI aborts its turn in the gateway; an exec step's command is stopped with its whole process group; http and ai steps are aborted; a custom step gets `ctx.abortSignal` and is abandoned if it ignores it. The run ends with status `cancelled`, the interrupted step's trace entry too, and nothing after it runs. A run paused at an approval gate, or left `running` by a restart, is marked cancelled and its approval dropped; a runner in another process that still holds a run stops before its next step. The route answers 404 (`run_not_found`) for an unknown run and 409 (`run_finished`) for a finished one. Each attempt of a step is stopped the same way when it times out, before any retry starts; an agent step with no `timeout` gets 120 s per attempt. -**Reading runs.** `flow_status` lists runs newest first, 20 per page (`limit`, then `cursor` from the previous page). One run comes back whole when it is small, and otherwise as a summary: status, error, each state key with its size, and the trace without outputs. Read any single value with `path`, e.g. `state.leads` or `trace.3.output`; arrays and long strings come back a page at a time (`offset`, `limit`). `flow_run` and `flow_resume` return the same summary when a run is too large to read in one tool result. The same functions (`listRuns`, `summarizeRun`, `readRunValue`, `sweepRuns`) are exported for other hosts. A small summary of each run is kept in `flow-state/_index/`, so a list reads a few hundred bytes per run instead of parsing every record; an entry older than its record is ignored, and the daily sweep fills in missing ones. +**Reading runs.** `flow_status` lists runs newest first, 20 per page (`limit`, then `cursor` from the previous page). One run comes back whole when it is small, and otherwise as a summary: status, error, each state key with its size, and the trace without outputs. Read any single value with `path`, e.g. `state.leads` or `trace.3.output`; arrays and long strings come back a page at a time (`offset`, `limit`). `flow_run` and `flow_resume` return the same summary when a run is too large to read in one tool result. The same functions (`listRuns`, `summarizeRun`, `readRunValue`, `sweepRuns`) are exported for other hosts. With JSON storage, a small summary of each run is kept in `flow-state/_index/`, so a list reads a few hundred bytes per run instead of parsing every record; an entry older than its record is ignored, and the daily sweep fills in missing ones. -This is the lightweight equivalent of Cloudflare Durable Objects' memoization. Cloudflare does it at the infrastructure level with global durability. We do it at the file system level for local/self-hosted use. +### SQLite and PostgreSQL + +Set `storage` in `plugins.entries.clawflow.config` (OpenClaw), or in the config +passed to `new FlowRunner(config)` (standalone). No configuration means JSON; +`{ "backend": "json" }` selects it explicitly. + +```json +{ + "stateDir": "/path/to/flow-state", + "storage": { "backend": "sqlite" } +} +``` + +SQLite uses Node's built-in `node:sqlite`, requiring **[Node 22.13+](https://nodejs.org/en/blog/release/v22.13.0)**. It creates +`stateDir/clawflow.sqlite`, or the file supplied as `storage.path`. Keep it on a +local filesystem. WAL, full synchronization and a 5-second busy timeout are +configured on each connection. New database files are mode `0600`. Back up the +SQLite database with a SQLite-aware backup tool, or stop every connection before +copying it; copying just the main file while WAL writers are active can lose +checkpoints. + +```json +{ + "storage": { + "backend": "postgres", + "connectionStringEnv": "CLAWFLOW_DATABASE_URL", + "namespace": "my-agent" + } +} +``` + +Set that environment variable through your deployment's secret manager to a +PostgreSQL connection URL. The URL is never written into a checkpoint. TLS options +are those of [node-postgres](https://node-postgres.com/features/ssl); require +certificate verification for remote connections. PostgreSQL is loaded only when +selected and uses at most four pooled connections per store. Empty configuration, +connection failures and unsupported schema versions fail the operation; they +never silently fall back to local files. + +Use a stable, distinct namespace for each independent installation sharing a +database. **A namespace separates keys, not permissions.** Use dedicated database +roles and databases or schema grants to isolate untrusted installations. The +configured role needs permission to create and use the `clawflow_*` tables and +indexes in its search path. Initial schema creation is transactional and versioned. + +SQL backends commit each step's checkpoint, memo and summary together; a pause +also commits its pending approval. Resume claims the paused run and consumes its +approval in one transaction, so competing resumes cannot both claim it. Run lists +read indexed summaries and use the same cursor/filter contract; retention deletes +finished roots, their child records and approvals. A new run cannot overwrite an +existing SQL run ID. `:` is reserved for engine-generated child IDs. + +Flow definitions, published versions, trigger schedules and `do: memory` files +remain in the workspace. This is execution storage, not a central history mirror +or a distributed scheduler. Event waiters and immediate cancellation remain +process-local (use `runner.sendEvent()` to select the store); another process's cancellation is observed at the next step. +Changing storage does not add automatic recovery of an interrupted running step +or exactly-once external side effects. Full checkpoint payloads are still rewritten +at each step; the SQL indexes bound list payloads, not individual checkpoint size. + +### Reading and closing a configured store + +The existing `StateStore`, `listRuns(stateDir)` and `sweepRuns(stateDir, days)` +exports keep their JSON-file behavior. Database-aware consumers must use the +configured store. Always await the methods on it, including those on `runner.getStore()` and +`runner.listApprovals()`; JSON implementations can return synchronously. + +```ts +import { createStateStore, readRunValue } from "@clawnify/clawflow"; + +const store = createStateStore(config.stateDir, config.storage); +try { + const page = await store.listRuns({ limit: 20, status: "paused" }); + if (page.runs[0]) { + const run = await store.get(page.runs[0].instanceId); + if (run) console.log(readRunValue(run, "state.inputs")); + } +} finally { + await store.close(); +} +``` + +For a standalone runner, call `await runner.close()` once its runs have settled. +This closes the SQLite handle or PostgreSQL pool. Close a reader when it is no +longer needed; avoid opening a new pool for each request in a long-lived host. + +### Importing existing JSON state + +There is no automatic conversion on startup. Before switching storage: + +1. Stop **every** ClawFlow writer, including the gateway, flow HTTP service and + scheduler. Keep the original state directory as the backup. +2. From a source checkout with dependencies installed, run one of: + + ```sh + npm run migrate:storage -- --writers-stopped --source /path/to/flow-state --sqlite /path/to/flow-state/clawflow.sqlite + npm run migrate:storage -- --writers-stopped --source /path/to/flow-state --postgres-env CLAWFLOW_DATABASE_URL --namespace my-agent + ``` + +3. Verify the imported runs and approvals using `createStateStore`. Update every + reader as well as the runner to use the same storage configuration, then restart. + +The importer copies all run records, including child runs, and pending approvals; +it ignores the derived `_index` directory. It never modifies or removes the JSON +source. Each run and its approvals commit atomically, bounded by one record at a +time. Identical records are accepted on retry; different destination records, +malformed JSON and orphaned approvals stop the import. Earlier committed runs +remain after an error, so repair the source while stopped and retry before +starting the destination. Do not import again after the destination has begun +executing: it will correctly refuse newer, conflicting records. Switching back +to old JSON files after SQL writes would lose those newer checkpoints; this +command does not export SQL back to JSON. + +The same importer is exported as `importJsonState(sourceDir, storageConfig)` for +hosts managing a stopped runtime. It copies execution state only; retain the +workspace's definitions, versions, schedules and memory files too. + +### Storage tests + +Run `npm test` on Node 22.13+ for JSON and native SQLite. Set +`CLAWFLOW_TEST_POSTGRES_URL` to an **isolated test database** to include the real +PostgreSQL suite. Tests create tables and records there. CI provisions PostgreSQL +16 and runs all three backends; production credentials are never needed. --- diff --git a/openclaw.plugin.json b/openclaw.plugin.json index 3528fd6..3a47b74 100644 --- a/openclaw.plugin.json +++ b/openclaw.plugin.json @@ -138,6 +138,60 @@ "description": "Retired in 1.6.1 and ignored: flow authoring (flow_create/edit/publish/delete) and schedule changes (flow_trigger) are gated by @clawnify/agent-permissions, the box's permission authority. Kept so existing configs still validate." } } + }, + "storage": { + "oneOf": [ + { + "type": "object", + "additionalProperties": false, + "properties": { + "backend": { + "const": "json" + } + }, + "required": [ + "backend" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "backend": { + "const": "sqlite" + }, + "path": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "backend" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "backend": { + "const": "postgres" + }, + "connectionStringEnv": { + "type": "string", + "minLength": 1 + }, + "namespace": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "backend", + "connectionStringEnv", + "namespace" + ] + } + ] } } }, @@ -177,6 +231,10 @@ "approval": { "label": "Approval Gate", "help": "Prompt the user before flow_run executes. Set enabled=false to disable entirely. Add session-key substrings to skipSessionPatterns to bypass for unattended hooks (e.g. \"email\" for inbound-email automation)." + }, + "storage": { + "label": "Execution Storage", + "help": "JSON (default), SQLite (Node 22.13+), or PostgreSQL. Stop the runtime and import existing JSON state before switching." } } } diff --git a/package-lock.json b/package-lock.json index 5a52d55..2bb6397 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,10 +9,12 @@ "version": "1.9.1", "license": "MIT", "dependencies": { - "croner": "^10.0.1" + "croner": "^10.0.1", + "pg": "^8.23.1" }, "devDependencies": { "@types/node": "^25.5.0", + "@types/pg": "^8.23.1", "tsx": "^4.21.0", "typescript": "^5.5.0" }, @@ -480,6 +482,17 @@ "undici-types": "~7.18.0" } }, + "node_modules/@types/pg": { + "version": "8.23.1", + "resolved": "https://registry.npmjs.org/@types/pg/-/pg-8.23.1.tgz", + "integrity": "sha512-fKVHpikPdg4GKks3JuLEhvwSyvwzF23hnabPy6DD8ljVbC7+6J5dQzdv4arV6jqq57djnMgs1HKBxX4P8aBI3A==", + "dev": true, + "dependencies": { + "@types/node": "*", + "pg-protocol": "*", + "pg-types": "^2.2.0" + } + }, "node_modules/croner": { "version": "10.0.1", "resolved": "https://registry.npmjs.org/croner/-/croner-10.0.1.tgz", @@ -576,6 +589,122 @@ "optional": true, "peer": true }, + "node_modules/pg": { + "version": "8.23.1", + "resolved": "https://registry.npmjs.org/pg/-/pg-8.23.1.tgz", + "integrity": "sha512-aL96AHANtWjPLDOLqnhx+ngp9+UK7ETEU8VJrDCGvsSSi/mGLcWYsS6Herg7lmaBJe4uwrfqsa7gTEFaSizDoQ==", + "dependencies": { + "pg-connection-string": "^2.14.1", + "pg-pool": "^3.14.0", + "pg-protocol": "^1.16.1", + "pg-types": "2.2.0", + "pgpass": "1.0.5" + }, + "engines": { + "node": ">= 16.0.0" + }, + "optionalDependencies": { + "pg-cloudflare": "^1.4.1" + }, + "peerDependencies": { + "pg-native": ">=3.0.1" + }, + "peerDependenciesMeta": { + "pg-native": { + "optional": true + } + } + }, + "node_modules/pg-cloudflare": { + "version": "1.4.1", + "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.1.tgz", + "integrity": "sha512-6PQbsFWZcp9EmJEwy5cGQ2La+AMWpP46lgbb8X+U/XsHIUweYDNCpeuKck5RxL2MdVFi7krbbEi5nX4Zh7JhrQ==", + "optional": true + }, + "node_modules/pg-connection-string": { + "version": "2.14.1", + "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.1.tgz", + "integrity": "sha512-qR3kGNPBLpCNtz0evbKA0Y/MRFXwSSdT+pTJvYp/bXTcReZbvX1kzF0IyTc1QnxqF7AZbOeBhNL8R5mYQZV/MA==" + }, + "node_modules/pg-int8": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz", + "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==", + "engines": { + "node": ">=4.0.0" + } + }, + "node_modules/pg-pool": { + "version": "3.14.0", + "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz", + "integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==", + "peerDependencies": { + "pg": ">=8.0" + } + }, + "node_modules/pg-protocol": { + "version": "1.16.1", + "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.16.1.tgz", + "integrity": "sha512-p9VOFMiHB/ZbJATetbg+99PxssTVSQRnyuPSQ67mN1+1KBOjZaZ83ZQzltnxPhJwSsC3nwVjJ10DVJlerbFzLg==" + }, + "node_modules/pg-types": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz", + "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==", + "dependencies": { + "pg-int8": "1.0.1", + "postgres-array": "~2.0.0", + "postgres-bytea": "~1.0.0", + "postgres-date": "~1.0.4", + "postgres-interval": "^1.1.0" + }, + "engines": { + "node": ">=4" + } + }, + "node_modules/pgpass": { + "version": "1.0.5", + "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz", + "integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==", + "dependencies": { + "split2": "^4.1.0" + } + }, + "node_modules/postgres-array": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz", + "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==", + "engines": { + "node": ">=4" + } + }, + "node_modules/postgres-bytea": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.1.tgz", + "integrity": "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-date": { + "version": "1.0.7", + "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz", + "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-interval": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz", + "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==", + "dependencies": { + "xtend": "^4.0.0" + }, + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/resolve-pkg-maps": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/resolve-pkg-maps/-/resolve-pkg-maps-1.0.0.tgz", @@ -586,6 +715,14 @@ "url": "https://github.com/privatenumber/resolve-pkg-maps?sponsor=1" } }, + "node_modules/split2": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz", + "integrity": "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==", + "engines": { + "node": ">= 10.x" + } + }, "node_modules/tsx": { "version": "4.21.0", "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.21.0.tgz", @@ -626,6 +763,14 @@ "integrity": "sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w==", "dev": true, "license": "MIT" + }, + "node_modules/xtend": { + "version": "4.0.2", + "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz", + "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==", + "engines": { + "node": ">=0.4" + } } } } diff --git a/package.json b/package.json index 0206cee..4cba0b0 100644 --- a/package.json +++ b/package.json @@ -21,7 +21,8 @@ "lint": "tsc --noEmit", "clean": "rm -rf dist", "migrate": "tsx scripts/migrate-to-1.0.ts", - "prepublishOnly": "npm run build" + "prepublishOnly": "npm run build", + "migrate:storage": "tsx scripts/migrate-storage.ts" }, "openclaw": { "extensions": [ @@ -46,7 +47,8 @@ "skills/**/*", "openclaw.plugin.json", "README.md", - "LICENSE" + "LICENSE", + "scripts/migrate-storage.ts" ], "keywords": [ "openclaw", @@ -72,6 +74,7 @@ }, "devDependencies": { "@types/node": "^25.5.0", + "@types/pg": "^8.23.1", "tsx": "^4.21.0", "typescript": "^5.5.0" }, @@ -87,6 +90,7 @@ "url": "https://github.com/clawnify/clawflow/issues" }, "dependencies": { - "croner": "^10.0.1" + "croner": "^10.0.1", + "pg": "^8.23.1" } } diff --git a/scripts/migrate-storage.ts b/scripts/migrate-storage.ts new file mode 100644 index 0000000..016a4a0 --- /dev/null +++ b/scripts/migrate-storage.ts @@ -0,0 +1,17 @@ +import { parseArgs } from "node:util"; +import { importJsonState } from "../src/core/migrate-storage.js"; + +const { values } = parseArgs({ options: { + source: { type: "string" }, sqlite: { type: "string" }, + "postgres-env": { type: "string" }, namespace: { type: "string" }, + "writers-stopped": { type: "boolean" }, +} }); +if (!values.source || !values["writers-stopped"] || (!!values.sqlite === !!values["postgres-env"]) || (values["postgres-env"] && !values.namespace)) { + console.error("Stop all ClawFlow writers first. Usage: npm run migrate:storage -- --writers-stopped --source (--sqlite | --postgres-env --namespace )"); + process.exitCode = 1; +} else { + const config = values.sqlite + ? { backend: "sqlite" as const, path: values.sqlite } + : { backend: "postgres" as const, connectionStringEnv: values["postgres-env"]!, namespace: values.namespace! }; + console.log(JSON.stringify(await importJsonState(values.source, config))); +} diff --git a/src/core/active-runs.ts b/src/core/active-runs.ts index 2551dd3..e0c59e1 100644 --- a/src/core/active-runs.ts +++ b/src/core/active-runs.ts @@ -21,7 +21,7 @@ const LIVE_RUNS_KEY = Symbol.for("@clawnify/clawflow/live-runs"); type LiveRunsHolder = { [LIVE_RUNS_KEY]?: Map }; const holder = globalThis as unknown as LiveRunsHolder; -/** instanceId of a top-level run -> its live state, for runs in this process. */ +/** Storage identity + instanceId -> live state, for runs in this process. */ export const liveRuns: Map = holder[LIVE_RUNS_KEY] ?? (holder[LIVE_RUNS_KEY] = new Map()); /** The error a cancelled step and run end with. */ diff --git a/src/core/migrate-storage.ts b/src/core/migrate-storage.ts new file mode 100644 index 0000000..95c24e4 --- /dev/null +++ b/src/core/migrate-storage.ts @@ -0,0 +1,47 @@ +import { opendir, readFile } from "node:fs/promises"; +import * as path from "node:path"; +import { createStateStore, type StorageConfig } from "./storage.js"; +import { SqlStateStore } from "./sql-store.js"; +import type { InstanceRecord } from "./store.js"; +import type { PendingApproval } from "./types.js"; + +/** + * Offline, restartable copy. Stop every writer first. Source files are never + * changed. Each run and its approvals commit together; an error stops the + * import, and re-running accepts identical records but refuses changed ones. + */ +export async function importJsonState(sourceDir: string, storage: Exclude): Promise<{ runs: number; approvals: number }> { + const store = createStateStore(sourceDir, storage) as SqlStateStore; + const result = { runs: 0, approvals: 0 }; + try { + let approvals: PendingApproval[] = []; + try { + approvals = JSON.parse(await readFile(path.join(sourceDir, "_pending-approvals.json"), "utf8")); + } catch (err) { if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err; } + if (!Array.isArray(approvals) || approvals.some((a) => !a || typeof a.token !== "string" || typeof a.instanceId !== "string" || typeof a.expiresAt !== "string")) { + throw new Error("Invalid pending approvals file; source was not modified"); + } + const pending = new Map(); + for (const a of approvals) pending.set(a.instanceId, [...(pending.get(a.instanceId) ?? []), a]); + // opendir and per-record transactions keep the copy bounded by one run, + // rather than loading the whole checkpoint directory into memory. + for await (const file of await opendir(sourceDir)) { + if (!file.isFile() || !file.name.endsWith(".json") || file.name === "_pending-approvals.json") continue; + const record = JSON.parse(await readFile(path.join(sourceDir, file.name), "utf8")) as InstanceRecord; + if (!record || typeof record.instanceId !== "string" || !record.instanceId || typeof record.flowName !== "string" + || !["running", "completed", "paused", "waiting", "failed", "cancelled"].includes(record.status) + || !record.state || typeof record.state !== "object" || Array.isArray(record.state) + || !record.completedNodes || typeof record.completedNodes !== "object" || Array.isArray(record.completedNodes) + || !Array.isArray(record.trace) || !Number.isFinite(Date.parse(record.createdAt)) || !Number.isFinite(Date.parse(record.updatedAt))) { + throw new Error(`Invalid instance record: ${file.name}; source was not modified`); + } + const forRun = pending.get(record.instanceId) ?? []; + await store.importRecords([record], forRun); + pending.delete(record.instanceId); + result.runs++; + result.approvals += forRun.length; + } + if (pending.size) throw new Error("Pending approvals reference missing runs; source was not modified"); + return result; + } finally { await store.close(); } +} diff --git a/src/core/runner.ts b/src/core/runner.ts index 064550d..456a439 100644 --- a/src/core/runner.ts +++ b/src/core/runner.ts @@ -30,7 +30,8 @@ import { NODE_FIELD_MODES, parseDuration, } from "./types.js"; -import { StateStore } from "./store.js"; +import { createStateStore, type ExecutionStore } from "./storage.js"; +import type { PendingApproval } from "./types.js"; import { validateFlow } from "./validate.js"; import { defaultRegistry, @@ -80,8 +81,14 @@ export function sendEvent( instanceId: string, eventType: string, payload: unknown, + storageIdentity?: string, ): boolean { - const waiters = eventBus.get(instanceId); + // Legacy unscoped callers work when the instance is unambiguous. A host + // serving multiple stores must use runner.sendEvent() to select its scope. + const keys = storageIdentity ? [`${storageIdentity}\0${instanceId}`] + : [...eventBus.keys()].filter((key) => key.endsWith(`\0${instanceId}`) && eventBus.get(key)?.has(eventType)); + if (keys.length !== 1) return false; + const waiters = eventBus.get(keys[0]); if (!waiters) return false; const waiter = waiters.get(eventType); if (!waiter) return false; @@ -118,15 +125,29 @@ function applyFilter(val: unknown, filter: string): unknown { export class FlowRunner { private cfg: PluginConfig; - private store: StateStore; + private store: ExecutionStore; private registry: StepRegistry; constructor(cfg: PluginConfig) { this.cfg = cfg; - this.store = new StateStore(cfg.stateDir); + this.store = createStateStore(cfg.stateDir, cfg.storage); this.registry = cfg.customSteps ?? defaultRegistry; } + sendEvent(instanceId: string, eventType: string, payload: unknown): boolean { + return sendEvent(instanceId, eventType, payload, this.store.identity); + } + + private runKey(id: string): string { return `${this.store.identity}\0${id}`; } + + /** Release database resources after runs have settled. */ + async close(): Promise { + if ([...liveRuns.keys()].some((key) => key.startsWith(`${this.store.identity}\0`))) { + throw new Error("Cannot close storage while flows are running"); + } + await this.store.close(); + } + // ---- Start a new run ---------------------------------------------------------- /** The agent session that started a run and is waiting on it (flow_run). @@ -188,8 +209,8 @@ export class FlowRunner { } if (missing.length > 0) { const error = `Missing required input${missing.length > 1 ? "s" : ""}: ${missing.join(", ")}`; - this.store.create(id, flow.flow, state); - this.store.update(id, { status: "failed", error }); + await this.store.create(id, flow.flow, state); + await this.store.update(id, { status: "failed", error }); return { ok: false, status: "failed", flowName: flow.flow ?? "unknown", instanceId: id, state, trace: [], error, @@ -198,7 +219,7 @@ export class FlowRunner { } // Create state record first — ensures a state file exists even if env resolution fails - this.store.create(id, flow.flow, state); + await this.store.create(id, flow.flow, state); // Seed state.env: flow defaults → shell-expand $(…) → process.env overrides if (flow.env) { @@ -222,7 +243,7 @@ export class FlowRunner { const resolved = stdout.trim(); if (!resolved) { const error = `env var "${k}": command "${v}" returned empty`; - this.store.update(id, { status: "failed", error }); + await this.store.update(id, { status: "failed", error }); return { ok: false, status: "failed", flowName: flow.flow ?? "unknown", instanceId: id, state, trace: [], error, @@ -231,7 +252,7 @@ export class FlowRunner { env[k] = resolved; } catch (err) { const error = `env var "${k}": failed to resolve "${v}": ${err instanceof Error ? err.message : String(err)}`; - this.store.update(id, { status: "failed", error }); + await this.store.update(id, { status: "failed", error }); return { ok: false, status: "failed", flowName: flow.flow ?? "unknown", instanceId: id, state, trace: [], error, @@ -248,7 +269,7 @@ export class FlowRunner { .map(([k]) => k); if (missing.length > 0) { const error = `Missing required env vars: ${missing.join(", ")}`; - this.store.update(id, { status: "failed", error }); + await this.store.update(id, { status: "failed", error }); return { ok: false, status: "failed", flowName: flow.flow ?? "unknown", instanceId: id, state, trace: [], error, @@ -268,77 +289,34 @@ export class FlowRunner { flow: FlowDefinition, approvedOrPayload: boolean | unknown = true, ): Promise { - // Token can be an instanceId (legacy wait nodes) or an approve token - let record = this.store.get(token); - let instanceId = token; - - // Check if it's an approve token - if (!record) { - const approval = this.store.resolveApproval(token); - if (approval) { - record = this.store.get(approval.instanceId); - instanceId = approval.instanceId; + let pausedAtIndex = 0; + // Pure transform under the store's transaction/lock: competing resumes + // cannot both claim the pause, and a failed write keeps the approval. + const record = await this.store.resume(token, (current) => { + if (current.status !== "paused" && current.status !== "waiting") { + throw new Error(`Instance "${current.instanceId}" is not paused (status: ${current.status})`); } - } - - if (!record) throw new Error(`Resume token not found: ${token}`); - if (record.status !== "paused" && record.status !== "waiting") { - throw new Error( - `Instance "${instanceId}" is not paused (status: ${record.status})`, - ); - } - - // Retrieve stored trace from before pause - const storedTrace = record.trace ?? []; - const pausedAtIndex = record.pausedAtIndex ?? 0; - - if (typeof approvedOrPayload === "boolean" && !approvedOrPayload) { - // Update the paused trace entry to "skipped" - const pausedEntry = storedTrace.find((t) => t.status === "paused"); - if (pausedEntry) pausedEntry.status = "skipped"; - this.store.update(instanceId, { status: "cancelled", trace: storedTrace }); - return { - ok: true, - status: "cancelled", - flowName: record.flowName, - instanceId, - state: record.state, - trace: storedTrace, - error: "Flow cancelled at approval gate", - }; - } - - // Memoize the paused node with the approval payload and set in state - const pausedNode = flow.nodes[pausedAtIndex]; - if (pausedNode) { - const isApproval = pausedNode.do === "wait" - && (pausedNode as WaitNode).for === "approval"; - const approvalOutput = isApproval - ? { approved: true, approvedAt: new Date().toISOString(), token } - : approvedOrPayload; - this.store.memoize(instanceId, pausedNode.name, approvalOutput); - if (pausedNode.output) record.state[pausedNode.output] = approvalOutput; - - // Update the paused trace entry to "ok" - const pausedEntry = storedTrace.find( - (t) => t.node === pausedNode.name && t.status === "paused", - ); - if (pausedEntry) { - pausedEntry.status = "ok"; - pausedEntry.output = approvalOutput; + pausedAtIndex = current.pausedAtIndex ?? 0; + const trace = current.trace ?? []; + const declined = typeof approvedOrPayload === "boolean" && !approvedOrPayload; + const pausedNode = flow.nodes[pausedAtIndex]; + if (declined) { + const entry = trace.find((t) => t.status === "paused"); + if (entry) entry.status = "skipped"; + } else if (pausedNode) { + const isApproval = pausedNode.do === "wait" && (pausedNode as WaitNode).for === "approval"; + const output = isApproval ? { approved: true, approvedAt: new Date().toISOString(), token } : approvedOrPayload; + current.completedNodes[pausedNode.name] = output; + if (pausedNode.output) current.state[pausedNode.output] = output; + const entry = trace.find((t) => t.node === pausedNode.name && t.status === "paused"); + if (entry) { entry.status = "ok"; entry.output = output; } } - } - - this.store.update(instanceId, { - status: "running", - trace: storedTrace, - resumeToken: undefined, - waitingFor: undefined, - pausedAtIndex: undefined, + return { ...current, trace, status: declined ? "cancelled" : "running", resumeToken: undefined, waitingFor: undefined, pausedAtIndex: undefined }; }); - - // Continue from the node AFTER the paused one, with preserved trace - return this.executeRun(flow, record.state, instanceId, pausedAtIndex + 1, storedTrace); + if (record.status === "cancelled") { + return { ok: true, status: "cancelled", flowName: record.flowName, instanceId: record.instanceId, state: record.state, trace: record.trace, error: "Flow cancelled at approval gate" }; + } + return this.executeRun(flow, record.state, record.instanceId, pausedAtIndex + 1, record.trace); } // ---- Cancel ------------------------------------------------------------------- @@ -357,8 +335,8 @@ export class FlowRunner { * and keeps that status. */ async cancel(instanceId: string, reason = "Cancelled"): Promise { - const live = liveRuns.get(instanceId); - const record = this.store.get(instanceId); + const live = liveRuns.get(this.runKey(instanceId)); + const record = await this.store.get(instanceId); if (!live && !record) return { ok: false, instanceId, error: "not_found" }; if (!live && record && FINISHED_STATUSES.has(record.status)) { return { ok: false, instanceId, error: "finished", status: record.status }; @@ -376,7 +354,7 @@ export class FlowRunner { }), ]); clearTimeout(bound); - const status = this.store.get(instanceId)?.status ?? "cancelled"; + const status = (await this.store.get(instanceId))?.status ?? "cancelled"; return { ok: true, instanceId, status: status === "cancelled" ? "cancelled" : "cancelling" }; } @@ -384,8 +362,8 @@ export class FlowRunner { const trace = current.trace ?? []; const paused = trace.find((t) => t.status === "paused"); if (paused) paused.status = "cancelled"; - if (current.resumeToken) this.store.resolveApproval(current.resumeToken); - this.store.update(instanceId, { + if (current.resumeToken) await this.store.resolveApproval(current.resumeToken); + await this.store.update(instanceId, { status: "cancelled", error: reason, trace, @@ -407,34 +385,34 @@ export class FlowRunner { priorTrace: TraceEntry[], ): Promise { const live: LiveRun = { controller: new AbortController(), flowName: flow.flow }; - liveRuns.set(instanceId, live); + liveRuns.set(this.runKey(instanceId), live); try { const done = this.execute(flow, state, instanceId, startIndex, priorTrace, live.controller.signal); live.done = done; return await done; } finally { - if (liveRuns.get(instanceId) === live) liveRuns.delete(instanceId); + if (liveRuns.get(this.runKey(instanceId)) === live) liveRuns.delete(this.runKey(instanceId)); } } /** End a run (or sub-flow) as cancelled, recording why. */ - private endCancelled( + private async endCancelled( flow: FlowDefinition, state: FlowState, instanceId: string, trace: TraceEntry[], reason?: string, - ): FlowResult { + ): Promise { const error = reason || "Cancelled"; - this.store.update(instanceId, { status: "cancelled", state, trace, error, running: undefined }); + await this.store.update(instanceId, { status: "cancelled", state, trace, error, running: undefined }); return { ok: false, status: "cancelled", flowName: flow.flow, instanceId, state, trace, error }; } /** A top-level run whose record was marked cancelled from outside this * process while it ran. Sub-flow records are never marked that way. */ - private cancelledOnRecord(instanceId: string): string | undefined { + private async cancelledOnRecord(instanceId: string): Promise { if (instanceId.includes(":")) return undefined; - const record = this.store.get(instanceId); + const record = await this.store.get(instanceId); return record?.status === "cancelled" ? (record.error || "Cancelled") : undefined; } @@ -457,7 +435,7 @@ export class FlowRunner { const t0 = Date.now(); // One read per step: whether the run was cancelled, and the memo below. - const record = this.store.get(instanceId); + const record = await this.store.get(instanceId); if (signal.aborted) return this.endCancelled(flow, state, instanceId, trace, abortReason(signal).message); if (record?.status === "cancelled" && !instanceId.includes(":")) { return this.endCancelled(flow, state, instanceId, trace, record.error); @@ -485,7 +463,8 @@ export class FlowRunner { try { // Say what this run is doing while the step runs; each exit below clears it. - this.store.update(instanceId, { running: this.runningStep(node, state) }); + const starting = await this.store.update(instanceId, { running: this.runningStep(node, state) }); + if (starting.status === "cancelled") return this.endCancelled(flow, state, instanceId, trace, starting.error); const result = await this.runWithRetry(node, state, flow, instanceId, signal); // ---- Paused for approval (do: wait, for: approval) ---------------------- @@ -497,7 +476,7 @@ export class FlowRunner { durationMs: Date.now() - t0, }; trace.push(entry); - this.store.update(instanceId, { + const paused = await this.store.checkpoint(instanceId, { status: "paused", state, trace, @@ -510,7 +489,8 @@ export class FlowRunner { preview: result.approve.preview, timeout: result.approve.timeout, }, - }); + }, undefined, result.approve.approval); + if (paused.status === "cancelled") return this.endCancelled(flow, state, instanceId, trace, paused.error); return { ok: true, status: "paused", @@ -531,7 +511,6 @@ export class FlowRunner { // ---- Normal completion -------------------------------------------------- if (node.output) state[node.output] = result.output; - this.store.memoize(instanceId, node.name, result.output); const entry: TraceEntry = { node: node.name, do: node.do, @@ -541,7 +520,7 @@ export class FlowRunner { durationMs: Date.now() - t0, }; trace.push(entry); - this.store.update(instanceId, { state, trace, running: undefined }); + await this.store.checkpoint(instanceId, { state, trace, running: undefined }, { node: node.name, output: result.output }); i++; } catch (err) { if (signal.aborted) { @@ -558,10 +537,10 @@ export class FlowRunner { durationMs: Date.now() - t0, }; trace.push(entry); - const cancelled = this.cancelledOnRecord(instanceId); + const cancelled = await this.cancelledOnRecord(instanceId); if (cancelled) return this.endCancelled(flow, state, instanceId, trace, cancelled); const error = `Node "${node.name}": ${message}`; - this.store.update(instanceId, { status: "failed", state, trace, error, running: undefined }); + await this.store.update(instanceId, { status: "failed", state, trace, error, running: undefined }); return { ok: false, status: "failed", @@ -574,9 +553,10 @@ export class FlowRunner { } } - const cancelled = this.cancelledOnRecord(instanceId); + const cancelled = await this.cancelledOnRecord(instanceId); if (cancelled) return this.endCancelled(flow, state, instanceId, trace, cancelled); - this.store.update(instanceId, { status: "completed", state, trace }); + const finished = await this.store.update(instanceId, { status: "completed", state, trace, error: undefined, running: undefined }); + if (finished.status === "cancelled") return this.endCancelled(flow, state, instanceId, trace, finished.error); return { ok: true, status: "completed", @@ -602,7 +582,7 @@ export class FlowRunner { signal: AbortSignal, ): Promise<{ output?: unknown; - approve?: { token: string; prompt: string; preview?: unknown; timeout?: string }; + approve?: { token: string; prompt: string; preview?: unknown; timeout?: string; approval: PendingApproval }; attempts?: number; }> { const policy: RetryPolicy = node.retry ?? { limit: 1, delay: 0 }; @@ -685,7 +665,7 @@ export class FlowRunner { signal: AbortSignal, ): Promise<{ output?: unknown; - approve?: { token: string; prompt: string; preview?: unknown; timeout?: string }; + approve?: { token: string; prompt: string; preview?: unknown; timeout?: string; approval: PendingApproval }; }> { // Interpolate every field classified as a template value in NODE_FIELD_MODES, // once, before dispatch — so handlers never re-decide "is this field dynamic?". @@ -1372,7 +1352,7 @@ export class FlowRunner { nodes: branch, }; const subId = `${instanceId}:branch:${node.name}:${pathName}`; - this.store.create(subId, subFlow.flow, state); + await this.store.create(subId, subFlow.flow, state); const subResult = await this.execute(subFlow, state, subId, 0, [], signal); if (!subResult.ok) { throw new Error( @@ -1410,7 +1390,7 @@ export class FlowRunner { nodes: branch, }; const subId = `${instanceId}:condition:${node.name}:${branchName}`; - this.store.create(subId, subFlow.flow, state); + await this.store.create(subId, subFlow.flow, state); const subResult = await this.execute(subFlow, state, subId, 0, [], signal); if (!subResult.ok) { throw new Error( @@ -1479,7 +1459,7 @@ export class FlowRunner { nodes: node.nodes, }; const subId = `${instanceId}:loop:${node.name}:${results.length}`; - this.store.create(subId, subFlow.flow, loopState); + await this.store.create(subId, subFlow.flow, loopState); const subResult = await this.execute( subFlow, loopState, @@ -1513,7 +1493,7 @@ export class FlowRunner { nodes: [subNode], }; const subId = `${instanceId}:parallel:${node.name}:${idx}`; - this.store.create(subId, subFlow.flow, branchState); + await this.store.create(subId, subFlow.flow, branchState); const subResult = await this.execute( subFlow, branchState, @@ -1676,7 +1656,7 @@ export class FlowRunner { ): Promise<{ output?: unknown; pause?: boolean; - approve?: { token: string; prompt: string; preview?: unknown; timeout?: string }; + approve?: { token: string; prompt: string; preview?: unknown; timeout?: string; approval: PendingApproval }; }> { if (node.for === "approval") { // prompt template already interpolated by resolveNodeFields @@ -1697,7 +1677,7 @@ export class FlowRunner { // Generate approval token and register const token = `cf-${crypto.randomUUID().slice(0, 8)}`; - this.store.addApproval({ + const approval: PendingApproval = { token, instanceId, flowName: flow.flow, @@ -1706,10 +1686,10 @@ export class FlowRunner { preview, createdAt: new Date().toISOString(), expiresAt, - }); + }; return { - approve: { token, prompt, preview, timeout }, + approve: { token, prompt, preview, timeout, approval }, }; } @@ -1724,7 +1704,7 @@ export class FlowRunner { // Cancelling the run (or this attempt timing out) drops the waiter. const onAbort = () => { clearTimeout(timeoutHandle); - eventBus.get(instanceId)?.delete(eventType); + eventBus.get(this.runKey(instanceId))?.delete(eventType); reject(abortReason(signal)); }; const timeoutHandle = setTimeout( @@ -1738,9 +1718,9 @@ export class FlowRunner { }, timeoutMs, ); - if (!eventBus.has(instanceId)) - eventBus.set(instanceId, new Map()); - eventBus.get(instanceId)!.set(eventType, { + if (!eventBus.has(this.runKey(instanceId))) + eventBus.set(this.runKey(instanceId), new Map()); + eventBus.get(this.runKey(instanceId))!.set(eventType, { resolve: (value) => { signal.removeEventListener("abort", onAbort); resolve(value); @@ -1752,7 +1732,7 @@ export class FlowRunner { else signal.addEventListener("abort", onAbort, { once: true }); }); - eventBus.get(instanceId)?.delete(eventType); + eventBus.get(this.runKey(instanceId))?.delete(eventType); return { output: payload }; } @@ -2259,7 +2239,7 @@ export class FlowRunner { // ---- Instance management ------------------------------------------------------ - getStore(): StateStore { + getStore(): ExecutionStore { return this.store; } diff --git a/src/core/runs.ts b/src/core/runs.ts index 818dfa4..dcece5c 100644 --- a/src/core/runs.ts +++ b/src/core/runs.ts @@ -66,7 +66,7 @@ export interface ListRunsOptions { * pages never shift a page under it, and runs created in the same millisecond * are neither skipped nor repeated. */ -export function listRuns(stateDir: string, opts: ListRunsOptions = {}): RunPage { +export function parseRunPage(opts: ListRunsOptions) { const limit = opts.limit ?? DEFAULT_PAGE; if (!Number.isInteger(limit) || limit < 1 || limit > MAX_PAGE) { throw new Error(`limit must be a whole number between 1 and ${MAX_PAGE}`); @@ -76,6 +76,11 @@ export function listRuns(stateDir: string, opts: ListRunsOptions = {}): RunPage throw new Error("cursor is not one this list returned; start again without it"); } + return { limit, after }; +} + +export function listRuns(stateDir: string, opts: ListRunsOptions = {}): RunPage { + const { limit, after } = parseRunPage(opts); const rows: RunRow[] = []; for (const row of readTopLevelRows(stateDir)) { if (opts.flow && row.flowName !== opts.flow) continue; @@ -102,7 +107,7 @@ function newestFirst(a: Keyed, b: Keyed): number { return 0; } -function encodeCursor(k: Keyed): string { +export function encodeCursor(k: Keyed): string { return btoa(encodeURIComponent(JSON.stringify([k.createdAt, k.instanceId]))); } diff --git a/src/core/sql-driver.ts b/src/core/sql-driver.ts new file mode 100644 index 0000000..ac27ee8 --- /dev/null +++ b/src/core/sql-driver.ts @@ -0,0 +1,188 @@ +import * as fs from "node:fs"; +import * as path from "node:path"; +import { createRequire } from "node:module"; +import type { DatabaseSync } from "node:sqlite"; +import type { Pool } from "pg"; + +type Row = Record; +export type Query = (sql: string, params?: unknown[]) => Promise; +export interface SqlDriver { + readonly lock: string; + transaction(fn: (query: Query) => Promise, write?: boolean): Promise; + close(): Promise; +} + +/** Resolve symlinked parents too when the database file does not exist yet. */ +export function canonicalStoragePath(file: string): string { + const absolute = path.resolve(file); + try { return fs.realpathSync(absolute); } + catch (err) { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err; + return path.join(canonicalStoragePath(path.dirname(absolute)), path.basename(absolute)); + } +} + +const schema = [ + `CREATE TABLE IF NOT EXISTS clawflow_storage (version INTEGER NOT NULL PRIMARY KEY)`, + `CREATE TABLE IF NOT EXISTS clawflow_runs ( + namespace TEXT NOT NULL, id TEXT NOT NULL, root_id TEXT NOT NULL, + flow_name TEXT NOT NULL, status TEXT NOT NULL, + created_at TEXT NOT NULL, updated_at TEXT NOT NULL, + record TEXT NOT NULL, summary TEXT NOT NULL, + PRIMARY KEY (namespace, id))`, + `CREATE INDEX IF NOT EXISTS clawflow_runs_page ON clawflow_runs (namespace, created_at DESC, id DESC) WHERE id = root_id`, + `CREATE INDEX IF NOT EXISTS clawflow_runs_flow_page ON clawflow_runs (namespace, flow_name, created_at DESC, id DESC) WHERE id = root_id`, + `CREATE INDEX IF NOT EXISTS clawflow_runs_status_page ON clawflow_runs (namespace, status, created_at DESC, id DESC) WHERE id = root_id`, + `CREATE INDEX IF NOT EXISTS clawflow_runs_retention ON clawflow_runs (namespace, status, updated_at) WHERE id = root_id`, + `CREATE INDEX IF NOT EXISTS clawflow_runs_root ON clawflow_runs (namespace, root_id)`, + `CREATE TABLE IF NOT EXISTS clawflow_approvals ( + namespace TEXT NOT NULL, token TEXT NOT NULL, instance_id TEXT NOT NULL, record TEXT NOT NULL, + PRIMARY KEY (namespace, token), + FOREIGN KEY (namespace, instance_id) REFERENCES clawflow_runs (namespace, id) ON DELETE CASCADE)`, + `CREATE INDEX IF NOT EXISTS clawflow_approvals_run ON clawflow_approvals (namespace, instance_id)`, +]; + +async function initialize(query: Query): Promise { + // All DDL and version checks are in one transaction. Future schema versions + // must migrate explicitly, never be opened by an older engine. + await query(schema[0]); + const versions = await query("SELECT version FROM clawflow_storage"); + if (versions.length && (versions.length !== 1 || versions[0].version !== 1)) { + throw new Error("Unsupported ClawFlow storage schema version"); + } + for (const sql of schema.slice(1)) await query(sql); + await query("INSERT INTO clawflow_storage (version) VALUES (1) ON CONFLICT DO NOTHING"); +} + +// DatabaseSync transactions must not interleave while their callbacks await. +// Share the queue across imports AND handles of the same file: otherwise a +// second handle's busy_timeout blocks the event loop holding the first lock. +const queueKey = Symbol.for("@clawnify/clawflow/sqlite-queues"); +const globalQueues = globalThis as unknown as { [queueKey]?: Map> }; +const queues = globalQueues[queueKey] ??= new Map(); + +export class SqliteDriver implements SqlDriver { + readonly lock = ""; + private db?: DatabaseSync; + private initialized = false; + private closed = false; + private readonly file: string; + constructor(file: string) { + if (!file || file === ":memory:") throw new Error("SQLite storage requires a durable file path"); + this.file = canonicalStoragePath(file); + } + + private serial(fn: () => Promise): Promise { + const previous = queues.get(this.file) ?? Promise.resolve(); + const next = previous.catch(() => {}).then(fn); + queues.set(this.file, next); + void next.finally(() => { if (queues.get(this.file) === next) queues.delete(this.file); }).catch(() => {}); + return next; + } + + transaction(fn: (query: Query) => Promise, write = true): Promise { + return this.serial(async () => { + if (this.closed) throw new Error("ClawFlow storage is closed"); + if (!this.db) { + // Lazy: JSON and PostgreSQL consumers can still run on Node 18/20. + const require = createRequire(import.meta.url); + let sqlite: typeof import("node:sqlite"); + try { sqlite = require("node:sqlite"); } + catch { throw new Error("SQLite storage requires Node.js 22.13 or newer"); } + fs.mkdirSync(path.dirname(this.file), { recursive: true }); + // Do not change permissions of an existing operator-owned file. + try { fs.closeSync(fs.openSync(this.file, "wx", 0o600)); } + catch (err) { if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err; } + const db = new sqlite.DatabaseSync(this.file); + try { db.exec("PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA foreign_keys=ON;"); } + catch (err) { db.close(); throw err; } + this.db = db; + } + const query: Query = async (sql, params = []) => { + const values: unknown[] = []; + const translated = sql.replace(/\$(\d+)/g, (_, n) => { values.push(params[Number(n) - 1]); return "?"; }); + return this.db!.prepare(translated).all(...values as import("node:sqlite").SQLInputValue[]) as Row[]; + }; + this.db.exec(write || !this.initialized ? "BEGIN IMMEDIATE" : "BEGIN"); + try { + if (!this.initialized) await initialize(query); + const result = await fn(query); + this.db.exec("COMMIT"); + this.initialized = true; + return result; + } catch (err) { + this.db.exec("ROLLBACK"); + throw err; + } + }); + } + + close(): Promise { + return this.serial(async () => { + if (!this.closed) this.db?.close(); + this.closed = true; + }); + } +} + +export class PostgresDriver implements SqlDriver { + readonly lock = " FOR UPDATE"; + private pool?: Pool; + private ready?: Promise; + private closed = false; + constructor(private readonly connectionString: string) {} + + private async getPool(): Promise { + if (this.closed) throw new Error("ClawFlow storage is closed"); + if (!this.ready) { + this.ready = (async () => { + const { Pool } = await import("pg"); + if (this.closed) throw new Error("ClawFlow storage is closed"); + this.pool = new Pool({ connectionString: this.connectionString, max: 4, connectionTimeoutMillis: 5000, idleTimeoutMillis: 10000, allowExitOnIdle: true }); + // pg removes failed idle clients itself. Do not log credentials or let + // EventEmitter's unhandled 'error' terminate unrelated active runs. + this.pool.on("error", () => console.warn("[clawflow] Idle PostgreSQL connection closed; the next operation will reconnect")); + const client = await this.pool.connect(); + try { + await client.query("BEGIN"); + // Serializes first-open DDL across processes, including pg_catalog. + await client.query("SELECT pg_advisory_xact_lock(18793, 1)"); + await initialize(async (sql, params) => (await client.query(sql, params)).rows); + await client.query("COMMIT"); + } catch (err) { + await client.query("ROLLBACK"); + throw err; + } finally { client.release(); } + })().catch(async (err) => { + await this.pool?.end(); + this.pool = undefined; + this.ready = undefined; + throw err; + }); + } + await this.ready; + return this.pool!; + } + + async transaction(fn: (query: Query) => Promise, write = true): Promise { + const pool = await this.getPool(); + const client = await pool.connect(); + let failed = false; + try { + await client.query(write ? "BEGIN" : "BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY"); + const result = await fn(async (sql, params) => (await client.query(sql, params)).rows); + await client.query("COMMIT"); + return result; + } catch (err) { + try { await client.query("ROLLBACK"); } catch { failed = true; } + throw err; + } finally { client.release(failed); } + } + + async close(): Promise { + this.closed = true; + await this.ready?.catch(() => {}); + await this.pool?.end(); + this.pool = undefined; + } +} diff --git a/src/core/sql-store.ts b/src/core/sql-store.ts new file mode 100644 index 0000000..6733207 --- /dev/null +++ b/src/core/sql-store.ts @@ -0,0 +1,199 @@ +import * as path from "node:path"; +import { createHash } from "node:crypto"; +import { runIndexEntry, type InstanceRecord } from "./store.js"; +import type { FlowState, PendingApproval } from "./types.js"; +import type { ExecutionStore, StorageConfig } from "./storage.js"; +import { encodeCursor, parseRunPage, type ListRunsOptions, type RunPage, type SweepResult } from "./runs.js"; +import { PostgresDriver, SqliteDriver, canonicalStoragePath, type Query, type SqlDriver } from "./sql-driver.js"; + +export class SqlStateStore implements ExecutionStore { + readonly identity: string; + private readonly driver: SqlDriver; + private readonly namespace: string; + constructor(readonly dir: string, config: Exclude) { + if (config.backend === "sqlite") { + const file = canonicalStoragePath(config.path ?? path.join(dir, "clawflow.sqlite")); + this.driver = new SqliteDriver(file); + this.namespace = "default"; + this.identity = `sqlite:${file}`; + } else { + const url = process.env[config.connectionStringEnv]!; + this.driver = new PostgresDriver(url); + this.namespace = config.namespace; + // No connection strings in live-run keys or diagnostic output. + this.identity = `postgres:${createHash("sha256").update(url).digest("hex")}:${this.namespace}`; + } + } + + private async read(query: Query, id: string, lock = false): Promise { + const rows = await query(`SELECT record FROM clawflow_runs WHERE namespace=$1 AND id=$2${lock ? this.driver.lock : ""}`, [this.namespace, id]); + return rows[0] ? JSON.parse(rows[0].record as string) : null; + } + + private async write(query: Query, record: InstanceRecord, insert = false): Promise { + const values = [this.namespace, record.instanceId, record.instanceId.split(":")[0], record.flowName, + record.status, record.createdAt, record.updatedAt, JSON.stringify(record), JSON.stringify(runIndexEntry(record))]; + if (insert) { + await query(`INSERT INTO clawflow_runs (namespace,id,root_id,flow_name,status,created_at,updated_at,record,summary) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)`, values); + } else { + await query(`UPDATE clawflow_runs SET root_id=$3,flow_name=$4,status=$5,created_at=$6,updated_at=$7,record=$8,summary=$9 + WHERE namespace=$1 AND id=$2`, values); + } + } + + create(instanceId: string, flowName: string, state: FlowState): Promise { + const now = new Date().toISOString(); + const record: InstanceRecord = { instanceId, flowName, status: "running", state, completedNodes: {}, trace: [], createdAt: now, updatedAt: now }; + return this.driver.transaction(async (q) => { + // A parent retry revisits deterministic sub-flow IDs. Reuse their memos + // so a successful child is not replayed after a sibling failed. + if (instanceId.includes(":")) { + const existing = await this.read(q, instanceId, true); + if (existing) return existing; + } + await this.write(q, record, true); + return record; + }); + } + + get(id: string): Promise { + return this.driver.transaction((q) => this.read(q, id), false); + } + + update(id: string, patch: Partial): Promise { + return this.checkpoint(id, patch); + } + + /** Record, derived summary, node memo and approval become durable together. */ + checkpoint(id: string, patch: Partial, memo?: { node: string; output: unknown }, approval?: PendingApproval): Promise { + return this.driver.transaction(async (q) => { + const existing = await this.read(q, id, true); + if (!existing) throw new Error(`Instance not found: ${id}`); + const updated = { ...existing, ...patch, instanceId: existing.instanceId, createdAt: existing.createdAt, updatedAt: new Date().toISOString() }; + // A concurrently finishing step must never resurrect a cancelled run. + if (existing.status === "cancelled") { + updated.status = "cancelled"; + updated.error = existing.error; + updated.running = undefined; + } + if (memo) updated.completedNodes = { ...updated.completedNodes, [memo.node]: memo.output }; + await this.write(q, updated); + if (approval && updated.status !== "cancelled") await this.insertApproval(q, approval); + return updated; + }); + } + + async memoize(id: string, node: string, output: unknown): Promise { + await this.checkpoint(id, {}, { node, output }); + } + + async getMemoized(id: string, node: string): Promise<{ found: boolean; output: unknown }> { + const r = await this.get(id); + return { found: !!r && Object.prototype.hasOwnProperty.call(r.completedNodes, node), output: r?.completedNodes[node] }; + } + + list(status?: string): Promise { + return this.driver.transaction(async (q) => { + const rows = await q(`SELECT record FROM clawflow_runs WHERE namespace=$1${status ? " AND status=$2" : ""}`, status ? [this.namespace, status] : [this.namespace]); + return rows.map((r) => JSON.parse(r.record as string)); + }, false); + } + + private async insertApproval(q: Query, approval: PendingApproval): Promise { + await q("INSERT INTO clawflow_approvals (namespace,token,instance_id,record) VALUES ($1,$2,$3,$4)", + [this.namespace, approval.token, approval.instanceId, JSON.stringify(approval)]); + } + + addApproval(approval: PendingApproval): Promise { + return this.driver.transaction((q) => this.insertApproval(q, approval)); + } + + /** Claim a paused run and consume its approval in the same transaction. */ + resume(token: string, transform: (record: InstanceRecord) => InstanceRecord): Promise { + return this.driver.transaction(async (q) => { + let record = await this.read(q, token, true); + if (!record) { + const approvals = await q("SELECT instance_id FROM clawflow_approvals WHERE namespace=$1 AND token=$2", [this.namespace, token]); + if (approvals[0]) record = await this.read(q, approvals[0].instance_id as string, true); + } + if (!record) throw new Error(`Resume token not found: ${token}`); + const updated = { ...transform(record), instanceId: record.instanceId, createdAt: record.createdAt, updatedAt: new Date().toISOString() }; + await this.write(q, updated); + await q("DELETE FROM clawflow_approvals WHERE namespace=$1 AND instance_id=$2", [this.namespace, record.instanceId]); + return updated; + }); + } + + resolveApproval(token: string): Promise { + return this.driver.transaction(async (q) => { + const rows = await q("DELETE FROM clawflow_approvals WHERE namespace=$1 AND token=$2 RETURNING record", [this.namespace, token]); + return rows[0] ? JSON.parse(rows[0].record as string) : null; + }); + } + + listApprovals(): Promise { + return this.driver.transaction(async (q) => (await q("SELECT record FROM clawflow_approvals WHERE namespace=$1 ORDER BY token", [this.namespace])).map((r) => JSON.parse(r.record as string)), false); + } + + listRuns(opts: ListRunsOptions = {}): Promise { + const { limit, after } = parseRunPage(opts); + return this.driver.transaction(async (q) => { + const params: unknown[] = [this.namespace]; + let where = "namespace=$1 AND id=root_id"; + if (opts.flow) { params.push(opts.flow); where += ` AND flow_name=$${params.length}`; } + if (opts.status) { params.push(opts.status); where += ` AND status=$${params.length}`; } + const count = await q(`SELECT COUNT(*) AS total FROM clawflow_runs WHERE ${where}`, params); + if (after) { + params.push(after.createdAt, after.instanceId); + where += ` AND (created_at < $${params.length - 1} OR (created_at = $${params.length - 1} AND id < $${params.length}))`; + } + params.push(limit + 1); + const rows = await q(`SELECT summary FROM clawflow_runs WHERE ${where} ORDER BY created_at DESC,id DESC LIMIT $${params.length}`, params); + const runs = rows.slice(0, limit).map((r) => JSON.parse(r.summary as string)); + return { runs, total: Number(count[0].total), next_cursor: rows.length > limit ? encodeCursor(runs[runs.length - 1]) : null }; + }, false); + } + + async sweep(retentionDays: number, now = Date.now()): Promise { + if (!Number.isFinite(retentionDays) || retentionDays < 0) throw new Error("retentionDays must be non-negative"); + if (!retentionDays) return { indexed: 0, runs: 0, files: 0 }; + const cutoff = new Date(now - retentionDays * 86_400_000).toISOString(); + const result = { indexed: 0, runs: 0, files: 0 }; + // Bound each transaction, including its locks. Always re-check under lock. + while (true) { + const batch = await this.driver.transaction(async (q) => { + const roots = await q(`SELECT id FROM clawflow_runs WHERE namespace=$1 AND id=root_id + AND status IN ('completed','failed','cancelled') AND updated_at < $2 LIMIT 100${this.driver.lock}`, [this.namespace, cutoff]); + let files = 0; + for (const root of roots) { + const rows = await q("DELETE FROM clawflow_runs WHERE namespace=$1 AND root_id=$2 RETURNING id", [this.namespace, root.id]); + files += rows.length; + } + return { runs: roots.length, files }; + }); + result.runs += batch.runs; + result.files += batch.files; + if (batch.runs < 100) return result; + await new Promise((resolve) => setImmediate(resolve)); + } + } + + /** Offline migration only: append missing records; refuse conflicts atomically. */ + importRecords(records: InstanceRecord[], approvals: PendingApproval[]): Promise { + return this.driver.transaction(async (q) => { + for (const record of records) { + const existing = await this.read(q, record.instanceId, true); + if (existing && JSON.stringify(existing) !== JSON.stringify(record)) throw new Error(`Migration conflict: ${record.instanceId}`); + if (!existing) await this.write(q, record, true); + } + for (const approval of approvals) { + const existing = await q("SELECT record FROM clawflow_approvals WHERE namespace=$1 AND token=$2", [this.namespace, approval.token]); + if (existing.length && existing[0].record !== JSON.stringify(approval)) throw new Error(`Migration approval conflict: ${approval.token}`); + if (!existing.length) await this.insertApproval(q, approval); + } + }); + } + + close(): Promise { return this.driver.close(); } +} diff --git a/src/core/storage.ts b/src/core/storage.ts new file mode 100644 index 0000000..c4a996d --- /dev/null +++ b/src/core/storage.ts @@ -0,0 +1,51 @@ +import * as path from "node:path"; +import { StateStore, type InstanceRecord } from "./store.js"; +import { SqlStateStore } from "./sql-store.js"; +import type { FlowState, PendingApproval } from "./types.js"; +import type { ListRunsOptions, RunPage, SweepResult } from "./runs.js"; + +export type StorageConfig = + | { backend: "json" } + | { backend: "sqlite"; path?: string } + | { backend: "postgres"; connectionStringEnv: string; namespace: string }; + +export type Awaitable = T | Promise; + +/** Await these methods: the legacy JSON store remains synchronous. */ +export interface ExecutionStore { + /** Local workspace path, NOT a database read API. */ + readonly dir: string; + /** Stable within a process; scopes cancellation to this database/namespace. */ + readonly identity: string; + create(id: string, flow: string, state: FlowState): Awaitable; + get(id: string): Awaitable; + update(id: string, patch: Partial): Awaitable; + checkpoint(id: string, patch: Partial, memo?: { node: string; output: unknown }, approval?: PendingApproval): Awaitable; + memoize(id: string, node: string, output: unknown): Awaitable; + getMemoized(id: string, node: string): Awaitable<{ found: boolean; output: unknown }>; + list(status?: string): Awaitable; + addApproval(approval: PendingApproval): Awaitable; + resume(token: string, transform: (record: InstanceRecord) => InstanceRecord): Awaitable; + resolveApproval(token: string): Awaitable; + listApprovals(): Awaitable; + listRuns(opts?: ListRunsOptions): Awaitable; + sweep(retentionDays: number, now?: number): Promise; + close(): Awaitable; +} + +/** Explicit opt-in; never falls back to another store when a database fails. */ +export function createStateStore(stateDir?: string, config?: StorageConfig): ExecutionStore { + const dir = stateDir ?? path.join(process.env.OPENCLAW_WORKSPACE ?? process.env.HOME ?? ".", "flow-state"); + if (!config || config.backend === "json") return new StateStore(dir); + if (config.backend === "sqlite") { + return new SqlStateStore(dir, { ...config, path: config.path ?? path.join(dir, "clawflow.sqlite") }); + } + if (config.backend === "postgres") { + if (!config.namespace?.trim()) throw new Error("PostgreSQL storage requires a namespace"); + if (!config.connectionStringEnv || !process.env[config.connectionStringEnv]) { + throw new Error("PostgreSQL storage requires a populated connectionStringEnv"); + } + return new SqlStateStore(dir, config); + } + throw new Error("Unknown ClawFlow storage backend"); +} diff --git a/src/core/store.ts b/src/core/store.ts index ff50804..77bb9a2 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -1,5 +1,6 @@ import * as fs from "fs"; import * as path from "path"; +import { listRuns, sweepRuns, type ListRunsOptions } from "./runs.js"; import type { FlowState, FlowResult, TraceEntry, PendingApproval, RunningStep } from "./types.js"; // ---- Durable State Store -------------------------------------------------------- @@ -132,6 +133,19 @@ export class StateStore { /** The state directory: one `.json` per run (see runs.ts for reads). */ readonly dir: string; + get identity(): string { return `json:${path.resolve(this.dir)}`; } + + listRuns(opts?: ListRunsOptions) { return listRuns(this.dir, opts); } + sweep(days: number, now?: number) { return sweepRuns(this.dir, days, now); } + close(): void {} + + checkpoint(id: string, patch: Partial, memo?: { node: string; output: unknown }, approval?: PendingApproval): InstanceRecord { + if (memo) this.memoize(id, memo.node, memo.output); + const record = this.update(id, patch); + if (approval) this.addApproval(approval); + return record; + } + constructor(stateDir?: string) { this.dir = stateDir ?? @@ -232,6 +246,15 @@ export class StateStore { fs.writeFileSync(this.approvalsFile, JSON.stringify(approvals, null, 2)); } + resume(token: string, transform: (record: InstanceRecord) => InstanceRecord): InstanceRecord { + const approval = this.listApprovals().find((a) => a.token === token); + const record = this.get(token) ?? (approval ? this.get(approval.instanceId) : null); + if (!record) throw new Error(`Resume token not found: ${token}`); + const updated = this.update(record.instanceId, transform(record)); + if (record.resumeToken) this.resolveApproval(record.resumeToken); + return updated; + } + resolveApproval(token: string): PendingApproval | null { const approvals = this.listApprovals(); const idx = approvals.findIndex((a) => a.token === token); diff --git a/src/core/types.ts b/src/core/types.ts index 39b51cc..e5a667f 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -549,6 +549,8 @@ export interface PluginConfig { memoryDir?: string; maxNodeDurationMs?: number; stateDir?: string; // where to persist flow state across restarts + /** Execution checkpoints and approvals. Defaults to the legacy JSON store. */ + storage?: import("./storage.js").StorageConfig; /** * Days to keep finished runs (completed, failed, cancelled) in `stateDir`. * Older ones are deleted by a daily sweep in the serving process; running, diff --git a/src/index.ts b/src/index.ts index f76a1c7..939c542 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,6 +8,10 @@ export { default } from "./plugin/index.js"; export { FlowRunner, sendEvent, targetsCallerSession, agentStepSessionKey } from "./core/runner.js"; export type { CallerSession, CancelResult } from "./core/runner.js"; export { StateStore } from "./core/store.js"; +export type { InstanceRecord, RunIndexEntry } from "./core/store.js"; +export { importJsonState } from "./core/migrate-storage.js"; +export { createStateStore } from "./core/storage.js"; +export type { ExecutionStore, StorageConfig } from "./core/storage.js"; export { transpileToCloudflare } from "./core/transpile.js"; export { validateFlow } from "./core/validate.js"; export type { diff --git a/src/plugin/index.ts b/src/plugin/index.ts index 7d39965..c07916e 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -1,7 +1,7 @@ import * as fs from "node:fs"; import * as path from "node:path"; -import { FlowRunner, sendEvent } from "../core/runner.js"; +import { FlowRunner } from "../core/runner.js"; import { startFlowServer } from "../core/serve.js"; import { TriggerStore } from "../core/triggers.js"; @@ -16,11 +16,10 @@ import { } from "../core/manage.js"; import { DEFAULT_RUN_RETENTION_DAYS, - listRuns, readRunValue, runView, - sweepRuns, } from "../core/runs.js"; +import type { ExecutionStore } from "../core/storage.js"; import type { FlowDefinition, FlowNode, PluginConfig, BranchNode, ConditionNode, LoopNode, ParallelNode } from "../core/types.js"; // OpenClaw calls register() more than once per gateway process (every config @@ -46,7 +45,7 @@ let retentionTimers: Array> = []; let sweeping = false; function armRetention( - stateDir: string, + store: ExecutionStore, retentionDays: number, logger?: { info: (msg: string) => void; warn: (msg: string) => void }, ): void { @@ -56,11 +55,11 @@ function armRetention( const sweep = () => { if (sweeping) return; sweeping = true; - sweepRuns(stateDir, retentionDays) + store.sweep(retentionDays) .then(({ indexed, runs, files }) => { if (indexed) logger?.info(`clawflow: indexed ${indexed} runs`); if (files) { - logger?.info(`clawflow: retention deleted ${runs} runs (${files} files) older than ${retentionDays} days`); + logger?.info(`clawflow: retention deleted ${runs} runs (${files} records) older than ${retentionDays} days`); } }) .catch((err) => { @@ -215,7 +214,7 @@ function register(api: PluginApi) { activeScheduler = scheduler; activeTriggerStore = triggerStore; scheduler.start(); - armRetention(store.dir, pluginCfg.runRetentionDays ?? DEFAULT_RUN_RETENTION_DAYS, api.logger); + armRetention(store, pluginCfg.runRetentionDays ?? DEFAULT_RUN_RETENTION_DAYS, api.logger); } // ---- Flow server (optional) --------------------------------------------------- @@ -981,7 +980,7 @@ payload is passed as the output of the wait node and into flow state.`, payload?: unknown; }, ) { - const delivered = sendEvent( + const delivered = runner.sendEvent( params.instanceId, params.eventType, params.payload ?? {}, @@ -1074,7 +1073,7 @@ Status values: running | completed | paused | waiting | failed | cancelled`, }); try { if (params.instanceId) { - const record = store.get(params.instanceId); + const record = await store.get(params.instanceId); if (!record) return { content: [ @@ -1089,7 +1088,7 @@ Status values: running | completed | paused | waiting | failed | cancelled`, } return { ...text(runView(record)), details: record }; } - const page = listRuns(store.dir, { + const page = await store.listRuns({ limit: params.limit, cursor: params.cursor, flow: params.flow, diff --git a/tests/storage.test.ts b/tests/storage.test.ts new file mode 100644 index 0000000..33d8813 --- /dev/null +++ b/tests/storage.test.ts @@ -0,0 +1,274 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { randomUUID } from "node:crypto"; +import { FlowRunner, StateStore, StepRegistry, createStateStore, importJsonState } from "../src/index.js"; +import { SqlStateStore } from "../src/core/sql-store.js"; +import type { StorageConfig, ExecutionStore, FlowDefinition, PendingApproval } from "../src/index.js"; +import plugin from "../src/plugin/index.js"; + +const postgres = !!process.env.CLAWFLOW_TEST_POSTGRES_URL; +for (const backend of ["sqlite", "postgres"] as const) { + describe(`${backend} execution storage`, { skip: backend === "postgres" && !postgres }, () => { + function setup(t: any) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "clawflow-sql-test-")); + const storage: StorageConfig = backend === "sqlite" + ? { backend, path: path.join(dir, "state.sqlite") } + : { backend, connectionStringEnv: "CLAWFLOW_TEST_POSTGRES_URL", namespace: `test-${randomUUID()}` }; + const handles: ExecutionStore[] = []; + const open = () => { const s = createStateStore(dir, storage); handles.push(s); return s; }; + t.after(async () => { for (const s of handles) await s.close(); fs.rmSync(dir, { recursive: true, force: true }); }); + return { dir, storage, open, handles }; + } + + it("reopens paused checkpoints, resumes once, and preserves memoized steps and trace", async (t) => { + const { dir, storage, handles } = setup(t); + const registry = new StepRegistry(); + let calls = 0; + registry.register({ name: "count", allowedKeys: [], run: () => { calls++; return { calls, text: "quote ' and NUL \0" }; } }); + const flow: FlowDefinition = { flow: "durable", nodes: [ + { name: "first", do: "count" as "code", output: "first" }, + { name: "gate", do: "wait", for: "approval", prompt: "Continue?", output: "approval" }, + { name: "last", do: "count" as "code", output: "last" }, + ] }; + const a = new FlowRunner({ stateDir: dir, storage, customSteps: registry }); + const paused = await a.run(flow, { nested: [null, true, 42] }, "restart"); + assert.equal(paused.status, "paused"); + assert.equal((await a.listApprovals()).length, 1); + const record = await a.getStore().get("restart"); + await a.close(); + const b = new FlowRunner({ stateDir: dir, storage, customSteps: registry }); + handles.push(b.getStore()); + assert.deepEqual(await b.getStore().get("restart"), record); + const done = await b.resume(paused.resumeToken!, flow); + assert.equal(done.status, "completed"); + assert.equal(calls, 2, "memoized first step must not execute again after reopening"); + assert.deepEqual(done.trace.map((r) => r.node), ["first", "gate", "last"]); + assert.equal((await b.listApprovals()).length, 0); + assert.equal((await b.getStore().get("restart"))?.resumeToken, undefined); + await assert.rejects(b.resume(paused.resumeToken!, flow), /not found/); + await assert.rejects(b.run(flow, {}, "restart"), /unique|duplicate/i, "must not overwrite a durable run"); + }); + + it("atomically claims concurrent resumes and keeps the token when a transition fails", async (t) => { + const { dir, storage, handles } = setup(t); + const a = new FlowRunner({ stateDir: dir, storage }); + const b = new FlowRunner({ stateDir: dir, storage }); + handles.push(a.getStore(), b.getStore()); + const flow: FlowDefinition = { flow: "claim", nodes: [ + { name: "gate", do: "wait", for: "approval" }, + { name: "end", do: "code", run: "42", output: "value" }, + ] }; + const paused = await a.run(flow, {}); + await assert.rejects(Promise.resolve().then(() => a.getStore().resume(paused.resumeToken!, () => { throw new Error("rollback"); })), /rollback/); + assert.equal((await a.listApprovals()).length, 1); + assert.equal((await a.getStore().get(paused.instanceId))!.status, "paused"); + const results = await Promise.allSettled([a.resume(paused.resumeToken!, flow), b.resume(paused.resumeToken!, flow)]); + assert.equal(results.filter((r) => r.status === "fulfilled").length, 1); + assert.equal((await a.getStore().get(paused.instanceId))!.state.value, 42); + assert.equal((await a.listApprovals()).length, 0); + }); + + it("executes parallel children and reuses a successful child when its parent retries", async (t) => { + const { dir, storage, handles } = setup(t); + const registry = new StepRegistry(); + let stable = 0, flaky = 0; + registry.register({ name: "stable", allowedKeys: [], run: () => ++stable }); + registry.register({ name: "flaky", allowedKeys: [], run: () => { if (++flaky === 1) throw new Error("retry"); return 2; } }); + const runner = new FlowRunner({ stateDir: dir, storage, customSteps: registry }); + handles.push(runner.getStore()); + const result = await runner.run({ flow: "children", nodes: [{ name: "branch", do: "condition", if: "true", retry: { limit: 2, delay: 0 }, then: [ + { name: "one", do: "stable" as "code", output: "one" }, + { name: "two", do: "flaky" as "code", output: "two" }, + ] }, { name: "parallel", do: "parallel", nodes: [ + { name: "a", do: "code", run: "1", output: "a" }, + { name: "b", do: "code", run: "2", output: "b" }, + ], output: "parallel" }] }, {}); + assert.equal(result.status, "completed", result.error); + assert.equal(stable, 1); + assert.equal(flaky, 2); + assert.equal((await runner.getStore().listRuns()).total, 1); + const records = await runner.getStore().list(); + assert.ok(records.length > 1); + assert.ok(records.every((r) => r.status === "completed" && !r.error)); + }); + + it("serializes concurrent mutations from separate handles without losing memoized outputs", async (t) => { + const { open } = setup(t); + const a = open(), b = open(); + await a.create("concurrent", "parallel", {}); + await Promise.all(Array.from({ length: 30 }, (_, n) => (n % 2 ? a : b).memoize("concurrent", `n${n}`, { n }))); + assert.equal(Object.keys((await b.get("concurrent"))!.completedNodes).length, 30); + await a.update("concurrent", { status: "cancelled", error: "stopped" }); + await b.update("concurrent", { status: "completed", error: undefined }); + assert.equal((await a.get("concurrent"))!.status, "cancelled"); + assert.equal((await a.get("concurrent"))!.error, "stopped"); + }); + + it("reports cancellation that races with the final checkpoint", async (t) => { + const { dir, storage, handles, open } = setup(t); + const runner = new FlowRunner({ stateDir: dir, storage }); + const external = open(); + const store = runner.getStore(); + handles.push(store); + const update = store.update.bind(store); + store.update = async (id, patch) => { + if (patch.status === "completed") await external.update(id, { status: "cancelled", error: "external cancellation" }); + return update(id, patch); + }; + const result = await runner.run({ flow: "race", nodes: [{ name: "one", do: "code", run: "1" }] }, {}); + assert.equal(result.status, "cancelled"); + assert.equal(result.error, "external cancellation"); + assert.equal((await store.get(result.instanceId))!.status, "cancelled"); + }); + + it("commits checkpoint plus approval atomically and only consumes a token once", async (t) => { + const { open } = setup(t); + const a = open(), b = open(); + await a.create("approved", "gate", {}); + const approval: PendingApproval = { token: "token", instanceId: "approved", flowName: "gate", node: "gate", prompt: "OK?", createdAt: new Date().toISOString(), expiresAt: new Date(Date.now() + 10000).toISOString() }; + await a.checkpoint("approved", { status: "paused", resumeToken: "token" }, { node: "first", output: 4 }, approval); + await assert.rejects(a.checkpoint("approved", { state: { lost: true } }, undefined, approval)); + assert.deepEqual((await a.get("approved"))!.state, {}, "failed approval insert must roll back the checkpoint"); + const consumed = await Promise.all([a.resolveApproval("token"), b.resolveApproval("token")]); + assert.equal(consumed.filter(Boolean).length, 1); + assert.deepEqual(consumed.find(Boolean), approval); + }); + + it("pages indexed summaries with tied timestamps, filters, subflows and SQL metacharacters", async (t) => { + const { open } = setup(t); + const s = open() as SqlStateStore; + const source = new StateStore(fs.mkdtempSync(path.join(os.tmpdir(), "clawflow-fixture-"))); + t.after(() => fs.rmSync(source.dir, { recursive: true, force: true })); + const template = source.create("a", "flow' OR 1=1 --", {}); + await s.importRecords(Array.from({ length: 23 }, (_, n) => ({ ...template, instanceId: `r${String(n).padStart(2, "0")}`, state: { large: "x".repeat(10000) }, completedNodes: { first: n }, status: n % 2 ? "completed" : "failed" })), []); + await s.create("r00:parallel:child:0", "child", {}); + const seen: string[] = []; + let cursor: string | undefined; + do { + const page = await s.listRuns({ limit: 4, cursor }); + assert.equal(page.total, 23); + assert.ok(JSON.stringify(page).length < 3000); + seen.push(...page.runs.map((r) => r.instanceId)); + cursor = page.next_cursor ?? undefined; + } while (cursor); + assert.equal(new Set(seen).size, 23); + assert.deepEqual(seen, [...seen].sort().reverse()); + assert.equal((await s.listRuns({ flow: "flow' OR 1=1 --", status: "failed" })).total, 12); + assert.equal((await s.listRuns({ flow: "other" })).total, 0); + assert.throws(() => s.listRuns({ cursor: "bad" }), /cursor/); + assert.throws(() => s.listRuns({ limit: 101 }), /limit/); + }); + + it("retains active runs and sweeps finished roots with their subflows and approvals", async (t) => { + const { open } = setup(t); + const s = open() as SqlStateStore; + const base = { flowName: "retained", state: {}, completedNodes: {}, trace: [], createdAt: "2020-01-01T00:00:00.000Z", updatedAt: "2020-01-01T00:00:00.000Z" }; + await s.importRecords([ + { ...base, instanceId: "old", status: "completed" }, + { ...base, instanceId: "old:parallel:child:0", status: "completed" }, + { ...base, instanceId: "old-prefix", status: "running" }, + { ...base, instanceId: "paused", status: "paused" }, + { ...base, instanceId: "waiting", status: "waiting" }, + ], [{ token: "orphan", instanceId: "old", flowName: "retained", node: "gate", prompt: "?", createdAt: base.createdAt, expiresAt: base.updatedAt }]); + assert.deepEqual(await s.sweep(0), { indexed: 0, runs: 0, files: 0 }); + assert.deepEqual(await s.sweep(30), { indexed: 0, runs: 1, files: 2 }); + assert.deepEqual((await s.list()).map((r) => r.instanceId).sort(), ["old-prefix", "paused", "waiting"]); + assert.equal((await s.listApprovals()).length, 0); + }); + + it("imports legacy JSON and approvals without changing files, retries safely and refuses conflicts", async (t) => { + const { dir, storage, open } = setup(t); + const json = new StateStore(dir); + const original = json.create("legacy", "imported", { inputs: { x: 1 } }); + json.memoize("legacy", "first", { x: 2 }); + json.update("legacy", { status: "paused", resumeToken: "legacy-token", pausedAtIndex: 1 }); + json.create("legacy:parallel:child:0", "child", {}); + json.addApproval({ token: "legacy-token", instanceId: "legacy", flowName: "imported", node: "gate", prompt: "?", createdAt: original.createdAt, expiresAt: original.updatedAt }); + const before = fs.readFileSync(path.join(dir, "legacy.json"), "utf8"); + assert.deepEqual(await importJsonState(dir, storage as any), { runs: 2, approvals: 1 }); + assert.deepEqual(await importJsonState(dir, storage as any), { runs: 2, approvals: 1 }); + assert.equal(fs.readFileSync(path.join(dir, "legacy.json"), "utf8"), before); + const sql = open(); + assert.deepEqual(await sql.get("legacy"), json.get("legacy")); + assert.deepEqual(await sql.listApprovals(), json.listApprovals()); + await sql.update("legacy", { status: "completed" }); + await assert.rejects(importJsonState(dir, storage as any), /Migration conflict/); + assert.equal((await sql.get("legacy"))!.status, "completed"); + fs.writeFileSync(path.join(dir, "broken.json"), "{bad"); + await assert.rejects(importJsonState(dir, storage as any)); + }); + + it("uses the selected backend for plugin run, status list, value reads and cancel", async (t) => { + const { dir, storage } = setup(t); + const defs = new Map(); + plugin.register({ + registerTool: (def: any, opts: any) => defs.set(typeof def === "function" ? opts.name : def.name, def), + config: { workspace: dir, plugins: { entries: { clawflow: { config: { stateDir: dir, storage } } } } }, + }); + const run = defs.get("flow_run")({}); + const paused = await run.execute("a", { flow: { flow: "tool", nodes: [{ name: "gate", do: "wait", for: "approval" }] } }); + assert.equal(paused.details.status, "paused"); + const listed = await defs.get("flow_status").execute("b", {}); + assert.equal(listed.details.runs[0].instanceId, paused.details.instanceId); + const value = await defs.get("flow_status").execute("c", { instanceId: paused.details.instanceId, path: "status" }); + assert.equal(JSON.parse(value.content[0].text).text, "paused"); + const cancelled = await defs.get("flow_cancel").execute("d", { instanceId: paused.details.instanceId }); + assert.equal(cancelled.details.status, "cancelled"); + }); + + it("scopes event delivery when two namespaces use the same instance ID", async (t) => { + const { dir, storage, handles } = setup(t); + const other: StorageConfig = backend === "sqlite" ? { backend, path: path.join(dir, "events.sqlite") } : { ...storage as any, namespace: `events-${randomUUID()}` }; + const a = new FlowRunner({ storage, stateDir: dir }); + const b = new FlowRunner({ storage: other, stateDir: dir }); + handles.push(a.getStore(), b.getStore()); + const flow: FlowDefinition = { flow: "events", nodes: [{ name: "wait", do: "wait", for: "event", event: "go", timeout: "3s", output: "event" }] }; + const aDone = a.run(flow, {}, "event-id"), bDone = b.run(flow, {}, "event-id"); + for (const runner of [a, b]) { + let sent = false; + for (let n = 0; !sent && n < 100; n++) { + sent = runner.sendEvent("event-id", "go", runner === a ? "a" : "b"); + if (!sent) await new Promise((r) => setTimeout(r, 10)); + } + assert.equal(sent, true); + } + assert.equal((await aDone).state.event, "a"); + assert.equal((await bDone).state.event, "b"); + }); + + it("isolates live cancellation from other databases or namespaces", async (t) => { + const { dir, storage, handles } = setup(t); + const other: StorageConfig = backend === "sqlite" ? { backend, path: path.join(dir, "other.sqlite") } : { ...storage as any, namespace: `other-${randomUUID()}` }; + const a = new FlowRunner({ storage, stateDir: dir }); + const b = new FlowRunner({ storage: other, stateDir: dir }); + handles.push(a.getStore(), b.getStore()); + const done = a.run({ flow: "slow", nodes: [{ name: "wait", do: "sleep", duration: "30s" }] }, {}, "same-id"); + for (let n = 0; !(await a.getStore().get("same-id"))?.running && n < 100; n++) await new Promise((r) => setTimeout(r, 10)); + assert.equal((await a.getStore().get("same-id"))?.running?.node, "wait"); + assert.deepEqual(await b.cancel("same-id"), { ok: false, instanceId: "same-id", error: "not_found" }); + assert.equal((await b.getStore().listRuns()).total, 0); + await a.cancel("same-id"); + assert.equal((await done).status, "cancelled"); + }); + }); +} + +it("requires explicit PostgreSQL configuration and never silently falls back", () => { + assert.throws(() => createStateStore(undefined, { backend: "postgres", connectionStringEnv: "CLAWFLOW_MISSING_TEST_ENV", namespace: "test" }), /connectionStringEnv/); + assert.throws(() => createStateStore(undefined, { backend: "postgres", connectionStringEnv: "x", namespace: "" }), /namespace/); + assert.throws(() => createStateStore(undefined, { backend: "unknown" } as any), /Unknown/); +}); + +// A database outage must not create a second, divergent local history. +it("fails a PostgreSQL write closed when its configured database is unavailable", async (t) => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "clawflow-unavailable-")); + const env = `CLAWFLOW_TEST_UNAVAILABLE_${randomUUID().replaceAll("-", "_")}`; + process.env[env] = `postgresql://clawflow@localhost/test?host=${encodeURIComponent(path.join(dir, "missing-socket"))}`; + const store = createStateStore(dir, { backend: "postgres", connectionStringEnv: env, namespace: "unavailable" }); + t.after(async () => { await store.close(); delete process.env[env]; fs.rmSync(dir, { recursive: true, force: true }); }); + await assert.rejects(store.create("lost", "unavailable", {})); + assert.deepEqual(fs.readdirSync(dir), []); +}); From a585131b2ff3da217ae44a1a97959c87484cb6a4 Mon Sep 17 00:00:00 2001 From: pallaoro Date: Sun, 8 Nov 2026 16:36:25 +0100 Subject: [PATCH 2/4] Expose authenticated native run management and paged descendants --- .github/workflows/ci.yml | 5 +- README.md | 35 ++++++++++ src/core/control.ts | 142 +++++++++++++++++++++++++++++++++++++++ src/core/runner.ts | 4 ++ src/core/runs.ts | 27 +++++++- src/core/serve.ts | 13 +++- src/core/sql-store.ts | 27 +++++++- src/core/storage.ts | 3 +- src/core/store.ts | 3 +- src/index.ts | 2 + src/plugin/index.ts | 3 +- tests/control.test.ts | 68 +++++++++++++++++++ 12 files changed, 324 insertions(+), 8 deletions(-) create mode 100644 src/core/control.ts create mode 100644 tests/control.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6828e82..81f4a35 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,6 +8,9 @@ on: jobs: check: + strategy: + matrix: + node: [22, 24] runs-on: ubuntu-latest services: postgres: @@ -28,7 +31,7 @@ jobs: - uses: actions/setup-node@v4 with: - node-version: 22 + node-version: ${{ matrix.node }} - run: npm ci diff --git a/README.md b/README.md index 0ff6a33..20a4b15 100644 --- a/README.md +++ b/README.md @@ -658,6 +658,41 @@ try { } ``` +`store.listSubRuns(parentId, { limit, cursor, node?, flow?, status? })` pages +all retained descendants, preserving their actual `:`-separated IDs. `node` +filters the immediate node beneath that parent, including its nested descendants. +It uses the same `(createdAt DESC, instanceId DESC)` order as `listRuns`. +SQL count and page share one read transaction; different pages and `get` calls +do not share a snapshot. JSON readers have no snapshot. Neither API provides an +update cursor, revision, complete inventory guarantee, or deletion tombstones. + +When the HTTP listener is enabled, authenticated management routes use the +same configured runner: + +- `GET /flows/runs?limit=20&cursor=...&flow=...&status=...` +- `GET /flows/runs/:instanceId` for the complete record; optional `path`, `offset`, + `limit`, `max_chars` use `readRunValue`. URL-encode actual instance IDs. +- `GET /flows/runs/:instanceId/subruns?limit=20&node=...` for descendant summaries. +- `GET /flows/runs/:instanceId?include_subruns=1&node=...` adds the last 20 + iterations per node and counts of omitted iterations for detail views. +- `POST /flows/runs/:instanceId/resume` with `{ "approved": true }` or `false`. + Approval resolves the definition through `resolveRunnableFlow` (published + version preferred); decline atomically cancels and consumes its approval. + A 202 is sent only after the durable resume claim, before later steps finish. + +The configured listener prefix replaces `/flows`. These new routes require +`Authorization: Bearer ` (or the host gateway's configured +token; standalone hosts may supply `controlToken` to `startFlowServer`). Without +a token they refuse access. Optional `X-ClawFlow-Backend: sqlite|postgres|json` +refuses a stale runner whose backend differs from the caller's configuration. +Storage errors return 503 without credentials or a JSON fallback. The existing +run/trigger/cancel routes retain their prior authentication contract. + +Run records do not yet bind an executed definition version/hash or trusted +actor/session access scope. Keep the relevant definition unchanged while a run +is paused. Do not infer historical provenance or access grants from missing +fields. Management reads are operator access, not a per-user authorization layer. + For a standalone runner, call `await runner.close()` once its runs have settled. This closes the SQLite handle or PostgreSQL pool. Close a reader when it is no longer needed; avoid opening a new pool for each request in a long-lived host. diff --git a/src/core/control.ts b/src/core/control.ts new file mode 100644 index 0000000..d7e1890 --- /dev/null +++ b/src/core/control.ts @@ -0,0 +1,142 @@ +import type { IncomingMessage, ServerResponse } from "node:http"; +import { timingSafeEqual } from "node:crypto"; +import type { FlowRunner } from "./runner.js"; +import type { FlowDefinition } from "./types.js"; +import { parseRunPage, readRunValue } from "./runs.js"; +import type { InstanceRecord, RunIndexEntry } from "./store.js"; + +/** Management reads/approval actions require the host's gateway bearer. The + * existing webhook listener may be behind a public tunnel: loopback alone is + * not an authorization boundary. No token configured means these routes close. */ +export async function controlRequest( + req: IncomingMessage, res: ServerResponse, url: URL, basePath: string, + runner: FlowRunner, token: string | undefined, + json: (res: ServerResponse, status: number, body: unknown) => void, + readBody: (req: IncomingMessage) => Promise, + resolveDefinition: (name: string) => FlowDefinition | null, +): Promise { + const base = `${basePath}/runs`; + const pathname = url.pathname.replace(/\/+$/, ""); + if (pathname !== base && !pathname.startsWith(`${base}/`)) return false; + const parts = pathname.slice(base.length).split("/").filter(Boolean); + // Preserve the older cancellation endpoint's contract in serve.ts. + if (parts.length === 2 && parts[1] === "cancel") return false; + const supplied = req.headers.authorization; + const expected = typeof token === "string" && token ? `Bearer ${token}` : ""; + if (!expected || typeof supplied !== "string" || Buffer.byteLength(supplied) !== Buffer.byteLength(expected) + || !timingSafeEqual(Buffer.from(supplied), Buffer.from(expected))) { + json(res, 401, { error: "Flow management authentication required", code: "flow_unauthorized" }); + return true; + } + if (req.headers["x-clawflow-backend"] && req.headers["x-clawflow-backend"] !== runner.storageBackend) { + json(res, 503, { error: "Restart the flow engine to apply its storage configuration", code: "flow_storage_mismatch" }); + return true; + } + let id: string | undefined; + try { id = parts[0] ? decodeURIComponent(parts[0]) : undefined; } + catch { json(res, 400, { error: "Invalid instance ID" }); return true; } + if (id && (id.length > 1024 || /[\x00-\x1f/\\]/.test(id))) { + json(res, 400, { error: "Invalid instance ID" }); return true; + } + const store = runner.getStore(); + const qs = url.searchParams; + let acknowledged = false; + try { + if (req.method === "GET" && (!id || (parts.length === 2 && parts[1] === "subruns"))) { + const opts = { limit: qs.has("limit") ? Number(qs.get("limit")) : undefined, cursor: qs.get("cursor") || undefined, + flow: qs.get("flow") || undefined, status: qs.get("status") || undefined, node: qs.get("node") || undefined }; + try { parseRunPage(opts); } catch (err) { json(res, 400, { error: (err as Error).message }); return true; } + const page = id ? await store.listSubRuns(id, opts) : await store.listRuns(opts); + json(res, 200, { ok: true, ...page }); + return true; + } + if (id && parts.length === 1 && req.method === "GET") { + const run = await store.get(id); + if (!run || run.instanceId !== id) { json(res, 404, { error: "Run not found", code: "run_not_found" }); return true; } + const valuePath = qs.get("path"); + // Full detail keeps the platform's existing per-node last-20 projection. + // Individual paths bypass the child scan unless they name that projection. + const detail = (!valuePath || valuePath.startsWith("subFlows")) && qs.get("include_subruns") === "1" + ? await runDetail(runner, run, qs.get("node") || undefined) : run; + if (valuePath) { + try { + json(res, 200, { ok: true, ...readRunValue(detail, valuePath, { + offset: qs.has("offset") ? Number(qs.get("offset")) : undefined, + limit: qs.has("limit") ? Number(qs.get("limit")) : undefined, + maxChars: qs.has("max_chars") ? Number(qs.get("max_chars")) : undefined, + }) }); + } catch (err) { json(res, 400, { error: (err as Error).message }); } + } else json(res, 200, { ok: true, run: detail }); + return true; + } + if (id && parts.length === 2 && parts[1] === "resume" && req.method === "POST") { + let body: { approved?: unknown }; + try { body = JSON.parse(await readBody(req)); } + catch { json(res, 400, { error: "Invalid JSON body" }); return true; } + if (!body || typeof body.approved !== "boolean") { json(res, 400, { error: "approved must be boolean" }); return true; } + const run = await store.get(id); + if (!run) { json(res, 404, { error: "Run not found", code: "run_not_found" }); return true; } + if (run.status !== "paused" && run.status !== "waiting") { + json(res, 409, { error: "Run is not paused", code: "run_not_paused" }); return true; + } + if (!body.approved) { + const result = await runner.resume(id, { flow: run.flowName, nodes: [] }, false); + json(res, 200, { ok: true, status: result.status, instanceId: id }); + return true; + } + const flow = resolveDefinition(run.flowName); + if (!flow || flow.flow !== run.flowName) { + json(res, 404, { error: "Flow definition not found", code: "flow_definition_required" }); return true; + } + // Acknowledge after the durable claim, before executing later steps. A + // database failure before the claim must not masquerade as accepted work. + await runner.resume(id, flow, true, (instanceId) => { + acknowledged = true; + json(res, 202, { ok: true, status: "resuming", instanceId }); + }); + return true; + } + json(res, 404, { error: "Not found", code: "flow_control_not_found" }); + } catch (err) { + if (!acknowledged) { + const raced = err instanceof Error && /is not paused|Resume token not found/.test(err.message); + json(res, raced ? 409 : 503, { error: raced ? "Run is no longer paused" : "Flow storage operation failed", code: raced ? "run_not_paused" : "flow_storage_unavailable" }); + } else console.warn("[clawflow] Resumed flow failed after acknowledgement; inspect its stored status"); + } + return true; +} + +async function runDetail(runner: FlowRunner, run: InstanceRecord, node?: string): Promise { + const store = runner.getStore(); + const groups = new Map(); + let cursor: string | undefined; + const iteration = (r: RunIndexEntry) => Number.parseInt(r.instanceId.slice(run.instanceId.length + 1).split(":")[2] ?? "0", 10) || 0; + do { + const page = await store.listSubRuns(run.instanceId, { limit: 100, cursor, node }); + for (const row of page.runs) { + const name = row.instanceId.slice(run.instanceId.length + 1).split(":")[1]; + const group = groups.get(name) ?? { total: 0, rows: [] }; + group.total++; + group.rows.push(row); + group.rows.sort((a, b) => iteration(a) - iteration(b)); + if (group.rows.length > 20) group.rows.shift(); + groups.set(name, group); + } + cursor = page.next_cursor ?? undefined; + } while (cursor); + const subFlows: Record = Object.create(null); + const omitted: Record = Object.create(null); + for (const [name, group] of groups) { + subFlows[name] = []; + for (const row of group.rows) { + const child = await store.get(row.instanceId); + if (!child) throw new Error("Sub-run changed during detail read"); + subFlows[name].push({ instanceId: child.instanceId, iteration: iteration(row), status: child.status, + trace: child.trace ?? [], completedNodes: child.completedNodes ?? {} }); + } + if (group.total > group.rows.length) omitted[name] = { total: group.total, returned: group.rows.length, + omitted: group.total - group.rows.length, note: "Oldest iterations omitted. Read the paged subruns endpoint for every retained instance." }; + } + return { ...run, subFlows, ...(node ? { subFlowsNodeFilter: node } : {}), + ...(Object.keys(omitted).length ? { subFlowsTruncated: omitted } : {}) } as InstanceRecord; +} diff --git a/src/core/runner.ts b/src/core/runner.ts index 456a439..52aa8e1 100644 --- a/src/core/runner.ts +++ b/src/core/runner.ts @@ -138,6 +138,8 @@ export class FlowRunner { return sendEvent(instanceId, eventType, payload, this.store.identity); } + get storageBackend(): string { return this.cfg.storage?.backend ?? "json"; } + private runKey(id: string): string { return `${this.store.identity}\0${id}`; } /** Release database resources after runs have settled. */ @@ -288,6 +290,7 @@ export class FlowRunner { token: string, flow: FlowDefinition, approvedOrPayload: boolean | unknown = true, + onClaimed?: (instanceId: string) => void, ): Promise { let pausedAtIndex = 0; // Pure transform under the store's transaction/lock: competing resumes @@ -313,6 +316,7 @@ export class FlowRunner { } return { ...current, trace, status: declined ? "cancelled" : "running", resumeToken: undefined, waitingFor: undefined, pausedAtIndex: undefined }; }); + onClaimed?.(record.instanceId); if (record.status === "cancelled") { return { ok: true, status: "cancelled", flowName: record.flowName, instanceId: record.instanceId, state: record.state, trace: record.trace, error: "Flow cancelled at approval gate" }; } diff --git a/src/core/runs.ts b/src/core/runs.ts index dcece5c..e62d7d3 100644 --- a/src/core/runs.ts +++ b/src/core/runs.ts @@ -80,13 +80,18 @@ export function parseRunPage(opts: ListRunsOptions) { } export function listRuns(stateDir: string, opts: ListRunsOptions = {}): RunPage { - const { limit, after } = parseRunPage(opts); + parseRunPage(opts); const rows: RunRow[] = []; for (const row of readTopLevelRows(stateDir)) { if (opts.flow && row.flowName !== opts.flow) continue; if (opts.status && row.status !== opts.status) continue; rows.push(row); } + return pageRunRows(rows, opts); +} + +export function pageRunRows(rows: RunRow[], opts: ListRunsOptions = {}): RunPage { + const { limit, after } = parseRunPage(opts); rows.sort(newestFirst); const rest = after ? rows.filter((r) => newestFirst(r, after) > 0) : rows; const page = rest.slice(0, limit); @@ -98,6 +103,26 @@ export function listRuns(stateDir: string, opts: ListRunsOptions = {}): RunPage }; } +export interface ListSubRunsOptions extends ListRunsOptions { node?: string } + +export function listSubRuns(stateDir: string, parentId: string, opts: ListSubRunsOptions = {}): RunPage { + parseRunPage(opts); + const rows: RunRow[] = []; + if (!fs.existsSync(stateDir)) return { runs: [], total: 0, next_cursor: null }; + const prefix = `${parentId}:`; + const filePrefix = prefix.replace(/[^a-zA-Z0-9_-]/g, "_"); + for (const file of fs.readdirSync(stateDir)) { + if (!file.startsWith(filePrefix) || !file.endsWith(".json")) continue; + const r = JSON.parse(fs.readFileSync(path.join(stateDir, file), "utf8")) as InstanceRecord; + if (!r.instanceId?.startsWith(prefix)) continue; + if (opts.node && r.instanceId.slice(prefix.length).split(":")[1] !== opts.node) continue; + if (opts.flow && r.flowName !== opts.flow) continue; + if (opts.status && r.status !== opts.status) continue; + rows.push(runIndexEntry(r)); + } + return pageRunRows(rows, opts); +} + type Keyed = { createdAt: string; instanceId: string }; /** (createdAt, instanceId) descending. Positive when `a` sorts after `b`. */ diff --git a/src/core/serve.ts b/src/core/serve.ts index ba35511..16639c3 100644 --- a/src/core/serve.ts +++ b/src/core/serve.ts @@ -5,6 +5,7 @@ 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 { controlRequest } from "./control.js"; import { resolveRunnableFlow } from "./manage.js"; // ---- Flow Server ---------------------------------------------------------------- @@ -20,6 +21,8 @@ import { resolveRunnableFlow } from "./manage.js"; export interface FlowServerOpts { runner: FlowRunner; + /** Bearer for management reads/resume. Defaults to OPENCLAW_GATEWAY_TOKEN. */ + controlToken?: () => string | undefined; serve: ServeConfig; /** * Workspace root — where published versions live (.clawflow/versions). @@ -46,6 +49,7 @@ const MAX_BODY_BYTES = 1_048_576; // 1 MB // Guard against double-init when register() is called more than once // (OpenClaw calls it during discovery and again at gateway startup). let activeServer: http.Server | null = null; +let activeOpts: FlowServerOpts | null = null; function resolveWorkspace(workspace?: string): string { return workspace ?? process.env.OPENCLAW_WORKSPACE ?? process.cwd(); @@ -87,9 +91,10 @@ function readBody(req: http.IncomingMessage): Promise { } export function startFlowServer(opts: FlowServerOpts): http.Server { + activeOpts = opts; if (activeServer) return activeServer; - const { runner, serve, logger } = opts; + const { serve, logger } = opts; const basePath = (serve.path ?? "/flows").replace(/\/+$/, ""); const workspace = resolveWorkspace(opts.workspace); const flowsDir = resolveFlowsDir(serve, workspace); @@ -100,9 +105,14 @@ export function startFlowServer(opts: FlowServerOpts): http.Server { }; const server = http.createServer(async (req, res) => { + const current = activeOpts ?? opts; + const runner = current.runner; const url = new URL(req.url ?? "/", `http://${req.headers.host ?? "localhost"}`); const pathname = url.pathname.replace(/\/+$/, "") || "/"; + if (await controlRequest(req, res, url, basePath, runner, current.controlToken?.() ?? process.env.OPENCLAW_GATEWAY_TOKEN, json, readBody, + (name) => resolveRunnableFlow(resolveWorkspace(current.workspace), resolveFlowsDir(current.serve, resolveWorkspace(current.workspace)), name)?.def ?? null)) return; + // Health check if (req.method === "GET" && pathname === `${basePath}/health`) { json(res, 200, { ok: true, flowsDir }); @@ -415,6 +425,7 @@ export function startFlowServer(opts: FlowServerOpts): http.Server { }); activeServer = server; + server.on("close", () => { if (activeServer === server) { activeServer = null; activeOpts = null; } }); server.on("error", (err: NodeJS.ErrnoException) => { if (err.code === "EADDRINUSE") { diff --git a/src/core/sql-store.ts b/src/core/sql-store.ts index 6733207..cfa9cec 100644 --- a/src/core/sql-store.ts +++ b/src/core/sql-store.ts @@ -3,7 +3,7 @@ import { createHash } from "node:crypto"; import { runIndexEntry, type InstanceRecord } from "./store.js"; import type { FlowState, PendingApproval } from "./types.js"; import type { ExecutionStore, StorageConfig } from "./storage.js"; -import { encodeCursor, parseRunPage, type ListRunsOptions, type RunPage, type SweepResult } from "./runs.js"; +import { encodeCursor, parseRunPage, type ListRunsOptions, type ListSubRunsOptions, type RunPage, type SweepResult } from "./runs.js"; import { PostgresDriver, SqliteDriver, canonicalStoragePath, type Query, type SqlDriver } from "./sql-driver.js"; export class SqlStateStore implements ExecutionStore { @@ -137,10 +137,33 @@ export class SqlStateStore implements ExecutionStore { } listRuns(opts: ListRunsOptions = {}): Promise { + return this.pageRuns(opts); + } + + listSubRuns(parentId: string, opts: ListSubRunsOptions = {}): Promise { + return this.pageRuns(opts, parentId); + } + + private pageRuns(opts: ListSubRunsOptions, parentId?: string): Promise { const { limit, after } = parseRunPage(opts); return this.driver.transaction(async (q) => { const params: unknown[] = [this.namespace]; - let where = "namespace=$1 AND id=root_id"; + let where = "namespace=$1"; + if (parentId === undefined) where += " AND id=root_id"; + else { + params.push(parentId.split(":")[0], `${parentId}:`); + where += " AND root_id=$2 AND substr(id, 1, length($3))=$3"; + if (opts.node) { + if (opts.node.includes(":")) where += " AND 1=0"; + else { + const filters = ["loop", "branch", "condition", "parallel"].map((kind) => { + params.push(`${parentId}:${kind}:${opts.node!}:`); + return `substr(id, 1, length($${params.length}))=$${params.length}`; + }); + where += ` AND (${filters.join(" OR ")})`; + } + } + } if (opts.flow) { params.push(opts.flow); where += ` AND flow_name=$${params.length}`; } if (opts.status) { params.push(opts.status); where += ` AND status=$${params.length}`; } const count = await q(`SELECT COUNT(*) AS total FROM clawflow_runs WHERE ${where}`, params); diff --git a/src/core/storage.ts b/src/core/storage.ts index c4a996d..e63871d 100644 --- a/src/core/storage.ts +++ b/src/core/storage.ts @@ -2,7 +2,7 @@ import * as path from "node:path"; import { StateStore, type InstanceRecord } from "./store.js"; import { SqlStateStore } from "./sql-store.js"; import type { FlowState, PendingApproval } from "./types.js"; -import type { ListRunsOptions, RunPage, SweepResult } from "./runs.js"; +import type { ListRunsOptions, ListSubRunsOptions, RunPage, SweepResult } from "./runs.js"; export type StorageConfig = | { backend: "json" } @@ -29,6 +29,7 @@ export interface ExecutionStore { resolveApproval(token: string): Awaitable; listApprovals(): Awaitable; listRuns(opts?: ListRunsOptions): Awaitable; + listSubRuns(parentId: string, opts?: ListSubRunsOptions): Awaitable; sweep(retentionDays: number, now?: number): Promise; close(): Awaitable; } diff --git a/src/core/store.ts b/src/core/store.ts index 77bb9a2..b9af1b4 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -1,6 +1,6 @@ import * as fs from "fs"; import * as path from "path"; -import { listRuns, sweepRuns, type ListRunsOptions } from "./runs.js"; +import { listRuns, listSubRuns, sweepRuns, type ListRunsOptions, type ListSubRunsOptions } from "./runs.js"; import type { FlowState, FlowResult, TraceEntry, PendingApproval, RunningStep } from "./types.js"; // ---- Durable State Store -------------------------------------------------------- @@ -136,6 +136,7 @@ export class StateStore { get identity(): string { return `json:${path.resolve(this.dir)}`; } listRuns(opts?: ListRunsOptions) { return listRuns(this.dir, opts); } + listSubRuns(parentId: string, opts?: ListSubRunsOptions) { return listSubRuns(this.dir, parentId, opts); } sweep(days: number, now?: number) { return sweepRuns(this.dir, days, now); } close(): void {} diff --git a/src/index.ts b/src/index.ts index 939c542..1c1efc7 100644 --- a/src/index.ts +++ b/src/index.ts @@ -80,6 +80,7 @@ export { export type { PublishResult } from "./core/manage.js"; export { listRuns, + listSubRuns, summarizeRun, runView, readRunValue, @@ -91,6 +92,7 @@ export type { RunRow, RunPage, ListRunsOptions, + ListSubRunsOptions, RunSummary, TraceSummary, ValueOutline, diff --git a/src/plugin/index.ts b/src/plugin/index.ts index c07916e..1f72812 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -157,7 +157,7 @@ interface PluginApi { ) => void; config?: { plugins?: { entries?: Record }; - gateway?: { port?: number; host?: string }; + gateway?: { port?: number; host?: string; auth?: { token?: string } }; session?: { mainKey?: string }; [key: string]: unknown; }; @@ -225,6 +225,7 @@ function register(api: PluginApi) { startFlowServer({ runner, serve: pluginCfg.serve, + controlToken: () => process.env.OPENCLAW_GATEWAY_TOKEN ?? api.config?.gateway?.auth?.token, // Same workspace the flow_* tools use, so a triggered run resolves the // published version exactly like flow_run does. workspace, diff --git a/tests/control.test.ts b/tests/control.test.ts new file mode 100644 index 0000000..2461cb6 --- /dev/null +++ b/tests/control.test.ts @@ -0,0 +1,68 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { once } from "node:events"; +import { randomUUID } from "node:crypto"; +import { FlowRunner, type StorageConfig } from "../src/index.js"; +import { startFlowServer } from "../src/core/serve.js"; + +for (const backend of ["json", "sqlite", "postgres"] as const) { + describe(`${backend} authenticated management`, { skip: backend === "postgres" && !process.env.CLAWFLOW_TEST_POSTGRES_URL }, () => { + it("reads actual child IDs, pages/filter descendants, resumes and declines durable approvals", async (t) => { + const dir = mkdtempSync(join(tmpdir(), "clawflow-control-")); + const storage: StorageConfig = backend === "postgres" + ? { backend, connectionStringEnv: "CLAWFLOW_TEST_POSTGRES_URL", namespace: randomUUID() } : { backend }; + const runner = new FlowRunner({ stateDir: dir, storage }); + mkdirSync(join(dir, "flows")); + const flow = { flow: "approve", nodes: [{ name: "gate", do: "wait", for: "approval" }, { name: "done", do: "code", run: "42", output: "answer" }] } as const; + writeFileSync(join(dir, "flows/approve.json"), JSON.stringify(flow)); + const server = startFlowServer({ runner, serve: { port: 0 }, workspace: dir, controlToken: () => "test-token", logger: { info() {}, warn() {}, error() {} } }); + await once(server, "listening"); + t.after(async () => { server.closeAllConnections(); await new Promise(r => server.close(() => r())); await runner.close(); rmSync(dir, { recursive: true, force: true }); }); + const base = `http://127.0.0.1:${(server.address() as any).port}/flows/runs`; + const request = (p: string, options: RequestInit = {}) => fetch(base + p, { ...options, headers: { Authorization: "Bearer test-token", "X-ClawFlow-Backend": backend, ...options.headers } }); + assert.equal((await fetch(base)).status, 401); + assert.equal((await request("", { headers: { "X-ClawFlow-Backend": "wrong" } })).status, 503); + const store = runner.getStore(); + await store.create("root_%", "example", {}); + for (let n = 0; n < 23; n++) await store.create(`root_%:loop:items:${n}`, "child", { n }); + await store.create("root_%:loop:other:0", "child", {}); + // Case-insensitive filesystems cannot retain this distinct legacy JSON ID. + if (backend !== "json") await store.create("root_%:loop:ITEMS:0", "child", {}); + await store.create("root_AB:loop:items:0", "unrelated", {}); + const seen: string[] = []; + let cursor: string | undefined; + do { + const p = await store.listSubRuns("root_%", { node: "items", limit: 4, cursor }); + assert.equal(p.total, 23); + seen.push(...p.runs.map(x => x.instanceId)); cursor = p.next_cursor ?? undefined; + } while (cursor); + assert.equal(new Set(seen).size, 23); + const child = await (await request("/" + encodeURIComponent("root_%:loop:items:22"))).json() as any; + assert.equal(child.run.state.n, 22); + const detail = await (await request("/root_%25?include_subruns=1&node=items")).json() as any; + assert.equal(detail.run.subFlows.items.length, 20); + assert.equal(detail.run.subFlows.items[0].iteration, 3); + assert.equal(detail.run.subFlowsTruncated.items.total, 23); + assert.equal((await request("?limit=101")).status, 400); + assert.equal((await request("/missing")).status, 404); + for (const approved of [true, false]) { + const paused = await runner.run(flow as any, {}); + const res = await request(`/${paused.instanceId}/resume`, { method: "POST", body: JSON.stringify({ approved }) }); + assert.equal(res.status, approved ? 202 : 200); + for (let n = 0; n < 100 && (await store.get(paused.instanceId))!.status === "running"; n++) await new Promise(r => setTimeout(r, 5)); + assert.equal((await store.get(paused.instanceId))!.status, approved ? "completed" : "cancelled"); + assert.equal((await runner.listApprovals()).length, 0); + assert.equal((await request(`/${paused.instanceId}/resume`, { method: "POST", body: '{"approved":true}' })).status, 409); + } + const get = store.get.bind(store); + store.get = () => { throw new Error("db secret must never appear"); }; + const failed = await request("/root_%25"); + assert.equal(failed.status, 503); + assert.ok(!(await failed.text()).includes("secret")); + store.get = get; + }); + }); +} From 392a6e50b148fe1958da0d99207c1342aa3e534d Mon Sep 17 00:00:00 2001 From: pallaoro Date: Sun, 8 Nov 2026 16:41:44 +0100 Subject: [PATCH 3/4] Require actual instance identity for managed approval actions --- src/core/control.ts | 2 +- tests/control.test.ts | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/src/core/control.ts b/src/core/control.ts index d7e1890..eba72c8 100644 --- a/src/core/control.ts +++ b/src/core/control.ts @@ -75,7 +75,7 @@ export async function controlRequest( catch { json(res, 400, { error: "Invalid JSON body" }); return true; } if (!body || typeof body.approved !== "boolean") { json(res, 400, { error: "approved must be boolean" }); return true; } const run = await store.get(id); - if (!run) { json(res, 404, { error: "Run not found", code: "run_not_found" }); return true; } + if (!run || run.instanceId !== id) { json(res, 404, { error: "Run not found", code: "run_not_found" }); return true; } if (run.status !== "paused" && run.status !== "waiting") { json(res, 409, { error: "Run is not paused", code: "run_not_paused" }); return true; } diff --git a/tests/control.test.ts b/tests/control.test.ts index 2461cb6..5a0f794 100644 --- a/tests/control.test.ts +++ b/tests/control.test.ts @@ -57,6 +57,10 @@ for (const backend of ["json", "sqlite", "postgres"] as const) { assert.equal((await runner.listApprovals()).length, 0); assert.equal((await request(`/${paused.instanceId}/resume`, { method: "POST", body: '{"approved":true}' })).status, 409); } + const alias = await runner.run(flow as any, {}, "alias!"); + assert.equal((await request("/alias_/resume", { method: "POST", body: '{"approved":false}' })).status, 404, + "a filename-safe alias must not cancel a different actual ID"); + await runner.resume(alias.instanceId, flow as any, false); const get = store.get.bind(store); store.get = () => { throw new Error("db secret must never appear"); }; const failed = await request("/root_%25"); From 8f5ac0ecee43e7684ea3315ca03aeeb6aa7c4731 Mon Sep 17 00:00:00 2001 From: pallaoro Date: Sun, 8 Nov 2026 16:59:19 +0100 Subject: [PATCH 4/4] Expose oldest matching checkpoint for running-flow diagnostics --- README.md | 6 +++++- src/core/control.ts | 2 +- src/core/runs.ts | 9 ++++++++- src/core/sql-store.ts | 5 ++++- tests/control.test.ts | 14 ++++++++++++++ 5 files changed, 32 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 20a4b15..3c492ec 100644 --- a/README.md +++ b/README.md @@ -669,7 +669,11 @@ update cursor, revision, complete inventory guarantee, or deletion tombstones. When the HTTP listener is enabled, authenticated management routes use the same configured runner: -- `GET /flows/runs?limit=20&cursor=...&flow=...&status=...` +- `GET /flows/runs?limit=20&cursor=...&flow=...&status=...`. + Optional `include_oldest=1` adds `oldest: { instanceId, updatedAt } | null` + across all matching runs, before cursor filtering. The store option is + `includeOldest: true`. SQL count, oldest checkpoint and page share one + transaction; this identifies old checkpoints without declaring runs dead. - `GET /flows/runs/:instanceId` for the complete record; optional `path`, `offset`, `limit`, `max_chars` use `readRunValue`. URL-encode actual instance IDs. - `GET /flows/runs/:instanceId/subruns?limit=20&node=...` for descendant summaries. diff --git a/src/core/control.ts b/src/core/control.ts index eba72c8..a0af0ce 100644 --- a/src/core/control.ts +++ b/src/core/control.ts @@ -44,7 +44,7 @@ export async function controlRequest( try { if (req.method === "GET" && (!id || (parts.length === 2 && parts[1] === "subruns"))) { const opts = { limit: qs.has("limit") ? Number(qs.get("limit")) : undefined, cursor: qs.get("cursor") || undefined, - flow: qs.get("flow") || undefined, status: qs.get("status") || undefined, node: qs.get("node") || undefined }; + flow: qs.get("flow") || undefined, status: qs.get("status") || undefined, node: qs.get("node") || undefined, includeOldest: qs.get("include_oldest") === "1" }; try { parseRunPage(opts); } catch (err) { json(res, 400, { error: (err as Error).message }); return true; } const page = id ? await store.listSubRuns(id, opts) : await store.listRuns(opts); json(res, 200, { ok: true, ...page }); diff --git a/src/core/runs.ts b/src/core/runs.ts index e62d7d3..aacd682 100644 --- a/src/core/runs.ts +++ b/src/core/runs.ts @@ -44,6 +44,8 @@ export type RunRow = RunIndexEntry; export interface RunPage { runs: RunRow[]; + /** Oldest updated matching record, before cursor filtering; present only when requested. */ + oldest?: { instanceId: string; updatedAt: string } | null; /** Runs matching the filters, across all pages. */ total: number; /** Pass back as `cursor` for the next page; null on the last page. */ @@ -58,6 +60,8 @@ export interface ListRunsOptions { /** Only runs of this flow (the definition's `flow` name). */ flow?: string; status?: string; + /** Include the oldest checkpoint identity across all matching records. */ + includeOldest?: boolean; } /** @@ -96,7 +100,10 @@ export function pageRunRows(rows: RunRow[], opts: ListRunsOptions = {}): RunPage const rest = after ? rows.filter((r) => newestFirst(r, after) > 0) : rows; const page = rest.slice(0, limit); const last = page[page.length - 1]; + const oldest = opts.includeOldest ? rows.reduce((best, row) => + !best || row.updatedAt < best.updatedAt || (row.updatedAt === best.updatedAt && row.instanceId < best.instanceId) ? row : best, undefined) : undefined; return { + ...(opts.includeOldest ? { oldest: oldest ? { instanceId: oldest.instanceId, updatedAt: oldest.updatedAt } : null } : {}), runs: page, total: rows.length, next_cursor: rest.length > limit && last ? encodeCursor(last) : null, @@ -108,7 +115,7 @@ export interface ListSubRunsOptions extends ListRunsOptions { node?: string } export function listSubRuns(stateDir: string, parentId: string, opts: ListSubRunsOptions = {}): RunPage { parseRunPage(opts); const rows: RunRow[] = []; - if (!fs.existsSync(stateDir)) return { runs: [], total: 0, next_cursor: null }; + if (!fs.existsSync(stateDir)) return pageRunRows([], opts); const prefix = `${parentId}:`; const filePrefix = prefix.replace(/[^a-zA-Z0-9_-]/g, "_"); for (const file of fs.readdirSync(stateDir)) { diff --git a/src/core/sql-store.ts b/src/core/sql-store.ts index cfa9cec..323281f 100644 --- a/src/core/sql-store.ts +++ b/src/core/sql-store.ts @@ -167,6 +167,9 @@ export class SqlStateStore implements ExecutionStore { if (opts.flow) { params.push(opts.flow); where += ` AND flow_name=$${params.length}`; } if (opts.status) { params.push(opts.status); where += ` AND status=$${params.length}`; } const count = await q(`SELECT COUNT(*) AS total FROM clawflow_runs WHERE ${where}`, params); + const oldestRows = opts.includeOldest + ? await q(`SELECT id,updated_at FROM clawflow_runs WHERE ${where} ORDER BY updated_at ASC,id ASC LIMIT 1`, params) : []; + const oldest = oldestRows[0] ? { instanceId: oldestRows[0].id as string, updatedAt: oldestRows[0].updated_at as string } : null; if (after) { params.push(after.createdAt, after.instanceId); where += ` AND (created_at < $${params.length - 1} OR (created_at = $${params.length - 1} AND id < $${params.length}))`; @@ -174,7 +177,7 @@ export class SqlStateStore implements ExecutionStore { params.push(limit + 1); const rows = await q(`SELECT summary FROM clawflow_runs WHERE ${where} ORDER BY created_at DESC,id DESC LIMIT $${params.length}`, params); const runs = rows.slice(0, limit).map((r) => JSON.parse(r.summary as string)); - return { runs, total: Number(count[0].total), next_cursor: rows.length > limit ? encodeCursor(runs[runs.length - 1]) : null }; + return { ...(opts.includeOldest ? { oldest } : {}), runs, total: Number(count[0].total), next_cursor: rows.length > limit ? encodeCursor(runs[runs.length - 1]) : null }; }, false); } diff --git a/tests/control.test.ts b/tests/control.test.ts index 5a0f794..46ef6d9 100644 --- a/tests/control.test.ts +++ b/tests/control.test.ts @@ -32,6 +32,20 @@ for (const backend of ["json", "sqlite", "postgres"] as const) { // Case-insensitive filesystems cannot retain this distinct legacy JSON ID. if (backend !== "json") await store.create("root_%:loop:ITEMS:0", "child", {}); await store.create("root_AB:loop:items:0", "unrelated", {}); + const old = { ...(await store.get("root_%"))!, instanceId: "oldest-checkpoint", + createdAt: "2001-01-01T00:00:00.000Z", updatedAt: "2002-01-01T00:00:00.000Z" }; + const childOld = { ...old, instanceId: "root_%:loop:other:999", updatedAt: "1900-01-01T00:00:00.000Z" }; + if (backend === "json") { + for (const record of [old, childOld]) writeFileSync(join(dir, record.instanceId.replace(/[^a-zA-Z0-9_-]/g, "_") + ".json"), JSON.stringify(record)); + } else await (store as any).importRecords([old, childOld], []); + const active = await (await request("?status=running&limit=1&include_oldest=1")).json() as any; + assert.equal(active.total, 2, "children are excluded from the root count"); + assert.equal(active.oldest.instanceId, "oldest-checkpoint"); + assert.notEqual(active.runs[0].instanceId, active.oldest.instanceId); + assert.equal(active.oldest.updatedAt, "2002-01-01T00:00:00.000Z"); + assert.equal(active.runs.length, 1); + const empty = await (await request("?status=paused&include_oldest=1")).json() as any; + assert.equal(empty.oldest, null); const seen: string[] = []; let cursor: string | undefined; do {