From 1d3443a754c8c939991ba4e1f624dc159f29bd0c Mon Sep 17 00:00:00 2001 From: os-zhuang Date: Sat, 4 Jul 2026 21:53:22 +0800 Subject: [PATCH] feat(automation): durable run history with failure reasons (run observability) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror every terminal automation run (completed / failed) to the sys_automation_run table so "did it run, and why did it fail?" survives a process restart and the in-memory ring-buffer eviction — previously runs were observable only in memory and failures surfaced no reason. - engine: `recordLog` fire-and-forget `recordTerminal` on terminal runs; `listRuns` merges durable history with the in-memory buffer (freshest wins). - store: `recordTerminal` / `listHistory` on both SuspendedRunStore impls, keyed by a `run_`-prefixed id disjoint from live suspended runs so the ADR-0019 suspend/resume path is untouched and resume sweeps skip history. - schema: sys_automation_run gains `finished_at` / `duration_ms` / `error` (all optional) and a `completed` / `failed` / `running` status alongside `paused`; becomes a durable run-history table, not suspend-only. - best-effort throughout: a history write/read failure is logged and swallowed, never breaking the run that produced it. Verified end-to-end on a clean showcase: completed + failed runs persisted durable rows (failure reason intact), a live paused run coexisted without collision, and the failed row survived a full process restart. New run-history.test.ts covers persistence, read-across-restart, best-effort isolation. Co-Authored-By: Claude Opus 4.8 --- .changeset/automation-run-history.md | 37 +++++++ .../services/service-automation/src/engine.ts | 96 ++++++++++++++++- .../src/run-history.test.ts | 102 ++++++++++++++++++ .../src/suspended-run-store.test.ts | 9 +- .../src/suspended-run-store.ts | 84 ++++++++++++++- .../src/sys-automation-run.object.ts | 35 ++++-- 6 files changed, 352 insertions(+), 11 deletions(-) create mode 100644 .changeset/automation-run-history.md create mode 100644 packages/services/service-automation/src/run-history.test.ts diff --git a/.changeset/automation-run-history.md b/.changeset/automation-run-history.md new file mode 100644 index 0000000000..513edd5af4 --- /dev/null +++ b/.changeset/automation-run-history.md @@ -0,0 +1,37 @@ +--- +"@objectstack/service-automation": minor +--- + +feat(automation): durable run history — every terminal run leaves a queryable record with its failure reason + +Automation runs were observable **only in memory**: the engine kept the last N +`ExecutionLogEntry` records in a ring buffer, so "did this flow run, and why did +it fail?" could not be answered after a process restart (or once the buffer +evicted the entry), and a failed run surfaced no reason at all. This was the +biggest silent-trust gap for anyone authoring automations — a flow could stop +firing or start failing with nothing durable to inspect. + +`sys_automation_run` — previously the ADR-0019 store for *live suspended* runs +only — becomes a durable **run-history** table. On every terminal run the engine +mirrors a row through the `SuspendedRunStore` (`recordTerminal`): `status` +(`completed` / `failed`), `finished_at`, `duration_ms`, and, for a failure, the +`error` message a designer needs to fix it. `listRuns()` merges this durable +history with the in-memory buffer (in-memory wins on id, newest-first) so the +Studio "Runs" surface shows runs that predate the current process. + +The design is **safe and additive**. Terminal history rows use a `run_`-prefixed +id, disjoint from live suspended runs (which key on the raw `runId` with +`status: 'paused'`), so the suspend save/load/delete/list path is untouched and +resume sweeps (`list()` filters `status: 'paused'`) never see history rows. +Persisting is **best-effort and fire-and-forget** — a history-write failure is +logged and swallowed, never breaking the run that produced it. New object fields +(`finished_at`, `duration_ms`, `error`) are all optional and the `status` enum +gains `running` / `completed` / `failed` alongside the existing `paused`. + +Verified end-to-end on a clean showcase instance: a schedule-triggered flow and +seven task-completion flows each left durable `completed` rows; a genuinely +failing flow (`showcase_resilient_sync`) left a `failed` row carrying its +`try_catch` failure reason; a live `paused` suspended run coexisted without +collision; and after a full process restart the `failed` row — reason intact — +was still queryable via `/api/v1/data/sys_automation_run`. New `run-history.test.ts` +covers completed/failed persistence, read-across-restart, and best-effort isolation. diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index d7fb293db2..cd123f0fec 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -353,6 +353,25 @@ export interface SuspendedRun { * serverless deployments where the process hibernates between suspend and * resume. */ +/** + * A terminal run summary persisted as durable run history (completed / failed) + * for the "Runs" observability surface — distinct from a live {@link SuspendedRun}. + */ +export interface RunRecord { + runId: string; + flowName: string; + flowVersion?: number; + status: 'completed' | 'failed'; + startedAt: string; + startTime?: number; + durationMs?: number; + /** Failure reason for a `failed` run — what a designer needs to fix it. */ + error?: string; + nodeId?: string; + organizationId?: string | null; + userId?: string | null; +} + export interface SuspendedRunStore { /** Persist (insert or replace) a suspended run. */ save(run: SuspendedRun): Promise; @@ -362,6 +381,16 @@ export interface SuspendedRunStore { delete(runId: string): Promise; /** List all currently-stored suspended runs. */ list(): Promise; + /** + * Persist a TERMINAL run (completed / failed) as durable history for the + * "Runs" observability surface. Optional — the in-memory / test defaults + * still work without it. Implementations MUST key history rows separately + * from live suspended runs (which are keyed by raw `runId`, status + * `paused`, and deleted on completion) so the two lifecycles never collide. + */ + recordTerminal?(record: RunRecord): Promise; + /** Newest terminal run-history records for a flow (for the Runs tab). */ + listHistory?(flowName: string, limit: number): Promise; } export class AutomationEngine implements IAutomationService { @@ -908,8 +937,45 @@ export class AutomationEngine implements IAutomationService { async listRuns(flowName: string, options?: { limit?: number; cursor?: string }): Promise { const limit = options?.limit ?? 20; - const logs = this.executionLogs.filter(l => l.flowName === flowName); - return logs.slice(-limit).reverse(); + const inMem = this.executionLogs.filter(l => l.flowName === flowName); + + // Merge durable run history so the "Runs" view survives a restart and + // ring-buffer eviction. In-memory entries are the freshest (they carry + // full step detail); durable rows backfill runs the process no longer + // holds. Best-effort: a history-read failure degrades to in-memory only. + let durable: ExecutionLogEntry[] = []; + if (this.store?.listHistory) { + try { + const rows = await this.store.listHistory(flowName, limit); + durable = rows.map(r => this.runRecordToLogEntry(r)); + } catch (err) { + this.logger.warn( + `[Automation] run-history read failed for '${flowName}': ${(err as Error)?.message}`, + ); + } + } + const byId = new Map(); + for (const e of durable) byId.set(e.id, e); + for (const e of inMem) byId.set(e.id, e); // freshest wins + return [...byId.values()] + .sort((a, b) => (b.startedAt ?? '').localeCompare(a.startedAt ?? '')) + .slice(0, limit); + } + + /** Rehydrate a durable {@link RunRecord} into an {@link ExecutionLogEntry} + * for the Runs list. Step-level detail isn't persisted in the summary. */ + private runRecordToLogEntry(r: RunRecord): ExecutionLogEntry { + return { + id: r.runId, + flowName: r.flowName, + flowVersion: r.flowVersion, + status: r.status, // 'completed' | 'failed' — both valid ExecutionLog statuses + startedAt: r.startedAt, + durationMs: r.durationMs, + trigger: { type: '', userId: r.userId ?? undefined }, + steps: [], + error: r.error, + }; } async getRun(runId: string): Promise { @@ -1581,6 +1647,32 @@ export class AutomationEngine implements IAutomationService { if (this.executionLogs.length > this.maxLogSize) { this.executionLogs.splice(0, this.executionLogs.length - this.maxLogSize); } + // Durable run history (observability): mirror every TERMINAL run to the + // store so "did it run / fail, and why?" survives a restart and the + // in-memory ring-buffer eviction. Best-effort + fire-and-forget: a + // history write must NEVER block or break the run that produced it. + const terminal = + entry.status === 'completed' || + entry.status === 'failed' || + entry.status === 'cancelled' || + entry.status === 'timed_out'; + if (terminal && this.store?.recordTerminal) { + const record: RunRecord = { + runId: entry.id, + flowName: entry.flowName, + flowVersion: entry.flowVersion, + status: entry.status === 'completed' ? 'completed' : 'failed', + startedAt: entry.startedAt, + durationMs: entry.durationMs, + error: entry.error, + userId: entry.trigger?.userId, + }; + void this.store.recordTerminal(record).catch((err) => { + this.logger.warn( + `[Automation] run-history persist failed for '${entry.flowName}': ${(err as Error)?.message}`, + ); + }); + } } /** diff --git a/packages/services/service-automation/src/run-history.test.ts b/packages/services/service-automation/src/run-history.test.ts new file mode 100644 index 0000000000..5959335aeb --- /dev/null +++ b/packages/services/service-automation/src/run-history.test.ts @@ -0,0 +1,102 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// Automation run observability: every TERMINAL run (completed / failed) must be +// mirrored to the durable store so "did it run / fail, and why?" survives a +// process restart and the in-memory ring-buffer eviction — and `listRuns` must +// merge that history. Persisting is best-effort: a history-write failure must +// never break the run that produced it. + +import { describe, it, expect } from 'vitest'; +import { AutomationEngine } from './engine.js'; +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; +import type { AutomationContext } from '@objectstack/spec/contracts'; + +const silent = { info() {}, warn() {}, error() {}, debug() {} } as never; + +function trivialFlow(name: string) { + return { + name, label: name, type: 'autolaunched', + nodes: [{ id: 'start', type: 'start', label: 's' }, { id: 'end', type: 'end', label: 'e' }], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + }; +} + +function failingFlow(name: string) { + return { + name, label: name, type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 's' }, + { id: 'boom', type: 'boom', label: 'b' }, + { id: 'end', type: 'end', label: 'e' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'boom' }, { id: 'e2', source: 'boom', target: 'end' }], + }; +} + +const flush = () => new Promise((r) => setTimeout(r, 0)); + +describe('automation run history (durable observability)', () => { + it('persists a completed run to the durable store', async () => { + const store = new InMemorySuspendedRunStore(); + const engine = new AutomationEngine(silent, store); + engine.registerFlow('ok_flow', trivialFlow('ok_flow') as never); + + const res = await engine.execute('ok_flow', { event: 'test' } as AutomationContext); + expect(res.success).toBe(true); + await flush(); // recordTerminal is fire-and-forget + + const hist = await store.listHistory('ok_flow', 10); + expect(hist).toHaveLength(1); + expect(hist[0].status).toBe('completed'); + expect(hist[0].flowName).toBe('ok_flow'); + }); + + it('persists a failed run WITH its error reason (the designer-facing "why")', async () => { + const store = new InMemorySuspendedRunStore(); + const engine = new AutomationEngine(silent, store); + engine.registerNodeExecutor({ + type: 'boom', + async execute() { throw new Error('kaboom'); }, + } as never); + engine.registerFlow('bad_flow', failingFlow('bad_flow') as never); + + const res = await engine.execute('bad_flow', { event: 'test' } as AutomationContext); + expect(res.success).toBe(false); + await flush(); + + const hist = await store.listHistory('bad_flow', 10); + expect(hist).toHaveLength(1); + expect(hist[0].status).toBe('failed'); + expect(hist[0].error ?? '').toMatch(/kaboom/); + }); + + it('listRuns merges durable history so runs survive a process restart', async () => { + const store = new InMemorySuspendedRunStore(); + const engineA = new AutomationEngine(silent, store); + engineA.registerFlow('surv', trivialFlow('surv') as never); + await engineA.execute('surv', { event: 'test' } as AutomationContext); + await flush(); + + // Simulate a restart: a fresh engine (empty in-memory logs) sharing the + // same durable store. Before this feature its listRuns would be empty. + const engineB = new AutomationEngine(silent, store); + const runs = await engineB.listRuns('surv', { limit: 10 }); + expect(runs).toHaveLength(1); + expect(runs[0].status).toBe('completed'); + expect(runs[0].id).toBeTruthy(); + }); + + it('a failing history store never breaks the run (best-effort isolation)', async () => { + const store = new InMemorySuspendedRunStore(); + // Override recordTerminal to reject — the run must still complete. + (store as unknown as { recordTerminal: () => Promise }).recordTerminal = async () => { + throw new Error('history db down'); + }; + const engine = new AutomationEngine(silent, store); + engine.registerFlow('resilient', trivialFlow('resilient') as never); + + const res = await engine.execute('resilient', { event: 'test' } as AutomationContext); + expect(res.success).toBe(true); // the run is unaffected by the persist failure + await flush(); + }); +}); diff --git a/packages/services/service-automation/src/suspended-run-store.test.ts b/packages/services/service-automation/src/suspended-run-store.test.ts index af606e0719..dd59d9d910 100644 --- a/packages/services/service-automation/src/suspended-run-store.test.ts +++ b/packages/services/service-automation/src/suspended-run-store.test.ts @@ -144,7 +144,12 @@ describe('ObjectStoreSuspendedRunStore', () => { expect(resumed.success).toBe(true); expect(ran).toContain('rejected'); expect(ran).not.toContain('approved'); - // Row removed on terminal completion. - expect(engine.rows.size).toBe(0); + // The live suspended row is removed on terminal completion; a durable + // terminal run-history row is kept in its place (run observability). + await new Promise((r) => setTimeout(r, 0)); // recordTerminal is fire-and-forget + const finalRows = [...engine.rows.values()]; + expect(finalRows.filter((r) => r.status === 'paused')).toHaveLength(0); + expect(finalRows).toHaveLength(1); + expect(finalRows[0].status).toBe('completed'); }); }); diff --git a/packages/services/service-automation/src/suspended-run-store.ts b/packages/services/service-automation/src/suspended-run-store.ts index 719aa276db..50a0f5eca6 100644 --- a/packages/services/service-automation/src/suspended-run-store.ts +++ b/packages/services/service-automation/src/suspended-run-store.ts @@ -1,7 +1,7 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. import type { Logger } from '@objectstack/spec/contracts'; -import type { SuspendedRun, SuspendedRunStore } from './engine.js'; +import type { RunRecord, SuspendedRun, SuspendedRunStore } from './engine.js'; /** * Durable persistence for suspended flow runs (ADR-0019). @@ -21,6 +21,9 @@ import type { SuspendedRun, SuspendedRunStore } from './engine.js'; */ const TABLE = 'sys_automation_run'; +/** Prefix for terminal run-history row ids, keeping them disjoint from live + * suspended runs (which use the raw `runId`). */ +const HISTORY_PREFIX = 'run_'; const SYSTEM_CTX = { isSystem: true, roles: [], permissions: [] } as const; /** Deep clone via JSON so a stored snapshot can't alias live engine state. */ @@ -49,6 +52,7 @@ function parseJson(raw: unknown, fallback: T): T { */ export class InMemorySuspendedRunStore implements SuspendedRunStore { private readonly runs = new Map(); + private readonly history = new Map(); async save(run: SuspendedRun): Promise { this.runs.set(run.runId, jsonClone(run)); @@ -66,6 +70,18 @@ export class InMemorySuspendedRunStore implements SuspendedRunStore { async list(): Promise { return [...this.runs.values()].map(jsonClone); } + + async recordTerminal(record: RunRecord): Promise { + this.history.set(record.runId, jsonClone(record)); + } + + async listHistory(flowName: string, limit: number): Promise { + return [...this.history.values()] + .filter((r) => r.flowName === flowName) + .sort((a, b) => (b.startedAt ?? '').localeCompare(a.startedAt ?? '')) + .slice(0, limit) + .map(jsonClone); + } } /** @@ -149,6 +165,72 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { return (Array.isArray(rows) ? rows : []).map(r => this.deserialize(r)); } + /** + * Persist a TERMINAL run (completed / failed) as durable history. Keyed by a + * `run_`-prefixed id so it NEVER collides with a live suspended run's row + * (id = raw `runId`, status `paused`) — the suspend save/load/delete/list + * path (which only touches raw ids and `status:'paused'` rows) is untouched. + * Upsert so a re-emitted terminal (e.g. a resumed run) updates in place. + */ + async recordTerminal(record: RunRecord): Promise { + const now = new Date().toISOString(); + const id = HISTORY_PREFIX + record.runId; + const row = { + id, + organization_id: record.organizationId ?? null, + flow_name: record.flowName, + flow_version: record.flowVersion ?? null, + node_id: record.nodeId ?? null, + status: record.status, + user_id: record.userId ?? null, + started_at: record.startedAt, + start_time: record.startTime ?? null, + finished_at: now, + duration_ms: record.durationMs ?? null, + error: record.error ?? null, + }; + const existing = await this.engine.find(TABLE, { + where: { id }, limit: 1, context: SYSTEM_CTX, + }); + if (Array.isArray(existing) && existing[0]) { + await this.engine.update(TABLE, { ...row, updated_at: now }, { where: { id }, context: SYSTEM_CTX }); + } else { + await this.engine.insert(TABLE, { ...row, created_at: now, updated_at: now }, { context: SYSTEM_CTX }); + } + } + + /** Newest terminal (`completed` / `failed`) run-history rows for one flow. */ + async listHistory(flowName: string, limit: number): Promise { + // Fetch the flow's rows and filter terminal in memory — avoids depending on + // IN-clause support in the driver's `where`. Paused rows are excluded. + const rows = await this.engine.find(TABLE, { + where: { flow_name: flowName }, limit: Math.max(limit * 4, 200), context: SYSTEM_CTX, + }); + return (Array.isArray(rows) ? rows : []) + .filter(r => r?.status === 'completed' || r?.status === 'failed') + .map(r => this.deserializeTerminal(r)) + .sort((a, b) => (b.startedAt ?? '').localeCompare(a.startedAt ?? '')) + .slice(0, limit); + } + + /** Rebuild a {@link RunRecord} from a terminal `sys_automation_run` row. */ + private deserializeTerminal(row: any): RunRecord { + const rawId = String(row.id ?? ''); + return { + runId: rawId.startsWith(HISTORY_PREFIX) ? rawId.slice(HISTORY_PREFIX.length) : rawId, + flowName: String(row.flow_name ?? ''), + flowVersion: typeof row.flow_version === 'number' ? row.flow_version : undefined, + status: row.status === 'failed' ? 'failed' : 'completed', + startedAt: row.started_at ?? row.created_at ?? '', + startTime: typeof row.start_time === 'number' ? row.start_time : undefined, + durationMs: typeof row.duration_ms === 'number' ? row.duration_ms : undefined, + error: row.error ?? undefined, + nodeId: row.node_id ?? undefined, + organizationId: row.organization_id ?? null, + userId: row.user_id ?? undefined, + }; + } + /** Flatten a run into a `sys_automation_run` row (state columns JSON-encoded). */ private serialize(run: SuspendedRun): Record { const ctx = (run.context ?? {}) as Record; diff --git a/packages/services/service-automation/src/sys-automation-run.object.ts b/packages/services/service-automation/src/sys-automation-run.object.ts index ed859e5ca3..e4b59a86fc 100644 --- a/packages/services/service-automation/src/sys-automation-run.object.ts +++ b/packages/services/service-automation/src/sys-automation-run.object.ts @@ -35,7 +35,7 @@ export const SysAutomationRun = ObjectSchema.create({ icon: 'pause-circle', isSystem: true, managedBy: 'system', - description: 'Durable state of a suspended automation flow run (ADR-0019)', + description: 'Durable automation run state: live suspended runs (resumable, ADR-0019) and terminal run history (completed / failed, for observability).', displayNameField: 'id', nameField: 'id', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) titleFormat: '{flow_name} · {node_id}', @@ -62,20 +62,20 @@ export const SysAutomationRun = ObjectSchema.create({ flow_version: Field.number({ label: 'Flow Version', required: false, group: 'Identity' }), node_id: Field.text({ - label: 'Paused Node', - required: true, + label: 'Node', + required: false, maxLength: 255, - description: 'Node the run is suspended at; resume continues from its out-edges.', + description: 'For a suspended run, the node it is paused at (resume continues from its out-edges); for a terminal run, the last node reached.', group: 'State', }), status: Field.select( - ['paused'], + ['running', 'paused', 'completed', 'failed'], { label: 'Status', required: true, defaultValue: 'paused', - description: 'Only suspended runs are persisted; the row is deleted on terminal completion.', + description: 'paused = a live suspended run (resumable); completed / failed = a terminal run kept as durable history.', group: 'State', }, ), @@ -133,6 +133,27 @@ export const SysAutomationRun = ObjectSchema.create({ group: 'State', }), + finished_at: Field.datetime({ + label: 'Finished At', + required: false, + description: 'When a terminal run (completed / failed) ended. Null while running / paused.', + group: 'Outcome', + }), + + duration_ms: Field.number({ + label: 'Duration (ms)', + required: false, + description: 'Wall-clock duration of a terminal run.', + group: 'Outcome', + }), + + error: Field.textarea({ + label: 'Error', + required: false, + description: 'Failure reason for a `failed` run — the message a designer needs to fix it.', + group: 'Outcome', + }), + created_at: Field.datetime({ label: 'Created At', required: true, @@ -148,6 +169,8 @@ export const SysAutomationRun = ObjectSchema.create({ // "Which runs are suspended for this flow?" — operability / resume sweeps. { fields: ['flow_name', 'status'] }, { fields: ['status', 'updated_at'] }, + // Run-history reads for the Studio "Runs" tab: newest terminal runs per flow. + { fields: ['flow_name', 'started_at'] }, // Look up a suspended run by the pausing node's correlation key. { fields: ['correlation'] }, ],