fix(spec,service-automation): ADR-0087 — wire conversions into the runtime flow-load seam + conflict guard

Close two third-party-facing risks in the P1 conversion layer.

Risk 1 (data-loss regression) — retiring the executor's `filters` fallback
while the conversion only ran on the build/validate seam meant a stored flow
rehydrated from the DB (engine.registerFlow → FlowSchema.parse, which bypassed
the conversion) kept carrying `config.filters`; with the fallback gone the
executor saw an empty `filter` and a `delete_record`/`update_record` widened to
the whole table. Fix: run the D2 conversion in registerFlow BEFORE parse (new
applyConversionsToFlow), the runtime load seam ADR-0087 D2 calls for. Stored
old-shape flows (`filters`, `webhook`/`http_request` callouts) are canonicalized
on rehydration; the executor only ever sees the canonical shape. This is the
gap config-aliases.ts already described (graph-lint can't reach never-republished
prod flows) — now closed for `filters` by the conversion, not a fallback.

Risk 2 (silent third-party clobber) — `flow.node.type` is an OPEN namespace
(ADR-0018 removed the enum gate), so a retired official name could be
re-registered by a third party. The node-type rename is now conflict-aware:
the runtime seam passes its live executor registry as reservedNodeTypes; if a
retired alias (`http_request`/`http_call`/`webhook`) is a live custom node, the
rewrite is REFUSED and a loud, structured OS_METADATA_CONVERSION_CONFLICT
diagnostic (node path, conversion id, rename hint) is emitted instead of a
silent clobber (ADR-0078). The pure build/validate seam has no registry, so the
historical alias converts as before.

Mechanism/policy split kept clean: spec provides the conflict-aware conversion
(reservedNodeTypes context on apply); service-automation supplies the data. The
crud-config-aliases test is rewritten to prove the stored-`filters` flow now
keeps its filter (the risk-1 fix); a new flow-load-conversion test covers the
rehydration rename and the conflict guard. New exports are additive.

Refs #2643, #2645 (ADR-0087 D2), ADR-0078.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YA9vjVpDrKLuaK6nbm1s37

Author: claude

.changeset/adr-0087-p1-conversion-layer.md

@@ -1,6 +1,6 @@
 ---
 '@objectstack/spec': minor
-'@objectstack/service-automation': patch
+'@objectstack/service-automation': minor
 ---
 
 ADR-0087 P1:元数据转换层(conversion layer,D2)——大多数破坏性变更对使用方零操作。
@@ -13,6 +13,10 @@ ADR-0087 P1:元数据转换层(conversion layer,D2)——大多数破坏性变
 - `page-kind-jsx-to-html`:页面 `kind: 'jsx'` → `'html'`(ADR-0080 规范拼写)。
 - `flow-node-crud-filter-alias`:CRUD 流程节点 `config.filters` → `config.filter`。
 
+**运行时加载 seam(存量流程零回归的关键)。** 转换不仅接在构建/校验入口,也接到运行时 `AutomationEngine.registerFlow`(在 `FlowSchema.parse` 之前跑,新增 `applyConversionsToFlow`)。这样从数据库 rehydrate 的**存量流程**也会被规范化——否则删掉 `filters` 执行器兜底会让存量 `delete_record` / `update_record` 的过滤条件被静默清空(退化成作用于全表)。这才真正兑现 D2 “applied at load, the same seam”。
+
+**开放命名空间的冲突守卫(第三方零静默误伤)。** `flow.node.type` 是开放命名空间(ADR-0018 移除了 enum gate),退役的官方名可能被第三方复用为自定义节点。转换层新增“保留名冲突”感知:运行时 seam 传入本环境已注册的执行器类型,若某退役别名(`http_request`/`http_call`/`webhook`)正被活的自定义执行器占用,则**拒绝改写并发出响亮的结构化告警 `OS_METADATA_CONVERSION_CONFLICT`**(带节点位置、conversion id、“请改名”的处置建议),而不是静默把它改成 `http` 破坏第三方节点。构建/校验入口无注册表上下文,历史别名照常转换。
+
 并落实 PD #12 退役路径示范:`filters` → `filter` 别名从 `service-automation` 执行器的 `readAliasedConfig` 兜底中删除,提升为上面这条声明式转换条目;执行器改为直接读取规范键 `cfg.filter`。
 
-新增导出(纯增量,无破坏):`applyConversions`、`collectConversionNotices`、`ALL_CONVERSIONS`、`CONVERSIONS_BY_MAJOR`、`CONVERSION_NOTICE_CODE`,以及类型 `MetadataConversion`、`ConversionNotice`、`ConversionApplication`、`ConversionFixture`、`ApplyConversionsOptions`、`NormalizeStackInputOptions`。`normalizeStackInput` 现接受可选第二参 `{ onConversionNotice }`(向后兼容)。
+新增导出(纯增量,无破坏):`applyConversions`、`applyConversionsToFlow`、`collectConversionNotices`、`ALL_CONVERSIONS`、`CONVERSIONS_BY_MAJOR`、`CONVERSION_NOTICE_CODE`、`CONVERSION_CONFLICT_CODE`,以及类型 `MetadataConversion`、`ConversionNotice`、`ConversionApplication`、`ConversionFixture`、`ConversionContext`、`ConversionConflictNotice`、`ConversionConflictDetail`、`ApplyConversionsOptions`、`NormalizeStackInputOptions`。`normalizeStackInput` 现接受可选第二参 `{ onConversionNotice, convert }`(向后兼容)。

packages/services/service-automation/src/builtin/config-aliases.ts

@@ -28,9 +28,12 @@
  * The `filters` → `filter` alias has since been **retired from this shim** and
  * promoted into the ADR-0087 D2 conversion layer (`@objectstack/spec` conversion
  * `flow-node-crud-filter-alias`): it is rewritten to the canonical key **at
- * load**, so the CRUD executors read `cfg.filter` directly. That is the PD #12
- * retirement path the ADR prescribes — a scattered consumer-side fallback
- * replaced by one declared, loud, tested, expiring conversion entry. The
+ * load** — including the runtime rehydration seam (`AutomationEngine.registerFlow`
+ * runs the conversion before parse), which is exactly the stored-prod-flow gap
+ * this shim describes above, now closed for `filters` by the conversion instead
+ * of an executor fallback. So the CRUD executors read `cfg.filter` directly. That
+ * is the PD #12 retirement path the ADR prescribes — a scattered consumer-side
+ * fallback replaced by one declared, loud, tested, expiring conversion entry. The
  * remaining `object` → `objectName` alias is the next candidate to graduate.
  *
  * @see crud-nodes.ts for the remaining call site (`objectName`).

packages/services/service-automation/src/builtin/crud-config-aliases.test.ts

@@ -90,30 +90,28 @@ describe('CRUD config-key aliases: object→objectName (executor shim) + filters
     expect(warns.length).toBe(before);
   });
 
-  it('no longer honors a raw `filters` alias in the executor (retired to the D2 conversion layer)', async () => {
+  it('a stored `filters` flow is canonicalized at registerFlow, so the filter is NOT dropped', async () => {
+    // Risk-1 regression guard (ADR-0087 D2 runtime load seam). Before the seam
+    // was wired, retiring the executor's `filters` fallback meant a stored
+    // `delete_record`/`get_record` flow carrying `filters` reached the executor
+    // with an empty `filter` — silently widening the query to the whole table.
+    // Now `registerFlow` runs the D2 conversion, so `filters`→`filter` happens
+    // on rehydration and the stored filter survives.
     const engine = new AutomationEngine(silentLogger());
     const { data, calls } = fakeData();
-    const warns: string[] = [];
-    registerCrudNodes(engine, ctxWith(data, collectingLogger(warns)));
+    registerCrudNodes(engine, ctxWith(data, silentLogger()));
 
-    // A flow that reached the executor WITHOUT going through the load conversion
-    // still carries `filters`. The executor now reads only the canonical `filter`,
-    // so `filters` is ignored and no `filters`→`filter` warning is emitted here.
     engine.registerFlow('gr', getRecordFlow({ objectName: 'crm_lead', filters: { id: 'L1' } }));
     await engine.execute('gr');
 
-    expect(calls[0].opts.where).toEqual({}); // `filters` no longer honored by the executor
-    const filterWarn = warns.find((w) => w.includes("'filters'"));
-    expect(filterWarn).toBeFalsy();
+    expect(calls[0].opts.where).toEqual({ id: 'L1' }); // filter preserved, not dropped
   });
 
-  it('the D2 conversion at load rewrites `filters` → `filter` so the flow works end-to-end', async () => {
+  it('the same conversion runs on the build/validate seam too (normalizeStackInput)', async () => {
     const engine = new AutomationEngine(silentLogger());
     const { data, calls } = fakeData();
     registerCrudNodes(engine, ctxWith(data, silentLogger()));
 
-    // Author with the deprecated `filters` key, then run it through the same load
-    // seam a real stack load uses. The conversion canonicalizes it to `filter`.
     const raw = getRecordFlow({ objectName: 'crm_lead', filters: { id: 'L1' }, outputVariable: 'lead' });
     const converted = (normalizeStackInput({ flows: [raw] }).flows as any[])[0];
     engine.registerFlow('gr', converted);

packages/services/service-automation/src/engine.ts

@@ -5,6 +5,7 @@ import type { ExecutionLog, ActionDescriptor } from '@objectstack/spec/automatio
 import type { AutomationContext, AutomationResult, ResumeSignal, IAutomationService, ScreenSpec } from '@objectstack/spec/contracts';
 import type { Logger } from '@objectstack/spec/contracts';
 import { FlowSchema, FLOW_STRUCTURAL_NODE_TYPES, validateControlFlow, findRegionEntry, defineActionDescriptor } from '@objectstack/spec/automation';
+import { applyConversionsToFlow } from '@objectstack/spec';
 import type { FlowRegionParsed } from '@objectstack/spec/automation';
 import type { Connector } from '@objectstack/spec/integration';
 import { ConnectorSchema } from '@objectstack/spec/integration';
@@ -884,7 +885,26 @@ export class AutomationEngine implements IAutomationService {
     // ── IAutomationService Contract Implementation ────────
 
     registerFlow(name: string, definition: unknown): void {
-        const parsed = FlowSchema.parse(definition);
+        // 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 +
+        // execution, so the executor only ever sees the canonical shape and a
+        // dropped-alias upgrade never silently changes behavior (e.g. an empty
+        // `filter` deleting a whole table). `reservedNodeTypes` is this engine's
+        // live executor registry: an open-namespace node-type rename over a type
+        // a custom executor owns becomes a loud, refused conflict — never a
+        // silent clobber of the third party's node.
+        const reservedNodeTypes = new Set<string>([
+            ...FLOW_STRUCTURAL_NODE_TYPES,
+            ...this.nodeExecutors.keys(),
+            ...this.actionDescriptors.keys(),
+        ]);
+        const converted = applyConversionsToFlow(definition, {
+            reservedNodeTypes,
+            onNotice: (n) => this.logger.warn(`[flow '${name}'] ${n.code}: ${n.message}`),
+            onConflict: (c) => this.logger.warn(`[flow '${name}'] ${c.code}: ${c.message}`),
+        });
+        const parsed = FlowSchema.parse(converted);
 
         // DAG cycle detection
         this.detectCycles(parsed);

packages/services/service-automation/src/flow-load-conversion.test.ts

@@ -0,0 +1,79 @@
+// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * ADR-0087 D2 runtime load seam: `AutomationEngine.registerFlow` canonicalizes a
+ * stored flow's shape on rehydration, BEFORE parse + execution — so a stored
+ * flow authored against an old protocol shape keeps running with zero action,
+ * and dropping a deprecated executor alias never silently changes behavior.
+ *
+ * Open-namespace node-type renames (`webhook`/`http_request`/`http_call` → `http`)
+ * are conflict-aware: if a live custom executor owns the retired name, the rename
+ * is refused and a loud diagnostic is logged instead of clobbering the node.
+ */
+import { describe, it, expect } from 'vitest';
+import { AutomationEngine } from './engine.js';
+
+function collectingLogger(warns: string[]): any {
+  const l: any = { info() {}, warn(m: string) { warns.push(m); }, error() {}, debug() {} };
+  l.child = () => l;
+  return l;
+}
+
+function flowWith(node: Record<string, unknown>) {
+  return {
+    name: 'f', label: 'F', type: 'autolaunched',
+    nodes: [
+      { id: 'start', type: 'start', label: 'Start' },
+      { id: 'n', label: 'N', ...node },
+      { id: 'end', type: 'end', label: 'End' },
+    ],
+    edges: [
+      { id: 'e1', source: 'start', target: 'n' },
+      { id: 'e2', source: 'n', target: 'end' },
+    ],
+  };
+}
+
+describe('registerFlow canonicalizes stored flows (ADR-0087 D2 runtime seam)', () => {
+  it('rewrites a retired callout node type to `http` on rehydration', async () => {
+    const warns: string[] = [];
+    const engine = new AutomationEngine(collectingLogger(warns));
+    engine.registerFlow('f', flowWith({ id: 'n', type: 'http_request', config: { url: 'https://x' } }));
+
+    const stored = await engine.getFlow('f');
+    expect(stored?.nodes.find((n) => n.id === 'n')?.type).toBe('http');
+    expect(warns.some((w) => w.includes("'http_request' → 'http'"))).toBe(true);
+  });
+
+  it('leaves a canonical flow untouched (no notice)', async () => {
+    const warns: string[] = [];
+    const engine = new AutomationEngine(collectingLogger(warns));
+    engine.registerFlow('f', flowWith({ id: 'n', type: 'http', config: { url: 'https://x' } }));
+
+    const stored = await engine.getFlow('f');
+    expect(stored?.nodes.find((n) => n.id === 'n')?.type).toBe('http');
+    expect(warns.some((w) => w.includes('OS_METADATA_CONVERTED') || w.includes('→'))).toBe(false);
+  });
+
+  it('refuses to rewrite a retired name a live custom executor owns, logging a conflict', async () => {
+    const warns: string[] = [];
+    const engine = new AutomationEngine(collectingLogger(warns));
+
+    // A third party registers a custom node under the retired official name.
+    let ran = false;
+    engine.registerNodeExecutor({
+      type: 'webhook',
+      async execute() { ran = true; return { success: true, output: {} }; },
+    });
+
+    engine.registerFlow('f', flowWith({ id: 'n', type: 'webhook', config: {} }));
+
+    const stored = await engine.getFlow('f');
+    // NOT clobbered to `http` — the custom node survives.
+    expect(stored?.nodes.find((n) => n.id === 'n')?.type).toBe('webhook');
+    expect(warns.some((w) => w.includes('OS_METADATA_CONVERSION_CONFLICT'))).toBe(true);
+
+    await engine.execute('f');
+    expect(ran).toBe(true); // the custom executor actually ran
+  });
+});

packages/spec/api-surface.json

@@ -13,12 +13,16 @@
     "BUILTIN_IDENTITY_PLATFORM_ADMIN (const)",
     "BuiltinIdentityName (type)",
     "CONVERSIONS_BY_MAJOR (const)",
+    "CONVERSION_CONFLICT_CODE (const)",
     "CONVERSION_NOTICE_CODE (const)",
     "ComposeStacksOptions (type)",
     "ComposeStacksOptionsSchema (const)",
     "ConflictStrategy (type)",
     "ConflictStrategySchema (const)",
     "ConversionApplication (interface)",
+    "ConversionConflictDetail (interface)",
+    "ConversionConflictNotice (interface)",
+    "ConversionContext (interface)",
     "ConversionFixture (interface)",
     "ConversionNotice (interface)",
     "CronExpressionInputSchema (const)",
@@ -91,6 +95,7 @@
     "Tool (type)",
     "ViewKeyCollision (interface)",
     "applyConversions (function)",
+    "applyConversionsToFlow (function)",
     "applyMetaMigrations (function)",
     "cel (function)",
     "collectConversionNotices (function)",

packages/spec/src/conversions/apply.ts

@@ -12,7 +12,13 @@
  */
 
 import { ALL_CONVERSIONS } from './registry.js';
-import { CONVERSION_NOTICE_CODE, type ConversionNotice } from './types.js';
+import {
+  CONVERSION_CONFLICT_CODE,
+  CONVERSION_NOTICE_CODE,
+  type ConversionConflictNotice,
+  type ConversionContext,
+  type ConversionNotice,
+} from './types.js';
 
 export interface ApplyConversionsOptions {
   /**
@@ -21,6 +27,18 @@ export interface ApplyConversionsOptions {
    * choice. `objectstack validate` passes a sink that prints them.
    */
   onNotice?: (notice: ConversionNotice) => void;
+  /**
+   * Sink for each structured **conflict** — a rename refused because its old
+   * token is a live name in this environment (see {@link ConversionContext}).
+   * Populated by the runtime load seam; absent on the build/validate seam.
+   */
+  onConflict?: (notice: ConversionConflictNotice) => void;
+  /**
+   * Node types that are live in this environment. Supplied by the runtime load
+   * seam so open-namespace renames can detect a collision with a live owner
+   * rather than silently clobber it.
+   */
+  reservedNodeTypes?: ReadonlySet<string>;
 }
 
 /**
@@ -35,34 +53,71 @@ export function applyConversions(
   stack: Record<string, unknown>,
   options: ApplyConversionsOptions = {},
 ): Record<string, unknown> {
-  const { onNotice } = options;
+  const { onNotice, onConflict, reservedNodeTypes } = options;
   let current = stack;
 
   for (const conversion of ALL_CONVERSIONS) {
     const retiresIn = conversion.toMajor + 1;
-    current = conversion.apply(current, (detail) => {
-      if (!onNotice) return;
-      onNotice({
-        code: CONVERSION_NOTICE_CODE,
-        conversionId: conversion.id,
-        surface: conversion.surface,
-        toMajor: conversion.toMajor,
-        retiresIn,
-        from: detail.from,
-        to: detail.to,
-        path: detail.path,
-        message:
-          `[protocol] converted ${conversion.surface} at ${detail.path}: ` +
-          `'${detail.from}' → '${detail.to}' (deprecated; ADR-0087 conversion ` +
-          `'${conversion.id}', retires from the load path in protocol ${retiresIn}). ` +
-          `Update the source to '${detail.to}'.`,
-      });
-    });
+    const context: ConversionContext = {
+      reservedNodeTypes,
+      reportConflict: onConflict
+        ? (detail) =>
+            onConflict({
+              code: CONVERSION_CONFLICT_CODE,
+              conversionId: conversion.id,
+              surface: conversion.surface,
+              token: detail.token,
+              path: detail.path,
+              message: `[protocol] ${detail.reason} (ADR-0087 conversion '${conversion.id}').`,
+            })
+        : undefined,
+    };
+    current = conversion.apply(
+      current,
+      (detail) => {
+        if (!onNotice) return;
+        onNotice({
+          code: CONVERSION_NOTICE_CODE,
+          conversionId: conversion.id,
+          surface: conversion.surface,
+          toMajor: conversion.toMajor,
+          retiresIn,
+          from: detail.from,
+          to: detail.to,
+          path: detail.path,
+          message:
+            `[protocol] converted ${conversion.surface} at ${detail.path}: ` +
+            `'${detail.from}' → '${detail.to}' (deprecated; ADR-0087 conversion ` +
+            `'${conversion.id}', retires from the load path in protocol ${retiresIn}). ` +
+            `Update the source to '${detail.to}'.`,
+        });
+      },
+      context,
+    );
   }
 
   return current;
 }
 
+/**
+ * Apply the conversion pass to a **single flow definition** — the shape the
+ * runtime automation engine loads one at a time via `registerFlow`.
+ *
+ * This is the runtime load seam ADR-0087 D2 calls for: a stored flow authored
+ * against an old shape (e.g. a `delete_record` node with `config.filters`, or a
+ * `webhook` callout node) is canonicalized on rehydration, so the executor only
+ * ever sees the canonical shape. Callers pass `reservedNodeTypes` (their live
+ * executor registry) so an open-namespace rename over a live custom node becomes
+ * a reported conflict, not a silent clobber. Non-object input is returned
+ * unchanged.
+ */
+export function applyConversionsToFlow<T>(flow: T, options: ApplyConversionsOptions = {}): T {
+  if (flow == null || typeof flow !== 'object' || Array.isArray(flow)) return flow;
+  const converted = applyConversions({ flows: [flow as Record<string, unknown>] }, options);
+  const flows = converted.flows;
+  return Array.isArray(flows) && flows.length > 0 ? (flows[0] as T) : flow;
+}
+
 /**
  * Collect the notices a stack would emit without needing an external sink —
  * convenience for `validate` / `lint` / the future MCP `spec_deprecations` tool.

packages/spec/src/conversions/conversions.test.ts

@@ -78,6 +78,36 @@ describe('conversion layer (ADR-0087 D2)', () => {
     });
   });
 
+  describe('flow-node-http-callout-rename — reserved-name conflict guard', () => {
+    it('refuses to rewrite an alias a live executor owns, reporting a conflict', () => {
+      const notices: ConversionNotice[] = [];
+      const conflicts: { token: string; path: string; conversionId: string }[] = [];
+      const out = applyConversions(
+        { flows: [{ name: 'f', nodes: [{ id: 'a', type: 'webhook' }] }] },
+        {
+          onNotice: (n) => notices.push(n),
+          onConflict: (c) => conflicts.push({ token: c.token, path: c.path, conversionId: c.conversionId }),
+          reservedNodeTypes: new Set(['webhook']), // a third-party custom node owns this name
+        },
+      );
+      // Not rewritten — the custom node is preserved.
+      expect((out.flows as any[])[0].nodes[0].type).toBe('webhook');
+      expect(notices).toHaveLength(0);
+      expect(conflicts).toEqual([
+        { token: 'webhook', path: 'flows[0].nodes[0].type', conversionId: 'flow-node-http-callout-rename' },
+      ]);
+    });
+
+    it('converts normally when the alias is not a live type (build/validate seam)', () => {
+      // No reservedNodeTypes → the historical alias converts as usual.
+      const { stack, notices } = collectConversionNotices({
+        flows: [{ name: 'f', nodes: [{ id: 'a', type: 'webhook' }] }],
+      });
+      expect((stack.flows as any[])[0].nodes[0].type).toBe('http');
+      expect(notices).toHaveLength(1);
+    });
+  });
+
   describe('flow-node-crud-filter-alias (PD #12 retirement)', () => {
     it('renames config.filters → config.filter only for CRUD node types', () => {
       const { stack, notices } = collectConversionNotices({