feat(automation): durable run history with failure reasons (run observability)

.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.

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<void>;
@@ -362,6 +381,16 @@ export interface SuspendedRunStore {
     delete(runId: string): Promise<void>;
     /** List all currently-stored suspended runs. */
     list(): Promise<SuspendedRun[]>;
+    /**
+     * 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<void>;
+    /** Newest terminal run-history records for a flow (for the Runs tab). */
+    listHistory?(flowName: string, limit: number): Promise<RunRecord[]>;
 }
 
 export class AutomationEngine implements IAutomationService {
@@ -908,8 +937,45 @@ export class AutomationEngine implements IAutomationService {
 
     async listRuns(flowName: string, options?: { limit?: number; cursor?: string }): Promise<ExecutionLogEntry[]> {
         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<string, ExecutionLogEntry>();
+        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<ExecutionLogEntry | null> {
@@ -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}`,
+                );
+            });
+        }
     }
 
     /**

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<void>((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<void> }).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();
+    });
+});

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');
     });
 });

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<T>(raw: unknown, fallback: T): T {
  */
 export class InMemorySuspendedRunStore implements SuspendedRunStore {
   private readonly runs = new Map<string, SuspendedRun>();
+  private readonly history = new Map<string, RunRecord>();
 
   async save(run: SuspendedRun): Promise<void> {
     this.runs.set(run.runId, jsonClone(run));
@@ -66,6 +70,18 @@ export class InMemorySuspendedRunStore implements SuspendedRunStore {
   async list(): Promise<SuspendedRun[]> {
     return [...this.runs.values()].map(jsonClone);
   }
+
+  async recordTerminal(record: RunRecord): Promise<void> {
+    this.history.set(record.runId, jsonClone(record));
+  }
+
+  async listHistory(flowName: string, limit: number): Promise<RunRecord[]> {
+    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<void> {
+    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<RunRecord[]> {
+    // 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<string, unknown> {
     const ctx = (run.context ?? {}) as Record<string, unknown>;