From 651e477780406d0d084ca484f5cbe4910ed975d6 Mon Sep 17 00:00:00 2001 From: pallaoro Date: Fri, 25 Sep 2026 16:37:50 +0200 Subject: [PATCH] Bounded run reads: paged run list, run summaries, path reads, retention Every run is one JSON file in the state directory and none was ever deleted, so the run list grows with every trigger that fires, and one record carries each node's output more than once (state, completedNodes, trace). Returned whole, either can overflow what an agent reads in one tool result; hosts cut larger results and keep only the start. - core/runs.ts, exported for other hosts like manage.ts: - listRuns: newest first, a page at a time, on a (createdAt, instanceId) cursor so runs that start while a caller pages never shift a page and runs created in the same millisecond are neither skipped nor repeated. Loop/branch instance files are skipped by name, before parsing. - summarizeRun / runView: the run whole when it fits, else its status, error, each state key with its size, and the trace without outputs. - readRunValue: one value by dot path. Arrays are paged by item and a page shrinks until it fits, strings are paged by character, and an object too big to return comes back as its outline with paths to pass back. - pruneRuns: deletes finished runs (completed, failed, cancelled) with their sub-flow files after the retention period. Running, paused and waiting runs are never deleted; resume reads their record. - flow_status lists runs a page at a time (limit, cursor, flow, filter) and reads one value of a run with path/offset/limit. flow_run and flow_resume return the summary once a run passes 16,000 characters. - Retention runs daily in the serving process, first a minute after start. runRetentionDays in the plugin config, default 30; 0 keeps every run. --- README.md | 6 +- openclaw.plugin.json | 4 + src/core/runs.ts | 456 +++++++++++++++++++++++++++++++++++++++++++ src/core/store.ts | 3 +- src/core/types.ts | 6 + src/index.ts | 19 ++ src/plugin/index.ts | 144 +++++++++++--- tests/runs.test.ts | 259 ++++++++++++++++++++++++ 8 files changed, 863 insertions(+), 34 deletions(-) create mode 100644 src/core/runs.ts create mode 100644 tests/runs.test.ts diff --git a/README.md b/README.md index 8584576..c02ffdd 100644 --- a/README.md +++ b/README.md @@ -570,6 +570,10 @@ Every flow run gets a unique `instanceId`. The runner persists state after every - 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. +**Retention.** Finished runs (completed, failed, cancelled) are deleted 30 days after their last update, together with their loop and branch instance files, by a daily sweep in the gateway process. Set `runRetentionDays` in the plugin config to change it; `0` keeps every run. Running, paused and waiting runs are never deleted. + +**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`, `pruneRuns`) are exported for other hosts. + 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. --- @@ -590,7 +594,7 @@ Eleven tools registered in OpenClaw: | `flow_run` | Execute a flow (uses latest published version by default) | | `flow_resume` | Resume after an approval gate | | `flow_send_event` | Push an event into a waiting flow | -| `flow_status` | Inspect any running or completed instance | +| `flow_status` | List instances a page at a time, or inspect one (a summary when large, one value by `path`) | | `flow_list` | List all flows with metadata, expected inputs, and version info | | `flow_read` | Read a flow definition (draft or specific version), inspect single nodes | | `flow_publish` | Publish current draft as a new numbered version | diff --git a/openclaw.plugin.json b/openclaw.plugin.json index 6ce145f..d5be03d 100644 --- a/openclaw.plugin.json +++ b/openclaw.plugin.json @@ -47,6 +47,10 @@ "stateDir": { "type": "string" }, + "runRetentionDays": { + "type": "number", + "minimum": 0 + }, "maxNodeDurationMs": { "type": "number" }, diff --git a/src/core/runs.ts b/src/core/runs.ts new file mode 100644 index 0000000..4177d50 --- /dev/null +++ b/src/core/runs.ts @@ -0,0 +1,456 @@ +import * as fs from "fs"; +import * as path from "path"; +import type { InstanceRecord } from "./store.js"; +import type { FlowResult, TraceEntry } from "./types.js"; + +// ---- Run reads + retention ------------------------------------------------------- +// How callers see stored runs: a paged list, a summary of one run, and one value +// of a run read by path, plus the retention sweep that deletes old finished runs. +// +// Used by the plugin's flow_status / flow_run / flow_resume tools in-process, and +// by out-of-process platform callers (e.g. the Clawnify hook server) that import +// these functions directly from dist, like manage.ts, so the run record format is +// read in exactly one place. +// +// Every run is one JSON file in the state directory, so the run list grows with +// every trigger that fires, and one record carries each node's output more than +// once (state, completedNodes, trace). Returned whole, either overflows what an +// agent can read in one tool result. So every read here is bounded: the list by +// a page, a record by a summary once it is too big, a value by a page of items or +// characters. + +/** + * Characters of one tool result an agent on the box reliably sees whole. Hosts + * cut larger tool results, keeping the start, so the on-box tools return a + * summary instead of letting the end of a record disappear. + */ +export const ON_BOX_RESULT_CHARS = 16_000; + +const DEFAULT_PAGE = 20; +const MAX_PAGE = 100; +const SCALAR_MAX_CHARS = 120; +const OUTLINE_MAX_DEPTH = 3; +const OUTLINE_MAX_ENTRIES = 60; +const DAY_MS = 86_400_000; + +/** Sub-flow instance files: `__[_].json`. */ +const SUB_FLOW_FILE = /_(loop|condition|branch|parallel)_/; + +// ---- List ------------------------------------------------------------------------ + +export interface RunRow { + instanceId: string; + flowName: string; + status: InstanceRecord["status"]; + createdAt: string; + updatedAt: string; + error?: string; + resumeToken?: string; + waitingFor?: InstanceRecord["waitingFor"]; + traceCount: number; + nodeCount: number; +} + +export interface RunPage { + runs: RunRow[]; + /** Runs matching the filters, across all pages. */ + total: number; + /** Pass back as `cursor` for the next page; null on the last page. */ + next_cursor: string | null; +} + +export interface ListRunsOptions { + /** Runs per page, 1..100. Default 20. */ + limit?: number; + /** `next_cursor` from the previous page. */ + cursor?: string; + /** Only runs of this flow (the definition's `flow` name). */ + flow?: string; + status?: string; +} + +/** + * Runs newest first, a page at a time. The cursor is keyed on + * (createdAt, instanceId), not an offset, so runs that start while a caller + * 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 { + 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}`); + } + const after = opts.cursor ? decodeCursor(opts.cursor) : null; + if (opts.cursor && !after) { + throw new Error("cursor is not one this list returned; start again without it"); + } + + const rows: RunRow[] = []; + for (const record of readTopLevelRecords(stateDir)) { + if (opts.flow && record.flowName !== opts.flow) continue; + if (opts.status && record.status !== opts.status) continue; + rows.push(toRow(record)); + } + rows.sort(newestFirst); + const rest = after ? rows.filter((r) => newestFirst(r, after) > 0) : rows; + const page = rest.slice(0, limit); + const last = page[page.length - 1]; + return { + runs: page, + total: rows.length, + next_cursor: rest.length > limit && last ? encodeCursor(last) : null, + }; +} + +function toRow(r: InstanceRecord): RunRow { + return { + instanceId: r.instanceId, + flowName: r.flowName, + status: r.status, + createdAt: r.createdAt, + updatedAt: r.updatedAt, + ...(r.error ? { error: r.error } : {}), + ...(r.resumeToken ? { resumeToken: r.resumeToken } : {}), + ...(r.waitingFor ? { waitingFor: r.waitingFor } : {}), + traceCount: Array.isArray(r.trace) ? r.trace.length : 0, + nodeCount: r.completedNodes ? Object.keys(r.completedNodes).length : 0, + }; +} + +type Keyed = { createdAt: string; instanceId: string }; + +/** (createdAt, instanceId) descending. Positive when `a` sorts after `b`. */ +function newestFirst(a: Keyed, b: Keyed): number { + if (a.createdAt !== b.createdAt) return a.createdAt < b.createdAt ? 1 : -1; + if (a.instanceId !== b.instanceId) return a.instanceId < b.instanceId ? 1 : -1; + return 0; +} + +function encodeCursor(k: Keyed): string { + return btoa(encodeURIComponent(JSON.stringify([k.createdAt, k.instanceId]))); +} + +function decodeCursor(cursor: string): Keyed | null { + try { + const v = JSON.parse(decodeURIComponent(atob(cursor))) as unknown; + return Array.isArray(v) && v.length === 2 && typeof v[0] === "string" && typeof v[1] === "string" + ? { createdAt: v[0], instanceId: v[1] } + : null; + } catch { + return null; + } +} + +/** Top-level run records; sub-flow files are skipped by name, before parsing. */ +function* readTopLevelRecords(stateDir: string): Generator { + if (!fs.existsSync(stateDir)) return; + for (const name of fs.readdirSync(stateDir)) { + if (!isTopLevelRunFile(name)) continue; + let record: InstanceRecord; + try { + record = JSON.parse(fs.readFileSync(path.join(stateDir, name), "utf8")) as InstanceRecord; + } catch { + continue; + } + if (!record || typeof record.instanceId !== "string" || record.instanceId.includes(":")) continue; + yield record; + } +} + +function isTopLevelRunFile(name: string): boolean { + return name.endsWith(".json") && !name.startsWith("_") && !SUB_FLOW_FILE.test(name); +} + +// ---- Summary --------------------------------------------------------------------- + +export interface ValueOutline { + path: string; + type: "object" | "array" | "string" | "number" | "boolean" | "null"; + /** Serialized size, for containers and long strings. */ + chars?: number; + /** Item count for arrays, key count for objects. */ + length?: number; + /** Short scalars are shown whole. */ + value?: string | number | boolean | null; +} + +export type TraceSummary = Omit & { outputChars?: number }; + +export interface RunSummary { + summary: true; + instanceId: string; + flowName: string; + status: string; + createdAt?: string; + updatedAt?: string; + error?: string; + pausedAt?: string; + resumeToken?: string; + waitingFor?: unknown; + /** One entry per state key (node outputs, inputs, env), with its size. */ + state: ValueOutline[]; + /** Every trace entry, with the output's size in place of the output. */ + trace: TraceSummary[]; + /** Characters of the full record this summarizes. */ + fullChars: number; + read_more: string; +} + +/** + * A run too big to return whole: status, error, where it paused, each state + * key with its size, and the trace without outputs. Accepts a stored record or + * the FlowResult flow_run returns. + */ +export function summarizeRun(run: InstanceRecord | FlowResult, fullChars?: number): RunSummary { + const r = run as Partial & Partial; + const state = (r.state ?? {}) as Record; + return { + summary: true, + instanceId: r.instanceId ?? "", + flowName: r.flowName ?? "", + status: r.status ?? "", + ...(r.createdAt ? { createdAt: r.createdAt } : {}), + ...(r.updatedAt ? { updatedAt: r.updatedAt } : {}), + ...(r.error ? { error: r.error } : {}), + ...(r.pausedAt ? { pausedAt: r.pausedAt } : {}), + ...(r.resumeToken ? { resumeToken: r.resumeToken } : {}), + ...(r.waitingFor ? { waitingFor: r.waitingFor } : {}), + state: Object.entries(state).map(([key, value]) => describe(`state.${key}`, value)), + trace: (r.trace ?? []).map((entry) => { + const { output, ...rest } = entry; + return output === undefined ? rest : { ...rest, outputChars: serializedLength(output) }; + }), + fullChars: fullChars ?? serializedLength(run), + read_more: + "Read one value with flow_status { instanceId, path }, e.g. path \"state.\" or " + + "\"trace..output\"; arrays and long strings come back a page at a time (offset, limit).", + }; +} + +/** + * The run whole when it fits in `maxChars` serialized, else its summary. What + * every tool that returns a run hands back. + */ +export function runView(run: InstanceRecord | FlowResult, maxChars = ON_BOX_RESULT_CHARS): unknown { + const chars = serializedLength(run); + return chars <= maxChars ? run : summarizeRun(run, chars); +} + +// ---- One value by path ----------------------------------------------------------- + +export interface ReadValueOptions { + /** Where a page starts: an item index for arrays, a character index for strings. */ + offset?: number; + /** Items (arrays, default 20, max 100) or characters (strings) per page. */ + limit?: number; + /** The largest serialized result to return; bigger objects come back as an outline. */ + maxChars?: number; +} + +/** + * One value of a run by dot path (`state.leads`, `trace.3.output`, + * `completedNodes.fetch`). Arrays come back a page of items at a time, strings a + * page of characters, and an object too big to return is replaced by its + * outline, with paths the caller can pass back. Keys containing a dot are not + * addressable. + */ +export function readRunValue( + run: InstanceRecord | FlowResult, + valuePath: string, + opts: ReadValueOptions = {}, +): Record { + const maxChars = opts.maxChars ?? ON_BOX_RESULT_CHARS; + const segments = valuePath.split(".").filter((s) => s !== ""); + if (!segments.length) throw new Error("path is empty"); + + let node: unknown = run; + for (let i = 0; i < segments.length; i++) { + const seg = segments[i]; + const at = segments.slice(0, i).join(".") || "the run"; + if (Array.isArray(node)) { + const index = Number(seg); + if (!Number.isInteger(index) || index < 0 || index >= node.length) { + throw new Error(`"${seg}" is not an index of ${at}, an array of ${node.length}`); + } + node = node[index]; + } else if (isRecord(node)) { + if (!(seg in node)) { + const keys = Object.keys(node); + throw new Error( + `"${seg}" is not a field of ${at}. Fields: ${keys.slice(0, 40).join(", ")}` + + (keys.length > 40 ? `, … (${keys.length} total)` : ""), + ); + } + node = node[seg]; + } else { + throw new Error(`${at} is ${node === null ? "null" : `a ${typeof node}`}, not an object or array`); + } + } + + const offset = opts.offset ?? 0; + if (!Number.isInteger(offset) || offset < 0) throw new Error("offset must be a whole number, 0 or more"); + + if (Array.isArray(node)) { + const limit = opts.limit ?? DEFAULT_PAGE; + if (!Number.isInteger(limit) || limit < 1 || limit > MAX_PAGE) { + throw new Error(`limit for an array must be between 1 and ${MAX_PAGE} items`); + } + // Fewer items when a page would not fit, down to one. + let end = Math.min(offset + limit, node.length); + while (end - offset > 1 && serializedLength(node.slice(offset, end)) > maxChars) { + end = offset + Math.ceil((end - offset) / 2); + } + if (end - offset === 1 && isRecord(node[offset]) && serializedLength(node[offset]) > maxChars) { + return withheldObject(`${valuePath}.${offset}`, node[offset] as Record, maxChars); + } + return { + path: valuePath, + type: "array", + total: node.length, + offset, + items: node.slice(offset, end), + next_offset: end < node.length ? end : null, + }; + } + if (typeof node === "string") { + const pageChars = Math.floor(maxChars * 0.75); + const limit = opts.limit ?? pageChars; + if (!Number.isInteger(limit) || limit < 1 || limit > maxChars) { + throw new Error(`limit for a string must be between 1 and ${maxChars} characters`); + } + const end = Math.min(offset + limit, node.length); + return { + path: valuePath, + type: "string", + total_chars: node.length, + offset, + text: node.slice(offset, end), + next_offset: end < node.length ? end : null, + }; + } + if (isRecord(node) && serializedLength(node) > maxChars) { + return withheldObject(valuePath, node, maxChars); + } + return { path: valuePath, type: node === null ? "null" : typeof node, value: node }; +} + +function withheldObject(at: string, node: Record, maxChars: number): Record { + const { entries, omitted } = outline(node, at); + return { + path: at, + type: "object", + withheld: true, + reason: `The object at "${at}" is over ${maxChars} characters; read one of its fields by a deeper path.`, + outline: entries, + ...(omitted ? { outline_omitted: omitted } : {}), + }; +} + +/** Every field down to three levels, shallowest first, capped at 60 entries. */ +function outline(value: Record, prefix: string): { entries: ValueOutline[]; omitted: number } { + const entries: ValueOutline[] = []; + let omitted = 0; + let level = Object.entries(value).map(([k, v]) => ({ path: `${prefix}.${k}`, node: v })); + for (let depth = 1; depth <= OUTLINE_MAX_DEPTH && level.length; depth++) { + const next: Array<{ path: string; node: unknown }> = []; + for (const { path: p, node } of level) { + if (entries.length >= OUTLINE_MAX_ENTRIES) { + omitted++; + continue; + } + entries.push(describe(p, node)); + if (isRecord(node)) next.push(...Object.entries(node).map(([k, v]) => ({ path: `${p}.${k}`, node: v }))); + } + level = next; + } + omitted += level.length; + entries.sort((a, b) => (a.path < b.path ? -1 : a.path > b.path ? 1 : 0)); + return { entries, omitted }; +} + +function describe(p: string, node: unknown): ValueOutline { + if (node === null || node === undefined) return { path: p, type: "null", value: null }; + if (Array.isArray(node)) return { path: p, type: "array", length: node.length, chars: serializedLength(node) }; + if (typeof node === "object") { + return { path: p, type: "object", length: Object.keys(node).length, chars: serializedLength(node) }; + } + if (typeof node === "string") { + return node.length <= SCALAR_MAX_CHARS + ? { path: p, type: "string", value: node } + : { path: p, type: "string", chars: node.length }; + } + if (typeof node === "number" || typeof node === "boolean") { + return { path: p, type: typeof node as "number" | "boolean", value: node }; + } + return { path: p, type: "null", value: null }; +} + +// ---- Retention ------------------------------------------------------------------- + +export const DEFAULT_RUN_RETENTION_DAYS = 30; + +const FINISHED = new Set(["completed", "failed", "cancelled"]); + +export interface PruneResult { + /** Top-level runs deleted. */ + runs: number; + /** Files deleted, including sub-flow instance files. */ + files: number; +} + +/** + * Delete finished runs (completed, failed, cancelled) last updated more than + * `retentionDays` ago, with their sub-flow files. Runs that are running, paused + * or waiting are never deleted: a paused or waiting run is resumed from its + * record. `retentionDays` 0 deletes nothing. + */ +export function pruneRuns(stateDir: string, retentionDays: number, now = Date.now()): PruneResult { + const result: PruneResult = { runs: 0, files: 0 }; + if (!(retentionDays > 0) || !fs.existsSync(stateDir)) return result; + const cutoff = now - retentionDays * DAY_MS; + const names = fs.readdirSync(stateDir); + const expired = new Set(); + + for (const name of names) { + if (!isTopLevelRunFile(name)) continue; + let record: Partial; + try { + record = JSON.parse(fs.readFileSync(path.join(stateDir, name), "utf8")) as Partial; + } catch { + continue; + } + const last = Date.parse(record.updatedAt ?? record.createdAt ?? ""); + if (!FINISHED.has(String(record.status)) || !(last < cutoff)) continue; + expired.add(name.slice(0, -".json".length)); + } + + for (const name of names) { + if (!name.endsWith(".json") || name.startsWith("_")) continue; + const base = name.slice(0, -".json".length); + const sub = SUB_FLOW_FILE.exec(base); + const owner = sub ? base.slice(0, sub.index) : base; + if (!expired.has(owner)) continue; + try { + fs.unlinkSync(path.join(stateDir, name)); + result.files++; + if (!sub) result.runs++; + } catch { + // Already gone, or not ours to delete: the next sweep retries. + } + } + return result; +} + +// ---- Shared ---------------------------------------------------------------------- + +function serializedLength(v: unknown): number { + try { + return JSON.stringify(v)?.length ?? 0; + } catch { + return 0; + } +} + +function isRecord(v: unknown): v is Record { + return typeof v === "object" && v !== null && !Array.isArray(v); +} diff --git a/src/core/store.ts b/src/core/store.ts index 4924a56..6411442 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -31,7 +31,8 @@ export interface InstanceRecord { } export class StateStore { - private dir: string; + /** The state directory: one `.json` per run (see runs.ts for reads). */ + readonly dir: string; constructor(stateDir?: string) { this.dir = diff --git a/src/core/types.ts b/src/core/types.ts index 01a8acf..561fc86 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -531,6 +531,12 @@ export interface PluginConfig { memoryDir?: string; maxNodeDurationMs?: number; stateDir?: string; // where to persist flow state across restarts + /** + * Days to keep finished runs (completed, failed, cancelled) in `stateDir`. + * Older ones are deleted by a daily sweep in the serving process; running, + * paused and waiting runs are never deleted. Default 30; 0 keeps every run. + */ + runRetentionDays?: number; /** Override for AI inference — used by tests and embedders */ inferenceFn?: InferenceFn; /** OpenClaw agent ID for do:agent nodes (e.g. "ops"). Falls back to --local if unset. */ diff --git a/src/index.ts b/src/index.ts index 1bd0e3f..d3ed76b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -72,3 +72,22 @@ export { resolveRunnableFlow, } from "./core/manage.js"; export type { PublishResult } from "./core/manage.js"; +export { + listRuns, + summarizeRun, + runView, + readRunValue, + pruneRuns, + ON_BOX_RESULT_CHARS, + DEFAULT_RUN_RETENTION_DAYS, +} from "./core/runs.js"; +export type { + RunRow, + RunPage, + ListRunsOptions, + RunSummary, + TraceSummary, + ValueOutline, + ReadValueOptions, + PruneResult, +} from "./core/runs.js"; diff --git a/src/plugin/index.ts b/src/plugin/index.ts index c479cfa..e831cfb 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -14,6 +14,13 @@ import { readVersion as readVersionCore, resolveFlowFile as resolveFlowFileCore, } from "../core/manage.js"; +import { + DEFAULT_RUN_RETENTION_DAYS, + listRuns, + pruneRuns, + readRunValue, + runView, +} from "../core/runs.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 @@ -30,6 +37,38 @@ export function activeTriggerScheduler(): TriggerScheduler | null { return activeScheduler; } +// Run retention: a daily sweep of finished runs older than runRetentionDays, +// armed only by the serving process (one per box). Module scope for the same +// reason as the scheduler: register() runs again on every reload, and each call +// must retire the previous sweep before arming its own. +let retentionTimers: Array> = []; + +function armRetention( + stateDir: string, + retentionDays: number, + logger?: { info: (msg: string) => void; warn: (msg: string) => void }, +): void { + for (const t of retentionTimers) clearTimeout(t); + retentionTimers = []; + if (!(retentionDays > 0)) return; + const sweep = () => { + try { + const { runs, files } = pruneRuns(stateDir, retentionDays); + if (files) { + logger?.info(`clawflow: retention deleted ${runs} runs (${files} files) older than ${retentionDays} days`); + } + } catch (err) { + logger?.warn(`clawflow: retention sweep failed: ${err instanceof Error ? err.message : String(err)}`); + } + }; + // First sweep a minute after start, off the startup path; then daily. + const first = setTimeout(sweep, 60_000); + const daily = setInterval(sweep, 24 * 60 * 60 * 1000); + first.unref?.(); + daily.unref?.(); + retentionTimers = [first, daily]; +} + /** * Whether this process should own the box's scheduler and flow server. * @@ -160,6 +199,7 @@ function register(api: PluginApi) { activeScheduler = scheduler; activeTriggerStore = triggerStore; scheduler.start(); + armRetention(store.dir, pluginCfg.runRetentionDays ?? DEFAULT_RUN_RETENTION_DAYS, api.logger); } // ---- Flow server (optional) --------------------------------------------------- @@ -752,9 +792,11 @@ Set version to run a specific published version.`, try { const result = await runner.run(flowDef, params.input ?? {}); const out = { ...result, _source: source }; + // Whole when it fits, else a summary the agent reads on from with + // flow_status { instanceId, path } (see core/runs.ts). return { content: [ - { type: "text", text: JSON.stringify(out, null, 2) }, + { type: "text", text: JSON.stringify(runView(out)) }, ], details: out, }; @@ -823,7 +865,7 @@ You must pass the original flow definition back so the runner can continue.`, ); return { content: [ - { type: "text", text: JSON.stringify(result, null, 2) }, + { type: "text", text: JSON.stringify(runView(result)) }, ], details: result, }; @@ -909,7 +951,9 @@ payload is passed as the output of the wait node and into flow state.`, api.registerTool( { name: "flow_status", - description: `Get the status and state of a flow instance, or list all instances. + description: `Get the status and state of a flow instance, or list instances. +Without instanceId: runs newest first, 20 per page (limit, then cursor = the returned next_cursor), filtered by flow and status. +With instanceId: the run whole when it is small, else a summary (status, error, each state key with its size, the trace without outputs). Read any one value with path, e.g. "state.leads" or "trace.3.output"; arrays and long strings come back a page at a time (offset, limit). Status values: running | completed | paused | waiting | failed | cancelled`, parameters: { @@ -917,7 +961,30 @@ Status values: running | completed | paused | waiting | failed | cancelled`, properties: { instanceId: { type: "string", - description: "Specific instance to inspect. Omit to list all.", + description: "Specific instance to inspect. Omit to list instances.", + }, + path: { + type: "string", + description: + "With instanceId: one value of the run as a dot path, e.g. \"state.leads\", \"trace.3.output\", \"completedNodes.fetch\".", + }, + offset: { + type: "number", + description: + "With path: where the page starts (an item index for an array, a character index for a string). Use next_offset from the previous page.", + }, + limit: { + type: "number", + description: + "Without instanceId: runs per page (default 20, max 100). With path: items (array) or characters (string) per page.", + }, + cursor: { + type: "string", + description: "Without instanceId: the next_cursor from the previous page.", + }, + flow: { + type: "string", + description: "Without instanceId: only runs of this flow.", }, filter: { type: "string", @@ -929,40 +996,53 @@ Status values: running | completed | paused | waiting | failed | cancelled`, async execute( _id: string, - params: { instanceId?: string; filter?: string }, + params: { + instanceId?: string; + path?: string; + offset?: number; + limit?: number; + cursor?: string; + flow?: string; + filter?: string; + }, ) { - if (params.instanceId) { - const record = store.get(params.instanceId); - if (!record) - return { - content: [ - { - type: "text", - text: `Instance not found: ${params.instanceId}`, - }, - ], - }; + const text = (value: unknown) => ({ + content: [{ type: "text", text: JSON.stringify(value) }], + }); + try { + if (params.instanceId) { + const record = store.get(params.instanceId); + if (!record) + return { + content: [ + { + type: "text", + text: `Instance not found: ${params.instanceId}`, + }, + ], + }; + if (params.path) { + return text(readRunValue(record, params.path, { offset: params.offset, limit: params.limit })); + } + return { ...text(runView(record)), details: record }; + } + const page = listRuns(store.dir, { + limit: params.limit, + cursor: params.cursor, + flow: params.flow, + status: params.filter, + }); + return { ...text(page), details: page }; + } catch (err) { return { content: [ - { type: "text", text: JSON.stringify(record, null, 2) }, + { + type: "text", + text: `Error: ${err instanceof Error ? err.message : String(err)}`, + }, ], - details: record, }; } - const records = store.list(params.filter); - const summary = records.map((r) => ({ - instanceId: r.instanceId, - flow: r.flowName, - status: r.status, - updatedAt: r.updatedAt, - waitingFor: r.waitingFor, - })); - return { - content: [ - { type: "text", text: JSON.stringify(summary, null, 2) }, - ], - details: summary, - }; }, }, { optional: true }, diff --git a/tests/runs.test.ts b/tests/runs.test.ts new file mode 100644 index 0000000..a2ba6e7 --- /dev/null +++ b/tests/runs.test.ts @@ -0,0 +1,259 @@ +import { describe, it, after } from "node:test"; +import assert from "node:assert/strict"; +import * as fs from "fs"; +import * as path from "path"; +import * as os from "os"; +import { randomUUID } from "crypto"; + +import plugin from "../src/plugin/index.js"; +import { listRuns, pruneRuns, readRunValue, runView, summarizeRun } from "../src/index.js"; +import type { RunPage, RunSummary } from "../src/index.js"; + +const roots: string[] = []; +after(() => { + for (const dir of roots) fs.rmSync(dir, { recursive: true, force: true }); +}); + +function stateDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "ocf-runs-")); + roots.push(dir); + return dir; +} + +interface Seed { + id?: string; + flow?: string; + status?: string; + createdAt: string; + updatedAt?: string; + state?: Record; + trace?: unknown[]; +} + +/** Write a stored run the way StateStore does: stateDir/.json. */ +function seed(dir: string, s: Seed): string { + const id = s.id ?? randomUUID(); + const record = { + instanceId: id, + flowName: s.flow ?? "digest", + status: s.status ?? "completed", + state: s.state ?? { inputs: {} }, + completedNodes: {}, + trace: s.trace ?? [], + createdAt: s.createdAt, + updatedAt: s.updatedAt ?? s.createdAt, + }; + fs.writeFileSync(path.join(dir, `${id}.json`), JSON.stringify(record)); + return id; +} + +const minute = (i: number) => new Date(Date.UTC(2026, 8, 25, 10, i)).toISOString(); + +describe("listRuns", () => { + it("returns the newest 20 by default, with the total and a cursor", () => { + const dir = stateDir(); + for (let i = 0; i < 45; i++) seed(dir, { createdAt: minute(i) }); + const page = listRuns(dir); + assert.equal(page.runs.length, 20); + assert.equal(page.runs[0].createdAt, minute(44)); + assert.equal(page.total, 45); + assert.equal(typeof page.next_cursor, "string"); + }); + + it("walks every run exactly once, including runs started in the same millisecond", () => { + const dir = stateDir(); + for (let i = 0; i < 30; i++) seed(dir, { createdAt: i >= 10 && i < 14 ? minute(10) : minute(i) }); + const seen: string[] = []; + let cursor: string | undefined; + for (let guard = 0; guard < 20; guard++) { + const page: RunPage = listRuns(dir, { limit: 7, cursor }); + seen.push(...page.runs.map((r) => r.instanceId)); + if (!page.next_cursor) break; + cursor = page.next_cursor; + } + assert.equal(seen.length, 30); + assert.equal(new Set(seen).size, 30); + }); + + it("filters by flow and status, and never lists sub-flow instances", () => { + const dir = stateDir(); + const parent = seed(dir, { flow: "digest", status: "failed", createdAt: minute(1) }); + seed(dir, { flow: "outreach", createdAt: minute(2) }); + // A loop iteration's instance file, as the runner writes it. + fs.writeFileSync( + path.join(dir, `${parent}_loop_each_0.json`), + JSON.stringify({ instanceId: `${parent}:loop:each:0`, flowName: "digest:loop:each", status: "completed", createdAt: minute(3) }), + ); + assert.deepEqual(listRuns(dir).runs.map((r) => r.instanceId).includes(parent), true); + assert.equal(listRuns(dir).total, 2); + assert.equal(listRuns(dir, { flow: "outreach" }).total, 1); + assert.equal(listRuns(dir, { status: "failed" }).runs[0].instanceId, parent); + }); + + it("rejects a page size out of range and a cursor it did not issue", () => { + const dir = stateDir(); + assert.throws(() => listRuns(dir, { limit: 0 })); + assert.throws(() => listRuns(dir, { limit: 101 })); + assert.throws(() => listRuns(dir, { cursor: "nope" })); + }); +}); + +function bigRun() { + const leads = Array.from({ length: 300 }, (_, i) => ({ id: i, name: `Lead ${i}`, note: "x".repeat(80) })); + return { + instanceId: "run-1", + flowName: "outreach", + status: "completed" as const, + state: { inputs: { max_pages: 2 }, leads }, + completedNodes: { fetch: leads }, + trace: [ + { node: "fetch", do: "http", status: "ok" as const, output: leads, durationMs: 40 }, + { node: "notify", do: "code", status: "ok" as const, output: "sent", durationMs: 1 }, + ], + createdAt: minute(1), + updatedAt: minute(2), + }; +} + +describe("runView / summarizeRun", () => { + it("returns a small run whole", () => { + const small = { ...bigRun(), state: { inputs: {} }, completedNodes: {}, trace: [] }; + assert.equal(runView(small), small); + }); + + it("summarizes a run too big to return: sizes instead of values, outputs dropped from the trace", () => { + const view = runView(bigRun()) as RunSummary; + assert.equal(view.summary, true); + assert.ok(JSON.stringify(view).length < 4_000); + const leads = view.state.find((e) => e.path === "state.leads"); + assert.equal(leads?.type, "array"); + assert.equal(leads?.length, 300); + assert.equal(view.trace[0].node, "fetch"); + assert.ok((view.trace[0].outputChars ?? 0) > 20_000); + assert.equal("output" in view.trace[0], false); + assert.ok(view.fullChars > 40_000); + }); + + it("summarizes the FlowResult flow_run returns as well as a stored record", () => { + const { completedNodes: _c, createdAt: _a, updatedAt: _u, ...result } = bigRun(); + const view = summarizeRun({ ...result, ok: true }); + assert.equal(view.instanceId, "run-1"); + assert.equal(view.status, "completed"); + }); +}); + +describe("readRunValue", () => { + it("pages an array, fewer items when a page would not fit", () => { + const run = bigRun(); + const first = readRunValue(run, "state.leads"); + assert.equal(first.total, 300); + assert.ok(JSON.stringify(first).length <= 16_000 + 200); + assert.ok((first.items as unknown[]).length >= 1); + assert.equal(first.next_offset, (first.items as unknown[]).length); + const last = readRunValue(run, "state.leads", { offset: 295, limit: 10 }); + assert.equal((last.items as unknown[]).length, 5); + assert.equal(last.next_offset, null); + + // 20 items of ~2K each would be ~40K: the page shrinks until it fits. + const wide = { ...run, state: { rows: Array.from({ length: 50 }, (_, i) => ({ i, body: "y".repeat(2_000) })) } }; + const page = readRunValue(wide, "state.rows"); + const count = (page.items as unknown[]).length; + assert.ok(count > 1 && count < 20, `expected a shrunk page, got ${count} items`); + assert.ok(JSON.stringify(page).length <= 16_000); + assert.equal(page.next_offset, count); + }); + + it("pages a long string by characters and walks indexes", () => { + const run = { ...bigRun(), state: { page: "abcdef".repeat(10_000) } }; + const part = readRunValue(run, "state.page", { offset: 59_990, limit: 50 }); + assert.equal(part.text, "cdefabcdef"); + assert.equal(part.next_offset, null); + assert.equal(readRunValue(bigRun(), "trace.1.output").text, "sent"); + }); + + it("returns an object too big to send as its outline, with paths to pass back", () => { + const read = readRunValue(bigRun(), "state"); + assert.equal(read.withheld, true); + const paths = (read.outline as Array<{ path: string }>).map((e) => e.path); + assert.ok(paths.includes("state.leads")); + }); + + it("names the fields that exist when a path is wrong", () => { + assert.throws(() => readRunValue(bigRun(), "state.nope"), /leads/); + assert.throws(() => readRunValue(bigRun(), "trace.9"), /not an index/); + }); +}); + +describe("pruneRuns", () => { + const now = Date.UTC(2026, 8, 25); + const daysAgo = (d: number) => new Date(now - d * 86_400_000).toISOString(); + + it("deletes finished runs past retention with their sub-flow files, and nothing else", () => { + const dir = stateDir(); + const old = seed(dir, { status: "completed", createdAt: daysAgo(40) }); + const oldFailed = seed(dir, { status: "failed", createdAt: daysAgo(31) }); + const recent = seed(dir, { status: "completed", createdAt: daysAgo(3) }); + const paused = seed(dir, { status: "paused", createdAt: daysAgo(90) }); + const waiting = seed(dir, { status: "waiting", createdAt: daysAgo(90) }); + fs.writeFileSync(path.join(dir, `${old}_loop_each_0.json`), "{}"); + fs.writeFileSync(path.join(dir, `${old}_loop_each_0_condition_ok_yes.json`), "{}"); + fs.writeFileSync(path.join(dir, `${recent}_loop_each_0.json`), "{}"); + fs.writeFileSync(path.join(dir, "_pending-approvals.json"), "[]"); + + assert.deepEqual(pruneRuns(dir, 30, now), { runs: 2, files: 4 }); + const left = fs.readdirSync(dir).sort(); + assert.deepEqual( + left, + [`${paused}.json`, `${recent}.json`, `${recent}_loop_each_0.json`, `${waiting}.json`, "_pending-approvals.json"].sort(), + ); + assert.equal(left.includes(`${oldFailed}.json`), false); + }); + + it("keeps everything when retention is 0", () => { + const dir = stateDir(); + seed(dir, { status: "completed", createdAt: daysAgo(400) }); + assert.deepEqual(pruneRuns(dir, 0, now), { runs: 0, files: 0 }); + assert.equal(fs.readdirSync(dir).length, 1); + }); +}); + +describe("flow_status tool", () => { + type ToolResult = { content: Array<{ type: string; text: string }> }; + type Tool = { name: string; execute: (id: string, p: Record) => Promise }; + + function flowStatus(dir: string): Tool { + const tools = new Map(); + const workspace = fs.mkdtempSync(path.join(os.tmpdir(), "ocf-runs-ws-")); + roots.push(workspace); + plugin.register({ + registerTool: (def: Tool) => tools.set(def.name, def), + registerHook: () => {}, + config: { workspace, plugins: { entries: { clawflow: { config: { stateDir: dir } } } } }, + logger: { info: () => {}, warn: () => {}, error: () => {} }, + } as never); + return tools.get("flow_status")!; + } + + it("pages the list, summarizes a big run, and reads one value by path", async () => { + const dir = stateDir(); + for (let i = 0; i < 25; i++) seed(dir, { createdAt: minute(i) }); + const run = bigRun(); + fs.writeFileSync(path.join(dir, `${run.instanceId}.json`), JSON.stringify(run)); + const tool = flowStatus(dir); + + const list = JSON.parse((await tool.execute("t", { limit: 10 })).content[0].text) as RunPage; + assert.equal(list.runs.length, 10); + assert.equal(list.total, 26); + + const one = JSON.parse((await tool.execute("t", { instanceId: "run-1" })).content[0].text) as RunSummary; + assert.equal(one.summary, true); + + const value = JSON.parse( + (await tool.execute("t", { instanceId: "run-1", path: "state.inputs.max_pages" })).content[0].text, + ) as { value: number }; + assert.equal(value.value, 2); + + const bad = (await tool.execute("t", { instanceId: "run-1", path: "state.nope" })).content[0].text; + assert.match(bad, /^Error: .*leads/); + }); +});