diff --git a/.changeset/automation-publish-rebind-flows.md b/.changeset/automation-publish-rebind-flows.md new file mode 100644 index 0000000000..5bc331913e --- /dev/null +++ b/.changeset/automation-publish-rebind-flows.md @@ -0,0 +1,39 @@ +--- +"@objectstack/service-automation": patch +"@objectstack/runtime": patch +--- + +fix(automation): bind a flow published while the server runs, without a restart + +Follow-up to #2560 (cold-boot flow binding). A flow **published while the server +is running** — the Studio online-authoring journey: author a record-triggered +automation, publish it, immediately update a matching record — did **not** fire. +Its trigger only bound on the next process restart. + +Two gaps, both fixed: + +1. **The publish path fired no rebind signal.** `POST /packages/:id/publish-drafts` + → `protocol.publishPackageDrafts` promotes the drafts to active but emitted no + event the automation service listens to. The runtime dispatcher now announces + `metadata:reloaded` after a successful publish — the same signal a dev artifact + reload fires (`MetadataPlugin._reloadAndAnnounce`) — so boot-cached consumers + re-sync without a restart. + +2. **The runtime re-sync read the wrong source.** The automation service's + `metadata:reloaded` re-sync pulled `metadata.list('flow')`, which returns 0 in a + real running server (it does not surface inline app flows), so even when the + hook fired it bound nothing. It now reads `protocol.getMetaItems({ type: 'flow' })` + — the same flattened flow view #2560's cold-boot bind and `GET /meta/flow` use — + while keeping the teardown of flows removed from the artifact. A failed or + unavailable protocol read is a no-op and never tears down live flows. + +Production is largely unaffected (a deploy reboots the process, so #2560's +cold-boot bind covers it); this closes the gap for dev and single-instance +Studio authoring. + +Verified end-to-end on a clean instance: authored a record-triggered flow in a +package, published it via `POST /packages/:id/publish-drafts` **without +restarting**, then updated a matching record and observed the flow fire (before +the fix it did not). New regression tests boot a kernel whose protocol serves a +flow only after boot and assert `metadata:reloaded` binds it — and that the +re-sync reads the protocol, not `metadata.list` — both failing on the pre-fix code. diff --git a/packages/runtime/src/http-dispatcher.test.ts b/packages/runtime/src/http-dispatcher.test.ts index 73517f64a1..27984ee528 100644 --- a/packages/runtime/src/http-dispatcher.test.ts +++ b/packages/runtime/src/http-dispatcher.test.ts @@ -971,6 +971,48 @@ describe('HttpDispatcher', () => { expect((result.response as any)?.body?.data?.publishedCount).toBe(3); }); + it('POST /packages/:id/publish-drafts announces metadata:reloaded so boot-cached consumers re-sync', async () => { + // #2560 follow-up: a flow published while the server runs must bind its + // trigger WITHOUT a restart. The publish path fires 'metadata:reloaded' + // — the same signal a dev artifact reload fires — so the automation + // service re-syncs the just-published flow from the protocol. + const publishPackageDrafts = vi.fn().mockResolvedValue({ + success: true, publishedCount: 1, failedCount: 0, + published: [{ type: 'flow', name: 'ticket_closed', version: '1' }], failed: [], + }); + (kernel as any).getService = vi.fn().mockImplementation((name: string) => { + if (name === 'protocol') return Promise.resolve({ publishPackageDrafts }); + if (name === 'objectql') return Promise.resolve({ registry: { getAllPackages: vi.fn().mockReturnValue([]) } }); + return null; + }); + const trigger = vi.fn().mockResolvedValue(undefined); + (kernel as any).context.trigger = trigger; + + const result = await dispatcher.handlePackages('/com.example.ops/publish-drafts', 'POST', {}, {}, { request: {} }); + + expect(result.response?.status).toBe(200); + expect(trigger).toHaveBeenCalledWith( + 'metadata:reloaded', + expect.objectContaining({ changed: expect.arrayContaining(['flow/ticket_closed']) }), + ); + }); + + it('POST /packages/:id/publish-drafts does NOT announce when nothing was published', async () => { + const publishPackageDrafts = vi.fn().mockResolvedValue({ + success: false, publishedCount: 0, failedCount: 0, published: [], failed: [], + }); + (kernel as any).getService = vi.fn().mockImplementation((name: string) => { + if (name === 'protocol') return Promise.resolve({ publishPackageDrafts }); + if (name === 'objectql') return Promise.resolve({ registry: { getAllPackages: vi.fn().mockReturnValue([]) } }); + return null; + }); + const trigger = vi.fn().mockResolvedValue(undefined); + (kernel as any).context.trigger = trigger; + + await dispatcher.handlePackages('/app.empty/publish-drafts', 'POST', {}, {}, { request: {} }); + expect(trigger).not.toHaveBeenCalled(); + }); + it('POST /packages/:id/publish-drafts returns 501 when protocol lacks the method', async () => { (kernel as any).getService = vi.fn().mockImplementation((name: string) => { if (name === 'protocol') return Promise.resolve({}); diff --git a/packages/runtime/src/http-dispatcher.ts b/packages/runtime/src/http-dispatcher.ts index fef3dec106..2b4c0d41ba 100644 --- a/packages/runtime/src/http-dispatcher.ts +++ b/packages/runtime/src/http-dispatcher.ts @@ -2173,6 +2173,31 @@ export class HttpDispatcher { } catch (e: any) { (result as any).unhideError = e?.message ?? 'visibility flip failed'; } + // A publish promoted drafts to active (or unhid an additive + // app) at RUNTIME — but boot-cached consumers still hold the + // pre-publish view. The load-bearing one is the automation + // engine: a record-triggered flow authored + published in the + // Studio does NOT bind its trigger (record-change automations + // never fire) until the next restart. Announce + // 'metadata:reloaded' — the same signal a dev artifact reload + // fires (MetadataPlugin._reloadAndAnnounce) — so subscribers + // re-sync WITHOUT a restart. #2560 covers the cold-boot bind; + // this covers publish-while-running. `this.kernel.context` is + // the same handle the service resolver uses above. Best-effort: + // a subscriber failure must never fail the publish (the drafts + // are already live), so it rides the response instead. + try { + const changed = [ + ...(((result as any)?.published ?? []) as Array<{ type: string; name: string }>) + .map((p) => `${p.type}/${p.name}`), + ...(((result as any)?.unhiddenApps ?? []) as string[]).map((n) => `app/${n}`), + ]; + if (changed.length > 0 && this.kernel?.context?.trigger) { + await this.kernel.context.trigger('metadata:reloaded', { changed }); + } + } catch (e: any) { + (result as any).rebindError = e?.message ?? 'metadata:reloaded announce failed'; + } return { handled: true, response: this.success(result) }; } catch (e: any) { return { handled: true, response: this.error(e.message, e.statusCode || 500) }; diff --git a/packages/services/service-automation/src/flow-hot-reload.test.ts b/packages/services/service-automation/src/flow-hot-reload.test.ts index bd419c0bcb..c2d5f8146e 100644 --- a/packages/services/service-automation/src/flow-hot-reload.test.ts +++ b/packages/services/service-automation/src/flow-hot-reload.test.ts @@ -9,8 +9,12 @@ // 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 +// tears down flows that vanished from the artifact. The re-sync reads the +// protocol's flattened flow view (`getMetaItems({type:'flow'})`) — the SAME +// source #2560's cold-boot bind uses — so the fake here is a `protocol` service, +// not a `metadata` one (`metadata.list('flow')` returns 0 in a real server; the +// same hook also fires on a Studio publish). 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'; @@ -84,20 +88,24 @@ function captureRunAs(engine: AutomationEngine): Array { return seen; } -/** A mutable fake `metadata` service exposing just the `list('flow')` the re-sync uses. */ -function fakeMetadataService(initial: unknown[]) { +/** + * A mutable fake `protocol` service exposing just the flattened + * `getMetaItems({type:'flow'})` view the re-sync reads (returned as the + * `{ items: [...] }` envelope the real protocol serves). + */ +function fakeProtocolService(initial: unknown[]) { let flows = initial; return { service: { - async list(type: string) { - return type === 'flow' ? flows : []; + async getMetaItems(q: { type: string }) { + return { items: q.type === 'flow' ? flows : [] }; }, }, setFlows: (next: unknown[]) => { flows = next; }, }; } -async function bootKernel(meta: { service: unknown }) { +async function bootKernel(proto: { service: unknown }) { const kernel = new LiteKernel({ logger: { level: 'silent' } } as never); const harness = { name: 'test.harness', @@ -105,7 +113,7 @@ async function bootKernel(meta: { service: unknown }) { version: '1.0.0', dependencies: [] as string[], async init(ctx: any) { - ctx.registerService('metadata', meta.service); + ctx.registerService('protocol', proto.service); }, async start() {}, }; @@ -119,8 +127,8 @@ const reload = (kernel: LiteKernel) => (kernel as any).context.trigger('metadata 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 proto = fakeProtocolService([scheduledFlow('sweep', 'user', 1000)]); + const kernel = await bootKernel(proto); const engine = kernel.getService('automation'); const seen = captureRunAs(engine); const sched = recordingScheduleTrigger(); @@ -136,7 +144,7 @@ describe("scheduled flow hot-reload re-bind (metadata:reloaded re-sync)", () => // Edit the flow: runAs:system + interval 5000. Recompile → reload. sched.events.length = 0; - meta.setFlows([scheduledFlow('sweep', 'system', 5000)]); + proto.setFlows([scheduledFlow('sweep', 'system', 5000)]); await reload(kernel); await flush(); @@ -155,8 +163,8 @@ describe("scheduled flow hot-reload re-bind (metadata:reloaded re-sync)", () => }); 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 proto = fakeProtocolService([scheduledFlow('sweep', 'user', 1000)]); + const kernel = await bootKernel(proto); const engine = kernel.getService('automation'); const sched = recordingScheduleTrigger(); engine.registerTrigger(sched.trigger); @@ -166,7 +174,7 @@ describe("scheduled flow hot-reload re-bind (metadata:reloaded re-sync)", () => expect(sched.has('sweep')).toBe(true); // Delete the flow file → recompiled artifact no longer carries it. - meta.setFlows([]); + proto.setFlows([]); await reload(kernel); await flush(); diff --git a/packages/services/service-automation/src/flow-publish-rebind.test.ts b/packages/services/service-automation/src/flow-publish-rebind.test.ts new file mode 100644 index 0000000000..f538a20e74 --- /dev/null +++ b/packages/services/service-automation/src/flow-publish-rebind.test.ts @@ -0,0 +1,203 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// Regression: a flow PUBLISHED while the server runs must bind its trigger +// WITHOUT a restart (follow-up to #2560's cold-boot bind). +// +// #2560 bound flows at 'kernel:ready' from `protocol.getMetaItems({type:'flow'})`, +// fixing the cold boot. But a flow authored + published in the Studio while the +// server was up still did NOT bind until the next restart: +// 1. the publish path (POST /packages/:id/publish-drafts) fired no rebind event +// the automation service listened to, and +// 2. the 'metadata:reloaded' re-sync read the WRONG source — +// `metadata.list('flow')`, which returns 0 in a real running server (it does +// not surface inline app flows) — so even when the hook DID fire it bound +// nothing. +// +// The fix routes the 'metadata:reloaded' re-sync through the SAME protocol view +// the cold-boot bind uses, and the runtime dispatcher fires 'metadata:reloaded' +// after publishPackageDrafts. These tests exercise the re-sync half directly: +// once 'metadata:reloaded' fires, a flow the protocol now serves binds, a flow it +// no longer serves is torn down, and neither depends on `metadata.list`. + +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 record-after-update triggered flow, the shape the protocol view serves. */ +function recordTriggeredFlow(name: string, object: string) { + return { + name, + label: name, + type: 'autolaunched', + nodes: [ + { + id: 'start', + type: 'start', + label: 'Start', + config: { objectName: object, triggerType: 'record-after-update', condition: 'status == "closed"' }, + }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + }; +} + +/** A recording FlowTrigger of type 'record_change' (stands in for the real one). */ +function recordingRecordChangeTrigger() { + const bound = new Map Promise>(); + const trigger: FlowTrigger = { + type: 'record_change', + start(binding: FlowTriggerBinding, cb: (ctx: AutomationContext) => Promise) { + bound.set(binding.flowName, cb); + }, + stop(flowName: string) { + bound.delete(flowName); + }, + }; + return { + trigger, + has: (n: string) => bound.has(n), + }; +} + +/** + * A MUTABLE protocol whose flow set can change after boot — models a Studio + * publish promoting a draft flow to active (or a delete removing one), which the + * flattened `getMetaItems({type:'flow'})` view then reflects. `setFail(true)` + * models a transient protocol read failure. + */ +function mutableProtocolService(initial: unknown[]) { + let flows = [...initial]; + let fail = false; + return { + service: { + async getMetaItems(q: { type: string }) { + if (fail) throw new Error('protocol unavailable'); + return { items: q.type === 'flow' ? flows : [] }; + }, + }, + setFlows: (next: unknown[]) => { + flows = [...next]; + }, + setFail: (v: boolean) => { + fail = v; + }, + }; +} + +/** + * Boot a kernel with the automation plugin, a mutable protocol service, and a + * recording record-change trigger (registered during init, before kernel:ready, + * the way the real RecordChangeTriggerPlugin registers its trigger). Captures the + * harness PluginContext so a test can fire 'metadata:reloaded' — the signal the + * dispatcher fires after a publish. Optionally registers a `metadata` service so + * a test can prove the re-sync does NOT depend on `metadata.list`. + */ +async function bootKernel( + proto: ReturnType, + rec: ReturnType, + opts: { metadata?: unknown } = {}, +) { + const kernel = new LiteKernel({ logger: { level: 'silent' } } as never); + kernel.use(new AutomationServicePlugin()); + let captured: any; + const harness = { + name: 'test.harness', + type: 'standard' as const, + version: '1.0.0', + dependencies: [] as string[], + async init(ctx: any) { + captured = ctx; + ctx.registerService('protocol', proto.service); + if (opts.metadata) ctx.registerService('metadata', opts.metadata); + // AutomationServicePlugin.init() ran first, so the engine service + // exists; register the trigger before kernel:ready so the bind finds it. + ctx.getService('automation').registerTrigger(rec.trigger); + }, + async start() {}, + }; + kernel.use(harness as never); + await kernel.bootstrap(); + return { kernel, ctx: () => captured }; +} + +describe('flow published at runtime binds on metadata:reloaded (publish rebind, #2560 follow-up)', () => { + it('binds a flow that appears in the protocol AFTER boot — no restart', async () => { + const rec = recordingRecordChangeTrigger(); + const proto = mutableProtocolService([]); // boot with NO flows + const { kernel, ctx } = await bootKernel(proto, rec); + await flush(); + expect(rec.has('ticket_closed'), 'nothing bound at boot').toBe(false); + + // Simulate the Studio publish: the draft flow is now active and served by + // the protocol, and the dispatcher fires 'metadata:reloaded'. + proto.setFlows([recordTriggeredFlow('ticket_closed', 'ticket')]); + await ctx().trigger('metadata:reloaded', { changed: ['flow/ticket_closed'] }); + await flush(); + + expect(rec.has('ticket_closed'), 'published flow bound without a restart').toBe(true); + const engine = kernel.getService('automation'); + expect(engine.getActiveTriggerBindings().map((b) => b.flowName)).toContain('ticket_closed'); + + await kernel.shutdown(); + }); + + it('tears down a flow the protocol no longer serves on re-sync', async () => { + const rec = recordingRecordChangeTrigger(); + const proto = mutableProtocolService([recordTriggeredFlow('temp_flow', 'ticket')]); + const { kernel, ctx } = await bootKernel(proto, rec); + await flush(); + expect(rec.has('temp_flow'), 'bound at kernel:ready').toBe(true); + + // The flow was deleted + the deletion published away. + proto.setFlows([]); + await ctx().trigger('metadata:reloaded', { changed: [] }); + await flush(); + + expect(rec.has('temp_flow'), 'removed flow unbound on re-sync').toBe(false); + await kernel.shutdown(); + }); + + it('reads the protocol, NOT metadata.list (which returns 0 in a real server)', async () => { + const rec = recordingRecordChangeTrigger(); + const proto = mutableProtocolService([]); + // A metadata service that ALWAYS serves 0 flows — the exact real-server + // behavior that made the pre-fix `metadata.list('flow')` re-sync a silent + // no-op. The re-sync must ignore it and read the protocol. + const metadata = { async list() { return []; } }; + const { kernel, ctx } = await bootKernel(proto, rec, { metadata }); + await flush(); + + proto.setFlows([recordTriggeredFlow('ticket_closed', 'ticket')]); + await ctx().trigger('metadata:reloaded', { changed: [] }); + await flush(); + + expect( + rec.has('ticket_closed'), + 'bound from the protocol despite an empty metadata.list', + ).toBe(true); + await kernel.shutdown(); + }); + + it('does not tear down live flows when the protocol read fails', async () => { + const rec = recordingRecordChangeTrigger(); + const proto = mutableProtocolService([recordTriggeredFlow('keep_me', 'ticket')]); + const { kernel, ctx } = await bootKernel(proto, rec); + await flush(); + expect(rec.has('keep_me'), 'bound at kernel:ready').toBe(true); + + // A transient protocol failure must be a no-op, never an unbind: a failed + // read is "couldn't read", not "zero flows". + proto.setFail(true); + await ctx().trigger('metadata:reloaded', { changed: [] }); + await flush(); + + expect(rec.has('keep_me'), 'live flow survives a failed protocol read').toBe(true); + await kernel.shutdown(); + }); +}); diff --git a/packages/services/service-automation/src/plugin.ts b/packages/services/service-automation/src/plugin.ts index bfc81831a0..bb8c9ae007 100644 --- a/packages/services/service-automation/src/plugin.ts +++ b/packages/services/service-automation/src/plugin.ts @@ -225,18 +225,21 @@ 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. + // ── Runtime re-bind: re-sync flow triggers on 'metadata:reloaded' ────── + // Fires on two RUNTIME events (never on a cold boot — the kernel:ready + // bind below covers that): a dev `os dev` artifact recompile (MetadataPlugin + // reloads dist/objectstack.json and announces), and a Studio package publish + // (the runtime dispatcher announces after publishPackageDrafts promotes the + // drafts to active). The engine still holds the flow definitions + trigger + // bindings it pulled ONCE above — including scheduled jobs. Without + // re-syncing, an edited schedule-triggered flow keeps firing its OLD + // definition (old runAs / schedule / logic), and a newly-published + // record-triggered flow never binds its trigger at all, until a full + // restart. Re-register every current flow (registerFlow re-binds its trigger + // idempotently — ScheduleTrigger.start cancels + reschedules) and unregister + // flows that vanished so their jobs stop. ctx.hook('metadata:reloaded', async () => { - await this.resyncFlowsFromMetadata(ctx); + await this.resyncFlowsFromProtocol(ctx); }); // ── Cold-boot bind via the PROTOCOL's flattened flow view ───────────── @@ -280,48 +283,84 @@ 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. + * Read the protocol's flattened flow view — `getMetaItems({ type: 'flow' })`, + * the same source `GET /meta/flow` serves and #2560's cold-boot bind uses. + * Returns the list of flow docs, or `null` when the protocol is unavailable + * or the read failed. Callers MUST treat `null` as "couldn't read", NOT as + * "zero flows" — tearing flows down on a failed read would unbind live + * automations. * - * 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. + * Unlike `registry.listItems('flow')` (the boot pull) this surfaces flows + * defined inline in an app manifest, and unlike `metadata.list('flow')` — the + * source this re-sync read before this fix — it is actually populated in a + * real running server (`metadata.list('flow')` returns 0 there, so the old + * re-sync bound nothing). */ - private async resyncFlowsFromMetadata(ctx: PluginContext): Promise { - if (!this.engine) return; - let metadata: { list?(type: string): Promise } | undefined; + private async readFlowDefsFromProtocol( + ctx: PluginContext, + ): Promise | null> { + let protocol: { getMetaItems?(q: { type: string }): Promise } | undefined; try { - metadata = ctx.getService('metadata'); + protocol = ctx.getService('protocol'); } catch { - // No metadata service (e.g. a bare engine / tests) — nothing to sync. - return; + return null; // no protocol service (bare engine / tests) — nothing to sync } - if (!metadata || typeof metadata.list !== 'function') return; + if (!protocol || typeof protocol.getMetaItems !== 'function') return null; - let defs: unknown[]; + let raw: unknown; try { - defs = await metadata.list('flow'); + raw = await protocol.getMetaItems({ type: 'flow' }); } catch (err) { ctx.logger.warn( - `[Automation] flow re-sync skipped: metadata.list('flow') failed: ${(err as Error).message}`, + `[Automation] flow read from protocol failed: getMetaItems('flow'): ${(err as Error).message}`, ); - return; + return null; } + // getMetaItems hands back a bare array or an `{ items: [...] }` envelope, + // and each entry is either the flow doc or an `{ item: }` wrapper. + const list = Array.isArray(raw) ? raw : (((raw as { items?: unknown[] })?.items) ?? []); + return list.map((entry) => + (entry && typeof entry === 'object' && 'item' in entry + ? (entry as { item: unknown }).item + : entry) as { name?: string }, + ); + } + + /** + * Re-pull flow definitions from the protocol and re-register them into the + * engine, so a RUNTIME metadata change re-binds flow triggers instead of + * leaving the engine executing the boot-time definitions. Driven by the + * `metadata:reloaded` hook, which fires on two runtime events: + * 1. a dev `os dev` artifact recompile — MetadataPlugin reloads the + * artifact from disk and announces; and + * 2. a Studio package publish — the runtime dispatcher announces after + * `publishPackageDrafts` promotes the drafts to active. Without (2), a + * flow authored + published while the server runs never bound its + * trigger (record-change automations never fired) until the next + * restart, even though #2560 fixed the cold-boot bind. + * + * Reads `protocol.getMetaItems({ type: 'flow' })` — the SAME source #2560's + * cold-boot bind and `GET /meta/flow` use. It does NOT read the ObjectQL + * schema registry (a boot-time cache the reload never refreshes) and — the + * bug this fixes — no longer reads `metadata.list('flow')`, which returns 0 + * in a real running server (it does not surface inline app flows), so the old + * re-sync was a silent no-op that bound nothing on publish. + * + * Idempotent and best-effort: registerFlow() re-binds the trigger + * (ScheduleTrigger.start cancels + reschedules), flows removed from the + * artifact are unregistered so their jobs/triggers stop firing, and any + * failure is logged without disturbing the rest of the runtime. A failed or + * unavailable protocol read is a no-op — it never tears down live flows. + */ + private async resyncFlowsFromProtocol(ctx: PluginContext): Promise { + if (!this.engine) return; + const defs = await this.readFlowDefsFromProtocol(ctx); + if (!defs) return; // unavailable / failed read — do not tear down live flows + const freshNames = new Set(); let resynced = 0; - for (const d of defs) { - const def = d as { name?: string }; + for (const def of defs) { if (!def?.name) continue; freshNames.add(def.name); try { @@ -353,43 +392,19 @@ export class AutomationServicePlugin implements Plugin { } /** - * Bind flows from the protocol's flattened flow view — `getMetaItems({ type: - * 'flow' })`, the same source `GET /meta/flow` serves. Unlike - * `registry.listItems('flow')` (the boot pull) this surfaces flows defined - * inline in an app manifest, and unlike `metadata.list('flow')` (the - * `metadata:reloaded` re-sync) it is populated on a cold boot once the app - * is registered. Called at kernel:ready so record-triggered automations - * actually bind on a fresh start. registerFlow is idempotent, so re-binding + * Bind flows from the protocol's flattened flow view at `kernel:ready`, so + * record-triggered automations actually bind on a fresh start (#2560). + * Additive by design — unlike {@link resyncFlowsFromProtocol} it never tears + * flows down, so a transient empty/failed read at boot can't unbind the flows + * the boot pull already registered. registerFlow is idempotent, so re-binding * a flow the boot pull already registered is harmless. */ private async syncFlowsFromProtocol(ctx: PluginContext): Promise { if (!this.engine) return; - let protocol: { getMetaItems?(q: { type: string }): Promise } | undefined; - try { - protocol = ctx.getService('protocol'); - } catch { - return; // no protocol service (bare engine / tests) — nothing to sync - } - if (!protocol || typeof protocol.getMetaItems !== 'function') return; - - let raw: unknown; - try { - raw = await protocol.getMetaItems({ type: 'flow' }); - } catch (err) { - ctx.logger.warn( - `[Automation] cold-boot flow bind skipped: getMetaItems('flow') failed: ${(err as Error).message}`, - ); - return; - } - - // getMetaItems hands back a bare array or an `{ items: [...] }` envelope, - // and each entry is either the flow doc or an `{ item: }` wrapper. - const list = Array.isArray(raw) ? raw : (((raw as { items?: unknown[] })?.items) ?? []); + const defs = await this.readFlowDefsFromProtocol(ctx); + if (!defs) return; let bound = 0; - for (const entry of list) { - const def = ( - entry && typeof entry === 'object' && 'item' in entry ? (entry as { item: unknown }).item : entry - ) as { name?: string } | undefined; + for (const def of defs) { if (!def?.name) continue; // registerFlow is idempotent, so re-binding is safe try { this.engine.registerFlow(def.name, def as never);