feat: add V55 dual-run migration controls

app/commands/migrate_n8n_workflow.py

@@ -0,0 +1,72 @@
+"""Convert one exported n8n workflow into governed migration artifacts."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from app.core.orchestration.migration.dual_run import (
+    MigrationBlocked,
+    N8nWorkflowConverter,
+)
+
+
+def convert_export(
+    workflow,
+    *,
+    dataflow_uid,
+    data_source_uids,
+):
+    try:
+        converted = N8nWorkflowConverter().convert(
+            workflow,
+            dataflow_uid=dataflow_uid,
+            data_source_uids=data_source_uids,
+        )
+    except MigrationBlocked as exc:
+        return exc.report
+    return {
+        "status": "converted",
+        "source_workflow_id": converted.source_workflow_id,
+        "definition_hash": converted.definition_hash,
+        "workflow_spec": converted.workflow_spec,
+        "schedule_plan": converted.schedule_plan,
+        "deployment_performed": False,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Convert one n8n export without deploying it or copying credentials."
+        )
+    )
+    parser.add_argument("--workflow-json", required=True)
+    parser.add_argument("--dataflow-uid", required=True)
+    parser.add_argument("--datasource-map-json", required=True)
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+
+    workflow = json.loads(
+        Path(args.workflow_json).read_text(encoding="utf-8")
+    )
+    mapping = json.loads(
+        Path(args.datasource_map_json).read_text(encoding="utf-8")
+    )
+    if not isinstance(mapping, dict):
+        raise ValueError("datasource map must be an object")
+    report = convert_export(
+        workflow,
+        dataflow_uid=args.dataflow_uid,
+        data_source_uids=mapping,
+    )
+    Path(args.output).write_text(
+        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    return 0 if report["status"] == "converted" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

app/commands/reconcile_dual_run.py

@@ -0,0 +1,60 @@
+"""Compare credential-safe n8n and Kestra execution snapshots."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from app.core.orchestration.migration.reconciliation import (
+    ExecutionSnapshot,
+    ReconciliationPolicy,
+    WorkflowReconciler,
+)
+
+
+def reconcile_exports(baseline, candidate, policy=None):
+    policy = policy or {}
+    if not isinstance(policy, dict):
+        raise ValueError("reconciliation policy must be an object")
+    return WorkflowReconciler().compare(
+        ExecutionSnapshot.from_dict(baseline),
+        ExecutionSnapshot.from_dict(candidate),
+        ReconciliationPolicy(
+            ignored_paths=policy.get("ignored_paths"),
+            numeric_tolerances=policy.get("numeric_tolerances"),
+        ),
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Reconcile bounded execution metrics without raw rows."
+    )
+    parser.add_argument("--n8n-snapshot", required=True)
+    parser.add_argument("--kestra-snapshot", required=True)
+    parser.add_argument("--policy-json")
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+
+    baseline = json.loads(
+        Path(args.n8n_snapshot).read_text(encoding="utf-8")
+    )
+    candidate = json.loads(
+        Path(args.kestra_snapshot).read_text(encoding="utf-8")
+    )
+    policy = (
+        json.loads(Path(args.policy_json).read_text(encoding="utf-8"))
+        if args.policy_json
+        else {}
+    )
+    report = reconcile_exports(baseline, candidate, policy)
+    Path(args.output).write_text(
+        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    return 0 if report["status"] == "passed" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

app/core/orchestration/migration/__init__.py

@@ -0,0 +1,5 @@
+"""Dual-run migration, reconciliation, cutover, and retirement controls."""
+
+from .dual_run import DualRunMode
+
+__all__ = ["DualRunMode"]

app/core/orchestration/migration/cutover.py

@@ -0,0 +1,300 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from app.core.orchestration.migration.dual_run import (
+    DualRunMode,
+    validate_mode_transition,
+)
+
+
+def _required_string(value: Any, label: str, maximum: int = 500) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{label} is required")
+    normalized = value.strip()
+    if len(normalized) > maximum:
+        raise ValueError(f"{label} exceeds {maximum} characters")
+    return normalized
+
+
+def _safe_engine_result(value: Any) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        return {"status": "completed"}
+    result = {}
+    for key in ("status", "execution_id", "revision"):
+        item = value.get(key)
+        if isinstance(item, (str, int, float, bool)) or item is None:
+            result[key] = item
+    return result or {"status": "completed"}
+
+
+@dataclass(frozen=True)
+class CutoverGates:
+    inventory_complete: bool
+    workflow_spec_valid: bool
+    rollback_target_present: bool
+    shadow_runs: int
+    required_shadow_runs: int
+    failure_recovery_observed: bool
+    reconciliation_passed: bool
+    connection_budget_passed: bool
+    sla_passed: bool
+    n8n_baseline_unchanged: bool
+
+    def __post_init__(self):
+        if isinstance(self.shadow_runs, bool) or not isinstance(
+            self.shadow_runs, int
+        ):
+            raise ValueError("shadow_runs must be an integer")
+        if isinstance(self.required_shadow_runs, bool) or not isinstance(
+            self.required_shadow_runs, int
+        ):
+            raise ValueError("required_shadow_runs must be an integer")
+        if self.shadow_runs < 0 or self.required_shadow_runs < 1:
+            raise ValueError("shadow run thresholds are invalid")
+
+    def as_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    def failed(self) -> list[str]:
+        checks = {
+            "inventory_complete": self.inventory_complete,
+            "workflow_spec_valid": self.workflow_spec_valid,
+            "rollback_target_present": self.rollback_target_present,
+            "shadow_runs": self.shadow_runs >= self.required_shadow_runs,
+            "failure_recovery_observed": self.failure_recovery_observed,
+            "reconciliation_passed": self.reconciliation_passed,
+            "connection_budget_passed": self.connection_budget_passed,
+            "sla_passed": self.sla_passed,
+            "n8n_baseline_unchanged": self.n8n_baseline_unchanged,
+        }
+        return [name for name, passed in checks.items() if not passed]
+
+
+class EngineRoleController:
+    def __init__(
+        self,
+        *,
+        n8n,
+        kestra,
+        n8n_namespace="dataops",
+        kestra_namespace="dataops",
+    ):
+        self.n8n = n8n
+        self.kestra = kestra
+        self.n8n_namespace = _required_string(
+            n8n_namespace, "n8n_namespace", 255
+        )
+        self.kestra_namespace = _required_string(
+            kestra_namespace, "kestra_namespace", 255
+        )
+
+    def deactivate_n8n(self, definition_id):
+        return self.n8n.deactivate(self.n8n_namespace, definition_id)
+
+    def activate_n8n(self, definition_id):
+        return self.n8n.activate(self.n8n_namespace, definition_id)
+
+    def deactivate_kestra(self, definition_id):
+        return self.kestra.deactivate(
+            self.kestra_namespace, definition_id
+        )
+
+    def activate_kestra(self, definition_id):
+        return self.kestra.activate(self.kestra_namespace, definition_id)
+
+
+class WorkflowCutoverService:
+    def __init__(self, *, store, engines):
+        self.store = store
+        self.engines = engines
+
+    def request_cutover(
+        self,
+        *,
+        dataflow_uid,
+        environment,
+        target_mode,
+        gates,
+        idempotency_key,
+        correlation_id,
+        retirement_decision=None,
+    ):
+        dataflow_uid = _required_string(dataflow_uid, "dataflow_uid", 255)
+        environment = _required_string(environment, "environment", 100)
+        idempotency_key = _required_string(
+            idempotency_key, "idempotency_key", 255
+        )
+        correlation_id = _required_string(correlation_id, "correlation_id", 255)
+        if not isinstance(gates, CutoverGates):
+            raise ValueError("cutover gates are required")
+
+        state = self.store.get_state(dataflow_uid, environment)
+        if not state:
+            raise ValueError("workflow migration state was not found")
+        source_mode = DualRunMode(state["mode"])
+        destination = DualRunMode(target_mode)
+        validate_mode_transition(source_mode, destination)
+
+        failed = gates.failed()
+        if failed:
+            raise ValueError(f"cutover gates failed: {', '.join(failed)}")
+        gate_evidence = gates.as_dict()
+        if destination is DualRunMode.KESTRA_PRIMARY:
+            valid_retirement = (
+                isinstance(retirement_decision, dict)
+                and retirement_decision.get("status") == "approved"
+                and retirement_decision.get("retirement_approved") is True
+                and retirement_decision.get("missing") == []
+            )
+            if not valid_retirement:
+                raise ValueError("independent retirement approval is required")
+            gate_evidence = {
+                "cutover": gate_evidence,
+                "retirement": {
+                    "status": "approved",
+                    "retirement_approved": True,
+                    "missing": [],
+                },
+            }
+
+        record = {
+            "dataflow_uid": dataflow_uid,
+            "environment": environment,
+            "source_mode": source_mode.value,
+            "target_mode": destination.value,
+            "n8n_definition_id": state.get("n8n_definition_id"),
+            "kestra_definition_id": state.get("kestra_definition_id"),
+            "gates": gate_evidence,
+            "idempotency_key": idempotency_key,
+            "correlation_id": correlation_id,
+        }
+        _, operation = self.store.request_transition(record)
+        return operation
+
+    def apply_requested_cutover(self, operation):
+        if not isinstance(operation, dict):
+            raise ValueError("cutover operation must be an object")
+        operation_id = _required_string(
+            operation.get("operation_id"), "operation_id", 255
+        )
+        source = DualRunMode(operation.get("source_mode"))
+        target = DualRunMode(operation.get("target_mode"))
+        validate_mode_transition(source, target)
+
+        primary_cutover = (
+            source is DualRunMode.N8N_PRIMARY_KESTRA_SHADOW
+            and target is DualRunMode.KESTRA_PRIMARY_N8N_STANDBY
+        )
+        final_retirement = (
+            source is DualRunMode.KESTRA_PRIMARY_N8N_STANDBY
+            and target is DualRunMode.KESTRA_PRIMARY
+        )
+        if not primary_cutover and not final_retirement:
+            raise ValueError("operation is not an executable Kestra cutover")
+
+        n8n_id = _required_string(
+            operation.get("n8n_definition_id"), "n8n_definition_id", 255
+        )
+        kestra_id = _required_string(
+            operation.get("kestra_definition_id"), "kestra_definition_id", 255
+        )
+        try:
+            n8n_result = self.engines.deactivate_n8n(n8n_id)
+            kestra_result = self.engines.activate_kestra(kestra_id)
+        except Exception:
+            if final_retirement:
+                result = {
+                    "status": "rolled_back",
+                    "mode": source.value,
+                    "reason_code": "retirement_finalization_failed",
+                }
+                self.store.rollback_transition(
+                    operation_id, source.value, result
+                )
+                return result
+            restore_result = self.engines.activate_n8n(n8n_id)
+            result = {
+                "status": "rolled_back",
+                "mode": source.value,
+                "n8n_restore": _safe_engine_result(restore_result),
+                "reason_code": "target_activation_failed",
+            }
+            self.store.rollback_transition(operation_id, source.value, result)
+            return result
+
+        result = {
+            "status": "succeeded",
+            "mode": target.value,
+            "n8n": _safe_engine_result(n8n_result),
+            "kestra": _safe_engine_result(kestra_result),
+        }
+        self.store.complete_transition(operation_id, target.value, result)
+        return result
+
+    def rollback_to_n8n(
+        self,
+        *,
+        dataflow_uid,
+        environment,
+        idempotency_key,
+        correlation_id,
+    ):
+        dataflow_uid = _required_string(dataflow_uid, "dataflow_uid", 255)
+        environment = _required_string(environment, "environment", 100)
+        state = self.store.get_state(dataflow_uid, environment)
+        if not state:
+            raise ValueError("workflow migration state was not found")
+
+        source = DualRunMode(state["mode"])
+        target = DualRunMode.N8N_PRIMARY_KESTRA_SHADOW
+        validate_mode_transition(source, target)
+        if source is not DualRunMode.KESTRA_PRIMARY_N8N_STANDBY:
+            raise ValueError("workflow is not in Kestra primary standby mode")
+
+        record = {
+            "dataflow_uid": dataflow_uid,
+            "environment": environment,
+            "source_mode": source.value,
+            "target_mode": target.value,
+            "n8n_definition_id": state.get("n8n_definition_id"),
+            "kestra_definition_id": state.get("kestra_definition_id"),
+            "gates": {},
+            "idempotency_key": _required_string(
+                idempotency_key, "idempotency_key", 255
+            ),
+            "correlation_id": _required_string(
+                correlation_id, "correlation_id", 255
+            ),
+        }
+        claimed, operation = self.store.request_transition(record)
+        if not claimed and operation.get("status") in {"succeeded", "rolled_back"}:
+            return {
+                "status": operation["status"],
+                "mode": operation.get("target_mode", target.value),
+            }
+
+        operation_id = _required_string(
+            operation.get("operation_id"), "operation_id", 255
+        )
+        kestra_result = self.engines.deactivate_kestra(
+            _required_string(
+                operation.get("kestra_definition_id"),
+                "kestra_definition_id",
+                255,
+            )
+        )
+        n8n_result = self.engines.activate_n8n(
+            _required_string(
+                operation.get("n8n_definition_id"), "n8n_definition_id", 255
+            )
+        )
+        result = {
+            "status": "succeeded",
+            "mode": target.value,
+            "kestra": _safe_engine_result(kestra_result),
+            "n8n": _safe_engine_result(n8n_result),
+        }
+        self.store.complete_transition(operation_id, target.value, result)
+        return result

app/core/orchestration/migration/dual_run.py

@@ -0,0 +1,400 @@
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Any
+
+from app.core.orchestration.spec import (
+    validate_schedule_plan,
+    validate_workflow_spec,
+)
+
+
+class DualRunMode(StrEnum):
+    N8N_PRIMARY = "n8n_primary"
+    N8N_PRIMARY_KESTRA_SHADOW = "n8n_primary_kestra_shadow"
+    KESTRA_PRIMARY_N8N_STANDBY = "kestra_primary_n8n_standby"
+    KESTRA_PRIMARY = "kestra_primary"
+
+
+_TRANSITIONS = {
+    DualRunMode.N8N_PRIMARY: {
+        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
+    },
+    DualRunMode.N8N_PRIMARY_KESTRA_SHADOW: {
+        DualRunMode.N8N_PRIMARY,
+        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+    },
+    DualRunMode.KESTRA_PRIMARY_N8N_STANDBY: {
+        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
+        DualRunMode.KESTRA_PRIMARY,
+    },
+    DualRunMode.KESTRA_PRIMARY: {
+        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+    },
+}
+
+
+def validate_mode_transition(current, target) -> bool:
+    try:
+        source = DualRunMode(current)
+        destination = DualRunMode(target)
+    except ValueError as exc:
+        raise ValueError("unsupported migration mode") from exc
+    if destination not in _TRANSITIONS[source]:
+        raise ValueError(
+            f"unsupported migration transition: {source.value} -> "
+            f"{destination.value}"
+        )
+    return True
+
+
+def _canonical(value: Any) -> str:
+    return json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    )
+
+
+def _hash(value: Any) -> str:
+    return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()
+
+
+def _required_string(value: Any, label: str, maximum: int = 500) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{label} is required")
+    normalized = value.strip()
+    if len(normalized) > maximum:
+        raise ValueError(f"{label} exceeds {maximum} characters")
+    return normalized
+
+
+def _node_id(name: str, used: set[str]) -> str:
+    candidate = re.sub(r"[^a-z0-9]+", "_", name.strip().lower()).strip("_")
+    if not candidate or not candidate[0].isalpha():
+        candidate = f"node_{candidate}" if candidate else "node"
+    candidate = candidate[:90]
+    unique = candidate
+    suffix = 2
+    while unique in used:
+        unique = f"{candidate}_{suffix}"
+        suffix += 1
+    used.add(unique)
+    return unique
+
+
+class MigrationBlocked(ValueError):
+    def __init__(self, report):
+        super().__init__("n8n workflow migration is blocked")
+        self.report = report
+
+
+@dataclass(frozen=True)
+class ConvertedWorkflow:
+    workflow_spec: dict[str, Any]
+    schedule_plan: dict[str, Any]
+    definition_hash: str
+    source_workflow_id: str
+
+
+class N8nWorkflowConverter:
+    _TRIGGER_TYPES = {
+        "n8n-nodes-base.manualTrigger": "manual",
+        "n8n-nodes-base.cron": "cron",
+        "n8n-nodes-base.scheduleTrigger": "cron",
+    }
+    _SUPPORTED_TASK_TYPES = {
+        "n8n-nodes-base.postgres",
+        "n8n-nodes-base.httpRequest",
+        "n8n-nodes-base.if",
+    }
+
+    @staticmethod
+    def _trigger(node):
+        kind = N8nWorkflowConverter._TRIGGER_TYPES[node["type"]]
+        if kind == "manual":
+            return {"type": "manual"}
+        expression = (
+            dict(node.get("parameters") or {}).get("cronExpression")
+            or dict(node.get("parameters") or {}).get("expression")
+        )
+        if not isinstance(expression, str) or len(expression.split()) not in {
+            5,
+            6,
+            7,
+        }:
+            raise ValueError("cron trigger requires an explicit cron expression")
+        return {"type": "cron", "expression": expression}
+
+    @staticmethod
+    def _postgres_node(node, node_id, data_source_uid):
+        parameters = dict(node.get("parameters") or {})
+        operation = parameters.get("operation")
+        if operation != "executeQuery":
+            raise ValueError("postgres operation is not safely convertible")
+        statement = _required_string(parameters.get("query"), "postgres query", 20000)
+        purpose = parameters.get("purpose")
+        if purpose != "read" or not re.match(r"(?is)^\s*select\b", statement):
+            raise ValueError(
+                "automatic n8n conversion supports only explicit read queries"
+            )
+        return {
+            "id": node_id,
+            "type": "sql.query",
+            "data_source_uid": data_source_uid,
+            "purpose": "read",
+            "config": {"statement": statement, "parameters": {}},
+        }
+
+    @staticmethod
+    def _http_node(node, node_id):
+        parameters = dict(node.get("parameters") or {})
+        method = str(parameters.get("method") or "GET").upper()
+        if method != "GET":
+            raise ValueError("automatic HTTP conversion supports only GET")
+        return {
+            "id": node_id,
+            "type": "http",
+            "purpose": "read",
+            "config": {
+                "method": "GET",
+                "url": _required_string(parameters.get("url"), "HTTP URL", 2000),
+            },
+        }
+
+    @staticmethod
+    def _condition_node(node, node_id):
+        parameters = dict(node.get("parameters") or {})
+        return {
+            "id": node_id,
+            "type": "condition",
+            "config": {"conditions": copy.deepcopy(parameters.get("conditions") or {})},
+        }
+
+    def convert(self, workflow, *, dataflow_uid, data_source_uids):
+        if not isinstance(workflow, dict):
+            raise ValueError("n8n workflow must be an object")
+        workflow_id = _required_string(workflow.get("id"), "n8n workflow id", 255)
+        workflow_name = _required_string(
+            workflow.get("name"), "n8n workflow name", 200
+        )
+        nodes = workflow.get("nodes")
+        if not isinstance(nodes, list) or not nodes:
+            raise ValueError("n8n workflow nodes must be a non-empty array")
+        if not isinstance(data_source_uids, dict):
+            raise ValueError("data source mapping must be an object")
+
+        blockers = []
+        triggers = []
+        used_ids: set[str] = set()
+        task_by_name = {}
+        task_nodes = []
+        trigger_names = set()
+        for node in nodes:
+            name = _required_string(node.get("name"), "n8n node name", 200)
+            node_type = _required_string(node.get("type"), "n8n node type", 300)
+            if node_type in self._TRIGGER_TYPES:
+                trigger_names.add(name)
+                try:
+                    triggers.append(self._trigger(node))
+                except ValueError:
+                    blockers.append(
+                        {
+                            "node_name": name,
+                            "node_type": node_type,
+                            "reason_code": "unsupported_trigger_configuration",
+                        }
+                    )
+                continue
+            if node_type not in self._SUPPORTED_TASK_TYPES:
+                blockers.append(
+                    {
+                        "node_name": name,
+                        "node_type": node_type,
+                        "reason_code": "unsupported_node_type",
+                    }
+                )
+                continue
+            node_id = _node_id(name, used_ids)
+            try:
+                if node_type == "n8n-nodes-base.postgres":
+                    source_uid = data_source_uids.get(name)
+                    if not source_uid:
+                        raise ValueError("postgres node data source mapping is required")
+                    converted = self._postgres_node(node, node_id, source_uid)
+                elif node_type == "n8n-nodes-base.httpRequest":
+                    converted = self._http_node(node, node_id)
+                else:
+                    converted = self._condition_node(node, node_id)
+            except ValueError:
+                blockers.append(
+                    {
+                        "node_name": name,
+                        "node_type": node_type,
+                        "reason_code": "unsupported_node_configuration",
+                    }
+                )
+                continue
+            task_by_name[name] = node_id
+            task_nodes.append(converted)
+
+        if blockers:
+            raise MigrationBlocked(
+                {
+                    "status": "blocked",
+                    "workflow_id": workflow_id,
+                    "workflow_name": workflow_name,
+                    "blockers": blockers,
+                }
+            )
+        if not task_nodes:
+            raise ValueError("n8n workflow has no convertible task nodes")
+        if not triggers:
+            triggers = [{"type": "manual"}]
+
+        edges = []
+        connections = workflow.get("connections") or {}
+        if not isinstance(connections, dict):
+            raise ValueError("n8n workflow connections must be an object")
+        for source_name, outputs in connections.items():
+            if source_name in trigger_names:
+                continue
+            source_id = task_by_name.get(source_name)
+            if source_id is None:
+                continue
+            for group in dict(outputs or {}).values():
+                for branch in group or []:
+                    for target in branch or []:
+                        target_id = task_by_name.get(target.get("node"))
+                        if target_id:
+                            edges.append({"from": source_id, "to": target_id})
+
+        spec = validate_workflow_spec(
+            {
+                "schema_version": "1.0",
+                "dataflow_uid": dataflow_uid,
+                "name": workflow_name,
+                "nodes": task_nodes,
+                "edges": edges,
+                "parameters": {},
+                "labels": {
+                    "migration_source": "n8n",
+                    "n8n_workflow_id": workflow_id,
+                },
+            }
+        )
+        plan = validate_schedule_plan(
+            {
+                "schema_version": "1.0",
+                "timezone": "Asia/Shanghai",
+                "triggers": triggers,
+                "max_concurrency": 1,
+                "conflict_policy": "skip",
+                "timeout_seconds": 3600,
+                "retry": {"max_attempts": 1, "delay_seconds": 0},
+                "backfill": {"max_days": 1, "max_runs": 1},
+            }
+        )
+        return ConvertedWorkflow(
+            workflow_spec=spec,
+            schedule_plan=plan,
+            definition_hash=_hash({"workflow_spec": spec, "schedule_plan": plan}),
+            source_workflow_id=workflow_id,
+        )
+
+
+@dataclass(frozen=True)
+class ShadowIsolation:
+    strategy: str
+    target_prefix: str | None = None
+
+    @classmethod
+    def read_only(cls):
+        return cls("read_only")
+
+    @classmethod
+    def isolated_target(cls, target_prefix):
+        prefix = _required_string(target_prefix, "shadow target prefix", 500)
+        return cls("isolated_target", prefix)
+
+    def validate(self, workflow_spec):
+        workflow = validate_workflow_spec(workflow_spec)
+        writes = [
+            node
+            for node in workflow["nodes"]
+            if node.get("purpose") == "write" or node["type"] == "sql.execute"
+        ]
+        if not writes:
+            return {"strategy": "read_only", "write_nodes": 0}
+        if self.strategy != "isolated_target" or not self.target_prefix:
+            raise ValueError("write shadow requires an isolated shadow target")
+        for node in writes:
+            target = node["config"].get("shadow_target")
+            if (
+                not isinstance(target, str)
+                or not target.startswith(self.target_prefix)
+                or target == self.target_prefix
+            ):
+                raise ValueError("write shadow requires an isolated shadow target")
+        return {
+            "strategy": "isolated_target",
+            "target_prefix": self.target_prefix,
+            "write_nodes": len(writes),
+        }
+
+
+class DualRunCoordinator:
+    def __init__(self, *, n8n_engine, kestra_engine, store):
+        self.n8n_engine = n8n_engine
+        self.kestra_engine = kestra_engine
+        self.store = store
+
+    def execute_shadow(
+        self,
+        *,
+        dataflow_uid,
+        environment,
+        mode,
+        n8n_definition_id,
+        kestra_definition_id,
+        workflow_spec,
+        inputs,
+        correlation_id,
+        isolation,
+    ):
+        if DualRunMode(mode) is not DualRunMode.N8N_PRIMARY_KESTRA_SHADOW:
+            raise ValueError("shadow execution requires n8n primary Kestra shadow mode")
+        if environment not in {"development", "test", "production"}:
+            raise ValueError("unsupported environment")
+        if not isinstance(inputs, dict):
+            raise ValueError("dual-run inputs must be an object")
+        isolation_result = isolation.validate(workflow_spec)
+        snapshot = copy.deepcopy(inputs)
+        input_hash = _hash(snapshot)
+        primary = self.n8n_engine.execute(n8n_definition_id, copy.deepcopy(snapshot))
+        shadow = self.kestra_engine.execute(
+            kestra_definition_id,
+            copy.deepcopy(snapshot),
+        )
+        record = {
+            "dataflow_uid": dataflow_uid,
+            "environment": environment,
+            "mode": DualRunMode(mode).value,
+            "primary_engine": "n8n",
+            "shadow_engine": "kestra",
+            "primary_execution_id": primary.get("execution_id"),
+            "shadow_execution_id": shadow.get("execution_id"),
+            "primary_status": primary.get("status"),
+            "shadow_status": shadow.get("status"),
+            "input_snapshot_hash": input_hash,
+            "isolation": isolation_result,
+            "correlation_id": correlation_id,
+            "formal_engine_changed": False,
+        }
+        return self.store.record_dual_run(record)

app/core/orchestration/migration/reconciliation.py

@@ -0,0 +1,453 @@
+from __future__ import annotations
+
+import copy
+import fnmatch
+from dataclasses import dataclass
+from typing import Any
+
+_CORE_PATHS = {
+    "status",
+    "nodes.*.partitions.*.row_count",
+    "nodes.*.partitions.*.primary_keys_hash",
+    "nodes.*.partitions.*.content_hash",
+    "nodes.*.partitions.*.error_count",
+}
+_CORE_PATH_EXAMPLES = {
+    "status",
+    "nodes.example.partitions.example.row_count",
+    "nodes.example.partitions.example.primary_keys_hash",
+    "nodes.example.partitions.example.content_hash",
+    "nodes.example.partitions.example.error_count",
+}
+_ROOT_KEYS = {
+    "engine",
+    "execution_id",
+    "status",
+    "duration_seconds",
+    "resource_usage",
+    "nodes",
+}
+_PARTITION_KEYS = {
+    "row_count",
+    "primary_keys_hash",
+    "content_hash",
+    "error_count",
+    "duration_seconds",
+    "resource_usage",
+    "aggregates",
+}
+
+
+def _required_string(value: Any, label: str, maximum: int = 500) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{label} is required")
+    normalized = value.strip()
+    if len(normalized) > maximum:
+        raise ValueError(f"{label} exceeds {maximum} characters")
+    return normalized
+
+
+def _number(value: Any, label: str) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise ValueError(f"{label} must be numeric")
+    return float(value)
+
+
+def _numeric_map(value: Any, label: str, maximum: int = 100) -> dict[str, float]:
+    if not isinstance(value, dict) or len(value) > maximum:
+        raise ValueError(f"{label} must be a bounded object")
+    return {
+        _required_string(key, f"{label} key", 200): _number(
+            item,
+            f"{label}.{key}",
+        )
+        for key, item in value.items()
+    }
+
+
+@dataclass
+class ExecutionSnapshot:
+    engine: str
+    execution_id: str
+    status: str
+    duration_seconds: float
+    resource_usage: dict[str, float]
+    nodes: dict[str, dict[str, Any]]
+
+    @classmethod
+    def from_dict(cls, value):
+        if not isinstance(value, dict):
+            raise ValueError("execution snapshot must be an object")
+        unknown = sorted(set(value) - _ROOT_KEYS)
+        if unknown:
+            raise ValueError(
+                f"execution snapshot contains unsupported fields: {', '.join(unknown)}"
+            )
+        engine = _required_string(value.get("engine"), "snapshot engine", 20)
+        if engine not in {"n8n", "kestra"}:
+            raise ValueError("unsupported snapshot engine")
+        status = _required_string(value.get("status"), "snapshot status", 30)
+        if status not in {
+            "success",
+            "failed",
+            "cancelled",
+            "killed",
+            "unknown",
+        }:
+            raise ValueError("unsupported snapshot status")
+        raw_nodes = value.get("nodes")
+        if not isinstance(raw_nodes, dict) or len(raw_nodes) > 100:
+            raise ValueError("snapshot nodes must be a bounded object")
+        nodes = {}
+        total_partitions = 0
+        for node_id, raw_node in raw_nodes.items():
+            normalized_node_id = _required_string(node_id, "snapshot node id", 100)
+            if not isinstance(raw_node, dict) or set(raw_node) != {"partitions"}:
+                raise ValueError("snapshot node must contain only partitions")
+            raw_partitions = raw_node["partitions"]
+            if not isinstance(raw_partitions, dict):
+                raise ValueError("snapshot partitions must be an object")
+            total_partitions += len(raw_partitions)
+            if total_partitions > 1000:
+                raise ValueError("snapshot partitions exceed limit")
+            partitions = {}
+            for partition, raw_partition in raw_partitions.items():
+                partition_id = _required_string(
+                    partition,
+                    "snapshot partition",
+                    300,
+                )
+                if not isinstance(raw_partition, dict):
+                    raise ValueError("snapshot partition metrics must be an object")
+                unknown_partition = sorted(set(raw_partition) - _PARTITION_KEYS)
+                if unknown_partition:
+                    raise ValueError(
+                        "snapshot partition contains unsupported fields: "
+                        + ", ".join(unknown_partition)
+                    )
+                if not {
+                    "row_count",
+                    "primary_keys_hash",
+                    "content_hash",
+                    "error_count",
+                }.issubset(raw_partition):
+                    raise ValueError("snapshot partition is missing core metrics")
+                row_count = raw_partition["row_count"]
+                error_count = raw_partition["error_count"]
+                if (
+                    isinstance(row_count, bool)
+                    or not isinstance(row_count, int)
+                    or row_count < 0
+                ):
+                    raise ValueError("row_count must be a non-negative integer")
+                if (
+                    isinstance(error_count, bool)
+                    or not isinstance(error_count, int)
+                    or error_count < 0
+                ):
+                    raise ValueError("error_count must be a non-negative integer")
+                partitions[partition_id] = {
+                    "row_count": row_count,
+                    "primary_keys_hash": _required_string(
+                        raw_partition["primary_keys_hash"],
+                        "primary_keys_hash",
+                        128,
+                    ),
+                    "content_hash": _required_string(
+                        raw_partition["content_hash"],
+                        "content_hash",
+                        128,
+                    ),
+                    "error_count": error_count,
+                    "duration_seconds": _number(
+                        raw_partition.get("duration_seconds", 0),
+                        "partition duration_seconds",
+                    ),
+                    "resource_usage": _numeric_map(
+                        raw_partition.get("resource_usage", {}),
+                        "partition resource_usage",
+                    ),
+                    "aggregates": _numeric_map(
+                        raw_partition.get("aggregates", {}),
+                        "partition aggregates",
+                    ),
+                }
+            nodes[normalized_node_id] = {"partitions": partitions}
+        return cls(
+            engine=engine,
+            execution_id=_required_string(
+                value.get("execution_id"),
+                "snapshot execution_id",
+                255,
+            ),
+            status=status,
+            duration_seconds=_number(
+                value.get("duration_seconds", 0),
+                "snapshot duration_seconds",
+            ),
+            resource_usage=_numeric_map(
+                value.get("resource_usage", {}),
+                "snapshot resource_usage",
+            ),
+            nodes=nodes,
+        )
+
+
+@dataclass(frozen=True, init=False)
+class ReconciliationPolicy:
+    ignored_paths: frozenset[str]
+    numeric_tolerances: dict[str, float]
+
+    def __init__(self, ignored_paths=None, numeric_tolerances=None):
+        ignored = frozenset(str(item).strip() for item in (ignored_paths or set()))
+        if "" in ignored or len(ignored) > 100:
+            raise ValueError("ignored paths must be a bounded set")
+        for path in ignored:
+            if (
+                path in _CORE_PATHS
+                or any(
+                    fnmatch.fnmatchcase(example, path)
+                    for example in _CORE_PATH_EXAMPLES
+                )
+                or any(
+                    fnmatch.fnmatchcase(path, core_pattern)
+                    for core_pattern in _CORE_PATHS
+                )
+            ):
+                raise ValueError("core reconciliation metric cannot be ignored")
+        tolerances = {}
+        for path, value in dict(numeric_tolerances or {}).items():
+            normalized_path = _required_string(
+                path,
+                "numeric tolerance path",
+                500,
+            )
+            tolerance = _number(value, f"numeric tolerance {normalized_path}")
+            if tolerance < 0:
+                raise ValueError("numeric tolerance cannot be negative")
+            tolerances[normalized_path] = tolerance
+        object.__setattr__(self, "ignored_paths", ignored)
+        object.__setattr__(self, "numeric_tolerances", tolerances)
+
+    def ignored(self, path):
+        return any(
+            fnmatch.fnmatchcase(path, pattern) for pattern in self.ignored_paths
+        )
+
+    def tolerance(self, path):
+        matches = [
+            value
+            for pattern, value in self.numeric_tolerances.items()
+            if fnmatch.fnmatchcase(path, pattern)
+        ]
+        return min(matches) if matches else 0.0
+
+    def as_dict(self):
+        return {
+            "ignored_paths": sorted(self.ignored_paths),
+            "numeric_tolerances": dict(sorted(self.numeric_tolerances.items())),
+        }
+
+
+class WorkflowReconciler:
+    @staticmethod
+    def _difference(
+        differences,
+        *,
+        node_id,
+        partition,
+        metric,
+        baseline,
+        candidate,
+        path,
+        policy,
+    ):
+        if policy.ignored(path):
+            return
+        if isinstance(baseline, (int, float)) and isinstance(
+            candidate,
+            (int, float),
+        ):
+            delta = float(candidate) - float(baseline)
+            tolerance = policy.tolerance(path)
+            if abs(delta) <= tolerance:
+                return
+            differences.append(
+                {
+                    "node_id": node_id,
+                    "partition": partition,
+                    "metric": metric,
+                    "baseline": baseline,
+                    "candidate": candidate,
+                    "delta": delta,
+                    "tolerance": tolerance,
+                }
+            )
+            return
+        if baseline != candidate:
+            differences.append(
+                {
+                    "node_id": node_id,
+                    "partition": partition,
+                    "metric": metric,
+                    "baseline": baseline,
+                    "candidate": candidate,
+                }
+            )
+
+    def compare(self, baseline, candidate, policy):
+        if not isinstance(baseline, ExecutionSnapshot) or not isinstance(
+            candidate,
+            ExecutionSnapshot,
+        ):
+            raise ValueError("reconciliation requires execution snapshots")
+        if not isinstance(policy, ReconciliationPolicy):
+            raise ValueError("reconciliation policy is required")
+        differences = []
+        self._difference(
+            differences,
+            node_id=None,
+            partition=None,
+            metric="status",
+            baseline=baseline.status,
+            candidate=candidate.status,
+            path="status",
+            policy=policy,
+        )
+        self._difference(
+            differences,
+            node_id=None,
+            partition=None,
+            metric="duration_seconds",
+            baseline=baseline.duration_seconds,
+            candidate=candidate.duration_seconds,
+            path="duration_seconds",
+            policy=policy,
+        )
+        for key in sorted(
+            set(baseline.resource_usage) | set(candidate.resource_usage)
+        ):
+            self._difference(
+                differences,
+                node_id=None,
+                partition=None,
+                metric=f"resource_usage.{key}",
+                baseline=baseline.resource_usage.get(key),
+                candidate=candidate.resource_usage.get(key),
+                path=f"resource_usage.{key}",
+                policy=policy,
+            )
+
+        compared_partitions = 0
+        baseline_rows = 0
+        candidate_rows = 0
+        node_ids = sorted(set(baseline.nodes) | set(candidate.nodes))
+        for node_id in node_ids:
+            baseline_partitions = baseline.nodes.get(node_id, {}).get(
+                "partitions",
+                {},
+            )
+            candidate_partitions = candidate.nodes.get(node_id, {}).get(
+                "partitions",
+                {},
+            )
+            partitions = sorted(
+                set(baseline_partitions) | set(candidate_partitions)
+            )
+            for partition in partitions:
+                before = baseline_partitions.get(partition)
+                after = candidate_partitions.get(partition)
+                if before is not None:
+                    baseline_rows += before["row_count"]
+                if after is not None:
+                    candidate_rows += after["row_count"]
+                if before is None or after is None:
+                    differences.append(
+                        {
+                            "node_id": node_id,
+                            "partition": partition,
+                            "metric": "partition_presence",
+                            "baseline": before is not None,
+                            "candidate": after is not None,
+                        }
+                    )
+                    continue
+                compared_partitions += 1
+                for metric in (
+                    "row_count",
+                    "primary_keys_hash",
+                    "content_hash",
+                    "error_count",
+                ):
+                    path = f"nodes.{node_id}.partitions.{partition}.{metric}"
+                    self._difference(
+                        differences,
+                        node_id=node_id,
+                        partition=partition,
+                        metric=metric,
+                        baseline=before[metric],
+                        candidate=after[metric],
+                        path=path,
+                        policy=policy,
+                    )
+                duration_path = (
+                    f"nodes.{node_id}.partitions.{partition}."
+                    "duration_seconds"
+                )
+                self._difference(
+                    differences,
+                    node_id=node_id,
+                    partition=partition,
+                    metric="duration_seconds",
+                    baseline=before["duration_seconds"],
+                    candidate=after["duration_seconds"],
+                    path=duration_path,
+                    policy=policy,
+                )
+                for group in ("aggregates", "resource_usage"):
+                    keys = sorted(set(before[group]) | set(after[group]))
+                    for key in keys:
+                        path = (
+                            f"nodes.{node_id}.partitions.{partition}."
+                            f"{group}.{key}"
+                        )
+                        self._difference(
+                            differences,
+                            node_id=node_id,
+                            partition=partition,
+                            metric=f"{group}.{key}",
+                            baseline=before[group].get(key),
+                            candidate=after[group].get(key),
+                            path=path,
+                            policy=policy,
+                        )
+
+        equivalent = not differences
+        return {
+            "status": "passed" if equivalent else "failed",
+            "equivalent": equivalent,
+            "baseline_execution_id": baseline.execution_id,
+            "candidate_execution_id": candidate.execution_id,
+            "baseline_engine": baseline.engine,
+            "candidate_engine": candidate.engine,
+            "difference_count": len(differences),
+            "differences": copy.deepcopy(differences),
+            "metrics": {
+                "compared_partitions": compared_partitions,
+                "baseline_row_count": baseline_rows,
+                "candidate_row_count": candidate_rows,
+                "row_count_delta": candidate_rows - baseline_rows,
+                "duration_delta_seconds": (
+                    candidate.duration_seconds - baseline.duration_seconds
+                ),
+                "resource_usage": {
+                    key: candidate.resource_usage.get(key, 0)
+                    - baseline.resource_usage.get(key, 0)
+                    for key in sorted(
+                        set(baseline.resource_usage)
+                        | set(candidate.resource_usage)
+                    )
+                },
+            },
+            "policy": policy.as_dict(),
+        }

app/core/orchestration/migration/repository.py

@@ -0,0 +1,549 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from sqlalchemy import text
+
+from app.core.common.identifiers import ensure_governance_uid, new_governance_uid
+from app.core.events.outbox import enqueue_outbox
+from app.core.orchestration.migration.dual_run import DualRunMode
+
+ENVIRONMENTS = {"development", "test", "production"}
+
+
+def _json(value: Any) -> str:
+    return json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    )
+
+
+def _required_string(value: Any, label: str, maximum: int = 500) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{label} is required")
+    normalized = value.strip()
+    if len(normalized) > maximum:
+        raise ValueError(f"{label} exceeds {maximum} characters")
+    return normalized
+
+
+def _uid(value: Any, label: str) -> str:
+    try:
+        return ensure_governance_uid({"uid": str(value)})
+    except ValueError as exc:
+        raise ValueError(f"{label} must be a valid UUIDv7") from exc
+
+
+def _state_row(row) -> dict[str, Any]:
+    return {
+        "state_id": row[0],
+        "dataflow_uid": row[1],
+        "environment": row[2],
+        "mode": row[3],
+        "status": row[4],
+        "n8n_definition_id": row[5],
+        "kestra_definition_id": row[6],
+        "migration_batch": row[7],
+        "observation_until": row[8],
+        "metadata": dict(row[9] or {}),
+    }
+
+
+def _operation_row(row, state) -> dict[str, Any]:
+    return {
+        "operation_id": row[0],
+        "dataflow_uid": state["dataflow_uid"],
+        "environment": state["environment"],
+        "source_mode": row[1],
+        "target_mode": row[2],
+        "n8n_definition_id": state["n8n_definition_id"],
+        "kestra_definition_id": state["kestra_definition_id"],
+        "gates": dict(row[3] or {}),
+        "idempotency_key": row[4],
+        "correlation_id": row[5],
+        "status": row[6],
+        "result": dict(row[7] or {}),
+    }
+
+
+class PostgresMigrationStore:
+    def __init__(self, engine):
+        self.engine = engine
+
+    def create_state(self, record):
+        dataflow_uid = _uid(record.get("dataflow_uid"), "dataflow_uid")
+        environment = _required_string(
+            record.get("environment"), "environment", 20
+        )
+        if environment not in ENVIRONMENTS:
+            raise ValueError("unsupported environment")
+        mode = DualRunMode(record.get("mode")).value
+        n8n_id = record.get("n8n_definition_id")
+        kestra_id = record.get("kestra_definition_id")
+        if mode != DualRunMode.KESTRA_PRIMARY.value:
+            n8n_id = _required_string(n8n_id, "n8n_definition_id", 255)
+        if mode != DualRunMode.N8N_PRIMARY.value:
+            kestra_id = _required_string(kestra_id, "kestra_definition_id", 255)
+        migration_batch = record.get("migration_batch")
+        if migration_batch is not None:
+            migration_batch = _required_string(
+                migration_batch, "migration_batch", 50
+            )
+        metadata = record.get("metadata") or {}
+        if not isinstance(metadata, dict):
+            raise ValueError("metadata must be an object")
+
+        state_id = new_governance_uid()
+        with self.engine.begin() as connection:
+            connection.execute(
+                text("SELECT pg_advisory_xact_lock(hashtext(:key))"),
+                {"key": f"{dataflow_uid}:{environment}:migration"},
+            )
+            connection.execute(
+                text(
+                    "INSERT INTO public.workflow_migration_states "
+                    "(id, dataflow_uid, environment, mode, status, "
+                    "n8n_definition_id, kestra_definition_id, "
+                    "migration_batch, metadata) "
+                    "VALUES (CAST(:id AS uuid), CAST(:dataflow_uid AS uuid), "
+                    ":environment, :mode, 'stable', :n8n_id, :kestra_id, "
+                    ":migration_batch, CAST(:metadata AS jsonb))"
+                ),
+                {
+                    "id": state_id,
+                    "dataflow_uid": dataflow_uid,
+                    "environment": environment,
+                    "mode": mode,
+                    "n8n_id": n8n_id,
+                    "kestra_id": kestra_id,
+                    "migration_batch": migration_batch,
+                    "metadata": _json(metadata),
+                },
+            )
+        return self.get_state(dataflow_uid, environment)
+
+    def get_state(self, dataflow_uid, environment):
+        dataflow_uid = _uid(dataflow_uid, "dataflow_uid")
+        environment = _required_string(environment, "environment", 20)
+        with self.engine.connect() as connection:
+            row = connection.execute(
+                text(
+                    "SELECT id::text, dataflow_uid::text, environment, mode, "
+                    "status, n8n_definition_id, kestra_definition_id, "
+                    "migration_batch, observation_until, metadata "
+                    "FROM public.workflow_migration_states "
+                    "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid) "
+                    "AND environment = :environment"
+                ),
+                {
+                    "dataflow_uid": dataflow_uid,
+                    "environment": environment,
+                },
+            ).one_or_none()
+        return _state_row(row) if row else None
+
+    @staticmethod
+    def _load_state(connection, dataflow_uid, environment):
+        row = connection.execute(
+            text(
+                "SELECT id::text, dataflow_uid::text, environment, mode, "
+                "status, n8n_definition_id, kestra_definition_id, "
+                "migration_batch, observation_until, metadata "
+                "FROM public.workflow_migration_states "
+                "WHERE dataflow_uid = CAST(:dataflow_uid AS uuid) "
+                "AND environment = :environment FOR UPDATE"
+            ),
+            {
+                "dataflow_uid": dataflow_uid,
+                "environment": environment,
+            },
+        ).one_or_none()
+        if row is None:
+            raise ValueError("workflow migration state was not found")
+        return _state_row(row)
+
+    @staticmethod
+    def _find_operation(connection, state, idempotency_key):
+        row = connection.execute(
+            text(
+                "SELECT id::text, source_mode, target_mode, gates, "
+                "idempotency_key, correlation_id::text, status, result "
+                "FROM public.workflow_cutover_operations "
+                "WHERE migration_state_id = CAST(:state_id AS uuid) "
+                "AND idempotency_key = :idempotency_key"
+            ),
+            {
+                "state_id": state["state_id"],
+                "idempotency_key": idempotency_key,
+            },
+        ).one_or_none()
+        return _operation_row(row, state) if row else None
+
+    def request_transition(self, record):
+        dataflow_uid = _uid(record.get("dataflow_uid"), "dataflow_uid")
+        environment = _required_string(
+            record.get("environment"), "environment", 20
+        )
+        source = DualRunMode(record.get("source_mode")).value
+        target = DualRunMode(record.get("target_mode")).value
+        idempotency_key = _required_string(
+            record.get("idempotency_key"), "idempotency_key", 500
+        )
+        correlation_id = _uid(record.get("correlation_id"), "correlation_id")
+        gates = record.get("gates") or {}
+        if not isinstance(gates, dict):
+            raise ValueError("gates must be an object")
+
+        with self.engine.begin() as connection:
+            connection.execute(
+                text("SELECT pg_advisory_xact_lock(hashtext(:key))"),
+                {"key": f"{dataflow_uid}:{environment}:migration"},
+            )
+            state = self._load_state(connection, dataflow_uid, environment)
+            previous = self._find_operation(
+                connection, state, idempotency_key
+            )
+            if previous:
+                return False, previous
+            if state["mode"] != source or state["status"] != "stable":
+                raise ValueError("workflow migration state changed")
+
+            operation_id = new_governance_uid()
+            connection.execute(
+                text(
+                    "INSERT INTO public.workflow_cutover_operations "
+                    "(id, migration_state_id, idempotency_key, source_mode, "
+                    "target_mode, status, gates, correlation_id) "
+                    "VALUES (CAST(:id AS uuid), CAST(:state_id AS uuid), "
+                    ":idempotency_key, :source_mode, :target_mode, 'claimed', "
+                    "CAST(:gates AS jsonb), CAST(:correlation_id AS uuid))"
+                ),
+                {
+                    "id": operation_id,
+                    "state_id": state["state_id"],
+                    "idempotency_key": idempotency_key,
+                    "source_mode": source,
+                    "target_mode": target,
+                    "gates": _json(gates),
+                    "correlation_id": correlation_id,
+                },
+            )
+            connection.execute(
+                text(
+                    "UPDATE public.workflow_migration_states "
+                    "SET status = 'transition_pending', "
+                    "updated_at = CURRENT_TIMESTAMP "
+                    "WHERE id = CAST(:state_id AS uuid)"
+                ),
+                {"state_id": state["state_id"]},
+            )
+            event_payload = {
+                "operation_id": operation_id,
+                "dataflow_uid": dataflow_uid,
+                "environment": environment,
+                "source_mode": source,
+                "target_mode": target,
+                "n8n_definition_id": state["n8n_definition_id"],
+                "kestra_definition_id": state["kestra_definition_id"],
+                "idempotency_key": idempotency_key,
+                "correlation_id": correlation_id,
+            }
+            enqueue_outbox(
+                connection,
+                aggregate_type="workflow_cutover",
+                aggregate_id=operation_id,
+                event_type="workflow.engine.cutover.requested",
+                payload=event_payload,
+                correlation_id=correlation_id,
+            )
+            operation = {
+                **event_payload,
+                "gates": gates,
+                "status": "claimed",
+            }
+        return True, operation
+
+    def _finish_transition(
+        self, operation_id, mode, status, result, *, expected_mode
+    ):
+        operation_id = _uid(operation_id, "operation_id")
+        mode = DualRunMode(mode).value
+        if not isinstance(result, dict):
+            raise ValueError("transition result must be an object")
+        with self.engine.begin() as connection:
+            row = connection.execute(
+                text(
+                    "SELECT migration_state_id::text "
+                    "FROM public.workflow_cutover_operations "
+                    "WHERE id = CAST(:id AS uuid) AND status = 'claimed' "
+                    "FOR UPDATE"
+                ),
+                {"id": operation_id},
+            ).one_or_none()
+            if row is None:
+                raise ValueError("cutover operation is not in claimed state")
+            state_id = row[0]
+            definitions = connection.execute(
+                text(
+                    "SELECT n8n_definition_id, kestra_definition_id "
+                    "FROM public.workflow_migration_states "
+                    "WHERE id = CAST(:state_id AS uuid)"
+                ),
+                {"state_id": state_id},
+            ).one()
+            connection.execute(
+                text(
+                    "UPDATE public.workflow_cutover_operations "
+                    "SET status = :status, result = CAST(:result AS jsonb), "
+                    "updated_at = CURRENT_TIMESTAMP "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {
+                    "id": operation_id,
+                    "status": status,
+                    "result": _json(result),
+                },
+            )
+            updated = connection.execute(
+                text(
+                    "UPDATE public.workflow_migration_states "
+                    "SET mode = :mode, status = 'stable', "
+                    "updated_at = CURRENT_TIMESTAMP "
+                    "WHERE id = CAST(:state_id AS uuid) "
+                    "AND mode = :expected_mode "
+                    "AND status = 'transition_pending'"
+                ),
+                {
+                    "state_id": state_id,
+                    "mode": mode,
+                    "expected_mode": expected_mode,
+                },
+            )
+            if updated.rowcount != 1:
+                raise ValueError("workflow migration state changed")
+            self._sync_engine_bindings(
+                connection,
+                mode,
+                n8n_definition_id=definitions[0],
+                kestra_definition_id=definitions[1],
+            )
+
+    @staticmethod
+    def _sync_engine_bindings(
+        connection,
+        mode,
+        *,
+        n8n_definition_id,
+        kestra_definition_id,
+    ):
+        roles = {
+            DualRunMode.N8N_PRIMARY.value: {
+                "n8n": ("primary", "enabled"),
+                "kestra": ("standby", "disabled"),
+            },
+            DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value: {
+                "n8n": ("primary", "enabled"),
+                "kestra": ("shadow", "disabled"),
+            },
+            DualRunMode.KESTRA_PRIMARY_N8N_STANDBY.value: {
+                "n8n": ("standby", "disabled"),
+                "kestra": ("primary", "enabled"),
+            },
+            DualRunMode.KESTRA_PRIMARY.value: {
+                "n8n": ("archived", "disabled"),
+                "kestra": ("primary", "enabled"),
+            },
+        }[mode]
+        definitions = {
+            "n8n": n8n_definition_id,
+            "kestra": kestra_definition_id,
+        }
+        engine_order = (
+            ("kestra", "n8n")
+            if mode
+            in {
+                DualRunMode.N8N_PRIMARY.value,
+                DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
+            }
+            else ("n8n", "kestra")
+        )
+        for engine_type in engine_order:
+            definition_id = definitions[engine_type]
+            if definition_id is None:
+                continue
+            role, binding_status = roles[engine_type]
+            connection.execute(
+                text(
+                    "UPDATE public.workflow_engine_bindings "
+                    "SET role = :role, status = :status, "
+                    "updated_at = CURRENT_TIMESTAMP "
+                    "WHERE engine_type = :engine_type "
+                    "AND engine_definition_id = :definition_id"
+                ),
+                {
+                    "role": role,
+                    "status": binding_status,
+                    "engine_type": engine_type,
+                    "definition_id": definition_id,
+                },
+            )
+
+    def complete_transition(self, operation_id, target_mode, result):
+        with self.engine.connect() as connection:
+            source = connection.execute(
+                text(
+                    "SELECT source_mode "
+                    "FROM public.workflow_cutover_operations "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": _uid(operation_id, "operation_id")},
+            ).scalar_one_or_none()
+        if source is None:
+            raise ValueError("cutover operation was not found")
+        self._finish_transition(
+            operation_id,
+            target_mode,
+            "succeeded",
+            result,
+            expected_mode=source,
+        )
+
+    def rollback_transition(self, operation_id, source_mode, result):
+        self._finish_transition(
+            operation_id,
+            source_mode,
+            "rolled_back",
+            result,
+            expected_mode=DualRunMode(source_mode).value,
+        )
+
+    def record_dual_run(self, record):
+        state_id = record.get("state_id")
+        if state_id is None:
+            state = self.get_state(
+                record.get("dataflow_uid"), record.get("environment")
+            )
+            if state is None:
+                raise ValueError("workflow migration state was not found")
+            state_id = state["state_id"]
+        state_id = _uid(state_id, "state_id")
+        isolation = record.get("isolation") or {}
+        isolation_mode = (
+            isolation.get("strategy")
+            if isinstance(isolation, dict)
+            else record.get("isolation_mode")
+        )
+        dual_run_id = new_governance_uid()
+        with self.engine.begin() as connection:
+            connection.execute(
+                text(
+                    "INSERT INTO public.workflow_dual_runs "
+                    "(id, migration_state_id, input_hash, n8n_execution_id, "
+                    "kestra_execution_id, n8n_status, kestra_status, "
+                    "isolation_mode, formal_engine_changed, correlation_id, "
+                    "safe_metrics) VALUES (CAST(:id AS uuid), "
+                    "CAST(:state_id AS uuid), :input_hash, :n8n_execution_id, "
+                    ":kestra_execution_id, :n8n_status, :kestra_status, "
+                    ":isolation_mode, :formal_engine_changed, "
+                    "CAST(:correlation_id AS uuid), CAST(:safe_metrics AS jsonb))"
+                ),
+                {
+                    "id": dual_run_id,
+                    "state_id": state_id,
+                    "input_hash": _required_string(
+                        record.get("input_hash")
+                        or record.get("input_snapshot_hash"),
+                        "input_hash",
+                        64,
+                    ),
+                    "n8n_execution_id": record.get("n8n_execution_id")
+                    or record.get("primary_execution_id"),
+                    "kestra_execution_id": record.get("kestra_execution_id")
+                    or record.get("shadow_execution_id"),
+                    "n8n_status": _required_string(
+                        record.get("n8n_status")
+                        or record.get("primary_status"),
+                        "n8n_status",
+                        30,
+                    ),
+                    "kestra_status": _required_string(
+                        record.get("kestra_status")
+                        or record.get("shadow_status"),
+                        "kestra_status",
+                        30,
+                    ),
+                    "isolation_mode": isolation_mode,
+                    "formal_engine_changed": bool(
+                        record.get("formal_engine_changed", False)
+                    ),
+                    "correlation_id": _uid(
+                        record.get("correlation_id"), "correlation_id"
+                    ),
+                    "safe_metrics": _json(record.get("safe_metrics") or {}),
+                },
+            )
+        return dual_run_id
+
+    def record_reconciliation(self, record):
+        state_id = _uid(record.get("state_id"), "state_id")
+        report_id = new_governance_uid()
+        with self.engine.begin() as connection:
+            connection.execute(
+                text(
+                    "INSERT INTO public.workflow_reconciliation_reports "
+                    "(id, migration_state_id, dual_run_id, status, policy, "
+                    "safe_metrics, differences, correlation_id) VALUES ("
+                    "CAST(:id AS uuid), CAST(:state_id AS uuid), "
+                    "CAST(:dual_run_id AS uuid), :status, "
+                    "CAST(:policy AS jsonb), CAST(:safe_metrics AS jsonb), "
+                    "CAST(:differences AS jsonb), "
+                    "CAST(:correlation_id AS uuid))"
+                ),
+                {
+                    "id": report_id,
+                    "state_id": state_id,
+                    "dual_run_id": record.get("dual_run_id"),
+                    "status": record.get("status"),
+                    "policy": _json(record.get("policy") or {}),
+                    "safe_metrics": _json(record.get("safe_metrics") or {}),
+                    "differences": _json(record.get("differences") or []),
+                    "correlation_id": _uid(
+                        record.get("correlation_id"), "correlation_id"
+                    ),
+                },
+            )
+        return report_id
+
+    def delete_test_state(self, state_id):
+        state_id = _uid(state_id, "state_id")
+        with self.engine.begin() as connection:
+            operation_ids = [
+                row[0]
+                for row in connection.execute(
+                    text(
+                        "SELECT id::text "
+                        "FROM public.workflow_cutover_operations "
+                        "WHERE migration_state_id = CAST(:state_id AS uuid)"
+                    ),
+                    {"state_id": state_id},
+                )
+            ]
+            if operation_ids:
+                connection.execute(
+                    text(
+                        "DELETE FROM public.outbox_events "
+                        "WHERE aggregate_type = 'workflow_cutover' "
+                        "AND aggregate_id = ANY(:operation_ids)"
+                    ),
+                    {"operation_ids": operation_ids},
+                )
+            connection.execute(
+                text(
+                    "DELETE FROM public.workflow_migration_states "
+                    "WHERE id = CAST(:state_id AS uuid)"
+                ),
+                {"state_id": state_id},
+            )

app/core/orchestration/migration/retirement.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+
+
+@dataclass(frozen=True)
+class N8nRetirementChecklist:
+    inventory_disposition_complete: bool
+    all_dataflows_kestra_primary: bool
+    business_cycle_and_recovery_complete: bool
+    standby_observation_no_rollback: bool
+    monitoring_and_alerting_complete: bool
+    dependency_failures_safe: bool
+    secure_archive_complete: bool
+    independent_review_approved: bool
+    final_l4_passed: bool
+
+
+class N8nRetirementGate:
+    @staticmethod
+    def evaluate(checklist: N8nRetirementChecklist):
+        if not isinstance(checklist, N8nRetirementChecklist):
+            raise ValueError("n8n retirement checklist is required")
+        evidence = asdict(checklist)
+        missing = [
+            name for name, passed in evidence.items() if passed is not True
+        ]
+        return {
+            "status": "blocked" if missing else "approved",
+            "retirement_approved": not missing,
+            "missing": missing,
+        }

app/core/orchestration/migration/rollout.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from enum import StrEnum
+
+
+class MigrationBatch(StrEnum):
+    READ_ONLY_LOW_FREQUENCY = "read_only_low_frequency"
+    IDEMPOTENT_PARTITION_WRITE = "idempotent_partition_write"
+    CORE_MULTI_NODE = "core_multi_node"
+    SPECIAL_HIGH_RISK = "special_high_risk"
+
+
+@dataclass(frozen=True)
+class RolloutEntryGates:
+    inventory_complete: bool
+    workflow_spec_valid: bool
+    policy_complete: bool
+    rollback_target_present: bool
+    shadow_runs: int
+    required_shadow_runs: int
+    failure_recovery_observed: bool
+    reconciliation_passed: bool
+    connection_budget_passed: bool
+    peak_concurrency_passed: bool
+    sla_passed: bool
+
+
+@dataclass(frozen=True)
+class ObservationWindow:
+    started_at: datetime
+    required_duration: timedelta
+    unexplained_differences: int
+    duplicate_writes: int
+    critical_sla_breaches: int
+    rollback_requests: int
+
+    def __post_init__(self):
+        if self.started_at.tzinfo is None:
+            raise ValueError("started_at must be timezone-aware")
+        if self.required_duration <= timedelta(0):
+            raise ValueError("required_duration must be positive")
+        for name in (
+            "unexplained_differences",
+            "duplicate_writes",
+            "critical_sla_breaches",
+            "rollback_requests",
+        ):
+            value = getattr(self, name)
+            if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+                raise ValueError(f"{name} must be a non-negative integer")
+
+
+@dataclass(frozen=True)
+class RolloutExitGates:
+    observation_window_complete: bool
+    no_unexplained_differences: bool
+    no_duplicate_writes: bool
+    sla_passed: bool
+    pause_retry_rollback_exercised: bool
+    n8n_standby: bool
+    n8n_unauthorized_changes: int
+
+
+class RolloutPolicy:
+    @staticmethod
+    def _raise_failed(label: str, checks: dict[str, bool]) -> None:
+        failed = [name for name, passed in checks.items() if passed is not True]
+        if failed:
+            raise ValueError(f"{label} failed: {', '.join(failed)}")
+
+    @classmethod
+    def assert_entry(cls, gates: RolloutEntryGates) -> None:
+        if not isinstance(gates, RolloutEntryGates):
+            raise ValueError("rollout entry gates are required")
+        runs_valid = (
+            isinstance(gates.shadow_runs, int)
+            and not isinstance(gates.shadow_runs, bool)
+            and isinstance(gates.required_shadow_runs, int)
+            and not isinstance(gates.required_shadow_runs, bool)
+            and gates.required_shadow_runs > 0
+            and gates.shadow_runs >= gates.required_shadow_runs
+        )
+        cls._raise_failed(
+            "rollout entry gates",
+            {
+                "inventory_complete": gates.inventory_complete,
+                "workflow_spec_valid": gates.workflow_spec_valid,
+                "policy_complete": gates.policy_complete,
+                "rollback_target_present": gates.rollback_target_present,
+                "shadow_runs": runs_valid,
+                "failure_recovery_observed": gates.failure_recovery_observed,
+                "reconciliation_passed": gates.reconciliation_passed,
+                "connection_budget_passed": gates.connection_budget_passed,
+                "peak_concurrency_passed": gates.peak_concurrency_passed,
+                "sla_passed": gates.sla_passed,
+            },
+        )
+
+    @classmethod
+    def assert_observation(
+        cls, window: ObservationWindow, *, now: datetime | None = None
+    ) -> None:
+        if not isinstance(window, ObservationWindow):
+            raise ValueError("observation window is required")
+        current = now or datetime.now(UTC)
+        if current.tzinfo is None:
+            raise ValueError("now must be timezone-aware")
+        cls._raise_failed(
+            "rollout observation",
+            {
+                "observation_window_complete": (
+                    current >= window.started_at + window.required_duration
+                ),
+                "unexplained_differences": (
+                    window.unexplained_differences == 0
+                ),
+                "duplicate_writes": window.duplicate_writes == 0,
+                "critical_sla_breaches": (
+                    window.critical_sla_breaches == 0
+                ),
+                "rollback_requests": window.rollback_requests == 0,
+            },
+        )
+
+    @classmethod
+    def assert_exit(cls, gates: RolloutExitGates) -> None:
+        if not isinstance(gates, RolloutExitGates):
+            raise ValueError("rollout exit gates are required")
+        cls._raise_failed(
+            "rollout exit gates",
+            {
+                "observation_window_complete": (
+                    gates.observation_window_complete
+                ),
+                "no_unexplained_differences": (
+                    gates.no_unexplained_differences
+                ),
+                "no_duplicate_writes": gates.no_duplicate_writes,
+                "sla_passed": gates.sla_passed,
+                "pause_retry_rollback_exercised": (
+                    gates.pause_retry_rollback_exercised
+                ),
+                "n8n_standby": gates.n8n_standby,
+                "n8n_unauthorized_changes": (
+                    isinstance(gates.n8n_unauthorized_changes, int)
+                    and not isinstance(gates.n8n_unauthorized_changes, bool)
+                    and gates.n8n_unauthorized_changes == 0
+                ),
+            },
+        )

docs/runbooks/kestra-dual-run-and-rollback.md

@@ -0,0 +1,92 @@
+# Kestra 双轨迁移与回切运行手册
+
+## 适用范围
+
+本手册用于把单个 DataFlow 从 n8n 正式执行逐步迁移到 Kestra。迁移是逐流程、
+逐环境进行的，不允许一次性改变全局正式引擎。
+
+四种受控状态如下：
+
+1. `n8n_primary`：n8n 是唯一正式引擎。
+2. `n8n_primary_kestra_shadow`：n8n 正式执行，Kestra 只读或写入隔离目标。
+3. `kestra_primary_n8n_standby`：Kestra 正式执行，n8n 保留为可回切目标。
+4. `kestra_primary`：仅在统一退役门禁独立批准后允许进入。
+
+状态只能沿相邻路径前进或回退，禁止从 `n8n_primary` 直接跳到
+`kestra_primary`。
+
+## 迁移前准备
+
+- 使用 `scripts/inventory_n8n_workflows.py` 取得目标环境的实时资产清单。空的本地
+  清单不能替代生产清单。
+- 导出一个目标 Workflow，并建立 n8n 节点到 DataOps 数据源 UID 的显式映射。
+  映射中不得包含密码、连接串或 API Key。
+- 使用 `app.commands.migrate_n8n_workflow` 生成 WorkflowSpec、SchedulePlan 和
+  阻断报告。命令只转换，不部署。
+- 未支持的社区节点、脚本节点和不能证明幂等性的写节点必须人工改造；转换器不会
+  生成任意脚本作为降级方案。
+- 为写流程声明影子 Schema、表或对象键前缀。只读流程使用 `read_only` 隔离策略。
+
+## 影子运行与对账
+
+同一次双跑必须使用不可变的同一输入快照，并记录 SHA-256 输入摘要。n8n 仍是
+唯一正式写入引擎，Kestra 只能：
+
+- 只读执行；或
+- 写入预先声明的隔离目标。
+
+每次执行用 `app.commands.reconcile_dual_run` 对账以下指标：
+
+- 状态、行数、主键摘要、内容摘要和异常数；
+- 预先声明的聚合值；
+- 执行耗时和资源消耗。
+
+对账文件只包含有界指标和摘要，不保存业务原始行。状态、行数、主键摘要、内容
+摘要和异常数不能配置为忽略项。非确定性时间戳或资源波动只能通过显式路径和数值
+容差处理。
+
+## 批次顺序和门禁
+
+迁移顺序固定为：
+
+1. 只读、低频、无下游依赖；
+2. 幂等分区写；
+3. 多节点核心流程；
+4. 高风险写、长任务和特殊节点。
+
+进入一个批次前，资产清单、WorkflowSpec、策略、回滚目标、规定次数的影子运行、
+一次失败恢复、对账、连接预算、峰值并发和 SLA 必须全部通过。任一门禁缺失即
+失败关闭。
+
+## 切换
+
+1. 调用切换服务创建带幂等键的切换请求。
+2. 数据库在同一事务中写入 `workflow_cutover_operations`、把迁移状态标记为
+   `transition_pending`，并写入 `workflow.engine.cutover.requested` Outbox 事件。
+3. 消费者先停用目标 n8n Workflow，再启用 Kestra Flow。
+4. 成功后把流程状态提交为 `kestra_primary_n8n_standby`。
+5. Kestra 激活失败时立即重新激活 n8n，并把操作记录为 `rolled_back`。原始异常
+   和引擎响应不会进入审计，只记录安全原因码与有界状态。
+
+不得在数据库事务中发起 n8n 或 Kestra HTTP 请求。
+
+## 观察期
+
+观察期必须达到流程预先声明的时长，并且满足：
+
+- 没有未解释差异、重复写入或关键 SLA 违约；
+- 自动暂停、有限重试和回切至少演练一次；
+- n8n 保持 Standby，且没有未授权外部修改；
+- 没有实际回切请求。
+
+观察期未结束或出现任一事件时，不允许推进下一批或退役 n8n。
+
+## 显式回切
+
+1. 确认当前状态是 `kestra_primary_n8n_standby`。
+2. 使用新的幂等键提交回切。
+3. 系统先停用 Kestra，再重新激活 n8n。
+4. 数据库状态返回 `n8n_primary_kestra_shadow`。
+5. 检查 n8n 已激活、Kestra 已禁用、Outbox 无重复事件，并重新执行针对性对账。
+
+若引擎状态不明确，暂停该流程，不允许同时启用两个正式写入引擎。

docs/runbooks/n8n-archive-and-retirement.md

@@ -0,0 +1,53 @@
+# n8n 归档与退役运行手册
+
+## 原则
+
+n8n 退役是独立变更，不随单流程切换自动发生。只有所有统一退出门禁和独立评审
+同时通过后，才允许从 `kestra_primary_n8n_standby` 进入
+`kestra_primary`。门禁未通过时必须保留 n8n 容器、API、前端兼容入口和历史数据。
+
+## 必须归档的内容
+
+- 全部 Workflow 定义、标签、激活状态和定义摘要；
+- Workflow 到 DataFlow、WorkflowSpec、Kestra namespace/flow ID 的版本映射；
+- 执行历史、关键失败和回切证据；
+- Webhook、脚本、前端入口和外部调用方清单；
+- 凭据类型、引用名和版本清单。
+
+归档中不得保存明文密码、API Key、访问令牌或完整连接串。凭据只记录 DataOps
+凭据引用和版本；归档包需加密、限制访问并验证恢复。
+
+## 统一退出门禁
+
+- 所有生产 n8n Workflow 已迁移、归档或有正式保留说明；
+- 所有正式 DataFlow 已在 `kestra_primary`，不存在未处理双轨差异；
+- 关键流程经过完整业务周期和至少一次恢复/回切演练；
+- Standby 观察期没有实际回切；
+- Kestra、Runner、Gateway、Context MCP、智能体和对账指标都有监控告警；
+- 模型离线、Kestra/Runner 重启和单数据源故障均不破坏正式状态；
+- 安全归档和恢复验证完成；
+- 独立评审确认没有 Webhook、前端、脚本或外部依赖；
+- 唯一一次最终 L4 验证通过。
+
+任何一项为假或缺少证据，退役决定都是 `blocked`。
+
+## 退役顺序
+
+1. 禁止新建和激活 n8n Workflow。
+2. 停止 n8n 调度，保留只读管理和归档访问。
+3. 完成 Standby 观察和归档恢复演练。
+4. 执行独立下线评审及唯一一次最终 L4。
+5. 单独变更 Docker、后端兼容 API、前端入口和部署文档。
+6. 最后移除运行容器；历史表和 `n8n_workflow_id` 字段通过后续独立迁移处理。
+
+禁止在同一变更中同时退役 n8n、重构 WorkflowSpec 和升级 Kestra。
+
+## V55 当前决定
+
+当前仓库只完成了双轨、对账、事务化切换、自动回切和本地受控演练。尚无生产
+Workflow 全量处置、完整业务周期、Standby 观察、完整依赖扫描、安全归档、独立
+评审和最终 L4 证据。因此当前退役状态是 `blocked`：
+
+- 不修改 `deploy/docker/docker-compose.yml` 移除 n8n；
+- 不删除 n8n API、前端入口、历史字段或执行历史；
+- V55 阶段性交付上限为 `kestra_primary_n8n_standby`。

docs/validation/kestra-v55-retirement-audit.json

@@ -0,0 +1,29 @@
+{
+  "schema_version": "1.0",
+  "evaluated_at": "2026-07-19",
+  "status": "blocked",
+  "retirement_approved": false,
+  "evidence": {
+    "inventory_disposition_complete": false,
+    "all_dataflows_kestra_primary": false,
+    "business_cycle_and_recovery_complete": false,
+    "standby_observation_no_rollback": false,
+    "monitoring_and_alerting_complete": false,
+    "dependency_failures_safe": false,
+    "secure_archive_complete": false,
+    "independent_review_approved": false,
+    "final_l4_passed": false
+  },
+  "missing": [
+    "inventory_disposition_complete",
+    "all_dataflows_kestra_primary",
+    "business_cycle_and_recovery_complete",
+    "standby_observation_no_rollback",
+    "monitoring_and_alerting_complete",
+    "dependency_failures_safe",
+    "secure_archive_complete",
+    "independent_review_approved",
+    "final_l4_passed"
+  ],
+  "decision": "Preserve n8n and stop at kestra_primary_n8n_standby."
+}

docs/validation/kestra-v55.md

@@ -0,0 +1,105 @@
+# Kestra V55 阶段验收记录
+
+- 代码分支：`codex/kestra-v55`
+- 基线提交：`945d680`
+- 验收日期：2026-07-19
+- Python：3.11.15
+- 结论：V55 双轨迁移、对账、事务化单流程切换、自动回切、批次门禁和退役审计
+  框架完成；本地可进入 `kestra_primary_n8n_standby`。n8n 统一退出门禁未满足，
+  不批准进入 `kestra_primary`，也不移除 n8n。
+
+## 完成范围
+
+### 双轨与转换
+
+- 实现四种迁移模式及相邻状态转换约束。
+- n8n 转换器只自动转换受控触发器、显式只读 SQL、GET HTTP 和条件节点；未知
+  节点生成阻断报告，不降级为任意脚本。
+- 数据源只通过 UID 映射，转换结果不复制凭据。
+- n8n 正式执行与 Kestra 影子执行共享同一不可变输入快照和摘要。
+- 影子写必须指向声明的隔离目标；否则失败关闭。
+
+### 对账、切换与持久化
+
+- 对账状态、行数、主键摘要、内容摘要、异常、聚合、耗时和资源消耗。
+- 对账策略允许显式忽略非确定性指标或声明数值容差，但核心正确性指标不可忽略。
+- 新增迁移状态、双跑、对账报告和切换操作四张 PostgreSQL 控制表。
+- 切换请求、状态更新和 Outbox 事件在同一事务提交；相同幂等键不会重复发事件。
+- 切换消费者先停 n8n 再启 Kestra；Kestra 激活失败自动恢复 n8n。
+- 显式回切先停 Kestra 再启 n8n。审计只保留有界状态和安全原因码。
+
+### 分批推广和退役保护
+
+- 固定四批顺序：只读低频、幂等分区写、核心多节点、特殊高风险。
+- 进入门禁覆盖资产、WorkflowSpec、策略、回滚目标、影子次数、失败恢复、对账、
+  连接预算、峰值并发和 SLA。
+- 退出门禁覆盖观察窗口、数据差异、重复写、SLA、暂停/重试/回切演练及 n8n
+  Standby 未被外部修改。
+- n8n 退役清单失败关闭，并要求独立评审及唯一一次最终 L4。
+
+## 验证结果
+
+### L1：V55 领域与命令
+
+```bash
+PYTHONPATH=. .venv/bin/python -m pytest -q \
+  tests/integration/test_n8n_kestra_dual_run.py \
+  tests/integration/test_workflow_reconciliation.py \
+  tests/integration/test_workflow_engine_cutover.py \
+  tests/integration/test_workflow_engine_rollback.py \
+  tests/integration/test_workflow_migration_rollout.py \
+  tests/integration/test_n8n_retirement_audit.py \
+  tests/test_v55_migration_commands.py
+```
+
+结果：`34 passed`。
+
+### L2：受影响边界与迁移
+
+- 最终合并执行 V55、V54 回归、受影响边界、真实持久化、真实切换回切和迁移测试：
+  `113 passed`。
+- 编排核心、Scheduling/Context MCP、安全边界、n8n 清单和客户端配置：
+  `48 passed`。
+- Alembic 全新临时库升级、重复升级和单版本回退：`3 passed`。
+- V54 Agent、MCP、真实 Kestra/Runner Canary 和模型离线回归：`40 passed`。
+- V55 引擎角色顺序、真实 PostgreSQL 状态/双跑/对账/Outbox：
+  `3 passed`。
+
+### L3：真实切换与回切
+
+使用本地 Docker 中运行的 PostgreSQL、n8n 和 Kestra 完成一个受控流程演练：
+
+1. n8n 已激活、Kestra Flow 已禁用；
+2. 先停用 n8n，再使用不存在的 Kestra Flow 模拟激活失败；
+3. 系统自动重新激活 n8n，操作落库为 `rolled_back`；
+4. 使用有效 Kestra Flow 再次切换，n8n 变为禁用、Kestra 变为启用；
+5. 显式回切先禁用 Kestra，再激活 n8n。
+
+结果：`1 passed`。演练后的临时 n8n Workflow、Kestra Flow、迁移状态和 Outbox
+事件均已清理；数据库处于 `20260719_90 (head)`。
+
+### 静态和安全检查
+
+- Ruff：通过。
+- V55 新增生产模块 Pyright：`0 errors, 0 warnings`。
+- `git diff --check`：通过。
+- 新增范围未发现 API Key、云访问密钥、私钥或本地测试密码硬编码。
+
+未执行全仓 Python、前端构建和完整栈 L4。V55 当前明确不退役 n8n，也没有修改
+前端、Docker Compose 或公共启动入口；按验证计划，只有正式移除 n8n 前才执行
+唯一一次 L4。
+
+## 未退役 n8n 的原因
+
+本地实时清单为零只能证明当前本地实例没有持久业务 Workflow，不能替代生产
+资产处置证据。当前也没有完整业务周期、Standby 观察期、生产依赖扫描、安全
+归档恢复、独立评审和最终 L4 证据。
+
+因此：
+
+- 不修改 Docker Compose 移除 n8n；
+- 不删除 n8n API、前端入口、历史执行或兼容字段；
+- 不宣称 n8n 已经下线；
+- 生产逐流程切换仍须为每个实际 Workflow 补齐真实双跑、对账和观察证据。
+
+详细缺口见 `docs/validation/kestra-v55-retirement-audit.json`。

migrations/versions/20260719_90_workflow_migration_control.py

@@ -0,0 +1,122 @@
+"""Add V55 dual-run, reconciliation, and cutover control records."""
+
+from alembic import op
+
+revision = "20260719_90"
+down_revision = "20260719_80"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE public.workflow_migration_states (
+            id UUID PRIMARY KEY,
+            dataflow_uid UUID NOT NULL,
+            environment VARCHAR(20) NOT NULL
+                CHECK (environment IN ('development', 'test', 'production')),
+            mode VARCHAR(50) NOT NULL CHECK (mode IN (
+                'n8n_primary',
+                'n8n_primary_kestra_shadow',
+                'kestra_primary_n8n_standby',
+                'kestra_primary'
+            )),
+            status VARCHAR(30) NOT NULL DEFAULT 'stable'
+                CHECK (status IN (
+                    'stable', 'transition_pending', 'blocked', 'retired'
+                )),
+            n8n_definition_id VARCHAR(255),
+            kestra_definition_id VARCHAR(255),
+            migration_batch VARCHAR(50),
+            observation_until TIMESTAMPTZ,
+            metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (dataflow_uid, environment)
+        );
+        CREATE INDEX idx_workflow_migration_mode
+            ON public.workflow_migration_states(
+                environment, mode, status, updated_at DESC
+            );
+
+        CREATE TABLE public.workflow_dual_runs (
+            id UUID PRIMARY KEY,
+            migration_state_id UUID NOT NULL
+                REFERENCES public.workflow_migration_states(id)
+                ON DELETE CASCADE,
+            input_hash CHAR(64) NOT NULL,
+            n8n_execution_id VARCHAR(255),
+            kestra_execution_id VARCHAR(255),
+            n8n_status VARCHAR(30) NOT NULL,
+            kestra_status VARCHAR(30) NOT NULL,
+            isolation_mode VARCHAR(30) NOT NULL
+                CHECK (isolation_mode IN ('read_only', 'isolated_target')),
+            formal_engine_changed BOOLEAN NOT NULL DEFAULT FALSE,
+            correlation_id UUID NOT NULL,
+            safe_metrics JSONB NOT NULL DEFAULT '{}'::jsonb,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE INDEX idx_workflow_dual_runs_state_created
+            ON public.workflow_dual_runs(
+                migration_state_id, created_at DESC
+            );
+
+        CREATE TABLE public.workflow_reconciliation_reports (
+            id UUID PRIMARY KEY,
+            migration_state_id UUID NOT NULL
+                REFERENCES public.workflow_migration_states(id)
+                ON DELETE CASCADE,
+            dual_run_id UUID
+                REFERENCES public.workflow_dual_runs(id)
+                ON DELETE SET NULL,
+            status VARCHAR(20) NOT NULL
+                CHECK (status IN ('passed', 'failed', 'blocked')),
+            policy JSONB NOT NULL DEFAULT '{}'::jsonb,
+            safe_metrics JSONB NOT NULL DEFAULT '{}'::jsonb,
+            differences JSONB NOT NULL DEFAULT '[]'::jsonb,
+            correlation_id UUID NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE INDEX idx_workflow_reconciliation_state_created
+            ON public.workflow_reconciliation_reports(
+                migration_state_id, status, created_at DESC
+            );
+
+        CREATE TABLE public.workflow_cutover_operations (
+            id UUID PRIMARY KEY,
+            migration_state_id UUID NOT NULL
+                REFERENCES public.workflow_migration_states(id)
+                ON DELETE CASCADE,
+            idempotency_key VARCHAR(500) NOT NULL,
+            source_mode VARCHAR(50) NOT NULL,
+            target_mode VARCHAR(50) NOT NULL,
+            status VARCHAR(30) NOT NULL
+                CHECK (status IN (
+                    'claimed', 'succeeded', 'rolled_back', 'failed'
+                )),
+            gates JSONB NOT NULL DEFAULT '{}'::jsonb,
+            result JSONB NOT NULL DEFAULT '{}'::jsonb,
+            safe_error VARCHAR(1000),
+            correlation_id UUID NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (migration_state_id, idempotency_key)
+        );
+        CREATE INDEX idx_workflow_cutover_state_created
+            ON public.workflow_cutover_operations(
+                migration_state_id, status, created_at DESC
+            );
+        """
+    )
+
+
+def downgrade() -> None:
+    op.execute(
+        """
+        DROP TABLE IF EXISTS public.workflow_cutover_operations;
+        DROP TABLE IF EXISTS public.workflow_reconciliation_reports;
+        DROP TABLE IF EXISTS public.workflow_dual_runs;
+        DROP TABLE IF EXISTS public.workflow_migration_states;
+        """
+    )

tests/core/orchestration/test_migration_repository.py

@@ -0,0 +1,54 @@
+from app.core.orchestration.migration.dual_run import DualRunMode
+from app.core.orchestration.migration.repository import PostgresMigrationStore
+
+
+class Connection:
+    def __init__(self):
+        self.parameters = []
+
+    def execute(self, _statement, parameters):
+        self.parameters.append(parameters)
+
+
+def test_cutover_disables_n8n_binding_before_enabling_kestra_primary():
+    connection = Connection()
+
+    PostgresMigrationStore._sync_engine_bindings(
+        connection,
+        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY.value,
+        n8n_definition_id="n8n-orders",
+        kestra_definition_id="kestra-orders",
+    )
+
+    assert connection.parameters == [
+        {
+            "role": "standby",
+            "status": "disabled",
+            "engine_type": "n8n",
+            "definition_id": "n8n-orders",
+        },
+        {
+            "role": "primary",
+            "status": "enabled",
+            "engine_type": "kestra",
+            "definition_id": "kestra-orders",
+        },
+    ]
+
+
+def test_rollback_disables_kestra_binding_before_restoring_n8n_primary():
+    connection = Connection()
+
+    PostgresMigrationStore._sync_engine_bindings(
+        connection,
+        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
+        n8n_definition_id="n8n-orders",
+        kestra_definition_id="kestra-orders",
+    )
+
+    assert [item["engine_type"] for item in connection.parameters] == [
+        "kestra",
+        "n8n",
+    ]
+    assert connection.parameters[0]["status"] == "disabled"
+    assert connection.parameters[1]["role"] == "primary"

tests/integration/test_n8n_kestra_dual_run.py

@@ -0,0 +1,209 @@
+import copy
+
+import pytest
+
+from app.core.orchestration.migration.dual_run import (
+    DualRunCoordinator,
+    DualRunMode,
+    MigrationBlocked,
+    N8nWorkflowConverter,
+    ShadowIsolation,
+    validate_mode_transition,
+)
+
+DATAFLOW_UID = "01900000-0000-7000-8000-000000000091"
+SOURCE_UID = "01900000-0000-7000-8000-000000000092"
+
+
+def n8n_read_workflow():
+    return {
+        "id": "n8n-read-orders",
+        "name": "Read orders",
+        "active": True,
+        "nodes": [
+            {
+                "id": "trigger-1",
+                "name": "Manual",
+                "type": "n8n-nodes-base.manualTrigger",
+                "parameters": {},
+            },
+            {
+                "id": "query-1",
+                "name": "Read Orders",
+                "type": "n8n-nodes-base.postgres",
+                "parameters": {
+                    "operation": "executeQuery",
+                    "query": "SELECT id, amount FROM orders ORDER BY id",
+                    "purpose": "read",
+                },
+            },
+        ],
+        "connections": {
+            "Manual": {
+                "main": [[{"node": "Read Orders", "type": "main", "index": 0}]]
+            }
+        },
+    }
+
+
+def test_dual_run_modes_allow_only_guarded_forward_and_rollback_paths():
+    assert validate_mode_transition(
+        DualRunMode.N8N_PRIMARY,
+        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
+    )
+    assert validate_mode_transition(
+        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
+        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+    )
+    assert validate_mode_transition(
+        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+        DualRunMode.KESTRA_PRIMARY,
+    )
+    assert validate_mode_transition(
+        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
+    )
+
+    with pytest.raises(ValueError, match="unsupported migration transition"):
+        validate_mode_transition(
+            DualRunMode.N8N_PRIMARY,
+            DualRunMode.KESTRA_PRIMARY,
+        )
+
+
+def test_converter_builds_read_only_workflow_spec_without_credentials():
+    converted = N8nWorkflowConverter().convert(
+        n8n_read_workflow(),
+        dataflow_uid=DATAFLOW_UID,
+        data_source_uids={"Read Orders": SOURCE_UID},
+    )
+
+    assert converted.workflow_spec["nodes"] == [
+        {
+            "id": "read_orders",
+            "type": "sql.query",
+            "data_source_uid": SOURCE_UID,
+            "purpose": "read",
+            "config": {
+                "statement": "SELECT id, amount FROM orders ORDER BY id",
+                "parameters": {},
+            },
+        }
+    ]
+    assert converted.workflow_spec["edges"] == []
+    assert converted.schedule_plan["triggers"] == [{"type": "manual"}]
+    assert converted.definition_hash
+    assert "credential" not in repr(converted).lower()
+
+
+def test_converter_blocks_community_nodes_with_actionable_report():
+    workflow = n8n_read_workflow()
+    workflow["nodes"].append(
+        {
+            "id": "community-1",
+            "name": "Community magic",
+            "type": "@vendor/n8n-nodes-secret-magic",
+            "parameters": {"operation": "run"},
+        }
+    )
+
+    with pytest.raises(MigrationBlocked) as error:
+        N8nWorkflowConverter().convert(
+            workflow,
+            dataflow_uid=DATAFLOW_UID,
+            data_source_uids={"Read Orders": SOURCE_UID},
+        )
+
+    assert error.value.report["status"] == "blocked"
+    assert error.value.report["workflow_id"] == "n8n-read-orders"
+    assert error.value.report["blockers"] == [
+        {
+            "node_name": "Community magic",
+            "node_type": "@vendor/n8n-nodes-secret-magic",
+            "reason_code": "unsupported_node_type",
+        }
+    ]
+    assert "script" not in repr(error.value.report).lower()
+
+
+class Engine:
+    def __init__(self, name):
+        self.name = name
+        self.calls = []
+
+    def execute(self, definition_id, inputs):
+        self.calls.append((definition_id, copy.deepcopy(inputs)))
+        return {
+            "execution_id": f"{self.name}-execution-1",
+            "status": "success",
+        }
+
+
+class Store:
+    def __init__(self):
+        self.records = []
+
+    def record_dual_run(self, record):
+        self.records.append(copy.deepcopy(record))
+        return copy.deepcopy(record)
+
+
+def test_shadow_execution_reuses_input_snapshot_and_never_changes_primary():
+    converted = N8nWorkflowConverter().convert(
+        n8n_read_workflow(),
+        dataflow_uid=DATAFLOW_UID,
+        data_source_uids={"Read Orders": SOURCE_UID},
+    )
+    n8n = Engine("n8n")
+    kestra = Engine("kestra")
+    store = Store()
+    coordinator = DualRunCoordinator(
+        n8n_engine=n8n,
+        kestra_engine=kestra,
+        store=store,
+    )
+
+    result = coordinator.execute_shadow(
+        dataflow_uid=DATAFLOW_UID,
+        environment="test",
+        mode=DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
+        n8n_definition_id="n8n-read-orders",
+        kestra_definition_id="kestra-read-orders",
+        workflow_spec=converted.workflow_spec,
+        inputs={"day": "2026-07-19"},
+        correlation_id="01900000-0000-7000-8000-000000000093",
+        isolation=ShadowIsolation.read_only(),
+    )
+
+    assert n8n.calls[0][1] == kestra.calls[0][1]
+    assert result["input_snapshot_hash"] == store.records[0]["input_snapshot_hash"]
+    assert result["primary_engine"] == "n8n"
+    assert result["shadow_engine"] == "kestra"
+    assert result["formal_engine_changed"] is False
+
+
+def test_shadow_write_requires_explicit_isolated_target():
+    write_spec = N8nWorkflowConverter().convert(
+        n8n_read_workflow(),
+        dataflow_uid=DATAFLOW_UID,
+        data_source_uids={"Read Orders": SOURCE_UID},
+    ).workflow_spec
+    write_spec["nodes"][0].update(
+        {
+            "type": "sql.execute",
+            "purpose": "write",
+            "idempotency": {
+                "strategy": "partition_replace",
+                "key": "orders:${parameters.day}",
+            },
+        }
+    )
+
+    with pytest.raises(ValueError, match="isolated shadow target"):
+        ShadowIsolation.read_only().validate(write_spec)
+
+    write_spec["nodes"][0]["config"]["shadow_target"] = (
+        "dataops_shadow.orders_20260719"
+    )
+    isolated = ShadowIsolation.isolated_target("dataops_shadow.")
+    assert isolated.validate(write_spec)["strategy"] == "isolated_target"

tests/integration/test_n8n_retirement_audit.py

@@ -0,0 +1,48 @@
+from app.core.orchestration.migration.retirement import (
+    N8nRetirementChecklist,
+    N8nRetirementGate,
+)
+
+
+def checklist(**overrides):
+    values = {
+        "inventory_disposition_complete": True,
+        "all_dataflows_kestra_primary": True,
+        "business_cycle_and_recovery_complete": True,
+        "standby_observation_no_rollback": True,
+        "monitoring_and_alerting_complete": True,
+        "dependency_failures_safe": True,
+        "secure_archive_complete": True,
+        "independent_review_approved": True,
+        "final_l4_passed": True,
+    }
+    values.update(overrides)
+    return N8nRetirementChecklist(**values)
+
+
+def test_retirement_is_blocked_with_explicit_missing_evidence():
+    decision = N8nRetirementGate.evaluate(
+        checklist(
+            inventory_disposition_complete=False,
+            independent_review_approved=False,
+            final_l4_passed=False,
+        )
+    )
+
+    assert decision["status"] == "blocked"
+    assert decision["retirement_approved"] is False
+    assert decision["missing"] == [
+        "inventory_disposition_complete",
+        "independent_review_approved",
+        "final_l4_passed",
+    ]
+
+
+def test_retirement_approval_requires_every_gate():
+    decision = N8nRetirementGate.evaluate(checklist())
+
+    assert decision == {
+        "status": "approved",
+        "retirement_approved": True,
+        "missing": [],
+    }

tests/integration/test_v55_real_cutover_rollback_l3.py

@@ -0,0 +1,195 @@
+"""Opt-in V55 PostgreSQL + n8n + Kestra cutover and rollback drill."""
+
+import os
+import uuid
+
+import pytest
+import requests
+from sqlalchemy import create_engine
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_factory.n8n_client import N8nClient
+from app.core.orchestration.engines.kestra import KestraAdapter
+from app.core.orchestration.engines.n8n import N8nAdapter
+from app.core.orchestration.migration.cutover import (
+    CutoverGates,
+    EngineRoleController,
+    WorkflowCutoverService,
+)
+from app.core.orchestration.migration.dual_run import DualRunMode
+from app.core.orchestration.migration.repository import PostgresMigrationStore
+
+pytestmark = pytest.mark.skipif(
+    os.getenv("RUN_V55_CUTOVER_L3") != "1",
+    reason="set RUN_V55_CUTOVER_L3=1 for the local cutover drill",
+)
+
+DATABASE_URL = os.getenv("DATAOPS_MCP_TEST_DATABASE_URL", "")
+KESTRA_URL = os.getenv(
+    "KESTRA_BASE_URL", "http://127.0.0.1:18080/api/v1"
+)
+KESTRA_AUTH = (
+    os.getenv("KESTRA_USERNAME", ""),
+    os.getenv("KESTRA_PASSWORD", ""),
+)
+KESTRA_NAMESPACE = "dataops.v55"
+
+
+def passing_gates():
+    return CutoverGates(
+        inventory_complete=True,
+        workflow_spec_valid=True,
+        rollback_target_present=True,
+        shadow_runs=3,
+        required_shadow_runs=3,
+        failure_recovery_observed=True,
+        reconciliation_passed=True,
+        connection_budget_passed=True,
+        sla_passed=True,
+        n8n_baseline_unchanged=True,
+    )
+
+
+def test_real_n8n_and_kestra_cutover_failure_restore_and_explicit_rollback():
+    engine = create_engine(DATABASE_URL, pool_pre_ping=True)
+    store = PostgresMigrationStore(engine)
+    n8n_client = N8nClient(
+        api_url="http://127.0.0.1:15678",
+        api_key=os.environ["N8N_API_KEY"],
+        timeout=20,
+    )
+    n8n = N8nAdapter(n8n_client)
+    kestra = KestraAdapter(
+        base_url=KESTRA_URL,
+        username=KESTRA_AUTH[0],
+        password=KESTRA_AUTH[1],
+    )
+    controller = EngineRoleController(
+        n8n=n8n,
+        kestra=kestra,
+        kestra_namespace=KESTRA_NAMESPACE,
+    )
+    service = WorkflowCutoverService(store=store, engines=controller)
+    suffix = uuid.uuid4().hex[:10]
+    flow_id = f"v55_cutover_{suffix}"
+    flow_url = f"{KESTRA_URL}/main/flows/{KESTRA_NAMESPACE}/{flow_id}"
+    workflow_id = None
+    state_ids = []
+
+    n8n_definition = {
+        "name": f"V55 rollback drill {suffix}",
+        "nodes": [
+            {
+                "parameters": {
+                    "rule": {
+                        "interval": [
+                            {"field": "hours", "hoursInterval": 24}
+                        ]
+                    }
+                },
+                "id": f"schedule-{suffix}",
+                "name": "Schedule Trigger",
+                "type": "n8n-nodes-base.scheduleTrigger",
+                "typeVersion": 1.2,
+                "position": [0, 0],
+            }
+        ],
+        "connections": {},
+        "settings": {"executionOrder": "v1"},
+    }
+    kestra_definition = f"""
+id: {flow_id}
+namespace: {KESTRA_NAMESPACE}
+description: V55 controlled cutover and rollback drill
+disabled: true
+tasks:
+  - id: evidence
+    type: io.kestra.plugin.core.log.Log
+    message: V55 cutover evidence
+""".strip()
+
+    try:
+        workflow = n8n.deploy_disabled(n8n_definition)
+        workflow_id = str(workflow["id"])
+        n8n.activate("dataops", workflow_id)
+        kestra.deploy_disabled(kestra_definition)
+
+        failure_state = store.create_state(
+            {
+                "dataflow_uid": new_governance_uid(),
+                "environment": "test",
+                "mode": DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
+                "n8n_definition_id": workflow_id,
+                "kestra_definition_id": f"missing_{suffix}",
+                "migration_batch": "read_only_low_frequency",
+            }
+        )
+        state_ids.append(failure_state["state_id"])
+        failed_operation = service.request_cutover(
+            dataflow_uid=failure_state["dataflow_uid"],
+            environment="test",
+            target_mode=DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+            gates=passing_gates(),
+            idempotency_key=f"v55-failure-{suffix}",
+            correlation_id=new_governance_uid(),
+        )
+        failure_result = service.apply_requested_cutover(failed_operation)
+
+        assert failure_result["status"] == "rolled_back"
+        assert n8n_client.get_workflow(workflow_id)["active"] is True
+        store.delete_test_state(failure_state["state_id"])
+        state_ids.remove(failure_state["state_id"])
+
+        success_state = store.create_state(
+            {
+                "dataflow_uid": new_governance_uid(),
+                "environment": "test",
+                "mode": DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
+                "n8n_definition_id": workflow_id,
+                "kestra_definition_id": flow_id,
+                "migration_batch": "read_only_low_frequency",
+            }
+        )
+        state_ids.append(success_state["state_id"])
+        operation = service.request_cutover(
+            dataflow_uid=success_state["dataflow_uid"],
+            environment="test",
+            target_mode=DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+            gates=passing_gates(),
+            idempotency_key=f"v55-success-{suffix}",
+            correlation_id=new_governance_uid(),
+        )
+        cutover_result = service.apply_requested_cutover(operation)
+
+        assert cutover_result["status"] == "succeeded"
+        assert n8n_client.get_workflow(workflow_id)["active"] is False
+        response = requests.get(
+            flow_url, auth=KESTRA_AUTH, timeout=20
+        )
+        response.raise_for_status()
+        assert response.json()["disabled"] is False
+
+        rollback_result = service.rollback_to_n8n(
+            dataflow_uid=success_state["dataflow_uid"],
+            environment="test",
+            idempotency_key=f"v55-rollback-{suffix}",
+            correlation_id=new_governance_uid(),
+        )
+
+        assert rollback_result["status"] == "succeeded"
+        assert n8n_client.get_workflow(workflow_id)["active"] is True
+        response = requests.get(
+            flow_url, auth=KESTRA_AUTH, timeout=20
+        )
+        response.raise_for_status()
+        assert response.json()["disabled"] is True
+    finally:
+        for state_id in state_ids:
+            store.delete_test_state(state_id)
+        if workflow_id is not None:
+            try:
+                n8n.deactivate("dataops", workflow_id)
+            finally:
+                n8n_client.delete_workflow(workflow_id)
+        requests.delete(flow_url, auth=KESTRA_AUTH, timeout=20)
+        engine.dispose()

tests/integration/test_workflow_engine_cutover.py

@@ -0,0 +1,294 @@
+import copy
+
+import pytest
+
+from app.core.orchestration.migration.cutover import (
+    CutoverGates,
+    EngineRoleController,
+    WorkflowCutoverService,
+)
+from app.core.orchestration.migration.dual_run import DualRunMode
+from app.core.orchestration.migration.retirement import (
+    N8nRetirementChecklist,
+    N8nRetirementGate,
+)
+
+
+class Store:
+    def __init__(self):
+        self.mode = DualRunMode.N8N_PRIMARY_KESTRA_SHADOW
+        self.operations = {}
+        self.events = []
+        self.completed = []
+        self.rolled_back = []
+
+    def get_state(self, dataflow_uid, environment):
+        return {
+            "dataflow_uid": dataflow_uid,
+            "environment": environment,
+            "mode": self.mode.value,
+            "n8n_definition_id": "n8n-orders",
+            "kestra_definition_id": "kestra-orders",
+            "status": "stable",
+        }
+
+    def request_transition(self, record):
+        previous = self.operations.get(record["idempotency_key"])
+        if previous:
+            return False, copy.deepcopy(previous)
+        operation = {
+            **copy.deepcopy(record),
+            "operation_id": "cutover-operation-1",
+            "status": "claimed",
+        }
+        self.operations[record["idempotency_key"]] = operation
+        self.events.append(
+            {
+                "event_type": "workflow.engine.cutover.requested",
+                "payload": copy.deepcopy(operation),
+            }
+        )
+        return True, copy.deepcopy(operation)
+
+    def complete_transition(self, operation_id, target_mode, result):
+        self.mode = DualRunMode(target_mode)
+        self.completed.append((operation_id, target_mode, copy.deepcopy(result)))
+
+    def rollback_transition(self, operation_id, source_mode, result):
+        self.mode = DualRunMode(source_mode)
+        self.rolled_back.append((operation_id, source_mode, copy.deepcopy(result)))
+
+
+class Engines:
+    def __init__(self):
+        self.calls = []
+        self.fail_kestra_activate = False
+
+    def deactivate_n8n(self, definition_id):
+        self.calls.append(("deactivate_n8n", definition_id))
+        return {"status": "disabled"}
+
+    def activate_n8n(self, definition_id):
+        self.calls.append(("activate_n8n", definition_id))
+        return {"status": "enabled"}
+
+    def deactivate_kestra(self, definition_id):
+        self.calls.append(("deactivate_kestra", definition_id))
+        return {"status": "disabled"}
+
+    def activate_kestra(self, definition_id):
+        self.calls.append(("activate_kestra", definition_id))
+        if self.fail_kestra_activate:
+            raise RuntimeError("Kestra unavailable")
+        return {"status": "enabled"}
+
+
+class Adapter:
+    def __init__(self):
+        self.calls = []
+
+    def activate(self, namespace, definition_id):
+        self.calls.append(("activate", namespace, definition_id))
+        return {"status": "enabled"}
+
+    def deactivate(self, namespace, definition_id):
+        self.calls.append(("deactivate", namespace, definition_id))
+        return {"status": "disabled"}
+
+
+def test_engine_role_controller_binds_fixed_namespaces():
+    n8n = Adapter()
+    kestra = Adapter()
+    controller = EngineRoleController(
+        n8n=n8n,
+        kestra=kestra,
+        n8n_namespace="dataops",
+        kestra_namespace="dataops.migration",
+    )
+
+    controller.deactivate_n8n("n8n-orders")
+    controller.activate_kestra("kestra-orders")
+
+    assert n8n.calls == [("deactivate", "dataops", "n8n-orders")]
+    assert kestra.calls == [
+        ("activate", "dataops.migration", "kestra-orders")
+    ]
+
+
+def passing_gates():
+    return CutoverGates(
+        inventory_complete=True,
+        workflow_spec_valid=True,
+        rollback_target_present=True,
+        shadow_runs=3,
+        required_shadow_runs=3,
+        failure_recovery_observed=True,
+        reconciliation_passed=True,
+        connection_budget_passed=True,
+        sla_passed=True,
+        n8n_baseline_unchanged=True,
+    )
+
+
+def test_cutover_request_is_transactional_outbox_only():
+    store = Store()
+    engines = Engines()
+    service = WorkflowCutoverService(store=store, engines=engines)
+
+    result = service.request_cutover(
+        dataflow_uid="01900000-0000-7000-8000-000000000094",
+        environment="test",
+        target_mode=DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+        gates=passing_gates(),
+        idempotency_key="cutover:orders:v55",
+        correlation_id="01900000-0000-7000-8000-000000000095",
+    )
+
+    assert result["status"] == "claimed"
+    assert result["source_mode"] == "n8n_primary_kestra_shadow"
+    assert result["target_mode"] == "kestra_primary_n8n_standby"
+    assert engines.calls == []
+    assert store.events[0]["event_type"] == "workflow.engine.cutover.requested"
+    assert "password" not in repr(store.events[0]).lower()
+
+
+def test_cutover_consumer_disables_n8n_before_enabling_kestra():
+    store = Store()
+    engines = Engines()
+    service = WorkflowCutoverService(store=store, engines=engines)
+    operation = service.request_cutover(
+        dataflow_uid="01900000-0000-7000-8000-000000000094",
+        environment="test",
+        target_mode=DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+        gates=passing_gates(),
+        idempotency_key="cutover:orders:v55",
+        correlation_id="01900000-0000-7000-8000-000000000095",
+    )
+
+    result = service.apply_requested_cutover(operation)
+
+    assert result["status"] == "succeeded"
+    assert result["mode"] == "kestra_primary_n8n_standby"
+    assert engines.calls == [
+        ("deactivate_n8n", "n8n-orders"),
+        ("activate_kestra", "kestra-orders"),
+    ]
+    assert store.completed[0][0] == "cutover-operation-1"
+
+
+def test_kestra_activation_failure_automatically_restores_n8n():
+    store = Store()
+    engines = Engines()
+    engines.fail_kestra_activate = True
+    service = WorkflowCutoverService(store=store, engines=engines)
+    operation = service.request_cutover(
+        dataflow_uid="01900000-0000-7000-8000-000000000094",
+        environment="test",
+        target_mode=DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+        gates=passing_gates(),
+        idempotency_key="cutover:orders:v55",
+        correlation_id="01900000-0000-7000-8000-000000000095",
+    )
+
+    result = service.apply_requested_cutover(operation)
+
+    assert result["status"] == "rolled_back"
+    assert result["mode"] == "n8n_primary_kestra_shadow"
+    assert engines.calls == [
+        ("deactivate_n8n", "n8n-orders"),
+        ("activate_kestra", "kestra-orders"),
+        ("activate_n8n", "n8n-orders"),
+    ]
+    assert store.rolled_back[0][1] == "n8n_primary_kestra_shadow"
+
+
+def test_same_cutover_key_is_idempotent():
+    store = Store()
+    service = WorkflowCutoverService(store=store, engines=Engines())
+    arguments = {
+        "dataflow_uid": "01900000-0000-7000-8000-000000000094",
+        "environment": "test",
+        "target_mode": DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+        "gates": passing_gates(),
+        "idempotency_key": "cutover:orders:v55",
+        "correlation_id": "01900000-0000-7000-8000-000000000095",
+    }
+
+    first = service.request_cutover(**arguments)
+    second = service.request_cutover(**arguments)
+
+    assert first == second
+    assert len(store.events) == 1
+
+
+def test_cutover_gates_fail_closed():
+    failed = CutoverGates(
+        **{
+            **passing_gates().as_dict(),
+            "reconciliation_passed": False,
+        }
+    )
+
+    with pytest.raises(ValueError, match="reconciliation_passed"):
+        WorkflowCutoverService(store=Store(), engines=Engines()).request_cutover(
+            dataflow_uid="01900000-0000-7000-8000-000000000094",
+            environment="test",
+            target_mode=DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
+            gates=failed,
+            idempotency_key="cutover:orders:v55",
+            correlation_id="01900000-0000-7000-8000-000000000095",
+        )
+
+
+def test_kestra_primary_requires_independent_retirement_approval():
+    store = Store()
+    store.mode = DualRunMode.KESTRA_PRIMARY_N8N_STANDBY
+    service = WorkflowCutoverService(store=store, engines=Engines())
+
+    with pytest.raises(ValueError, match="retirement approval"):
+        service.request_cutover(
+            dataflow_uid="01900000-0000-7000-8000-000000000094",
+            environment="test",
+            target_mode=DualRunMode.KESTRA_PRIMARY,
+            gates=passing_gates(),
+            idempotency_key="retire:orders:v55",
+            correlation_id="01900000-0000-7000-8000-000000000095",
+        )
+
+
+def test_approved_retirement_can_finalize_kestra_primary_mode():
+    store = Store()
+    store.mode = DualRunMode.KESTRA_PRIMARY_N8N_STANDBY
+    engines = Engines()
+    service = WorkflowCutoverService(store=store, engines=engines)
+    decision = N8nRetirementGate.evaluate(
+        N8nRetirementChecklist(
+            inventory_disposition_complete=True,
+            all_dataflows_kestra_primary=True,
+            business_cycle_and_recovery_complete=True,
+            standby_observation_no_rollback=True,
+            monitoring_and_alerting_complete=True,
+            dependency_failures_safe=True,
+            secure_archive_complete=True,
+            independent_review_approved=True,
+            final_l4_passed=True,
+        )
+    )
+
+    operation = service.request_cutover(
+        dataflow_uid="01900000-0000-7000-8000-000000000094",
+        environment="test",
+        target_mode=DualRunMode.KESTRA_PRIMARY,
+        gates=passing_gates(),
+        idempotency_key="retire:orders:v55",
+        correlation_id="01900000-0000-7000-8000-000000000095",
+        retirement_decision=decision,
+    )
+    result = service.apply_requested_cutover(operation)
+
+    assert result["status"] == "succeeded"
+    assert result["mode"] == "kestra_primary"
+    assert engines.calls == [
+        ("deactivate_n8n", "n8n-orders"),
+        ("activate_kestra", "kestra-orders"),
+    ]

tests/integration/test_workflow_engine_rollback.py

@@ -0,0 +1,25 @@
+from app.core.orchestration.migration.cutover import WorkflowCutoverService
+from app.core.orchestration.migration.dual_run import DualRunMode
+from tests.integration.test_workflow_engine_cutover import Engines, Store
+
+
+def test_explicit_rollback_disables_kestra_before_reactivating_n8n():
+    store = Store()
+    store.mode = DualRunMode.KESTRA_PRIMARY_N8N_STANDBY
+    engines = Engines()
+    service = WorkflowCutoverService(store=store, engines=engines)
+
+    result = service.rollback_to_n8n(
+        dataflow_uid="01900000-0000-7000-8000-000000000094",
+        environment="test",
+        idempotency_key="rollback:orders:v55",
+        correlation_id="01900000-0000-7000-8000-000000000095",
+    )
+
+    assert result["status"] == "succeeded"
+    assert result["mode"] == "n8n_primary_kestra_shadow"
+    assert engines.calls == [
+        ("deactivate_kestra", "kestra-orders"),
+        ("activate_n8n", "n8n-orders"),
+    ]
+    assert store.mode is DualRunMode.N8N_PRIMARY_KESTRA_SHADOW

tests/integration/test_workflow_migration_persistence.py

@@ -0,0 +1,106 @@
+import os
+
+import pytest
+from sqlalchemy import create_engine, text
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.orchestration.migration.dual_run import DualRunMode
+from app.core.orchestration.migration.repository import PostgresMigrationStore
+
+pytestmark = pytest.mark.skipif(
+    os.getenv("RUN_V55_PERSISTENCE_INTEGRATION") != "1",
+    reason="set RUN_V55_PERSISTENCE_INTEGRATION=1",
+)
+
+
+def test_cutover_state_and_outbox_commit_atomically_and_idempotently():
+    engine = create_engine(
+        os.environ["DATAOPS_MCP_TEST_DATABASE_URL"], pool_pre_ping=True
+    )
+    store = PostgresMigrationStore(engine)
+    dataflow_uid = new_governance_uid()
+    correlation_id = new_governance_uid()
+    idempotency_key = f"cutover:{dataflow_uid}:v55"
+    state = None
+    operation = None
+    try:
+        state = store.create_state(
+            {
+                "dataflow_uid": dataflow_uid,
+                "environment": "test",
+                "mode": DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
+                "n8n_definition_id": f"n8n-{dataflow_uid}",
+                "kestra_definition_id": f"kestra-{dataflow_uid}",
+                "migration_batch": "low_risk",
+            }
+        )
+        dual_run_id = store.record_dual_run(
+            {
+                "dataflow_uid": dataflow_uid,
+                "environment": "test",
+                "input_snapshot_hash": "a" * 64,
+                "primary_execution_id": "n8n-v55-persistence",
+                "shadow_execution_id": "kestra-v55-persistence",
+                "primary_status": "success",
+                "shadow_status": "success",
+                "isolation": {"strategy": "read_only"},
+                "formal_engine_changed": False,
+                "correlation_id": correlation_id,
+            }
+        )
+        report_id = store.record_reconciliation(
+            {
+                "state_id": state["state_id"],
+                "dual_run_id": dual_run_id,
+                "status": "passed",
+                "policy": {},
+                "safe_metrics": {"row_count_delta": 0},
+                "differences": [],
+                "correlation_id": correlation_id,
+            }
+        )
+        assert dual_run_id
+        assert report_id
+        record = {
+            "dataflow_uid": dataflow_uid,
+            "environment": "test",
+            "source_mode": DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
+            "target_mode": DualRunMode.KESTRA_PRIMARY_N8N_STANDBY.value,
+            "n8n_definition_id": state["n8n_definition_id"],
+            "kestra_definition_id": state["kestra_definition_id"],
+            "gates": {"reconciliation_passed": True},
+            "idempotency_key": idempotency_key,
+            "correlation_id": correlation_id,
+        }
+
+        claimed, operation = store.request_transition(record)
+        replay_claimed, replay = store.request_transition(record)
+
+        assert claimed is True
+        assert replay_claimed is False
+        assert replay["operation_id"] == operation["operation_id"]
+        with engine.connect() as connection:
+            event = connection.execute(
+                text(
+                    "SELECT event_type, payload "
+                    "FROM public.outbox_events "
+                    "WHERE aggregate_id = :operation_id"
+                ),
+                {"operation_id": operation["operation_id"]},
+            ).one()
+        assert event[0] == "workflow.engine.cutover.requested"
+        assert event[1]["idempotency_key"] == idempotency_key
+        assert "password" not in repr(event[1]).lower()
+
+        store.complete_transition(
+            operation["operation_id"],
+            DualRunMode.KESTRA_PRIMARY_N8N_STANDBY.value,
+            {"status": "succeeded"},
+        )
+        current = store.get_state(dataflow_uid, "test")
+        assert current["mode"] == "kestra_primary_n8n_standby"
+        assert current["status"] == "stable"
+    finally:
+        if state:
+            store.delete_test_state(state["state_id"])
+        engine.dispose()

tests/integration/test_workflow_migration_rollout.py

@@ -0,0 +1,79 @@
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from app.core.orchestration.migration.rollout import (
+    MigrationBatch,
+    ObservationWindow,
+    RolloutEntryGates,
+    RolloutExitGates,
+    RolloutPolicy,
+)
+
+
+def test_batches_are_fixed_from_low_to_high_risk():
+    assert [batch.value for batch in MigrationBatch] == [
+        "read_only_low_frequency",
+        "idempotent_partition_write",
+        "core_multi_node",
+        "special_high_risk",
+    ]
+
+
+def test_entry_gates_require_shadow_failure_recovery_and_capacity():
+    gates = RolloutEntryGates(
+        inventory_complete=True,
+        workflow_spec_valid=True,
+        policy_complete=True,
+        rollback_target_present=True,
+        shadow_runs=2,
+        required_shadow_runs=3,
+        failure_recovery_observed=True,
+        reconciliation_passed=True,
+        connection_budget_passed=True,
+        peak_concurrency_passed=True,
+        sla_passed=True,
+    )
+
+    with pytest.raises(ValueError, match="shadow_runs"):
+        RolloutPolicy.assert_entry(gates)
+
+
+def test_observation_window_fails_closed_before_deadline_or_on_incident():
+    now = datetime(2026, 7, 19, tzinfo=UTC)
+    window = ObservationWindow(
+        started_at=now - timedelta(hours=2),
+        required_duration=timedelta(hours=4),
+        unexplained_differences=0,
+        duplicate_writes=0,
+        critical_sla_breaches=0,
+        rollback_requests=0,
+    )
+    with pytest.raises(ValueError, match="observation_window_complete"):
+        RolloutPolicy.assert_observation(window, now=now)
+
+    incident = ObservationWindow(
+        started_at=now - timedelta(hours=5),
+        required_duration=timedelta(hours=4),
+        unexplained_differences=0,
+        duplicate_writes=1,
+        critical_sla_breaches=0,
+        rollback_requests=0,
+    )
+    with pytest.raises(ValueError, match="duplicate_writes"):
+        RolloutPolicy.assert_observation(incident, now=now)
+
+
+def test_exit_gates_require_recovery_and_unchanged_n8n_standby():
+    gates = RolloutExitGates(
+        observation_window_complete=True,
+        no_unexplained_differences=True,
+        no_duplicate_writes=True,
+        sla_passed=True,
+        pause_retry_rollback_exercised=False,
+        n8n_standby=True,
+        n8n_unauthorized_changes=0,
+    )
+
+    with pytest.raises(ValueError, match="pause_retry_rollback_exercised"):
+        RolloutPolicy.assert_exit(gates)

tests/integration/test_workflow_reconciliation.py

@@ -0,0 +1,187 @@
+import pytest
+
+from app.core.orchestration.migration.reconciliation import (
+    ExecutionSnapshot,
+    ReconciliationPolicy,
+    WorkflowReconciler,
+)
+
+
+def snapshot(
+    *,
+    engine,
+    execution_id,
+    content_hash="content-a",
+    amount_sum=30.0,
+    duration=12.0,
+    partition_duration=None,
+    cpu_seconds=2.0,
+):
+    partition_duration = (
+        duration if partition_duration is None else partition_duration
+    )
+    return ExecutionSnapshot.from_dict(
+        {
+            "engine": engine,
+            "execution_id": execution_id,
+            "status": "success",
+            "duration_seconds": duration,
+            "resource_usage": {
+                "cpu_seconds": cpu_seconds,
+                "peak_memory_mb": 128.0,
+            },
+            "nodes": {
+                "read_orders": {
+                    "partitions": {
+                        "2026-07-19": {
+                            "row_count": 2,
+                            "primary_keys_hash": "keys-a",
+                            "content_hash": content_hash,
+                            "error_count": 0,
+                            "duration_seconds": partition_duration,
+                            "resource_usage": {"rows_scanned": 2},
+                            "aggregates": {
+                                "amount_sum": amount_sum,
+                                "generated_epoch": 1000,
+                            },
+                        }
+                    }
+                }
+            },
+        }
+    )
+
+
+def test_equivalent_runs_pass_all_core_dimensions_without_raw_rows():
+    report = WorkflowReconciler().compare(
+        snapshot(engine="n8n", execution_id="n8n-1"),
+        snapshot(engine="kestra", execution_id="kestra-1"),
+        ReconciliationPolicy(),
+    )
+
+    assert report["status"] == "passed"
+    assert report["equivalent"] is True
+    assert report["difference_count"] == 0
+    assert report["metrics"]["compared_partitions"] == 1
+    assert report["metrics"]["row_count_delta"] == 0
+    assert "rows" not in repr(report).lower()
+
+
+def test_report_locates_hash_and_aggregate_difference_to_node_partition():
+    report = WorkflowReconciler().compare(
+        snapshot(engine="n8n", execution_id="n8n-1"),
+        snapshot(
+            engine="kestra",
+            execution_id="kestra-1",
+            content_hash="content-b",
+            amount_sum=32.0,
+        ),
+        ReconciliationPolicy(
+            numeric_tolerances={
+                "nodes.*.partitions.*.aggregates.amount_sum": 0.5
+            }
+        ),
+    )
+
+    assert report["status"] == "failed"
+    assert report["equivalent"] is False
+    assert {
+        (
+            difference["node_id"],
+            difference["partition"],
+            difference["metric"],
+        )
+        for difference in report["differences"]
+    } >= {
+        ("read_orders", "2026-07-19", "content_hash"),
+        ("read_orders", "2026-07-19", "aggregates.amount_sum"),
+    }
+
+
+def test_explicit_tolerance_and_nondeterministic_ignore_can_pass():
+    policy = ReconciliationPolicy(
+        ignored_paths={
+            "duration_seconds",
+            "resource_usage.cpu_seconds",
+            "nodes.*.partitions.*.duration_seconds",
+            "nodes.*.partitions.*.aggregates.generated_epoch",
+        },
+        numeric_tolerances={
+            "nodes.*.partitions.*.aggregates.amount_sum": 0.1
+        },
+    )
+    report = WorkflowReconciler().compare(
+        snapshot(engine="n8n", execution_id="n8n-1"),
+        snapshot(
+            engine="kestra",
+            execution_id="kestra-1",
+            amount_sum=30.05,
+            duration=30.0,
+            cpu_seconds=5.0,
+        ),
+        policy,
+    )
+
+    assert report["status"] == "passed"
+    assert report["policy"]["ignored_paths"] == sorted(policy.ignored_paths)
+    assert report["policy"]["numeric_tolerances"] == {
+        "nodes.*.partitions.*.aggregates.amount_sum": 0.1
+    }
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "status",
+        "nodes.*.partitions.*.row_count",
+        "nodes.*.partitions.*.primary_keys_hash",
+        "nodes.*.partitions.*.content_hash",
+        "nodes.*.partitions.*.error_count",
+        "nodes.read_orders.partitions.2026-07-19.row_count",
+        "nodes.*.partitions.*",
+    ],
+)
+def test_policy_cannot_ignore_core_correctness_metrics(path):
+    with pytest.raises(ValueError, match="core reconciliation metric"):
+        ReconciliationPolicy(ignored_paths={path})
+
+
+def test_partition_duration_is_reconciled_independently():
+    report = WorkflowReconciler().compare(
+        snapshot(engine="n8n", execution_id="n8n-1"),
+        snapshot(
+            engine="kestra",
+            execution_id="kestra-1",
+            partition_duration=15.0,
+        ),
+        ReconciliationPolicy(),
+    )
+
+    assert {
+        difference["metric"] for difference in report["differences"]
+    } == {"duration_seconds"}
+    assert report["differences"][0]["node_id"] == "read_orders"
+
+
+def test_missing_partition_is_reported_without_silent_truncation():
+    baseline = snapshot(engine="n8n", execution_id="n8n-1")
+    candidate = snapshot(engine="kestra", execution_id="kestra-1")
+    candidate.nodes["read_orders"]["partitions"] = {}
+
+    report = WorkflowReconciler().compare(
+        baseline,
+        candidate,
+        ReconciliationPolicy(),
+    )
+
+    assert report["status"] == "failed"
+    assert report["differences"] == [
+        {
+            "node_id": "read_orders",
+            "partition": "2026-07-19",
+            "metric": "partition_presence",
+            "baseline": True,
+            "candidate": False,
+        }
+    ]
+    assert report["metrics"]["row_count_delta"] == -2

tests/test_database_migrations.py

@@ -31,6 +31,10 @@ EXPECTED_UPGRADED_TABLES = {
     "runner_task_executions",
     "workflow_canary_evidence",
     "workflow_gateway_operations",
+    "workflow_migration_states",
+    "workflow_dual_runs",
+    "workflow_reconciliation_reports",
+    "workflow_cutover_operations",
 }
 
 

tests/test_v55_migration_commands.py

@@ -0,0 +1,30 @@
+from app.commands.migrate_n8n_workflow import convert_export
+from app.commands.reconcile_dual_run import reconcile_exports
+from tests.integration.test_n8n_kestra_dual_run import (
+    DATAFLOW_UID,
+    SOURCE_UID,
+    n8n_read_workflow,
+)
+from tests.integration.test_workflow_reconciliation import snapshot
+
+
+def test_conversion_command_is_dry_run_and_credential_free():
+    report = convert_export(
+        n8n_read_workflow(),
+        dataflow_uid=DATAFLOW_UID,
+        data_source_uids={"Read Orders": SOURCE_UID},
+    )
+
+    assert report["status"] == "converted"
+    assert report["deployment_performed"] is False
+    assert "credential" not in repr(report).lower()
+
+
+def test_reconciliation_command_uses_bounded_metrics_only():
+    report = reconcile_exports(
+        snapshot(engine="n8n", execution_id="n8n-1").__dict__,
+        snapshot(engine="kestra", execution_id="kestra-1").__dict__,
+    )
+
+    assert report["status"] == "passed"
+    assert "rows" not in repr(report).lower()