diff --git a/.changeset/dev-reload-scheduled-flow-rebind.md b/.changeset/dev-reload-scheduled-flow-rebind.md new file mode 100644 index 0000000000..7912e03cdc --- /dev/null +++ b/.changeset/dev-reload-scheduled-flow-rebind.md @@ -0,0 +1,26 @@ +--- +"@objectstack/service-automation": patch +"@objectstack/metadata": patch +--- + +fix(automation): re-bind scheduled-flow jobs on `os dev` hot-reload + +Editing a schedule-triggered flow under `objectstack dev` silently kept firing +the OLD definition until a full server restart. The dev watcher recompiles +`dist/objectstack.json` and MetadataPlugin reloads it into the MetadataManager +(so GET /meta reads + UI HMR are fresh), but the AutomationEngine pulls its flow +definitions and trigger/job bindings ONCE at boot — nothing re-registered them +on reload. So the scheduled job bound at boot kept running the pre-edit flow +(old `runAs`, schedule, or logic) on its timer, with no signal that the edit had +no effect. + +Fix: MetadataPlugin now fires a generic `metadata:reloaded` hook after each +artifact reload (the HMR POST handler and the server-side artifact-file watcher; +never on the initial boot load). AutomationServicePlugin subscribes and re-syncs +the engine from the metadata service — re-registering every current flow +(idempotent: `registerFlow` re-binds the trigger, and `ScheduleTrigger.start` +cancels + reschedules the job) and unregistering flows removed from the artifact +so their jobs stop firing. This covers all auto-triggered flow types +(schedule / record-change / api), not just scheduled ones, since record-change +flows were also executing their boot-time definitions after an edit. Production +deployments are unaffected — nothing reloads the artifact there. diff --git a/packages/metadata/src/plugin-hmr-reload.test.ts b/packages/metadata/src/plugin-hmr-reload.test.ts new file mode 100644 index 0000000000..82335e24cd --- /dev/null +++ b/packages/metadata/src/plugin-hmr-reload.test.ts @@ -0,0 +1,94 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// Regression: the `os dev` artifact-reload path (HMR POST handler + server-side +// artifact-file watcher) must announce a generic `metadata:reloaded` hook AFTER +// re-loading the artifact into the MetadataManager. Runtime consumers that cached +// boot-time metadata re-sync on that signal — notably the automation engine, +// which re-binds flow triggers (incl. scheduled jobs) it pulled once at boot. +// Without the announce, an edited schedule-triggered flow keeps firing its +// pre-edit definition until a full restart. + +import { describe, it, expect, vi } from 'vitest'; +import { mkdtempSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { MetadataPlugin } from './plugin'; +import type { NodeMetadataManager } from './node-metadata-manager'; + +function fakeCtx() { + return { + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + trigger: vi.fn(async () => {}), + } as any; +} + +function writeArtifact(flowName: string): string { + const dir = mkdtempSync(join(tmpdir(), 'os-hmr-')); + const file = join(dir, 'objectstack.json'); + const artifact = { + id: 'com.example.test', + name: 'test', + version: '0.0.0', + type: 'app', + scope: 'app', + namespace: 'test', + defaultDatasource: 'memory', + flows: [ + { + name: flowName, + label: flowName, + type: 'schedule', + runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'Start', config: { schedule: { type: 'interval', intervalMs: 1000 } } }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + }, + ], + }; + writeFileSync(file, JSON.stringify(artifact), 'utf8'); + return file; +} + +describe('MetadataPlugin._reloadAndAnnounce — fires metadata:reloaded after reload', () => { + it('reloads the artifact into the manager THEN announces metadata:reloaded', async () => { + const plugin = new MetadataPlugin({ + watch: false, + config: { bootstrap: 'eager' }, + environmentId: 'proj_test', + }); + const mgr = (plugin as any).manager as NodeMetadataManager; + const ctx = fakeCtx(); + const file = writeArtifact('sweep'); + + await (plugin as any)._reloadAndAnnounce(ctx, { path: file, fetchTimeoutMs: undefined }, [file]); + + // The fresh flow landed in the metadata manager (so the re-sync's + // metadata.list('flow') will see the edited definition)… + const registered = await mgr.get('flow', 'sweep'); + expect(registered).toBeDefined(); + + // …and the generic reload signal fired with the changed path. + expect(ctx.trigger).toHaveBeenCalledTimes(1); + expect(ctx.trigger).toHaveBeenCalledWith('metadata:reloaded', { changed: [file] }); + }); + + it('still announces even if a subscriber throws (reload must not break)', async () => { + const plugin = new MetadataPlugin({ + watch: false, + config: { bootstrap: 'eager' }, + environmentId: 'proj_test', + }); + const ctx = fakeCtx(); + ctx.trigger = vi.fn(async () => { throw new Error('subscriber boom'); }); + const file = writeArtifact('sweep2'); + + await expect( + (plugin as any)._reloadAndAnnounce(ctx, { path: file, fetchTimeoutMs: undefined }, [file]), + ).resolves.toBeUndefined(); + + expect(ctx.trigger).toHaveBeenCalledTimes(1); + expect(ctx.logger.warn).toHaveBeenCalled(); + }); +}); diff --git a/packages/metadata/src/plugin.ts b/packages/metadata/src/plugin.ts index 26fafdd78b..e3f22c8590 100644 --- a/packages/metadata/src/plugin.ts +++ b/packages/metadata/src/plugin.ts @@ -354,7 +354,7 @@ export class MetadataPlugin implements Plugin { const src = this.options.artifactSource; if (src?.mode === 'local-file') { try { - await this._loadFromLocalFile(ctx, src.path, src.fetchTimeoutMs); + await this._reloadAndAnnounce(ctx, src, body?.changed ?? [src.path]); ctx.logger.info('[MetadataPlugin] artifact reloaded via HMR POST', { path: src.path, reason: body?.reason, @@ -405,7 +405,7 @@ export class MetadataPlugin implements Plugin { if (pending) return; pending = true; try { - await this._loadFromLocalFile(ctx, src.path, src.fetchTimeoutMs); + await this._reloadAndAnnounce(ctx, src, [src.path]); hub.broadcastReload('artifact-file-changed', [src.path]); ctx.logger.info('[MetadataPlugin] artifact auto-reloaded (file watcher)', { path: src.path, @@ -595,6 +595,32 @@ export class MetadataPlugin implements Plugin { return totalRegistered; } + /** + * Reload the artifact from disk into the MetadataManager, then announce a + * generic `metadata:reloaded` hook. Used by BOTH reload paths (the HMR POST + * handler and the server-side artifact-file watcher) — but NOT the initial + * boot load, which other plugins already consume directly. + * + * Runtime consumers that cached boot-time metadata re-sync on this signal. + * The automation engine subscribes to re-bind flow triggers it pulled ONCE + * at boot — notably scheduled jobs: without this, an edited + * schedule-triggered flow keeps firing its pre-edit definition (old runAs / + * schedule / logic) until a full process restart. A subscriber failure is + * logged but never blocks the reload. + */ + private async _reloadAndAnnounce( + ctx: PluginContext, + src: { path: string; fetchTimeoutMs?: number }, + changed: string[], + ): Promise { + await this._loadFromLocalFile(ctx, src.path, src.fetchTimeoutMs); + try { + await ctx.trigger('metadata:reloaded', { changed }); + } catch (e: any) { + ctx.logger.warn('[MetadataPlugin] metadata:reloaded subscriber failed', { error: e?.message }); + } + } + private async _loadFromLocalFile(ctx: PluginContext, filePath: string, fetchTimeoutMs?: number): Promise { const isUrl = /^https?:\/\//i.test(filePath); ctx.logger.info( diff --git a/packages/services/service-automation/src/flow-hot-reload.test.ts b/packages/services/service-automation/src/flow-hot-reload.test.ts new file mode 100644 index 0000000000..bd419c0bcb --- /dev/null +++ b/packages/services/service-automation/src/flow-hot-reload.test.ts @@ -0,0 +1,178 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// Regression: `os dev` recompiles dist/objectstack.json on a src edit, and +// MetadataPlugin reloads it into the metadata service + fires 'metadata:reloaded'. +// The automation engine, however, pulled its flow definitions + trigger bindings +// ONCE at boot — so without re-syncing, an edited SCHEDULE-triggered flow keeps +// firing its OLD definition (old runAs / schedule / logic) until a full restart. +// +// This proves the fix end-to-end on the automation side: the 'metadata:reloaded' +// hook re-registers every current flow (re-binding its trigger — for a scheduled +// flow that is the job cancel + reschedule the ScheduleTrigger performs) and +// tears down flows that vanished from the artifact. A recording trigger stands in +// for the concrete ScheduleTrigger (whose idempotent job re-bind is covered by +// trigger-schedule's schedule-runas-e2e test) so this stays dependency-light. + +import { describe, it, expect } from 'vitest'; +import { LiteKernel } from '@objectstack/core'; +import { AutomationEngine } from './engine.js'; +import { AutomationServicePlugin } from './plugin.js'; +import type { FlowTrigger, FlowTriggerBinding } from './engine.js'; +import type { AutomationContext } from '@objectstack/spec/contracts'; + +const flush = () => new Promise((r) => setTimeout(r, 0)); + +/** A schedule-triggered flow that touches data, parameterized by runAs + interval. */ +function scheduledFlow(name: string, runAs: 'system' | 'user', intervalMs: number) { + return { + name, + label: name, + type: 'schedule', + runAs, + nodes: [ + { id: 'start', type: 'start', label: 'Start', config: { schedule: { type: 'interval', intervalMs } } }, + { id: 'mk', type: 'create_record', label: 'Create', config: { objectName: 'thing', fields: { a: 1 } } }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'mk' }, + { id: 'e2', source: 'mk', target: 'end' }, + ], + }; +} + +/** + * A FlowTrigger of type 'schedule' that records start/stop and, like the real + * ScheduleTrigger, keeps exactly one binding per flow (re-bind drops the prior). + * `fire()` invokes the engine callback the way a cron tick would. + */ +function recordingScheduleTrigger() { + const bound = new Map Promise }>(); + const events: Array<{ op: 'start' | 'stop'; flow: string }> = []; + const trigger: FlowTrigger = { + type: 'schedule', + start(binding: FlowTriggerBinding, cb: (ctx: AutomationContext) => Promise) { + const schedule = binding.schedule ?? (binding.config as any)?.schedule; + bound.set(binding.flowName, { schedule, cb }); + events.push({ op: 'start', flow: binding.flowName }); + }, + stop(flowName: string) { + if (bound.delete(flowName)) events.push({ op: 'stop', flow: flowName }); + }, + }; + return { + trigger, + has: (n: string) => bound.has(n), + intervalOf: (n: string) => bound.get(n)?.schedule?.intervalMs, + fire: async (n: string) => { + await bound.get(n)?.cb({ event: 'schedule', params: { flowName: n } } as AutomationContext); + }, + events, + }; +} + +/** Stub `create_record` executor that captures the runAs it is handed. */ +function captureRunAs(engine: AutomationEngine): Array { + const seen: Array = []; + engine.registerNodeExecutor({ + type: 'create_record', + async execute(_node: unknown, _vars: unknown, context: AutomationContext) { + seen.push(context.runAs); + return { success: true, output: {} }; + }, + } as never); + return seen; +} + +/** A mutable fake `metadata` service exposing just the `list('flow')` the re-sync uses. */ +function fakeMetadataService(initial: unknown[]) { + let flows = initial; + return { + service: { + async list(type: string) { + return type === 'flow' ? flows : []; + }, + }, + setFlows: (next: unknown[]) => { flows = next; }, + }; +} + +async function bootKernel(meta: { service: unknown }) { + const kernel = new LiteKernel({ logger: { level: 'silent' } } as never); + const harness = { + name: 'test.harness', + type: 'standard' as const, + version: '1.0.0', + dependencies: [] as string[], + async init(ctx: any) { + ctx.registerService('metadata', meta.service); + }, + async start() {}, + }; + kernel.use(harness as never); + kernel.use(new AutomationServicePlugin()); + await kernel.bootstrap(); + return kernel; +} + +const reload = (kernel: LiteKernel) => (kernel as any).context.trigger('metadata:reloaded', {}); + +describe("scheduled flow hot-reload re-bind (metadata:reloaded re-sync)", () => { + it('re-binds an edited scheduled flow to its NEW definition without a restart', async () => { + const meta = fakeMetadataService([scheduledFlow('sweep', 'user', 1000)]); + const kernel = await bootKernel(meta); + const engine = kernel.getService('automation'); + const seen = captureRunAs(engine); + const sched = recordingScheduleTrigger(); + engine.registerTrigger(sched.trigger); + + // First reload binds v1 (runAs:user, interval 1000) — like the boot bind. + await reload(kernel); + await flush(); + expect(sched.has('sweep'), 'flow bound to the schedule trigger').toBe(true); + expect(sched.intervalOf('sweep')).toBe(1000); + await sched.fire('sweep'); + expect(seen[seen.length - 1], 'v1 runs as user').toBe('user'); + + // Edit the flow: runAs:system + interval 5000. Recompile → reload. + sched.events.length = 0; + meta.setFlows([scheduledFlow('sweep', 'system', 5000)]); + await reload(kernel); + await flush(); + + // The OLD binding was torn down and a NEW one created (not left stale). + expect(sched.events).toEqual([ + { op: 'stop', flow: 'sweep' }, + { op: 'start', flow: 'sweep' }, + ]); + expect(sched.intervalOf('sweep'), 'schedule re-bound to the new interval').toBe(5000); + + // The fired job now runs the NEW definition — the actual footgun fixed. + await sched.fire('sweep'); + expect(seen[seen.length - 1], 'v2 runs elevated as system after reload').toBe('system'); + + await kernel.shutdown(); + }); + + it('tears down a scheduled flow that was deleted from the artifact', async () => { + const meta = fakeMetadataService([scheduledFlow('sweep', 'user', 1000)]); + const kernel = await bootKernel(meta); + const engine = kernel.getService('automation'); + const sched = recordingScheduleTrigger(); + engine.registerTrigger(sched.trigger); + + await reload(kernel); + await flush(); + expect(sched.has('sweep')).toBe(true); + + // Delete the flow file → recompiled artifact no longer carries it. + meta.setFlows([]); + await reload(kernel); + await flush(); + + expect(sched.has('sweep'), 'deleted flow unbound — its job stops firing').toBe(false); + expect(await engine.listFlows()).not.toContain('sweep'); + + await kernel.shutdown(); + }); +}); diff --git a/packages/services/service-automation/src/plugin.ts b/packages/services/service-automation/src/plugin.ts index 6e39955874..012b7e6d83 100644 --- a/packages/services/service-automation/src/plugin.ts +++ b/packages/services/service-automation/src/plugin.ts @@ -72,6 +72,15 @@ export class AutomationServicePlugin implements Plugin { private engine?: AutomationEngine; private readonly options: AutomationServicePluginOptions; + /** + * Flow names this plugin has registered into the engine from the + * artifact / ObjectQL registry, tracked so a `metadata:reloaded` re-sync + * can tear down flows that were removed from the artifact (stopping their + * triggers/jobs). Seeded by the boot pull, then replaced on each re-sync. + * Plugin-contributed node packs never enter this set, so re-sync never + * unregisters them. + */ + private syncedFlowNames = new Set(); constructor(options: AutomationServicePluginOptions = {}) { this.options = options; @@ -201,6 +210,7 @@ export class AutomationServicePlugin implements Plugin { if (!def?.name) continue; try { this.engine.registerFlow(def.name, def as never); + this.syncedFlowNames.add(def.name); registered++; } catch (e) { const msg = e instanceof Error ? e.message : String(e); @@ -215,6 +225,20 @@ export class AutomationServicePlugin implements Plugin { ctx.logger.warn(`[Automation] flow pull from ObjectQL registry failed: ${msg}`); } + // ── Dev hot-reload: re-bind flow triggers when the artifact recompiles ── + // `os dev` recompiles dist/objectstack.json on a src edit; MetadataPlugin + // reloads it into the metadata service and fires 'metadata:reloaded'. The + // engine, however, still holds the flow definitions + trigger bindings it + // pulled ONCE above — including scheduled jobs bound through the schedule + // trigger. Without re-syncing, an edited schedule-triggered flow keeps + // firing its OLD definition (old runAs / schedule / logic) until a full + // restart. Re-register every current flow (registerFlow re-binds its + // trigger idempotently — ScheduleTrigger.start cancels + reschedules) and + // unregister flows that vanished from the artifact so their jobs stop. + ctx.hook('metadata:reloaded', async () => { + await this.resyncFlowsFromMetadata(ctx); + }); + // ADR-0019 follow-up: re-arm auto-resume timers for runs that were // suspended at a timer-`wait` node when the process went down. Must run // *after* the flow pull above — resume() needs the flow definitions @@ -235,6 +259,79 @@ export class AutomationServicePlugin implements Plugin { } } + /** + * Re-pull flow definitions from the metadata service and re-register them + * into the engine, so an `os dev` artifact recompile re-binds flow triggers + * (notably scheduled jobs) instead of leaving the engine executing the + * boot-time definitions. Driven by the `metadata:reloaded` hook that + * MetadataPlugin fires after reloading the artifact from disk. + * + * Pulls from the metadata service — NOT the ObjectQL schema registry used by + * the boot pull. The schema registry is a boot-time cache the artifact reload + * does not refresh; `MetadataManager.register()` (the reload's write path) IS + * refreshed, so the metadata service is the only source carrying the edited + * definitions. + * + * Idempotent and best-effort: registerFlow() re-binds the trigger + * (ScheduleTrigger.start cancels + reschedules), flows removed from the + * artifact are unregistered so their jobs stop firing, and any failure is + * logged without disturbing the rest of the runtime. + */ + private async resyncFlowsFromMetadata(ctx: PluginContext): Promise { + if (!this.engine) return; + let metadata: { list?(type: string): Promise } | undefined; + try { + metadata = ctx.getService('metadata'); + } catch { + // No metadata service (e.g. a bare engine / tests) — nothing to sync. + return; + } + if (!metadata || typeof metadata.list !== 'function') return; + + let defs: unknown[]; + try { + defs = await metadata.list('flow'); + } catch (err) { + ctx.logger.warn( + `[Automation] flow re-sync skipped: metadata.list('flow') failed: ${(err as Error).message}`, + ); + return; + } + + const freshNames = new Set(); + let resynced = 0; + for (const d of defs) { + const def = d as { name?: string }; + if (!def?.name) continue; + freshNames.add(def.name); + try { + this.engine.registerFlow(def.name, def as never); + resynced++; + } catch (err) { + ctx.logger.warn( + `[Automation] flow re-sync: failed to register ${def.name}: ${(err as Error).message}`, + ); + } + } + + // Tear down flows that were synced from a prior artifact but are gone + // now, so their triggers/jobs (e.g. a scheduled job) stop firing. + for (const prev of this.syncedFlowNames) { + if (!freshNames.has(prev)) { + try { + this.engine.unregisterFlow(prev); + } catch { + /* best-effort */ + } + } + } + this.syncedFlowNames = freshNames; + + if (resynced > 0) { + ctx.logger.info(`[Automation] Re-synced ${resynced} flow(s) after metadata reload`); + } + } + async destroy(): Promise { this.engine = undefined; }