|
| 1 | +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. |
| 2 | +// |
| 3 | +// [#20677] ADR-0126 §7.2 — a switched-off flow stays UNBOUND whenever its |
| 4 | +// trigger type arrives, and a restart does not re-arm it. |
| 5 | +// |
| 6 | +// ## The boot order this pins |
| 7 | +// |
| 8 | +// `AutomationServicePlugin.start()` pulls the flows (`registerFlow`) and then |
| 9 | +// applies the activation ledger (`hydrateFlowActivations()`), which unbinds |
| 10 | +// every flow a row switches off. The trigger plugins register their triggers |
| 11 | +// LATER, at `kernel:ready` — and `registerTrigger` arms every registered, |
| 12 | +// unbound flow whose trigger type matches. Before the fix it did so without |
| 13 | +// asking whether the flow may run at all, so a flow the administrator switched |
| 14 | +// off came back `bound: true` after every cold boot, and each matching event |
| 15 | +// then logged an ERROR claiming a run-history row that was never written. |
| 16 | +// |
| 17 | +// ## The shape of the fix these pins hold |
| 18 | +// |
| 19 | +// ONE gate, in `activateFlowTrigger` itself: it refuses to arm a flow |
| 20 | +// `isFlowEnabled` answers `false` for, so `registerFlow`, `registerTrigger`, |
| 21 | +// the enable toggle and any arming path added later all inherit it. The pins |
| 22 | +// therefore drive the arming paths from outside rather than asserting on |
| 23 | +// `registerTrigger`'s body. |
| 24 | + |
| 25 | +import { describe, it, expect, vi } from 'vitest'; |
| 26 | +import { AutomationEngine } from './engine.js'; |
| 27 | +import type { FlowTrigger, FlowTriggerBinding, FlowActivationStore } from './engine.js'; |
| 28 | +import { InMemoryFlowActivationStore } from './flow-activation-store.js'; |
| 29 | +import type { AutomationContext } from '@objectstack/spec/contracts'; |
| 30 | +import { withScheduledWorkOn } from './deployment-switch.test-support.js'; |
| 31 | + |
| 32 | +function createTestLogger(): any { |
| 33 | + const l: any = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }; |
| 34 | + l.child = () => l; |
| 35 | + return l; |
| 36 | +} |
| 37 | + |
| 38 | +/** A packaged, runnable flow whose `start` config decides its trigger kind. */ |
| 39 | +function packagedFlow(name: string, startConfig: Record<string, unknown>, extra: Record<string, unknown> = {}) { |
| 40 | + return { |
| 41 | + name, |
| 42 | + label: name, |
| 43 | + type: 'autolaunched', |
| 44 | + nodes: [ |
| 45 | + { id: 'start', type: 'start', label: 'Start', config: startConfig }, |
| 46 | + { id: 'end', type: 'end', label: 'End' }, |
| 47 | + ], |
| 48 | + edges: [{ id: 'e1', source: 'start', target: 'end' }], |
| 49 | + _packageId: 'crm', |
| 50 | + ...extra, |
| 51 | + }; |
| 52 | +} |
| 53 | + |
| 54 | +/** A recording trigger: binding state is read off the trigger, never inferred. */ |
| 55 | +function recordingTrigger(type: string) { |
| 56 | + const bound = new Map<string, (ctx: AutomationContext) => Promise<void>>(); |
| 57 | + const trigger: FlowTrigger = { |
| 58 | + type, |
| 59 | + start(binding: FlowTriggerBinding, cb: (ctx: AutomationContext) => Promise<void>) { |
| 60 | + bound.set(binding.flowName, cb); |
| 61 | + }, |
| 62 | + stop(flowName: string) { |
| 63 | + bound.delete(flowName); |
| 64 | + }, |
| 65 | + }; |
| 66 | + return { trigger, isBound: (n: string) => bound.has(n), callbackOf: (n: string) => bound.get(n) }; |
| 67 | +} |
| 68 | + |
| 69 | +/** The runtime-state row `GET /automation/_status` serves for one flow. */ |
| 70 | +function stateOf(engine: AutomationEngine, name: string) { |
| 71 | + const row = engine.getFlowRuntimeStates().find((s) => s.name === name); |
| 72 | + return row && { enabled: row.enabled, bound: row.bound }; |
| 73 | +} |
| 74 | + |
| 75 | +const RECORD_CHANGE = { objectName: 'task', triggerType: 'record-after-create' }; |
| 76 | + |
| 77 | +// Time-triggered kinds arm only where the deployment runs package-authored |
| 78 | +// scheduled work (off by default); without this every schedule / time-relative |
| 79 | +// assertion below would fail for a reason unrelated to the ledger. |
| 80 | +withScheduledWorkOn(); |
| 81 | + |
| 82 | +describe('a trigger type registered AFTER hydrateFlowActivations() does not arm a ledger-disabled flow', () => { |
| 83 | + // Every kind a trigger plugin registers at `kernel:ready`. The disabled |
| 84 | + // flow and its enabled sibling share the kind, so the sibling is the |
| 85 | + // control that the trigger registration really did arm what it should. |
| 86 | + const kinds: Array<[string, Record<string, unknown>]> = [ |
| 87 | + ['record_change', RECORD_CHANGE], |
| 88 | + ['schedule', { schedule: '0 9 * * *' }], |
| 89 | + ['time_relative', { timeRelative: { object: 'task', field: 'due_at' }, schedule: '0 * * * *' }], |
| 90 | + // An `api` flow registers only with its per-flow secret (ADR-0041). |
| 91 | + ['api', { triggerType: 'api', secret: 'hook-secret' }], |
| 92 | + ]; |
| 93 | + |
| 94 | + for (const [kind, startConfig] of kinds) { |
| 95 | + it(`${kind}: the switched-off flow stays unbound, its enabled sibling arms`, async () => { |
| 96 | + const store = new InMemoryFlowActivationStore(); |
| 97 | + await store.setActive({ name: 'off', packageId: 'crm', active: false }); |
| 98 | + |
| 99 | + const engine = new AutomationEngine(createTestLogger()); |
| 100 | + engine.setFlowActivationStore(store); |
| 101 | + engine.registerFlow('off', packagedFlow('off', startConfig)); |
| 102 | + engine.registerFlow('on', packagedFlow('on', startConfig)); |
| 103 | + expect(await engine.hydrateFlowActivations()).toEqual(['off']); |
| 104 | + |
| 105 | + // The trigger plugin's `kernel:ready` registration, after hydration. |
| 106 | + const trigger = recordingTrigger(kind); |
| 107 | + engine.registerTrigger(trigger.trigger); |
| 108 | + |
| 109 | + expect(trigger.isBound('off')).toBe(false); |
| 110 | + expect(stateOf(engine, 'off')).toEqual({ enabled: false, bound: false }); |
| 111 | + // The control: the same registration armed the enabled sibling. |
| 112 | + expect(trigger.isBound('on')).toBe(true); |
| 113 | + expect(stateOf(engine, 'on')).toEqual({ enabled: true, bound: true }); |
| 114 | + }); |
| 115 | + } |
| 116 | + |
| 117 | + it('a STATUS-disabled flow is not armed by a later trigger registration either — the gate is `isFlowEnabled`, both dimensions', async () => { |
| 118 | + const engine = new AutomationEngine(createTestLogger()); |
| 119 | + engine.registerFlow('retired', packagedFlow('retired', RECORD_CHANGE, { status: 'obsolete' })); |
| 120 | + engine.registerFlow('live', packagedFlow('live', RECORD_CHANGE, { status: 'active' })); |
| 121 | + |
| 122 | + const trigger = recordingTrigger('record_change'); |
| 123 | + engine.registerTrigger(trigger.trigger); |
| 124 | + |
| 125 | + expect(trigger.isBound('retired')).toBe(false); |
| 126 | + expect(stateOf(engine, 'retired')).toEqual({ enabled: false, bound: false }); |
| 127 | + expect(trigger.isBound('live')).toBe(true); |
| 128 | + }); |
| 129 | +}); |
| 130 | + |
| 131 | +describe('the enable path still arms', () => { |
| 132 | + it('re-enabling a flow hydrated as disabled arms it on the trigger that registered after hydration', async () => { |
| 133 | + const store = new InMemoryFlowActivationStore(); |
| 134 | + await store.setActive({ name: 'f', packageId: 'crm', active: false }); |
| 135 | + const engine = new AutomationEngine(createTestLogger()); |
| 136 | + engine.setFlowActivationStore(store); |
| 137 | + engine.registerFlow('f', packagedFlow('f', RECORD_CHANGE)); |
| 138 | + await engine.hydrateFlowActivations(); |
| 139 | + const trigger = recordingTrigger('record_change'); |
| 140 | + engine.registerTrigger(trigger.trigger); |
| 141 | + expect(trigger.isBound('f')).toBe(false); |
| 142 | + |
| 143 | + await engine.toggleFlow('f', true); |
| 144 | + |
| 145 | + expect(trigger.isBound('f')).toBe(true); |
| 146 | + expect(stateOf(engine, 'f')).toEqual({ enabled: true, bound: true }); |
| 147 | + expect(await store.list()).toEqual([{ name: 'f', packageId: 'crm', active: true }]); |
| 148 | + expect((await engine.execute('f')).success).toBe(true); |
| 149 | + }); |
| 150 | + |
| 151 | + it('re-enabling the LEDGER bit of a flow whose STATUS is obsolete does not arm it — it still cannot run', async () => { |
| 152 | + const engine = new AutomationEngine(createTestLogger()); |
| 153 | + engine.setFlowActivationStore(new InMemoryFlowActivationStore()); |
| 154 | + const trigger = recordingTrigger('record_change'); |
| 155 | + engine.registerTrigger(trigger.trigger); |
| 156 | + engine.registerFlow('retired', packagedFlow('retired', RECORD_CHANGE, { status: 'obsolete' })); |
| 157 | + |
| 158 | + await engine.toggleFlow('retired', true); |
| 159 | + |
| 160 | + // Armed, it would refuse every event it fired (`FLOW_DISABLED`, the |
| 161 | + // status dimension) — the same armed-but-cannot-run state as a |
| 162 | + // ledger-disabled flow, reached through the toggle instead. |
| 163 | + expect(trigger.isBound('retired')).toBe(false); |
| 164 | + expect(stateOf(engine, 'retired')).toEqual({ enabled: false, bound: false }); |
| 165 | + expect((await engine.execute('retired')).code).toBe('FLOW_DISABLED'); |
| 166 | + }); |
| 167 | +}); |
| 168 | + |
| 169 | +describe('a cold restart over the same store keeps a switched-off flow enabled:false, bound:false', () => { |
| 170 | + /** |
| 171 | + * One process lifetime, in `AutomationServicePlugin`'s boot order: |
| 172 | + * `start()` pulls the flows and hydrates the ledger; at `kernel:ready` the |
| 173 | + * protocol sync re-registers every flow and the trigger plugin registers |
| 174 | + * its trigger; `kernel:bootstrapped` seals the node-type vocabulary. Only |
| 175 | + * the STORE outlives the engine, as the `sys_metadata_activation` rows |
| 176 | + * outlive the process. |
| 177 | + */ |
| 178 | + async function boot(store: FlowActivationStore) { |
| 179 | + const engine = new AutomationEngine(createTestLogger()); |
| 180 | + engine.setFlowActivationStore(store); |
| 181 | + const defs = [packagedFlow('urgent_alert', RECORD_CHANGE), packagedFlow('welcome', RECORD_CHANGE)]; |
| 182 | + for (const def of defs) engine.registerFlow(def.name, def); // the boot pull |
| 183 | + await engine.hydrateFlowActivations(); |
| 184 | + for (const def of defs) engine.registerFlow(def.name, def); // kernel:ready protocol sync |
| 185 | + const trigger = recordingTrigger('record_change'); |
| 186 | + engine.registerTrigger(trigger.trigger); // kernel:ready trigger plugin |
| 187 | + engine.sealNodeTypeVocabulary(); // kernel:bootstrapped |
| 188 | + return { engine, trigger }; |
| 189 | + } |
| 190 | + |
| 191 | + it('survives two consecutive cold boots; the enabled sibling arms on each', async () => { |
| 192 | + const store = new InMemoryFlowActivationStore(); |
| 193 | + |
| 194 | + const first = await boot(store); |
| 195 | + expect(stateOf(first.engine, 'urgent_alert')).toEqual({ enabled: true, bound: true }); |
| 196 | + await first.engine.toggleFlow('urgent_alert', false); |
| 197 | + expect(stateOf(first.engine, 'urgent_alert')).toEqual({ enabled: false, bound: false }); |
| 198 | + |
| 199 | + for (const restart of [1, 2]) { |
| 200 | + const { engine, trigger } = await boot(store); |
| 201 | + expect(stateOf(engine, 'urgent_alert'), `restart ${restart}`).toEqual({ enabled: false, bound: false }); |
| 202 | + expect(trigger.isBound('urgent_alert'), `restart ${restart}`).toBe(false); |
| 203 | + expect(stateOf(engine, 'welcome'), `restart ${restart}`).toEqual({ enabled: true, bound: true }); |
| 204 | + // The binding audit reads the same bookkeeping: a disabled flow is |
| 205 | + // not reported as a binding failure either. |
| 206 | + expect(engine.getTriggerBindingAudit(), `restart ${restart}`).toEqual([]); |
| 207 | + } |
| 208 | + }); |
| 209 | +}); |
| 210 | + |
| 211 | +describe('the trigger-fired failure line claims a run-history row only when one is written', () => { |
| 212 | + const flush = () => new Promise<void>((r) => setTimeout(r, 0)); |
| 213 | + |
| 214 | + function loggedLines(logger: any, level: 'info' | 'error') { |
| 215 | + return logger[level].mock.calls.map((c: any[]) => String(c[0])); |
| 216 | + } |
| 217 | + |
| 218 | + it('a FLOW_DISABLED refusal reaching the callback (an event in flight when the flow was switched off) is not logged as a failure', async () => { |
| 219 | + const logger = createTestLogger(); |
| 220 | + const engine = new AutomationEngine(logger); |
| 221 | + engine.setFlowActivationStore(new InMemoryFlowActivationStore()); |
| 222 | + const trigger = recordingTrigger('record_change'); |
| 223 | + engine.registerTrigger(trigger.trigger); |
| 224 | + engine.registerFlow('f', packagedFlow('f', RECORD_CHANGE)); |
| 225 | + // The callback the trigger already holds — an event dispatched before |
| 226 | + // the switch-off is delivered to it after. |
| 227 | + const inFlight = trigger.callbackOf('f'); |
| 228 | + expect(inFlight).toBeTypeOf('function'); |
| 229 | + await engine.toggleFlow('f', false); |
| 230 | + |
| 231 | + await inFlight!({ event: 'record-after-create', object: 'task', record: { id: 't1' } } as AutomationContext); |
| 232 | + await flush(); |
| 233 | + |
| 234 | + expect(loggedLines(logger, 'error').filter((m: string) => m.includes('Trigger-fired run'))).toEqual([]); |
| 235 | + const told = loggedLines(logger, 'info').filter((m: string) => m.includes("flow 'f'") && m.includes('disabled')); |
| 236 | + expect(told).toHaveLength(1); |
| 237 | + expect(told[0]).not.toContain('recorded in the flow\'s run history'); |
| 238 | + // Nothing ran, so nothing may claim a row: there is none. |
| 239 | + expect(await engine.listRuns('f')).toEqual([]); |
| 240 | + }); |
| 241 | + |
| 242 | + it('a run that dispatched and failed still logs at ERROR and still says its failure is in the run history — which holds it', async () => { |
| 243 | + const logger = createTestLogger(); |
| 244 | + const engine = new AutomationEngine(logger); |
| 245 | + const trigger = recordingTrigger('record_change'); |
| 246 | + engine.registerTrigger(trigger.trigger); |
| 247 | + engine.registerNodeExecutor({ type: 'exploder', async execute() { throw new Error('boom'); } } as never); |
| 248 | + engine.registerFlow('failing', { |
| 249 | + ...packagedFlow('failing', RECORD_CHANGE), |
| 250 | + nodes: [ |
| 251 | + { id: 'start', type: 'start', label: 'Start', config: RECORD_CHANGE }, |
| 252 | + { id: 'boom', type: 'exploder', label: 'Boom' }, |
| 253 | + { id: 'end', type: 'end', label: 'End' }, |
| 254 | + ], |
| 255 | + edges: [ |
| 256 | + { id: 'e1', source: 'start', target: 'boom' }, |
| 257 | + { id: 'e2', source: 'boom', target: 'end' }, |
| 258 | + ], |
| 259 | + }); |
| 260 | + |
| 261 | + await trigger.callbackOf('failing')!({ event: 'record-after-create', object: 'task', record: { id: 't1' } } as AutomationContext); |
| 262 | + await flush(); |
| 263 | + |
| 264 | + const errors = loggedLines(logger, 'error').filter((m: string) => m.includes('Trigger-fired run')); |
| 265 | + expect(errors).toHaveLength(1); |
| 266 | + expect(errors[0]).toContain("recorded in the flow's run history"); |
| 267 | + const runs = await engine.listRuns('failing'); |
| 268 | + expect(runs.map((r) => r.status)).toEqual(['failed']); |
| 269 | + }); |
| 270 | + |
| 271 | + it('a failure refused before it dispatched makes no run-history claim — no row was written', async () => { |
| 272 | + const logger = createTestLogger(); |
| 273 | + const engine = new AutomationEngine(logger); |
| 274 | + const trigger = recordingTrigger('record_change'); |
| 275 | + engine.registerTrigger(trigger.trigger); |
| 276 | + engine.registerFlow('gone', packagedFlow('gone', RECORD_CHANGE)); |
| 277 | + const inFlight = trigger.callbackOf('gone'); |
| 278 | + // Unregistered while an event is in flight: `execute()` answers the |
| 279 | + // never-dispatched "not found" exit, which records nothing. |
| 280 | + engine.unregisterFlow('gone'); |
| 281 | + |
| 282 | + await inFlight!({ event: 'record-after-create', object: 'task', record: { id: 't1' } } as AutomationContext); |
| 283 | + await flush(); |
| 284 | + |
| 285 | + const errors = loggedLines(logger, 'error').filter((m: string) => m.includes('Trigger-fired run')); |
| 286 | + expect(errors).toHaveLength(1); |
| 287 | + expect(errors[0]).not.toContain("recorded in the flow's run history"); |
| 288 | + expect(await engine.listRuns('gone')).toEqual([]); |
| 289 | + }); |
| 290 | +}); |
0 commit comments