fix(automation): re-bind scheduled-flow jobs on os dev hot-reload

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

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

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<void> {
+        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<void> {
         const isUrl = /^https?:\/\//i.test(filePath);
         ctx.logger.info(

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<void>((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<string, { schedule: any; cb: (ctx: AutomationContext) => Promise<void> }>();
+    const events: Array<{ op: 'start' | 'stop'; flow: string }> = [];
+    const trigger: FlowTrigger = {
+        type: 'schedule',
+        start(binding: FlowTriggerBinding, cb: (ctx: AutomationContext) => Promise<void>) {
+            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<string | undefined> {
+    const seen: Array<string | undefined> = [];
+    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<AutomationEngine>('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<AutomationEngine>('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();
+    });
+});