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/); + }); +});