diff --git a/.changeset/paused-run-variables-snapshot.md b/.changeset/paused-run-variables-snapshot.md new file mode 100644 index 0000000000..5a133b6e9f --- /dev/null +++ b/.changeset/paused-run-variables-snapshot.md @@ -0,0 +1,37 @@ +--- +"@objectstack/service-automation": patch +--- + +fix(services): a paused run's variable snapshot is readable on run-detail (#7639) + +While an automation run was **paused**, `GET /api/v1/automation/{flow}/runs/{runId}` +carried no `variables` key at all — so a run stopped at an approval, a screen or a +wait, which is precisely the state an operator most often needs to inspect, +answered with no variable state. "What did the previous node actually produce, and +why did the next one route the way it did?" was not answerable from the product; +it could only be inferred backwards from whatever the next node happened to +resolve. + +This was structural, not a data gap. `ExecutionLogSchema` has declared +`variables` ("Final state of flow variables") since the schema was written, and +the engine's own log entry declared it too — with no producer anywhere, so the +key the run-detail read publishes was never populated. The engine already held +the answer: both `status: 'paused'` `recordLog` call sites sit a few lines below +the suspend bookkeeping that computes `Object.fromEntries(variables)` for the +continuation. The snapshot simply never reached the surface a caller can read. + +Both paused sites now write it — the initial-execution suspend **and** the +resume-path re-suspend, so a multi-stage approval is readable at every stage +rather than only the first. Each site takes ONE snapshot expression and hands the +same object to the continuation and to the log entry, so what an operator reads +can never disagree with the state the run will resume from. + +The snapshot is **point-in-time at the suspend**, not a live read: the variable +map is dead by then (the run has unwound; resume rebuilds a fresh map from the +continuation), so there is nothing later to diverge from. + +Nothing about the exposure envelope changes: the run-detail read serves the log +entry verbatim — no projection, redaction or masking on any field — and +`variables` receives exactly that same treatment, under the same anonymous +baseline that already gates the whole `/automation` domain. Terminal runs keep +exactly the fields they had; only `paused` gains the key. diff --git a/packages/runtime/src/domains/automation-run-detail-passthrough.test.ts b/packages/runtime/src/domains/automation-run-detail-passthrough.test.ts new file mode 100644 index 0000000000..b761efa814 --- /dev/null +++ b/packages/runtime/src/domains/automation-run-detail-passthrough.test.ts @@ -0,0 +1,130 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #7639 — the wire half of "a paused run's variables are readable on + * run-detail": `GET /automation/:name/runs/:runId` must carry `variables` + * through untouched, exactly as it already carries `output` and `steps`. + * + * The engine half (the two `status: 'paused'` `recordLog` call sites finally + * writing the snapshot they already hold) is pinned one package over, in + * `packages/services/service-automation/src/paused-run-variables.test.ts`. This + * file pins the surface that serves it, and the two claims that made the change + * dispatchable rather than a disclosure decision: + * + * 1. IDENTICAL ENVELOPE. `output`, `steps` and `variables` are not three + * policies — they are one object handed to `deps.success(run)`. There is no + * per-field projection, redaction or masking anywhere on this path, so a + * field the engine records is a field the caller reads. The test drives one + * entry carrying all three and asserts each survives byte-for-byte. + * 2. IDENTICAL ACCESS CONTROL. The one gate on this read is the #5519 + * anonymous baseline, which covers the WHOLE `/automation` domain — so + * whoever could already read a completed run's `output` here is exactly + * whoever can now read a paused run's `variables`, and an anonymous caller + * gets neither. + * + * If a future change starts shaping one of these fields, the deep-equal + * assertions below fail — which is the point: the shaping policy for the three + * must stay one policy. + */ + +import { describe, it, expect, vi } from 'vitest'; + +import { HttpDispatcher } from '../http-dispatcher.js'; + +/** A paused run as the engine records it since #7639 — snapshot and all. */ +const PAUSED_RUN = { + id: 'run_7', + flowName: 'approval_flow', + flowVersion: 3, + status: 'paused', + startedAt: '2026-08-12T02:00:00.000Z', + durationMs: 42, + trigger: { type: 'record_change', object: 'crm_order', recordId: 'ord_1', userId: 'user_1' }, + steps: [ + { nodeId: 'start', nodeType: 'start', status: 'success', startedAt: '2026-08-12T02:00:00.000Z' }, + { nodeId: 'stage1', nodeType: 'approval', status: 'success', startedAt: '2026-08-12T02:00:00.010Z' }, + ], + variables: { + 'stage1.pending_approvers': ['user_ops', 'user_finance'], + 'stage1.decision': { route: 'dual', weights: { ops: 1, finance: 2 }, note: null }, + record: { id: 'ord_1', amount: 90_000 }, + $runId: 'run_7', + $flowName: 'approval_flow', + }, +} as const; + +/** A terminal run, whose `output` this surface has always carried. */ +const COMPLETED_RUN = { + ...PAUSED_RUN, + id: 'run_8', + status: 'completed', + completedAt: '2026-08-12T02:00:01.000Z', + output: { approved: true, decision: { route: 'dual', weights: { ops: 1, finance: 2 }, note: null } }, +} as const; + +function makeDispatcher(run: unknown) { + const getRun = vi.fn(async () => run); + const services: Record = { automation: { getRun, handlerReady: true } }; + const resolve = (name: string) => services[name]; + const kernel: any = { + getService: resolve, + getServiceAsync: async (name: string) => resolve(name), + context: { getService: resolve }, + }; + return { dispatcher: new HttpDispatcher(kernel), getRun }; +} + +const CTX = () => ({ request: {}, executionContext: { userId: 'user_1' } } as any); +const ANON_CTX = () => ({ request: {}, executionContext: {} } as any); + +/** Drive `GET /automation/:flow/runs/:runId` and hand back the raw response. */ +async function getRunDetail(run: unknown, context = CTX()) { + const { dispatcher, getRun } = makeDispatcher(run); + const { response } = await dispatcher.handleAutomation( + `approval_flow/runs/${(run as { id: string }).id}`, 'GET', undefined, context, undefined, + ); + return { response: response as any, getRun }; +} + +/** The run payload out of the success envelope, whatever the envelope's shape. */ +const payloadOf = (response: any) => response?.data ?? response?.body?.data ?? response; + +describe('#7639 — GET /automation/:name/runs/:runId serves a paused run WITH its variable snapshot', () => { + it('passes `variables` through untouched', async () => { + const { response } = await getRunDetail(PAUSED_RUN); + const run = payloadOf(response); + + expect(run.status).toBe('paused'); + // The defect this closes: the key was absent from the response entirely. + expect(run.variables).toBeDefined(); + // Untouched — nested objects, arrays and a null all survive intact. A + // projection or a redaction anywhere on this path breaks this line. + expect(run.variables).toEqual(PAUSED_RUN.variables); + }); + + it('shapes `variables` exactly as it shapes `output` and `steps` — not at all', async () => { + const paused = payloadOf((await getRunDetail(PAUSED_RUN)).response); + const completed = payloadOf((await getRunDetail(COMPLETED_RUN)).response); + + // One policy for the three fields, which is the whole basis on which + // adding `variables` is consistency rather than a new disclosure + // surface: the handler answers with the log entry as recorded. + expect(completed.output).toEqual(COMPLETED_RUN.output); + expect(completed.steps).toEqual(COMPLETED_RUN.steps); + expect(paused.steps).toEqual(PAUSED_RUN.steps); + + // The identical nested value reads back the same whether it arrives via + // `output` (terminal run) or via `variables` (paused run). + expect(paused.variables['stage1.decision']).toEqual(completed.output.decision); + }); + + it('gates the snapshot behind the same anonymous baseline as the rest of the domain (#5519)', async () => { + const { response, getRun } = await getRunDetail(PAUSED_RUN, ANON_CTX()); + + // ADR-0112: the refusal's `code` AND its `status`, never just one. + expect(response.body?.error?.code ?? response.body?.error?.details?.code).toBe('UNAUTHENTICATED'); + expect(response.status).toBe(401); + // The gate fires ahead of the service, so no snapshot is even read. + expect(getRun).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 5f0b4a0d33..075fa5f185 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -815,6 +815,34 @@ interface ExecutionLogEntry { // ↑ built by `buildRunTrigger` at EVERY site — see its doc comment for why // that is a chokepoint rather than eight object literals. steps: StepLogEntry[]; + /** + * #7639: the run's variable map, written at the two `paused` sites only — + * the point-in-time snapshot the run suspended holding, and the SAME object + * handed to {@link AutomationEngine.persistSuspendedRun}. + * + * Not new vocabulary. `ExecutionLogSchema` has declared + * `variables` ("Final state of flow variables") since the schema was + * written, and this interface has declared it for as long — with no + * producer anywhere, so the key `GET /automation/:name/runs/:runId` + * publishes was never populated by anything. That is the same + * declared-with-no-writer shape as `StepLogEntry.retryAttempt` (#7546): + * consume the existing declaration rather than invent a second spelling. + * + * Why `paused` and not every status: a paused run is the one an operator + * cannot otherwise inspect. A terminal run has already produced its + * `output`, and its step log says what ran; a run stopped at an approval or + * a screen has produced neither, so "what did the previous node actually + * resolve, and why did the next one route the way it did?" was answerable + * only by inference. Widening to `completed`/`failed` would be a disclosure + * change with no card behind it — those runs keep exactly the fields they + * had. + * + * SNAPSHOT, not a live read: taken at the suspend, never refreshed. The map + * itself is dead by then (the run unwound; resume rebuilds a fresh one from + * the continuation), so there is nothing later to diverge from — and + * because the continuation gets this very object, the snapshot an operator + * reads is by construction the state the run will resume from. + */ variables?: Record; output?: unknown; error?: string; @@ -3011,13 +3039,19 @@ export class AutomationEngine implements IAutomationService { // caller can later `resume()` it. This is NOT a failure. if (isSuspendSignal(err)) { const durationMs = Date.now() - startTime; + // #7639 — ONE snapshot expression feeding BOTH consumers: the + // continuation the run will resume from, and the `paused` log + // entry run-detail serves. Same object, so what an operator + // reads can never disagree with what the run holds. See + // {@link ExecutionLogEntry.variables} for why the log carries it. + const variablesSnapshot = Object.fromEntries(variables); await this.persistSuspendedRun({ runId, flowName, flowVersion: flow.version, nodeId: err.nodeId, nodeType: err.nodeType, - variables: Object.fromEntries(variables), + variables: variablesSnapshot, steps, context: runContext, startedAt, @@ -3034,6 +3068,7 @@ export class AutomationEngine implements IAutomationService { durationMs, trigger: buildRunTrigger(context), steps, + variables: variablesSnapshot, }); return { success: true, @@ -3766,11 +3801,18 @@ export class AutomationEngine implements IAutomationService { // Re-suspended at a downstream node: persist a fresh continuation. if (isSuspendSignal(err)) { const durationMs = Date.now() - run.startTime; + // #7639 — the re-suspend half of the same rule as the + // initial-execution site above: one snapshot, both consumers. + // A multi-stage approval re-pauses HERE on every stage but the + // first, so covering only the other site would leave every + // stage after stage 1 — the ones an operator actually needs to + // inspect — unreadable. + const variablesSnapshot = Object.fromEntries(variables); await this.persistSuspendedRun({ ...run, nodeId: err.nodeId, nodeType: err.nodeType, - variables: Object.fromEntries(variables), + variables: variablesSnapshot, steps, correlation: err.correlation, screen: err.screen, @@ -3784,6 +3826,7 @@ export class AutomationEngine implements IAutomationService { durationMs, trigger: buildRunTrigger(context), steps, + variables: variablesSnapshot, }); return { success: true, status: 'paused', runId, durationMs, screen: err.screen }; } diff --git a/packages/services/service-automation/src/paused-run-variables.test.ts b/packages/services/service-automation/src/paused-run-variables.test.ts new file mode 100644 index 0000000000..7a4d355063 --- /dev/null +++ b/packages/services/service-automation/src/paused-run-variables.test.ts @@ -0,0 +1,270 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #7639 — a SUSPENDED run's variable snapshot must be readable on run-detail. + * + * `GET /api/v1/automation/:name/runs/:runId` serves an `ExecutionLogEntry` + * verbatim (`packages/runtime/src/domains/automation.ts` → `deps.success(run)`), + * and `ExecutionLogSchema` has declared a `variables` key — "Final state of flow + * variables" — since the schema was written. Nothing in the repo ever wrote it. + * The two `status: 'paused'` `recordLog` call sites in `engine.ts` passed + * `id`/`flowName`/`flowVersion`/`startedAt`/`durationMs`/`trigger`/`steps` and + * stopped there, so a run stopped at an approval or a screen — the state an + * operator most often needs to inspect — answered with no variable state at all. + * + * The information was never lost: a few lines above each of those sites the + * suspend bookkeeping already computed `Object.fromEntries(variables)` for the + * continuation. It simply never reached the surface a caller can read. + * + * What is pinned here: + * + * 1. BOTH paused sites — the initial-execution suspend and the resume-path + * re-suspend. A multi-stage approval re-pauses at the second site on every + * stage after the first, so covering one site is half a fix. + * 2. The snapshot is the RUN's, not a fixture's: node outputs written under + * `.`, the triggering record, and declared flow variables all + * read back. + * 3. SHAPING PARITY (the redaction question). The run-detail read applies + * **no** field-level shaping to anything today — `output` and `steps` go out + * byte-for-byte as the engine recorded them, and there is no redaction, + * masking or projection anywhere on that path to mirror. So `variables` + * must receive exactly that same nil treatment: the test drives one and the + * same nested value through `output` (terminal run) and `variables` (paused + * run) and asserts the two are deep-equal. This does NOT invent a redaction + * policy — it pins that no new one was invented, and it FAILS if a future + * change starts shaping one field without the other. + * 4. Snapshot semantics: point-in-time at the suspend, and the same object the + * continuation carries — so the state read is the state the run resumes from. + * + * The wire half (the handler passing `variables` through untouched, beside + * `output`) is pinned one package over, in + * `packages/runtime/src/domains/automation-run-detail-passthrough.test.ts`. + */ + +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'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; + +const silent = { info() {}, warn() {}, error() {}, debug() {} } as never; + +/** + * `resumeAuthority: 'any'` is required of a pausing fixture since #5561 — these + * tests continue their pause through the public `resume` door. Nothing here is + * about the resume gate (`resume-authority-gate.test.ts` owns that); the fixture + * states the posture it relies on, as the pausing built-ins do. + */ +const pauser = (type: string) => defineActionDescriptor({ + type, version: '1.0.0', name: type, + supportsPause: true, resumeAuthority: 'any', +}); + +/** start → stage1 (pauses) → stage2 (pauses again) → end. */ +function twoStageFlow(name: string) { + return { + name, label: name, type: 'autolaunched', + variables: [ + // `isInput` is what makes `context.params` seed a declared variable + // (`seedDeclaredVariables`); without it the trigger's value is + // dropped and the fixture would assert nothing. + { name: 'ticket', type: 'text', isInput: true, isOutput: true }, + { name: 'internal_note', type: 'text', isInput: true }, + ], + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'stage1', type: 'stage1_pause', label: 'Stage 1' }, + { id: 'stage2', type: 'stage2_pause', label: 'Stage 2' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'stage1' }, + { id: 'e2', source: 'stage1', target: 'stage2' }, + { id: 'e3', source: 'stage2', target: 'end' }, + ], + }; +} + +/** + * A node that resolves a non-trivial value on entry and THEN suspends. The + * engine writes `result.output` into variables under `.` before it + * honours `suspend`, so this is exactly the "what did the previous node + * actually produce?" the card is about. + */ +function registerStage(engine: AutomationEngine, type: string, output: Record) { + engine.registerNodeExecutor({ + type, + descriptor: pauser(type), + async execute() { + return { success: true, suspend: true, correlation: `${type}:req`, output }; + }, + } as never); +} + +const STAGE1_OUTPUT = { + // Deliberately non-scalar and nested: the shaping-parity assertion below is + // worthless against a bare string, which survives any projection. + pending_approvers: ['user_ops', 'user_finance'], + decision: { route: 'dual', reason: 'amount over threshold', weights: { ops: 1, finance: 2 } }, +}; + +describe('#7639 — a paused run carries its variable snapshot on run-detail', () => { + it('writes the snapshot at the INITIAL-EXECUTION suspend site', async () => { + const engine = new AutomationEngine(silent, new InMemorySuspendedRunStore()); + registerStage(engine, 'stage1_pause', STAGE1_OUTPUT); + registerStage(engine, 'stage2_pause', { stage: 2 }); + engine.registerFlow('approval_flow', twoStageFlow('approval_flow') as never); + + const res = await engine.execute('approval_flow', { + event: 'test', + object: 'crm_order', + record: { id: 'ord_1', amount: 90_000, owner: 'user_ops' }, + params: { ticket: 'TKT-1', internal_note: 'do not page on-call' }, + } as unknown as AutomationContext); + expect(res.status).toBe('paused'); + + const run = await engine.getRun(res.runId as string); + expect(run?.status).toBe('paused'); + + // The defect: this key was absent entirely, not empty. + expect(run?.variables).toBeDefined(); + const vars = run!.variables!; + + // The pausing node's own resolved output — the value the QA oracle + // wanted to read and could not (`approvals.dynamic-approver-routing`). + expect(vars['stage1.pending_approvers']).toEqual(['user_ops', 'user_finance']); + expect(vars['stage1.decision']).toEqual(STAGE1_OUTPUT.decision); + + // The triggering record and the declared flow variables the run holds. + expect(vars.record).toEqual({ id: 'ord_1', amount: 90_000, owner: 'user_ops' }); + expect(vars.ticket).toBe('TKT-1'); + expect(vars.internal_note).toBe('do not page on-call'); + + // Run identity, so the snapshot is self-describing. + expect(vars.$runId).toBe(res.runId); + expect(vars.$flowName).toBe('approval_flow'); + }); + + it('writes the snapshot at the RESUME-PATH re-suspend site too', async () => { + const engine = new AutomationEngine(silent, new InMemorySuspendedRunStore()); + registerStage(engine, 'stage1_pause', STAGE1_OUTPUT); + registerStage(engine, 'stage2_pause', { approver_pool: ['user_cfo'] }); + engine.registerFlow('approval_flow', twoStageFlow('approval_flow') as never); + + const first = await engine.execute('approval_flow', { + event: 'test', + record: { id: 'ord_2', amount: 10 }, + } as unknown as AutomationContext); + expect(first.status).toBe('paused'); + + // Stage 1 decided; the run continues and re-pauses at stage 2. This is + // the SECOND `status: 'paused'` recordLog site — reached only here. + const second = await engine.resume(first.runId as string, { + output: { verdict: 'approved' }, + } as never); + expect(second.status).toBe('paused'); + expect(second.runId).toBe(first.runId); + + const run = await engine.getRun(first.runId as string); + expect(run?.status).toBe('paused'); + expect(run?.variables).toBeDefined(); + const vars = run!.variables!; + + // Everything stage 1 produced survives the resume… + expect(vars['stage1.pending_approvers']).toEqual(['user_ops', 'user_finance']); + // …the resume signal's own write is there… + expect(vars['stage1.verdict']).toBe('approved'); + // …and so is what stage 2 resolved before it paused in turn. + expect(vars['stage2.approver_pool']).toEqual(['user_cfo']); + }); + + it('is the SAME object the continuation resumes from (snapshot, not a re-read)', async () => { + const store = new InMemorySuspendedRunStore(); + const engine = new AutomationEngine(silent, store); + registerStage(engine, 'stage1_pause', STAGE1_OUTPUT); + registerStage(engine, 'stage2_pause', { stage: 2 }); + engine.registerFlow('approval_flow', twoStageFlow('approval_flow') as never); + + const res = await engine.execute('approval_flow', { + event: 'test', + record: { id: 'ord_3' }, + } as unknown as AutomationContext); + + const run = await engine.getRun(res.runId as string); + const suspended = await store.load(res.runId as string); + + // What run-detail shows and what the run will actually resume from are + // one snapshot expression, so they cannot drift apart. + expect(run?.variables).toEqual(suspended?.variables); + }); + + it('applies to `variables` exactly the shaping `output` already gets — none', async () => { + // The parity fixture: ONE value, driven down both paths. + // + // Path A — a run that COMPLETES: the value leaves through `output`, + // which the completed `recordLog` sites have always carried. + // Path B — a run that PAUSES holding the identical value: it leaves + // through `variables`. + // + // Deep-equal is the whole assertion. The run-detail read projects, + // redacts and masks nothing today (the handler answers `deps.success(run)` + // with the log entry as recorded), so parity means "unshaped, both". If + // a future change starts shaping either field, this fails. + const payload = { + pending_approvers: ['user_ops', 'user_finance'], + decision: { route: 'dual', nested: { deeper: [1, 2, { leaf: true }] } }, + blank: null, + }; + + const engine = new AutomationEngine(silent, new InMemorySuspendedRunStore()); + engine.registerNodeExecutor({ + type: 'emit', async execute() { return { success: true, output: { payload } }; }, + } as never); + registerStage(engine, 'hold', {}); + const nodes = (extra: unknown[]) => [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'emit', type: 'emit', label: 'Emit' }, + ...extra, + { id: 'end', type: 'end', label: 'End' }, + ]; + + // Path A — terminal run, value surfaced through `output`. + engine.registerFlow('done_flow', { + name: 'done_flow', label: 'done', type: 'autolaunched', + variables: [{ name: 'carried', type: 'object', isOutput: true, defaultValue: payload }], + nodes: nodes([]), + edges: [ + { id: 'e1', source: 'start', target: 'emit' }, + { id: 'e2', source: 'emit', target: 'end' }, + ], + } as never); + const doneRes = await engine.execute('done_flow', { event: 'test' } as AutomationContext); + const doneRun = await engine.getRun(doneRes.runId as string ?? ''); + const viaOutput = (doneRun?.output as Record | undefined)?.carried + ?? (doneRes.output as Record).carried; + + // Path B — paused run, the same value surfaced through `variables`. + engine.registerFlow('paused_flow', { + name: 'paused_flow', label: 'paused', type: 'autolaunched', + variables: [{ name: 'carried', type: 'object', defaultValue: payload }], + nodes: nodes([{ id: 'hold', type: 'hold', label: 'Hold' }]), + edges: [ + { id: 'e1', source: 'start', target: 'emit' }, + { id: 'e2', source: 'emit', target: 'hold' }, + { id: 'e3', source: 'hold', target: 'end' }, + ], + } as never); + const pausedRes = await engine.execute('paused_flow', { event: 'test' } as AutomationContext); + expect(pausedRes.status).toBe('paused'); + const pausedRun = await engine.getRun(pausedRes.runId as string); + const viaVariables = pausedRun?.variables?.carried; + + expect(viaVariables).toEqual(viaOutput); + expect(viaVariables).toEqual(payload); + + // Same for a NODE-produced value: `emit.payload` rides the variables + // snapshot unshaped, exactly as `output` carried it on the other run. + expect(pausedRun?.variables?.['emit.payload']).toEqual(payload); + }); +});