diff --git a/.changeset/stored-migration-covers-flows.md b/.changeset/stored-migration-covers-flows.md new file mode 100644 index 0000000000..f4f8d07d42 --- /dev/null +++ b/.changeset/stored-migration-covers-flows.md @@ -0,0 +1,67 @@ +--- +"@objectstack/service-automation": minor +"@objectstack/metadata-protocol": minor +"@objectstack/cli": patch +--- + +feat(automation,migrate): `os migrate meta --stored` now covers flow rows too (#4454) + +#4327 gave the stored-metadata conversion chain a finish line for every +metadata type except `flow` — the one type where the most stored dialect +actually lives, since the graduated conversions `flow-node-crud-filter-alias`, +`flow-node-crud-object-alias`, `flow-node-notify-config-aliases` and +`flow-node-script-config-aliases` are all flow-node entries. Flow-node +conversions carry ADR-0078's open-namespace conflict guard, which has to consult +the *live* executor registry to tell a rename from a clobber, and the metadata +layer has no way to obtain one. Flows were reported `skipped` with that reason. +They are now converted. + +**One canonicalization policy, two shapes.** +`AutomationEngine.canonicalizeStoredFlow` is the single implementation and +`registerFlow` calls it, so the load seam and the migration can never disagree +about what "canonical" means. It returns `parsed` (for execution — the +`FlowSchema.parse` + #4347 region output, schema defaults materialized) and +`storable` (for persistence). + +**`storable` excludes schema defaults, and that is the load-bearing decision.** +Measured rather than assumed: driving a pre-17 flow through all three steps +*removes* nothing — `FlowSchema` is strict since #4001, so an unrecognized key +throws instead of being silently dropped, which means the +`graftNormalizedOperators` precedent (it exists because the *view* parse strips +Studio-only auxiliary keys) does not transfer — and *adds* only defaults: +`version`, `runAs`, per-edge `type` / `isDefault`. Persisting a default the +author never wrote would pin every migrated row to today's value while untouched +rows follow tomorrow's: two populations with different behaviour, which is +exactly the drift this pass exists to remove. So the write-back is the +conversion result plus the `{dialect, source}` envelopes the schema derives for +edge conditions, and nothing else. + +One subtlety worth knowing if you extend this: that envelope is a schema +transform, not a conversion, so it emits **no** notice while still changing the +body. Reading notices alone — correct for every other metadata type — would call +such a row canonical and leave it re-deriving on every boot. Both passes are +copy-on-write, so identity is the exact test for flows. + +**New: `AutomationServicePluginOptions.armRuntime`** (default `true`, so every +server, dev stack and test host is unaffected). Set `false` and the plugin +brings up the engine and the complete node registry — built-ins plus whatever +`automation:ready` contributes, because a *partial* registry would make the +conflict guard read a live custom node type as unowned and rewrite over it — and +then stops before anything is armed: + +| Skipped when `armRuntime: false` | Why it must be | +|---|---| +| flow pull + `kernel:ready` / `metadata:reloaded` re-sync | `registerFlow` calls `activateFlowTrigger` — record triggers and scheduled jobs would go live | +| declarative connector materialization | opens real connections; an MCP provider spawns a child process | +| suspended-run wait-timer re-arm | would resume someone's paused approval mid-migration | + +`os migrate meta --stored` boots the plugin in that mode. A migration process +must not become a second server. + +A refused rename — the guard firing because the old node-type token is a live +name something else owns in this environment — fails that row loudly, naming the +token and its owner. Never a silent skip, never a clobber. A flow that cannot +canonicalize at all (a strict-schema violation, a malformed control-flow region) +is reported as failed with the parse message rather than persisted as a guess; +such a row cannot register today either, so the report is telling you about a +flow that is already broken at runtime. diff --git a/content/docs/deployment/cli.mdx b/content/docs/deployment/cli.mdx index 3a57e1e869..f1420526ae 100644 --- a/content/docs/deployment/cli.mdx +++ b/content/docs/deployment/cli.mdx @@ -809,9 +809,21 @@ counting as done: | Not rewritten | Why | | :--- | :--- | -| `flow` rows | Flow-node conversions carry a conflict guard that needs the automation engine's live executor registry; flows canonicalize at their own seam when the engine loads them | | Types with no repository write path (`agent`) | Their write path records no history and would force a draft live — a half-write is worse than leaving the row to the read path | | Rows that still fail the current schema after conversion | That is a genuine contract violation, not chain-owned history. The write path's rejection is correct; fix the row in Studio | +| A flow whose rename the conflict guard refused | The old node-type token is a live name something else owns here. Rewriting would clobber that owner, so the row fails loudly naming the token — never a silent skip | + +**Flows are covered, and cost one extra plugin.** Flow-node conversions carry an +open-namespace conflict guard that has to consult the *live* executor registry +to tell a rename from a clobber, so this run boots the automation engine — in an +inert mode that installs the node registry and then arms nothing: no flow +registered, no record trigger or scheduled job bound, no connector +materialized, no suspended run resumed. A migration process must not become a +second server. What gets written back for a flow is the conversion result plus +the `condition` envelopes the schema derives, and deliberately **not** the +schema's defaults (`version`, `runAs`, per-edge `type`) — persisting a default +the author never wrote would pin that row to today's value while untouched rows +follow tomorrow's, which is the drift this command exists to remove. `--apply` is the only writing mode, and it rewrites **metadata** — each affected diff --git a/docs/adr/0087-metadata-protocol-upgrade-contract.md b/docs/adr/0087-metadata-protocol-upgrade-contract.md index 9c98420c8f..58764b6dcc 100644 --- a/docs/adr/0087-metadata-protocol-upgrade-contract.md +++ b/docs/adr/0087-metadata-protocol-upgrade-contract.md @@ -450,10 +450,40 @@ writing mode. never rewritten. Canonicalizing a past version's body would break the checksum↔body pairing this contract depends on — the migration is a new commit, not a rewrite of history. -- **What the pass does not cover, it names.** Flows (their seam is - `AutomationEngine.registerFlow`, which holds the executor registry the - conflict guard needs) and types with no repository write path are reported as - `skipped` with the reason, never counted as done. Giving flows the same finish - line needs a canonicalization entry point on the automation engine — tracked - as #4454, and worth doing precisely because the graduated flow-node - conversions are where the most stored dialect lives. +- **What the pass does not cover, it names.** Types with no repository write + path are reported as `skipped` with the reason, never counted as done. + +## Addendum (2026-08-01b) — flows reach the finish line too (#4454) + +The pass above initially skipped `flow` rows, which was the largest hole in it: +the graduated flow-node conversions are where the most stored dialect lives. +Closing it needed three decisions. + +- **One canonicalization policy, two shapes.** + `AutomationEngine.canonicalizeStoredFlow` is now the single implementation and + `registerFlow` calls it, so the load seam and the migration cannot disagree + about what canonical means. It returns `parsed` (for execution — schema + defaults materialized) and `storable` (for persistence). +- **`storable` excludes schema defaults, and this is load-bearing.** Measured, + not assumed: driving a pre-17 flow through parse + the region pass *removes* + nothing (`FlowSchema` is strict since #4001 — an unknown key throws rather + than being dropped, so the `graftNormalizedOperators` precedent does not + transfer) and *adds* only defaults: `version`, `runAs`, per-edge `type` / + `isDefault`. Persisting a default the author never wrote would pin every + migrated row to today's value while untouched rows follow tomorrow's — two + populations with different behaviour, which is the drift this pass exists to + remove. So the write-back is conversions plus the schema's `condition` + envelopes, and nothing else. +- **The engine is borrowed, not started.** `AutomationServicePlugin` gains + `armRuntime: false`: built-in nodes installed and `automation:ready` fired + (the registry must be COMPLETE, or the conflict guard reads a live custom node + type as unowned and rewrites over it), then a hard stop before anything is + armed — no flow registered, no trigger or schedule bound, no connector + materialized, no suspended run resumed. `registerFlow` arms triggers as a side + effect, so skipping only the boot pull would not have been enough; the + `kernel:ready` and `metadata:reloaded` re-syncs are skipped for the same + reason. A migration process must not become a second server. + +A refused rename — the guard firing because the old token is a live name owned +by something else — fails that row loudly with the token and its owner. Never a +silent skip, never a clobber; that is the whole reason the guard exists. diff --git a/packages/cli/src/commands/migrate/meta.ts b/packages/cli/src/commands/migrate/meta.ts index a5a4aaa17f..90c442291b 100644 --- a/packages/cli/src/commands/migrate/meta.ts +++ b/packages/cli/src/commands/migrate/meta.ts @@ -30,6 +30,7 @@ import { bootSchemaStack } from '../../utils/schema-migrate.js'; import { buildDataMigrationPlugins } from '../../utils/data-migration-plugins.js'; import { OCCUPANCY_HINT, probeMigrationTarget } from '../../utils/migrate-occupancy-gate.js'; import { describeOccupancy } from '../../utils/sqlite-occupancy.js'; +import type { IAutomationService } from '@objectstack/spec/contracts'; async function confirm(question: string): Promise { if (!process.stdin.isTTY) return false; // non-interactive → require --yes @@ -511,12 +512,15 @@ export default class MigrateMeta extends Command { let stack; try { - // `PlatformObjectsPlugin` only — this pass needs `sys_metadata` and its - // history/audit siblings, which the protocol assembly registers itself. - // No storage adapter: unlike the file migration, nothing here reads bytes. + // `PlatformObjectsPlugin` for `sys_metadata` and its history/audit + // siblings, plus the automation engine in INERT mode so `flow` rows are + // covered too (#4454) — flow-node conversions need its executor registry + // for the conflict guard, and `armRuntime: false` means taking it arms + // nothing. No storage adapter: unlike the file migration, nothing here + // reads bytes. stack = await bootSchemaStack({ ...(flags['database-url'] ? { databaseUrl: flags['database-url'] } : {}), - extraPlugins: await buildDataMigrationPlugins(), + extraPlugins: await buildDataMigrationPlugins({ automation: true }), }); } catch (error: any) { if (flags.json) { await emitJson({ error: error.message }, 0, { compact: true }); this.exit(1); return; } @@ -541,10 +545,23 @@ export default class MigrateMeta extends Command { const { formatStoredMigrationReport, storedMigrationClean } = await import('@objectstack/metadata-protocol'); + // The automation engine canonicalizes `flow` rows — it holds the executor + // registry ADR-0078's conflict guard needs (#4454). It is booted inert, so + // this is the only thing it does in this process. Absent (an older stack, + // or a boot that skipped it), flow rows keep reporting `skipped` with the + // reason rather than being silently counted done. + // `SchemaStack.kernel` is untyped, so the slot's contract is stated on the + // result rather than as a type argument — narrowing, not erasing. + let automation: IAutomationService | undefined; + try { automation = stack.kernel.getService('automation') as IAutomationService | undefined; } + catch { /* not registered — flows stay reported as skipped */ } + const canonicalize = automation?.canonicalizeStoredFlow?.bind(automation); + const report = await protocol.migrateStoredMetadata({ apply, ...(flags.type && flags.type.length > 0 ? { types: flags.type } : {}), actor: 'os migrate meta --stored', + ...(canonicalize ? { canonicalizeFlow: canonicalize } : {}), }); const clean = storedMigrationClean(report); if (!clean) exitCode = 1; diff --git a/packages/cli/src/utils/data-migration-plugins.ts b/packages/cli/src/utils/data-migration-plugins.ts index 8e5019b322..3d9ba98b61 100644 --- a/packages/cli/src/utils/data-migration-plugins.ts +++ b/packages/cli/src/utils/data-migration-plugins.ts @@ -24,11 +24,27 @@ import { resolveStorageCapabilityArg } from '../commands/serve.js'; * where the server would. */ export async function buildDataMigrationPlugins( - opts: { storage?: boolean } = {}, + opts: { storage?: boolean; automation?: boolean } = {}, ): Promise { const plugins: unknown[] = []; const { PlatformObjectsPlugin } = await import('@objectstack/platform-objects/plugin'); plugins.push(new PlatformObjectsPlugin()); + if (opts.automation === true) { + // `os migrate meta --stored` needs the automation ENGINE, never the + // automation RUNTIME (#4454). Flow-node conversions carry ADR-0078's + // open-namespace conflict guard, which consults the live executor registry + // to tell a rename from a clobber — and only this plugin has that registry. + // + // `armRuntime: false` is what makes taking it safe: the engine and the full + // node registry come up (built-ins plus whatever `automation:ready` + // contributes, because a PARTIAL registry would make the guard rewrite over + // a live custom node type instead of refusing), and then nothing is armed — + // no flow registered, no record trigger or scheduled job bound, no + // declarative connector materialized, no suspended run resumed. A migration + // process must not become a second server. + const { AutomationServicePlugin } = await import('@objectstack/service-automation'); + plugins.push(new AutomationServicePlugin({ armRuntime: false, suspendedRunStore: 'memory' })); + } if (opts.storage === true) { try { const { SettingsServicePlugin } = await import('@objectstack/service-settings'); diff --git a/packages/metadata-protocol/src/index.ts b/packages/metadata-protocol/src/index.ts index 303677250a..390e00ce11 100644 --- a/packages/metadata-protocol/src/index.ts +++ b/packages/metadata-protocol/src/index.ts @@ -17,6 +17,7 @@ export type { export { formatStoredMigrationReport, storedMigrationClean } from './stored-migration.js'; export type { + StoredFlowCanonicalization, StoredMigrationNotice, StoredMigrationOutcome, StoredMigrationReport, diff --git a/packages/metadata-protocol/src/protocol.stored-migration.test.ts b/packages/metadata-protocol/src/protocol.stored-migration.test.ts index 8155d82d9c..5ee1c189df 100644 --- a/packages/metadata-protocol/src/protocol.stored-migration.test.ts +++ b/packages/metadata-protocol/src/protocol.stored-migration.test.ts @@ -288,6 +288,156 @@ describe('migrateStoredMetadata — apply (#4327)', () => { }); }); +describe('migrateStoredMetadata — flow rows via the canonicalizeFlow hook (#4454)', () => { + // A body the write path's schema gate accepts — the hook canonicalizes the + // shape, it does not exempt the row from validation. + const flowBody = (config: Record) => ({ + name: 'purge_flow', + label: 'Purge Stale Leads', + type: 'autolaunched', + status: 'active', + nodes: [{ id: 'n1', type: 'delete_record', label: 'Purge', config }], + edges: [], + }); + const flowRow = { + type: 'flow', + name: 'purge_flow', + metadata: flowBody({ objectName: 'lead', filters: { status: 'stale' } }), + }; + /** Stands in for `AutomationEngine.canonicalizeStoredFlow` — same contract. */ + const canonicalizeFlow = (_name: string, body: any) => { + const node = body?.nodes?.[0]; + if (!node || !('filters' in (node.config ?? {}))) { + // Copy-on-write: an unchanged body comes back BY REFERENCE, which is + // what the pass reads as "already canonical". + return { storable: body, notices: [], conflicts: [] }; + } + const { filters, ...rest } = node.config; + return { + storable: { ...body, nodes: [{ ...node, config: { ...rest, filter: filters } }] }, + notices: [{ + conversionId: 'flow-node-crud-filter-alias', + surface: 'flow.node.config.filter', + from: 'filters', + to: 'filter', + path: 'flows[0].nodes[0].config', + message: 'filters → filter', + }], + conflicts: [], + }; + }; + + it('rewrites a flow row when the caller supplies the engine hook', async () => { + const { engine, tables } = makeStubEngine([flowRow]); + const protocol = new ObjectStackProtocolImplementation(engine); + + const report = await protocol.migrateStoredMetadata({ apply: true, canonicalizeFlow }); + + expect(report.rewritten).toBe(1); + expect(report.skipped).toBe(0); + const stored = JSON.parse(metaRows(tables)[0]!.metadata); + expect(stored.nodes[0].config.filter).toEqual({ status: 'stale' }); + expect('filters' in stored.nodes[0].config).toBe(false); + expect(historyRows(tables)[0]).toMatchObject({ type: 'flow', source: 'migrate-stored' }); + }); + + it('still skips — with the reason — when no hook is supplied', async () => { + const { engine, tables } = makeStubEngine([flowRow]); + const protocol = new ObjectStackProtocolImplementation(engine); + + const report = await protocol.migrateStoredMetadata({ apply: true }); + + expect(report.skipped).toBe(1); + expect(report.rows[0]!.reason).toMatch(/registerFlow/); + expect(historyRows(tables)).toHaveLength(0); + }); + + it('counts an already-canonical flow as canonical, not as a rewrite', async () => { + const canonicalFlow = { + type: 'flow', + name: 'purge_flow', + metadata: flowBody({ objectName: 'lead', filter: { status: 'stale' } }), + }; + const { engine, tables } = makeStubEngine([canonicalFlow]); + const protocol = new ObjectStackProtocolImplementation(engine); + + const report = await protocol.migrateStoredMetadata({ apply: true, canonicalizeFlow }); + + expect(report.canonical).toBe(1); + expect(report.rewritten).toBe(0); + expect(historyRows(tables)).toHaveLength(0); + }); + + it('rewrites a flow the hook changed WITHOUT emitting a notice — the condition envelope case', async () => { + // The `{dialect, source}` envelope is a schema transform, not a + // conversion, so it reports no notice while still changing the body. + // Reading notices alone would call this row canonical and leave it + // re-deriving on every boot — the exact thing the pass exists to end. + const envelopeOnly = (_n: string, body: any) => ({ + storable: { + ...body, + edges: [{ ...body.edges[0], condition: { dialect: 'cel', source: "x == 'y'" } }], + }, + notices: [], + conflicts: [], + }); + const row = { + type: 'flow', + name: 'purge_flow', + metadata: { + ...flowBody({ objectName: 'lead', filter: { status: 'stale' } }), + edges: [{ id: 'e1', source: 'n1', target: 'n1', condition: "x == 'y'" }], + }, + }; + const { engine, tables } = makeStubEngine([row]); + const protocol = new ObjectStackProtocolImplementation(engine); + + const report = await protocol.migrateStoredMetadata({ apply: true, canonicalizeFlow: envelopeOnly }); + + expect(report.rewritten).toBe(1); + expect(JSON.parse(metaRows(tables)[0]!.metadata).edges[0].condition) + .toEqual({ dialect: 'cel', source: "x == 'y'" }); + }); + + it('fails the row loudly when the guard refuses a rename over a live name', async () => { + const conflicting = (_n: string, body: any) => ({ + storable: body, + notices: [], + conflicts: [{ + conversionId: 'flow-node-type-rename', + token: 'webhook', + path: 'flows[0].nodes[0].type', + message: "'webhook' is registered by a custom executor in this environment.", + }], + }); + const { engine, tables } = makeStubEngine([flowRow]); + const protocol = new ObjectStackProtocolImplementation(engine); + + const report = await protocol.migrateStoredMetadata({ apply: true, canonicalizeFlow: conflicting }); + + expect(report.failed).toBe(1); + expect(report.rewritten).toBe(0); + expect(report.rows[0]!.reason).toMatch(/live name/); + expect(report.rows[0]!.reason).toMatch(/webhook/); + // Never a silent skip and never a clobber — the owner's node survives. + expect(historyRows(tables)).toHaveLength(0); + expect(storedMigrationClean(report)).toBe(false); + }); + + it('reports a flow that cannot canonicalize instead of persisting a guess', async () => { + const throwing = () => { throw new Error('Unrecognized key(s) on this flow: `_uiPosition`'); }; + const { engine, tables } = makeStubEngine([flowRow]); + const protocol = new ObjectStackProtocolImplementation(engine); + + const report = await protocol.migrateStoredMetadata({ apply: true, canonicalizeFlow: throwing }); + + expect(report.failed).toBe(1); + expect(report.rows[0]!.reason).toMatch(/does not canonicalize/); + expect(report.rows[0]!.reason).toMatch(/_uiPosition/); + expect(historyRows(tables)).toHaveLength(0); + }); +}); + describe('migrateStoredMetadata — what it declines to touch, loudly (#4327)', () => { it('skips flow rows and names the seam that owns them', async () => { const legacyFlow = { diff --git a/packages/metadata-protocol/src/protocol.ts b/packages/metadata-protocol/src/protocol.ts index 8ec5dede16..09254caef5 100644 --- a/packages/metadata-protocol/src/protocol.ts +++ b/packages/metadata-protocol/src/protocol.ts @@ -50,6 +50,7 @@ import { type MetadataDiagnostics, } from './metadata-diagnostics.js'; import type { + StoredFlowCanonicalization, StoredMigrationNotice, StoredMigrationReport, StoredMigrationRow, @@ -6456,6 +6457,22 @@ export class ObjectStackProtocolImplementation implements types?: string[]; /** Recorded as the writer on the history + audit rows. */ actor?: string; + /** + * Canonicalize a stored `flow` body (#4454). + * + * Supplied only by a caller that holds a live automation engine — + * `AutomationEngine.canonicalizeStoredFlow` is the implementation. Flow + * conversions carry ADR-0078's open-namespace conflict guard, which + * needs the engine's executor registry to tell a rename from a clobber, + * and this layer has no way to obtain one. Omit it and flow rows are + * reported `skipped` with that reason rather than quietly counted done. + * + * Must return the **storable** shape — conversions and the schema's + * `condition` envelopes, without schema defaults. Throwing is a valid + * answer for a row that cannot canonicalize; the row is reported + * `failed` with the message. + */ + canonicalizeFlow?: (name: string, body: unknown) => StoredFlowCanonicalization; } = {}): Promise { const apply = request.apply === true; const typeFilter = request.types && request.types.length > 0 @@ -6524,14 +6541,47 @@ export class ObjectStackProtocolImplementation implements continue; } + // Flow rows need the automation engine's live executor registry for + // ADR-0078's open-namespace conflict guard, which this layer does + // not have — so they are canonicalized through a caller-supplied + // hook, or reported `skipped` when no caller can supply one (#4454). + let flowResult: StoredFlowCanonicalization | undefined; if (singular === 'flow') { - record({ - ...base, - outcome: 'skipped', - reason: 'flows canonicalize at AutomationEngine.registerFlow — the node-type ' - + 'conflict guard needs the live executor registry this layer does not have', - }); - continue; + if (!request.canonicalizeFlow) { + record({ + ...base, + outcome: 'skipped', + reason: 'flows canonicalize at AutomationEngine.registerFlow — the node-type ' + + 'conflict guard needs the live executor registry this caller did not supply', + }); + continue; + } + try { + flowResult = request.canonicalizeFlow(base.name, body); + } catch (e: any) { + // `FlowSchema` is strict (#4001) and the region validator + // hard-fails, so this is a row that cannot register at all — + // already broken at runtime. Report it; never persist a guess. + record({ + ...base, + outcome: 'failed', + reason: `the flow does not canonicalize: ${e?.message ?? String(e)}`, + }); + continue; + } + if (flowResult.conflicts.length > 0) { + // A rename refused because its old token is a LIVE name owned + // by something else. Rewriting would clobber that owner, and + // skipping quietly would hide it — the guard exists to be loud. + const first = flowResult.conflicts[0]!; + record({ + ...base, + outcome: 'failed', + reason: `conversion refused — '${first.token}' at ${first.path} is a live name in ` + + `this environment (${flowResult.conflicts.length} conflict(s)). ${first.message}`, + }); + continue; + } } const overlayAllowed = ObjectStackProtocolImplementation.isOverlayAllowed(singular); @@ -6546,8 +6596,23 @@ export class ObjectStackProtocolImplementation implements continue; } - const { item, notices } = this.convertStoredItemDetailed(singular, body); - if (notices.length === 0) { + // A flow's canonical body was already computed above (it needs the + // engine); everything else converts here. + // + // The change signal differs by type, and the difference is real. + // For a non-flow item every rewrite comes from a conversion, and a + // conversion always emits a notice (ADR-0087 D2 "loud"), so notices + // are exact. A flow additionally gains the `{dialect, source}` + // envelope the schema derives for edge conditions — that is a + // schema transform, not a conversion, so it emits NO notice while + // still changing the body. Both passes are copy-on-write, so + // identity is the precise test there: `storable === body` exactly + // when nothing was rewritten at all. + const { item, notices } = flowResult + ? { item: flowResult.storable, notices: flowResult.notices } + : this.convertStoredItemDetailed(singular, body); + const changed = flowResult ? item !== body : notices.length > 0; + if (!changed) { record({ ...base, outcome: 'canonical' }); continue; } diff --git a/packages/metadata-protocol/src/stored-migration.ts b/packages/metadata-protocol/src/stored-migration.ts index 0cccec8a19..b71cbe9155 100644 --- a/packages/metadata-protocol/src/stored-migration.ts +++ b/packages/metadata-protocol/src/stored-migration.ts @@ -27,6 +27,37 @@ * that reports every row canonical is the evidence, and it costs one command. */ +/** + * What a caller with a live automation engine hands back for a stored `flow` + * body (#4454) — structurally `AutomationEngine.canonicalizeStoredFlow`'s + * result, declared here so `metadata-protocol` states the contract it consumes + * without depending on the automation service. + * + * `storable` is the shape to PERSIST: conversions plus the `{dialect, source}` + * envelopes the flow schema derives for edge conditions, and deliberately not + * the schema's defaults — persisting a default the author never wrote would pin + * that row to today's value while untouched rows follow tomorrow's. + */ +export interface StoredFlowCanonicalization { + /** The canonical body to write back. Identical (by reference) to the input when nothing changed. */ + storable: unknown; + /** Conversions that fired, in the spec's notice shape. */ + notices: Array<{ + conversionId: string; + surface: string; + from: string; + to: string; + path: string; + message: string; + }>; + /** + * Renames the guard REFUSED because the old token is a live name owned by + * something else. A non-empty list fails the row loudly — rewriting would + * clobber that owner, and skipping quietly would hide it. + */ + conflicts: Array<{ conversionId: string; token: string; path: string; message: string }>; +} + /** What the pass did with (or would do with) one `sys_metadata` row. */ export type StoredMigrationOutcome = /** The chain was a no-op — the row is already on protocol. Not itemised. */ diff --git a/packages/services/service-automation/src/canonicalize-stored-flow.test.ts b/packages/services/service-automation/src/canonicalize-stored-flow.test.ts new file mode 100644 index 0000000000..b241807dfe --- /dev/null +++ b/packages/services/service-automation/src/canonicalize-stored-flow.test.ts @@ -0,0 +1,159 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #4454 — `AutomationEngine.canonicalizeStoredFlow`: one policy, two shapes. + * + * `os migrate meta --stored` (#4327) covers every metadata type except `flow`, + * because flow-node conversions carry ADR-0078's open-namespace conflict guard + * and that needs the engine's live executor registry. This method is the entry + * point that lets a caller outside the load seam ask for a flow's canonical + * shape without registering (and thereby arming) it. + * + * The interesting half is what `storable` deliberately does NOT contain. + * `FlowSchema.parse` materializes defaults — `version`, `runAs`, per-edge + * `type` / `isDefault` — and persisting a default the author never wrote pins + * that row to today's value while untouched rows follow tomorrow's. Two + * populations with different behaviour is precisely the drift a + * canonicalization pass exists to remove, so these pin that it stays out. + */ +import { describe, expect, it, vi } from 'vitest'; +import { AutomationEngine } from './engine.js'; + +const silentLogger = { + info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn(), +} as any; + +/** + * A pre-17 flow: the retired `filters` alias on a crud node at top level AND + * inside a `loop` body, plus bare-string edge conditions in both places (the + * #4347 nesting case). + */ +const legacyFlow = () => ({ + name: 'sweep_stale', + label: 'Sweep Stale Leads', + type: 'autolaunched', + status: 'active', + nodes: [ + { id: 'start', type: 'start', label: 'Start', config: { objectName: 'lead', triggerType: 'record-after-update' } }, + { id: 'n1', type: 'delete_record', label: 'Delete Stale', config: { objectName: 'lead', filters: { status: 'stale' } } }, + { + id: 'loop1', + type: 'loop', + label: 'Per Item', + config: { + collection: '{record.items}', + // A well-formed region: single entry (b1), single exit (b2), + // acyclic — `validateControlFlow` rejects anything else, and it + // runs before the region canonicalization under test. + body: { + nodes: [ + { id: 'b1', type: 'update_record', label: 'Touch', config: { objectName: 'lead', filters: { id: '{item.id}' } } }, + { id: 'b2', type: 'update_record', label: 'Mark', config: { objectName: 'lead', filter: { id: '{item.id}' } } }, + ], + edges: [{ id: 'be1', source: 'b1', target: 'b2', condition: "status == 'x'" }], + }, + }, + }, + ], + edges: [ + { id: 'e0', source: 'start', target: 'n1' }, + { id: 'e1', source: 'n1', target: 'loop1', condition: "status == 'y'" }, + ], +}); + +describe('canonicalizeStoredFlow — the storable shape (#4454)', () => { + it('lowers the retired filters alias, including inside a loop body', () => { + const engine = new AutomationEngine(silentLogger); + const { storable } = engine.canonicalizeStoredFlow('sweep_stale', legacyFlow()); + const s = storable as any; + + expect(s.nodes[1].config.filter).toEqual({ status: 'stale' }); + expect('filters' in s.nodes[1].config).toBe(false); + // The conversion pass already reaches into regions — only the condition + // envelope needs the schema's help. + expect(s.nodes[2].config.body.nodes[0].config.filter).toEqual({ id: '{item.id}' }); + expect('filters' in s.nodes[2].config.body.nodes[0].config).toBe(false); + }); + + it('lifts the condition envelope at BOTH nesting depths', () => { + const engine = new AutomationEngine(silentLogger); + const { storable } = engine.canonicalizeStoredFlow('sweep_stale', legacyFlow()); + const s = storable as any; + + expect(s.edges[1].condition).toEqual({ dialect: 'cel', source: "status == 'y'" }); + // #4347's asymmetry is what made this worth doing: the identical + // predicate one level in used to keep its bare-string shape. + expect(s.nodes[2].config.body.edges[0].condition).toEqual({ dialect: 'cel', source: "status == 'x'" }); + }); + + it('does NOT persist schema defaults — a migrated row must not freeze on today\'s values', () => { + const engine = new AutomationEngine(silentLogger); + const { storable, parsed } = engine.canonicalizeStoredFlow('sweep_stale', legacyFlow()); + const s = storable as any; + + // The parse materializes these; the stored shape must not carry them, + // or every migrated row is pinned to today's default while untouched + // rows follow tomorrow's. + expect('version' in s).toBe(false); + expect('runAs' in s).toBe(false); + expect('type' in s.edges[0]).toBe(false); + expect('isDefault' in s.edges[0]).toBe(false); + + // …while the EXECUTION shape does carry them, which is the whole + // reason the two shapes are distinct. + expect((parsed as any).version).toBeDefined(); + expect((parsed as any).edges[0].type).toBeDefined(); + }); + + it('leaves an already-canonical flow byte-identical — the pass is a no-op on protocol', () => { + const engine = new AutomationEngine(silentLogger); + const canonical = engine.canonicalizeStoredFlow('sweep_stale', legacyFlow()).storable; + const second = engine.canonicalizeStoredFlow('sweep_stale', canonical).storable; + + expect(JSON.stringify(second)).toBe(JSON.stringify(canonical)); + // …and reports nothing, which is what lets a re-run report "canonical". + expect(engine.canonicalizeStoredFlow('sweep_stale', canonical).notices).toEqual([]); + }); + + it('reports the conversions it applied, so a migration can name them per row', () => { + const engine = new AutomationEngine(silentLogger); + const { notices } = engine.canonicalizeStoredFlow('sweep_stale', legacyFlow()); + + expect(notices.length).toBeGreaterThan(0); + expect(notices.some((n) => n.from === 'filters' && n.to === 'filter')).toBe(true); + }); + + it('leaves a node config predicate alone — the parse never lowers an open z.record', () => { + const engine = new AutomationEngine(silentLogger); + const flow = legacyFlow(); + (flow.nodes[0].config as any).condition = 'title != previous.title'; + const { storable } = engine.canonicalizeStoredFlow('sweep_stale', flow); + + // A start node's record-change predicate is config, not an edge — no + // envelope exists on the parsed side, so the graft must not invent one. + expect((storable as any).nodes[0].config.condition).toBe('title != previous.title'); + }); + + it('throws on an unrecognized key rather than silently dropping it (#4001)', () => { + const engine = new AutomationEngine(silentLogger); + const flow = { ...legacyFlow(), _uiPosition: { x: 1, y: 2 } }; + + // FlowSchema is strict. A stored row carrying this cannot be registered + // at all, so a migration must report it failed — never persist a guess. + expect(() => engine.canonicalizeStoredFlow('sweep_stale', flow)).toThrow(); + }); +}); + +describe('registerFlow still behaves identically (#4454 refactor)', () => { + it('registers the converted flow and serves the parsed shape', async () => { + const engine = new AutomationEngine(silentLogger); + engine.registerFlow('sweep_stale', legacyFlow()); + + const flow: any = await engine.getFlow('sweep_stale'); + expect(flow).not.toBeNull(); + // Execution sees canonical config… + expect(flow.nodes[1].config.filter).toEqual({ status: 'stale' }); + // …and the defaults it needs. + expect(flow.version).toBeDefined(); + }); +}); diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 937f825df3..789201dc92 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -14,7 +14,7 @@ import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; import type { Logger } from '@objectstack/spec/contracts'; import { FlowSchema, FLOW_STRUCTURAL_NODE_TYPES, validateControlFlow, normalizeControlFlowRegions, collectFlowGraphs, findRegionEntry, defineActionDescriptor } from '@objectstack/spec/automation'; import { resolveFlowNodeExpressions } from '@objectstack/spec/automation'; -import { applyConversionsToFlow } from '@objectstack/spec'; +import { applyConversionsToFlow, type ConversionNotice, type ConversionConflictNotice } from '@objectstack/spec'; import type { FlowRegionParsed } from '@objectstack/spec/automation'; import type { Connector, @@ -916,6 +916,69 @@ export interface SuspendedRunStore { loadTerminal?(runId: string): Promise; } +/** + * Lift the `{ dialect, source }` envelopes the flow schema derives for edge + * `condition`s back onto the conversion output — and take nothing else with + * them (#4454). + * + * This is the persistence half of {@link AutomationEngine.canonicalizeStoredFlow}. + * A stored flow that is written back must end up in the shape the load seam + * would produce, or the seam keeps re-deriving it on every boot and the + * migration was pointless. But `FlowSchema.parse` also materializes defaults + * (`version`, `runAs`, per-edge `type` / `isDefault`), and persisting a default + * the author never wrote pins that row to today's value forever — so the graft + * is deliberately narrow: it copies the lowered `condition`, nothing more. + * + * Structural alignment is by position, which is sound because neither the parse + * nor `normalizeControlFlowRegions` reorders or drops array members — both are + * copy-on-write maps. Where the two sides disagree in shape (a caller passed a + * mismatched pair), the converted side is returned untouched: this only ever + * lifts a value it can positively match. + * + * Node `config.condition` (e.g. a start node's record-change predicate) is + * left alone by construction — `FlowNodeSchema.config` is an open `z.record`, + * so the parse never lowers it, so there is no envelope on the parsed side to + * copy and the recursion finds a string facing a string. + */ +function graftConditionEnvelopes(converted: unknown, parsed: unknown): unknown { + if (Array.isArray(converted)) { + if (!Array.isArray(parsed)) return converted; + let changed = false; + const out = converted.map((entry, i) => { + const next = graftConditionEnvelopes(entry, parsed[i]); + if (next !== entry) changed = true; + return next; + }); + return changed ? out : converted; + } + if ( + converted && typeof converted === 'object' + && parsed && typeof parsed === 'object' && !Array.isArray(parsed) + ) { + const parsedRec = parsed as Record; + let changed = false; + const out: Record = {}; + for (const [key, value] of Object.entries(converted as Record)) { + if (key === 'condition' && typeof value === 'string') { + const lowered = parsedRec[key]; + if ( + lowered && typeof lowered === 'object' && !Array.isArray(lowered) + && typeof (lowered as { source?: unknown }).source === 'string' + ) { + out[key] = lowered; + changed = true; + continue; + } + } + const next = graftConditionEnvelopes(value, parsedRec[key]); + if (next !== value) changed = true; + out[key] = next; + } + return changed ? out : converted; + } + return converted; +} + export class AutomationEngine implements IAutomationService { /** * ADR-0044: maximum times a single node may be (re-)entered at the top @@ -1611,7 +1674,37 @@ export class AutomationEngine implements IAutomationService { // ── IAutomationService Contract Implementation ──────── - registerFlow(name: string, definition: unknown): void { + /** + * Canonicalize a flow definition the way the load seam does — the ONE + * policy, exposed so a caller that is not registering the flow can still + * ask "what is this flow's canonical shape?" (#4454). + * + * Two consumers, two shapes, one pass — because they share every expensive + * step and must never drift apart: + * + * - `parsed` is for **execution**: `FlowSchema.parse` output with the + * region pass applied, i.e. schema defaults materialized. This is what + * {@link registerFlow} runs and stores in `this.flows`. + * - `storable` is for **persistence** (`os migrate meta --stored`, #4327): + * the conversion output plus the `condition` envelopes the schema lowers, + * and *deliberately nothing else*. Schema defaults (`version`, `runAs`, + * per-edge `type` / `isDefault`) are excluded on purpose — writing values + * the author never wrote would freeze every migrated row on today's + * defaults while untouched rows follow tomorrow's, i.e. two populations + * with different behaviour. That is exactly the drift a canonicalization + * pass exists to remove, so the pass must not become a source of it. + * + * Throws whatever the parse throws. `FlowSchema` is **strict** (#4001), so + * a flow carrying an unrecognized key is a hard error here rather than a + * silent drop; a caller migrating stored rows reports that row as failed + * instead of persisting a guess. + */ + canonicalizeStoredFlow(name: string, definition: unknown): { + parsed: FlowParsed; + storable: unknown; + notices: ConversionNotice[]; + conflicts: ConversionConflictNotice[]; + } { // ADR-0087 D2 — the runtime load seam. A stored flow authored against an // old shape (a `webhook`/`http_request` callout node, a `delete_record` // with `config.filters`) is canonicalized on rehydration, BEFORE parse + @@ -1636,11 +1729,19 @@ export class AutomationEngine implements IAutomationService { ...this.nodeExecutors.keys(), ...this.actionDescriptors.keys(), ]); + const notices: ConversionNotice[] = []; + const conflicts: ConversionConflictNotice[] = []; const converted = applyConversionsToFlow(definition, { reservedNodeTypes, includeRetired: true, - onNotice: (n) => this.logger.warn(`[flow '${name}'] ${n.code}: ${n.message}`), - onConflict: (c) => this.logger.warn(`[flow '${name}'] ${c.code}: ${c.message}`), + onNotice: (n) => { + notices.push(n); + this.logger.warn(`[flow '${name}'] ${n.code}: ${n.message}`); + }, + onConflict: (c) => { + conflicts.push(c); + this.logger.warn(`[flow '${name}'] ${c.code}: ${c.message}`); + }, }); const flowShell = FlowSchema.parse(converted); @@ -1662,6 +1763,20 @@ export class AutomationEngine implements IAutomationService { // still reported by the validator that owns that message. const parsed = normalizeControlFlowRegions(flowShell); + return { + parsed, + storable: graftConditionEnvelopes(converted, parsed), + notices, + conflicts, + }; + } + + registerFlow(name: string, definition: unknown): void { + // One canonicalization policy, shared with the stored-row migration so + // the two can never disagree about what "canonical" means (#4454). + // Execution takes the parsed shape (schema defaults materialized). + const { parsed } = this.canonicalizeStoredFlow(name, definition); + // ADR-0018 §M1 — validate node types against the live action registry. // The protocol no longer gates `type` with a closed enum; membership is // checked here instead. Soft-fail (warn, don't throw): a flow authored diff --git a/packages/services/service-automation/src/inert-mode.test.ts b/packages/services/service-automation/src/inert-mode.test.ts new file mode 100644 index 0000000000..6e26850deb --- /dev/null +++ b/packages/services/service-automation/src/inert-mode.test.ts @@ -0,0 +1,119 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #4454 — `armRuntime: false`: an engine, and nothing armed. + * + * `os migrate meta --stored` needs this plugin for exactly one read-only thing: + * the live executor registry, so ADR-0078's open-namespace conflict guard can + * tell a flow-node rename from a clobber. It must not get the runtime that + * normally rides along — booting a migration process that arms record triggers, + * fires scheduled jobs, opens connector connections, or resumes a paused + * approval is indefensible. + * + * The two halves are equally load-bearing and are tested as such: nothing is + * armed, AND the registry is complete. A partial registry would not fail + * loudly — it would make the guard read a live custom node type as unowned and + * rewrite over it, which is the exact silent clobber the guard exists to stop. + */ +import { describe, expect, it, vi } from 'vitest'; +import { AutomationServicePlugin } from './plugin.js'; + +/** A minimal PluginContext that records what the plugin did with it. */ +function makeCtx() { + const services = new Map(); + const hooks: string[] = []; + const triggered: string[] = []; + const ctx: any = { + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + registerService: (name: string, svc: unknown) => services.set(name, svc), + getService: (name: string) => { + if (services.has(name)) return services.get(name); + throw new Error(`no service ${name}`); + }, + getServices: () => services, + hook: (name: string) => { hooks.push(name); }, + trigger: async (name: string) => { triggered.push(name); }, + }; + return { ctx, services, hooks, triggered }; +} + +async function boot(options: Record) { + const h = makeCtx(); + const plugin = new AutomationServicePlugin(options as any); + await plugin.init(h.ctx); + await plugin.start(h.ctx); + return { ...h, plugin, engine: h.services.get('automation') as any }; +} + +describe('armRuntime: false — nothing is armed (#4454)', () => { + it('registers no flow, so no trigger or schedule is bound', async () => { + const { engine } = await boot({ armRuntime: false, suspendedRunStore: 'memory' }); + + expect(await engine.listFlows()).toEqual([]); + // The audit is the engine's own account of what it would fire. + expect(engine.getFlowRuntimeStates()).toEqual([]); + }); + + it('arms none of the runtime lifecycle hooks that would register flows later', async () => { + // Skipping only the boot pull would be a half-measure: `kernel:ready` + // and `metadata:reloaded` both re-register flows, so a long-lived + // inert process would arm them a moment later. + const { hooks } = await boot({ armRuntime: false, suspendedRunStore: 'memory' }); + + expect(hooks).not.toContain('kernel:ready'); + expect(hooks).not.toContain('metadata:reloaded'); + }); + + it('still fires automation:ready — a partial registry would corrupt the guard', async () => { + // This is the one thing inert mode must NOT skip. Third-party executors + // register on this hook; without them `reservedNodeTypes` is short, and + // the conflict guard silently rewrites over a live custom node type + // instead of refusing. + const { triggered } = await boot({ armRuntime: false, suspendedRunStore: 'memory' }); + + expect(triggered).toContain('automation:ready'); + }); + + it('has the built-in node registry populated, which is what the migration needs', async () => { + const { engine } = await boot({ armRuntime: false, suspendedRunStore: 'memory' }); + + const types = engine.getRegisteredNodeTypes(); + expect(types.length).toBeGreaterThan(0); + expect(types).toContain('delete_record'); + // …and the method the migration actually calls works off it. + expect(typeof engine.canonicalizeStoredFlow).toBe('function'); + }); + + it('canonicalizes a stored flow — the whole point of booting it at all', async () => { + const { engine } = await boot({ armRuntime: false, suspendedRunStore: 'memory' }); + + const { storable, notices } = engine.canonicalizeStoredFlow('purge', { + name: 'purge', + label: 'Purge', + type: 'autolaunched', + status: 'active', + nodes: [{ id: 'n1', type: 'delete_record', label: 'Purge', config: { objectName: 'lead', filters: { status: 'stale' } } }], + edges: [], + }); + + expect((storable as any).nodes[0].config.filter).toEqual({ status: 'stale' }); + expect(notices.some((n: any) => n.from === 'filters')).toBe(true); + // Still nothing registered — canonicalizing is not registering. + expect(await engine.listFlows()).toEqual([]); + }); +}); + +describe('the default is unchanged (#4454)', () => { + it('arms the runtime hooks when armRuntime is not set', async () => { + const { hooks, triggered } = await boot({ suspendedRunStore: 'memory' }); + + expect(triggered).toContain('automation:ready'); + expect(hooks).toContain('kernel:ready'); + expect(hooks).toContain('metadata:reloaded'); + }); + + it('arms them when armRuntime is explicitly true', async () => { + const { hooks } = await boot({ armRuntime: true, suspendedRunStore: 'memory' }); + expect(hooks).toContain('kernel:ready'); + }); +}); diff --git a/packages/services/service-automation/src/plugin.ts b/packages/services/service-automation/src/plugin.ts index ba9881ea33..2888086b0d 100644 --- a/packages/services/service-automation/src/plugin.ts +++ b/packages/services/service-automation/src/plugin.ts @@ -60,6 +60,28 @@ export function parseObjectFieldSchema( export interface AutomationServicePluginOptions { /** Enable debug logging for flow execution */ debug?: boolean; + /** + * Bring up the automation **runtime**, not just the engine. Default `true` + * — every server, dev stack and test host wants this and is unaffected. + * + * Set `false` for a one-shot tool that needs the engine as a *reference* + * rather than a runtime — today that is `os migrate meta --stored` (#4454), + * which needs the live executor registry so ADR-0078's open-namespace + * conflict guard can tell a flow-node rename from a clobber, and needs + * nothing else. Inert mode still installs the built-in nodes and still + * fires `automation:ready` (so third-party executors register and the + * registry is COMPLETE — a partial one would make the guard's verdict + * wrong), then stops before anything is armed: + * + * | Skipped in inert mode | Why it must be | + * |---|---| + * | flow pull + `kernel:ready` / `metadata:reloaded` re-sync | `registerFlow` calls `activateFlowTrigger` — record triggers and scheduled jobs would go live | + * | declarative connector materialization | opens real connections; an MCP provider spawns a child process | + * | suspended-run wait-timer re-arm | would RESUME someone's paused approval mid-migration | + * + * A migration process must not become a second server. + */ + armRuntime?: boolean; /** * Durable suspended-run persistence (ADR-0019): * - `'auto'` (default): persist to `sys_automation_run` via the ObjectQL @@ -478,6 +500,37 @@ export class AutomationServicePlugin implements Plugin { `[Automation] Engine started with ${nodeTypes.length} node types: ${nodeTypes.join(', ') || '(none)'}`, ); + // ── Inert mode (#4454) — an engine, and nothing armed ───────────────── + // A one-shot tool (`os migrate meta --stored`) needs this engine for one + // read-only thing: `reservedNodeTypes`, the live executor registry that + // ADR-0078's open-namespace conflict guard consults to tell a rename + // from a clobber. It does NOT want the runtime this plugin normally + // brings up, and everything below this line arms something: + // + // • the flow pull calls `registerFlow`, which calls + // `activateFlowTrigger` — record triggers and scheduled jobs go live; + // • `materializeDeclaredConnectors` opens real connections (an MCP + // provider spawns a child process); + // • the `metadata:reloaded` / `kernel:ready` hooks re-register flows, + // so a long-lived process would arm them later even if we skipped + // the boot pull; + // • `rearmSuspendedWaitTimers` RESUMES suspended runs — a migration + // that silently continues someone's paused approval is indefensible. + // + // The return is placed AFTER `automation:ready` deliberately: that hook + // is how third-party plugins contribute node executors, and a partial + // registry would make the conflict guard's answer wrong — it would read + // a live custom node type as unowned and rewrite over it. Registry + // population is the one thing inert mode must NOT skip. + if (this.options.armRuntime === false) { + ctx.logger.info( + '[Automation] inert mode (armRuntime: false) — engine and node registry are up; ' + + 'no flow registered, no trigger or schedule armed, no connector materialized, ' + + 'no suspended run resumed.', + ); + return; + } + // Upgrade to durable suspended-run persistence when an ObjectQL engine is // present (ADR-0019). The engine was constructed in init() before // services were wired, so we attach the DB-backed store here. Without an diff --git a/packages/spec/src/contracts/automation-service.ts b/packages/spec/src/contracts/automation-service.ts index 6b5d386c23..ea09968fcf 100644 --- a/packages/spec/src/contracts/automation-service.ts +++ b/packages/spec/src/contracts/automation-service.ts @@ -17,6 +17,7 @@ import type { FlowParsed } from '../automation/flow.zod'; import type { ExecutionLog, FlowRunSummary } from '../automation/execution.zod'; import type { ActionDescriptor } from '../automation/node-executor.zod'; import type { ConnectorDescriptor } from '../integration/connector-descriptor'; +import type { ConversionNotice, ConversionConflictNotice } from '../conversions/types'; /** * Context passed to a flow/script execution @@ -340,6 +341,38 @@ export interface IAutomationService { */ registerFlow?(name: string, definition: unknown): void; + /** + * Canonicalize a flow definition WITHOUT registering it (#4454). + * + * The same ADR-0087 conversion policy {@link registerFlow} applies, exposed + * for a caller that needs a flow's canonical shape but must not arm it — + * `os migrate meta --stored` rewriting stored `sys_metadata` rows is the + * reason this is on the contract rather than only on the implementation. + * + * Only an implementation holding the live executor registry can offer this: + * flow-node conversions carry ADR-0078's open-namespace conflict guard, and + * deciding a rename from a clobber requires knowing which node types are + * actually owned here. Hence optional — a caller falls back to leaving flow + * rows alone rather than guessing. + * + * @param name - Flow name (snake_case), used for diagnostics + * @param definition - The stored/authored flow body + * @returns `parsed` (execution shape — schema defaults materialized) and + * `storable` (persistence shape — conversions plus the schema's + * `condition` envelopes, deliberately WITHOUT schema defaults, so a + * written-back row is not frozen on today's default values), plus the + * conversions applied and any rewrite the guard refused. + * @throws when the definition cannot be canonicalized at all (a strict-schema + * violation, a malformed control-flow region) — such a flow cannot be + * registered either, so a caller reports it rather than persisting a guess. + */ + canonicalizeStoredFlow?(name: string, definition: unknown): { + parsed: FlowParsed; + storable: unknown; + notices: ConversionNotice[]; + conflicts: ConversionConflictNotice[]; + }; + /** * Unregister a flow by name * @param name - Flow name (snake_case)