fix: make governed dataflow creation recoverable

.superpowers/sdd/task-8-report.md

@@ -117,11 +117,10 @@ Fail-first:
 
 Final:
 
-- Task 8 API/repository/frontend and legacy-cutover contracts:
-  `90 passed, 3 skipped, 59 subtests passed`;
+- Task 8 API/repository/frontend, Saga and legacy-cutover contracts:
+  `64 passed`;
 - full repository suite:
-  `666 passed, 30 skipped, 59 subtests passed`;
-- changed Python Ruff: `All checks passed!`;
+  `672 passed, 31 skipped, 59 subtests passed`;
 - `git diff --check`: passed;
 - frontend production build: completed with `0 errors`;
 - build retained 20 pre-existing `no-console` warnings plus existing CSS
@@ -218,6 +217,63 @@ Real PostgreSQL acceptance confirmed:
 - replay, cross-actor and expired receipt consumption are rejected;
 - acceptance fixtures are removed after the test.
 
+## Recoverable create and trusted-schema closeout
+
+The third review replaced the one-shot PostgreSQL-consume/Neo4j-create
+boundary with a recoverable create Saga. Historical migration 210 remains
+unchanged; forward-only migration `20260724_220` adds:
+
+- `reserved`, `creating`, `completed`, and `failed` states;
+- an expiring create lease, attempt counter and error code;
+- the Neo4j node ID, bounded result JSON and canonical SHA-256 digest;
+- create-started, failed, completed and updated timestamps.
+
+`begin_dataflow_create` now uses one conditional `UPDATE ... RETURNING` to
+claim a lease. The claim is committed before the graph side effect. Governed
+DataFlow creation installs the idempotent `DataFlow.uid` unique constraint,
+uses `MERGE` only on the server-reserved UID, and verifies the persisted name,
+UID, script type and canonical governed requirement without overwriting an
+existing node. Completion conditionally owns the lease and persists the exact
+response. A retry after response loss returns that stored response; an expired
+lease reclaims the same UID, observes the existing graph node and finalizes it.
+Name/UID or immutable-property conflicts fail closed.
+
+Catalog compatibility no longer trusts matching reference strings alone:
+
+- the current input/output references are resolved through `SchemaResolver`;
+- published rules compare the current snapshot hashes with the hashes frozen
+  into their published logical plan;
+- compile and test evidence IDs must be tied to that same logical plan and
+  hash set;
+- a Standard expands every fixed `standard_rule_bindings` entry and requires
+  the same trusted rule plan/evidence/hash chain for every clause;
+- resolver failure returns `unknown`; same-reference hash drift and any bound
+  Standard rule drift return `incompatible`;
+- reader evidence contains only stable IDs, hashes, snapshot IDs and source
+  revisions.
+
+The Data Flow editor stores the reservation expiry, renews it when less than
+60 seconds remain, and changes to a new server UID only when the server
+returns the explicit `draft_reservation_refresh_required` signal. It retries
+that safe pre-side-effect case once. Network errors and in-progress leases do
+not cause a new UID; a completed response-loss retry is replayed by the server.
+
+Additional acceptance evidence:
+
+- local PostgreSQL upgraded from 210 to `20260724_220 (head)`;
+- two real concurrent PostgreSQL sessions produced one claim and one
+  `dataflow_create_in_progress`; completion replay returned the identical
+  result, expired drafts were rejected, and an expired create lease was
+  reclaimed as attempt 2;
+- real Neo4j installed the `data_flow_uid` unique constraint, two same-UID
+  merges produced one node, and immutable drift was rejected;
+- fault-injection tests cover Neo4j failure after a durable PG claim,
+  PostgreSQL finalize failure after Neo4j success, completed-response replay,
+  conflict closure and lease recovery;
+- real PostgreSQL and Neo4j integration: `2 passed`;
+- rebuilt backend and frontend images report migration 220, healthy database
+  and Neo4j checks, application health code 200, and frontend HTTP 200.
+
 ## Residual scope
 
 - Production-line cross-station compatibility remains ultimately authoritative

app/api/data_rules/routes.py

@@ -88,6 +88,15 @@ def _release_service() -> ProductionLineReleaseService:
     return ProductionLineReleaseService(repository, schema_resolver=resolver)
 
 
+def _schema_resolver() -> SchemaResolver:
+    configured = current_app.extensions.get("data_rule_schema_resolver")
+    if configured is not None:
+        return configured
+    repository = _repository()
+    catalog = current_app.extensions.get("data_rule_metadata_catalog")
+    return SchemaResolver(catalog or Neo4jSchemaMetadataCatalog(), repository)
+
+
 def _receipt_signer() -> GenerationReceiptSigner:
     configured = current_app.extensions.get("generation_receipt_signer")
     if configured is not None:
@@ -515,6 +524,7 @@ def catalog_asset(asset_type: str, version_id: str):
                     version_id=version_id,
                     input_schema_refs=inputs,
                     output_schema_ref=output,
+                    schema_resolver=_schema_resolver() if inputs is not None else None,
                 )
             )
         )
@@ -567,6 +577,7 @@ def published_rule_catalog():
                 {
                     "input_schema_refs": inputs,
                     "output_schema_ref": output,
+                    "schema_resolver": _schema_resolver(),
                 }
             )
         return jsonify(

app/core/data_flow/dataflows.py

@@ -139,6 +139,51 @@ class DataFlowService:
             "migration_metadata": copy.deepcopy(metadata),
         }
 
+    @staticmethod
+    def _merge_governed_dataflow(node_data: dict[str, Any]) -> tuple[int, dict]:
+        """Idempotently create by stable UID and reject immutable conflicts."""
+        immutable = {
+            key: node_data[key]
+            for key in ("uid", "name_zh", "script_type", "script_requirement")
+        }
+        driver = connect_graph()
+        try:
+            with driver.session() as session:
+                session.run(
+                    "CREATE CONSTRAINT data_flow_uid IF NOT EXISTS "
+                    "FOR (n:DataFlow) REQUIRE n.uid IS UNIQUE"
+                )
+                conflict = session.run(
+                    "MATCH (n:DataFlow {name_zh: $name_zh}) "
+                    "WHERE n.uid IS NULL OR n.uid <> $uid "
+                    "RETURN n.uid AS uid LIMIT 1",
+                    {"name_zh": node_data["name_zh"], "uid": node_data["uid"]},
+                ).single()
+                if conflict is not None:
+                    raise ValueError("dataflow_uid_conflict")
+                record = session.run(
+                    "MERGE (n:DataFlow {uid: $uid}) "
+                    "ON CREATE SET n = $properties "
+                    "RETURN n, id(n) AS node_id",
+                    {"uid": node_data["uid"], "properties": node_data},
+                ).single()
+                if record is None:
+                    raise RuntimeError(
+                        "governed DataFlow MERGE returned no node"
+                    )
+                persisted = dict(record["n"])
+                for key, expected in immutable.items():
+                    if persisted.get(key) != expected:
+                        raise ValueError("dataflow_uid_conflict")
+                node_id = record["node_id"]
+                if isinstance(node_id, bool) or not isinstance(node_id, int):
+                    raise RuntimeError("governed DataFlow node id is invalid")
+                result = dict(persisted)
+                result["id"] = node_id
+                return node_id, result
+        finally:
+            driver.close()
+
     @staticmethod
     def get_dataflows(
         page: int = 1,
@@ -354,6 +399,8 @@ class DataFlowService:
             governed = DataFlowService._signals_governed(
                 data, decoded_requirement
             )
+            saga_claim = None
+            receipt = None
             if governed:
                 DataFlowService._reject_governed_execution_fields(data)
                 if repository is None:
@@ -370,9 +417,12 @@ class DataFlowService:
                         "governed DataFlow requires an authenticated actor"
                     )
                 receipt = data.get("draft_reservation")
-                reserved_uid = repository.consume_dataflow_draft(
+                saga_claim = repository.begin_dataflow_create(
                     receipt, actor_uid=actor_uid
                 )
+                if saga_claim["status"] == "completed":
+                    return copy.deepcopy(saga_claim["result"])
+                reserved_uid = saga_claim["dataflow_uid"]
                 if (
                     reserved_uid
                     != decoded_requirement["dataflow_spec"]["dataflow_uid"]
@@ -380,6 +430,7 @@ class DataFlowService:
                     raise ValueError(
                         "draft reservation does not match dataflow_uid"
                     )
+                repository.commit_dataflow_create_claim()
                 script_requirement = decoded_requirement
 
             # 处理 script_requirement，将其转换为 JSON 字符串
@@ -387,7 +438,10 @@ class DataFlowService:
                 # 如果是字典或列表，转换为 JSON 字符串
                 if isinstance(script_requirement, (dict, list)):
                     script_requirement_str = json.dumps(
-                        script_requirement, ensure_ascii=False
+                        script_requirement,
+                        ensure_ascii=False,
+                        sort_keys=governed,
+                        separators=(",", ":") if governed else None,
                     )
                 else:
                     # 如果已经是字符串，直接使用
@@ -419,12 +473,29 @@ class DataFlowService:
                 node_data["script_type"] = "governed"
             ensure_governance_uid(node_data)
 
-            # 创建或获取数据流节点
-            dataflow_id = get_node("DataFlow", name=dataflow_name)
-            if dataflow_id:
-                raise ValueError(f"数据流 '{dataflow_name}' 已存在")
-
-            dataflow_id = create_or_get_node("DataFlow", **node_data)
+            # Governed nodes use the reservation UID as the only create key.
+            if governed:
+                try:
+                    dataflow_id, result = (
+                        DataFlowService._merge_governed_dataflow(node_data)
+                    )
+                except Exception as graph_error:
+                    try:
+                        repository.commit_dataflow_create_failure(
+                            reservation_id=receipt["reservation_id"],
+                            lease_token=saga_claim["lease_token"],
+                            error_code="neo4j_create_failed",
+                        )
+                    except Exception:
+                        logger.exception(
+                            "failed to persist governed DataFlow saga failure"
+                        )
+                    raise graph_error
+            else:
+                dataflow_id = get_node("DataFlow", name=dataflow_name)
+                if dataflow_id:
+                    raise ValueError(f"数据流 '{dataflow_name}' 已存在")
+                dataflow_id = create_or_get_node("DataFlow", **node_data)
 
             # 处理标签关系（支持多标签数组）
             tag_list = data.get("tag", [])
@@ -458,27 +529,39 @@ class DataFlowService:
                 except Exception as pg_error:
                     logger.error(f"写入PG数据库失败: {str(pg_error)}")
 
-            # 返回创建的数据流信息
-            # 查询创建的节点获取完整信息
-            query = "MATCH (n:DataFlow {name_zh: $name_zh}) RETURN n, id(n) as node_id"
-            with connect_graph().session() as session:
-                id_result = session.run(query, name_zh=dataflow_name).single()
-                if id_result:
-                    dataflow_node = id_result["n"]
-                    node_id = id_result["node_id"]
-
-                    # 将节点属性转换为字典
-                    result = dict(dataflow_node)
-                    result["id"] = node_id
-                else:
-                    # 如果查询失败，返回基本信息
-                    result = {
-                        "id": (dataflow_id if isinstance(dataflow_id, int) else None),
-                        "uid": node_data["uid"],
-                        "name_zh": dataflow_name,
-                        "name_en": name_en,
-                        "created_at": get_formatted_time(),
-                    }
+            if governed:
+                result = repository.complete_dataflow_create(
+                    reservation_id=receipt["reservation_id"],
+                    dataflow_uid=node_data["uid"],
+                    lease_token=saga_claim["lease_token"],
+                    dataflow_node_id=dataflow_id,
+                    result=result,
+                )
+            else:
+                # 查询创建的节点获取完整信息
+                query = (
+                    "MATCH (n:DataFlow {name_zh: $name_zh}) "
+                    "RETURN n, id(n) as node_id"
+                )
+                with connect_graph().session() as session:
+                    id_result = session.run(query, name_zh=dataflow_name).single()
+                    if id_result:
+                        dataflow_node = id_result["n"]
+                        node_id = id_result["node_id"]
+                        result = dict(dataflow_node)
+                        result["id"] = node_id
+                    else:
+                        result = {
+                            "id": (
+                                dataflow_id
+                                if isinstance(dataflow_id, int)
+                                else None
+                            ),
+                            "uid": node_data["uid"],
+                            "name_zh": dataflow_name,
+                            "name_en": name_en,
+                            "created_at": get_formatted_time(),
+                        }
 
             if not governed:
                 try:

app/core/data_rules/repository.py

@@ -166,9 +166,10 @@ class DataRuleRepository:
             ),
         }
 
-    def consume_dataflow_draft(
-        self, receipt: dict[str, Any], *, actor_uid: str
-    ) -> str:
+    @staticmethod
+    def _draft_receipt(
+        receipt: dict[str, Any], *, actor_uid: str
+    ) -> dict[str, str]:
         if not isinstance(receipt, dict) or set(receipt) != {
             "reservation_id",
             "dataflow_uid",
@@ -182,31 +183,205 @@ class DataRuleRepository:
         actor = _uid(actor_uid, "actor_uid")
         nonce = _text(receipt.get("nonce"), "draft nonce", 200)
         nonce_hash = hashlib.sha256(nonce.encode("utf-8")).hexdigest()
-        consumed = self.session.execute(
+        return {
+            "id": reservation_id,
+            "dataflow_uid": dataflow_uid,
+            "actor_uid": actor,
+            "nonce_hash": nonce_hash,
+        }
+
+    def begin_dataflow_create(
+        self,
+        receipt: dict[str, Any],
+        *,
+        actor_uid: str,
+        lease_seconds: int = 60,
+    ) -> dict[str, Any]:
+        """Atomically claim a create attempt or replay its committed result."""
+        if (
+            isinstance(lease_seconds, bool)
+            or not isinstance(lease_seconds, int)
+            or lease_seconds < 10
+            or lease_seconds > 300
+        ):
+            raise ValueError("dataflow create lease is invalid")
+        values = self._draft_receipt(receipt, actor_uid=actor_uid)
+        lease_token = new_governance_uid()
+        claimed = (
+            self.session.execute(
+                text(
+                    "UPDATE public.dataflow_draft_reservations "
+                    "SET state = 'creating', lease_token = CAST(:lease_token AS uuid), "
+                    "lease_expires_at = CURRENT_TIMESTAMP + "
+                    "(:lease_seconds * INTERVAL '1 second'), "
+                    "attempt_count = attempt_count + 1, "
+                    "create_started_at = COALESCE("
+                    "create_started_at, CURRENT_TIMESTAMP), "
+                    "error_code = NULL, updated_at = CURRENT_TIMESTAMP "
+                    "WHERE id = CAST(:id AS uuid) "
+                    "AND dataflow_uid = CAST(:dataflow_uid AS uuid) "
+                    "AND actor_uid = CAST(:actor_uid AS uuid) "
+                    "AND nonce_hash = :nonce_hash "
+                    "AND consumed_at IS NULL "
+                    "AND (expires_at > CURRENT_TIMESTAMP OR attempt_count > 0) "
+                    "AND (state IN ('reserved', 'failed') OR "
+                    "(state = 'creating' AND lease_expires_at <= CURRENT_TIMESTAMP)) "
+                    "RETURNING dataflow_uid::text, attempt_count "
+                    "/* begin_dataflow_create */"
+                ),
+                {
+                    **values,
+                    "lease_token": lease_token,
+                    "lease_seconds": lease_seconds,
+                },
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if claimed is not None:
+            return {
+                "status": "claimed",
+                "dataflow_uid": str(claimed["dataflow_uid"]),
+                "lease_token": lease_token,
+                "attempt": int(claimed["attempt_count"]),
+            }
+
+        row = (
+            self.session.execute(
+                text(
+                    "SELECT state, dataflow_uid::text, result, "
+                    "result_digest, lease_expires_at, expires_at "
+                    "FROM public.dataflow_draft_reservations "
+                    "WHERE id = CAST(:id AS uuid) "
+                    "AND dataflow_uid = CAST(:dataflow_uid AS uuid) "
+                    "AND actor_uid = CAST(:actor_uid AS uuid) "
+                    "AND nonce_hash = :nonce_hash "
+                    "/* inspect_dataflow_create */"
+                ),
+                values,
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if row is not None and row["state"] == "completed":
+            result = _object(row["result"], "dataflow create result")
+            if _canonical_hash(result) != str(row["result_digest"]):
+                raise RuntimeError("dataflow create result integrity check failed")
+            return {
+                "status": "completed",
+                "dataflow_uid": str(row["dataflow_uid"]),
+                "result": copy.deepcopy(result),
+            }
+        if row is not None and row["state"] == "creating":
+            raise ValueError("dataflow_create_in_progress")
+        raise ValueError("draft_reservation_refresh_required")
+
+    def complete_dataflow_create(
+        self,
+        *,
+        reservation_id: str,
+        dataflow_uid: str,
+        lease_token: str,
+        dataflow_node_id: int,
+        result: dict[str, Any],
+    ) -> dict[str, Any]:
+        reservation = _uid(reservation_id, "reservation_id")
+        flow_uid = _uid(dataflow_uid, "dataflow_uid")
+        lease = _uid(lease_token, "lease_token")
+        if (
+            isinstance(dataflow_node_id, bool)
+            or not isinstance(dataflow_node_id, int)
+            or dataflow_node_id < 0
+        ):
+            raise ValueError("dataflow_node_id is invalid")
+        normalized_result = _object(result, "dataflow create result")
+        result_digest = _canonical_hash(normalized_result)
+        completed = (
+            self.session.execute(
+                text(
+                    "UPDATE public.dataflow_draft_reservations "
+                    "SET state = 'completed', result = CAST(:result AS jsonb), "
+                    "result_digest = :result_digest, "
+                    "dataflow_node_id = :dataflow_node_id, "
+                    "completed_at = CURRENT_TIMESTAMP, "
+                    "consumed_at = CURRENT_TIMESTAMP, lease_token = NULL, "
+                    "lease_expires_at = NULL, updated_at = CURRENT_TIMESTAMP "
+                    "WHERE id = CAST(:id AS uuid) "
+                    "AND dataflow_uid = CAST(:dataflow_uid AS uuid) "
+                    "AND state = 'creating' "
+                    "AND lease_token = CAST(:lease_token AS uuid) "
+                    "RETURNING result /* complete_dataflow_create */"
+                ),
+                {
+                    "id": reservation,
+                    "dataflow_uid": flow_uid,
+                    "lease_token": lease,
+                    "result": _json(normalized_result),
+                    "result_digest": result_digest,
+                    "dataflow_node_id": dataflow_node_id,
+                },
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if completed is None:
+            replay = (
+                self.session.execute(
+                    text(
+                        "SELECT result, result_digest "
+                        "FROM public.dataflow_draft_reservations "
+                        "WHERE id = CAST(:id AS uuid) "
+                        "AND dataflow_uid = CAST(:dataflow_uid AS uuid) "
+                        "AND state = 'completed' "
+                        "/* replay_completed_dataflow_create */"
+                    ),
+                    {"id": reservation, "dataflow_uid": flow_uid},
+                )
+                .mappings()
+                .one_or_none()
+            )
+            if replay is None:
+                raise RuntimeError("dataflow create lease was lost")
+            replay_result = _object(replay["result"], "dataflow create result")
+            if _canonical_hash(replay_result) != str(replay["result_digest"]):
+                raise RuntimeError("dataflow create result integrity check failed")
+            return copy.deepcopy(replay_result)
+        return copy.deepcopy(normalized_result)
+
+    def fail_dataflow_create(
+        self, *, reservation_id: str, lease_token: str, error_code: str
+    ) -> None:
+        self.session.execute(
             text(
                 "UPDATE public.dataflow_draft_reservations "
-                "SET consumed_at = CURRENT_TIMESTAMP "
+                "SET state = 'failed', error_code = :error_code, "
+                "failed_at = CURRENT_TIMESTAMP, lease_token = NULL, "
+                "lease_expires_at = NULL, updated_at = CURRENT_TIMESTAMP "
                 "WHERE id = CAST(:id AS uuid) "
-                "AND dataflow_uid = CAST(:dataflow_uid AS uuid) "
-                "AND actor_uid = CAST(:actor_uid AS uuid) "
-                "AND nonce_hash = :nonce_hash "
-                "AND consumed_at IS NULL "
-                "AND expires_at > CURRENT_TIMESTAMP "
-                "RETURNING dataflow_uid::text "
-                "/* consume_dataflow_draft */"
+                "AND state = 'creating' "
+                "AND lease_token = CAST(:lease_token AS uuid) "
+                "/* fail_dataflow_create */"
             ),
             {
-                "id": reservation_id,
-                "dataflow_uid": dataflow_uid,
-                "actor_uid": actor,
-                "nonce_hash": nonce_hash,
+                "id": _uid(reservation_id, "reservation_id"),
+                "lease_token": _uid(lease_token, "lease_token"),
+                "error_code": _text(error_code, "error_code", 100),
             },
-        ).scalar_one_or_none()
-        if consumed is None or str(consumed) != dataflow_uid:
-            raise ValueError(
-                "dataflow draft receipt is invalid, expired, or already used"
-            )
-        return dataflow_uid
+        )
+
+    def commit_dataflow_create_claim(self) -> None:
+        """Durably publish the lease before the Neo4j side effect starts."""
+        self.session.commit()
+
+    def commit_dataflow_create_failure(
+        self, *, reservation_id: str, lease_token: str, error_code: str
+    ) -> None:
+        self.fail_dataflow_create(
+            reservation_id=reservation_id,
+            lease_token=lease_token,
+            error_code=error_code,
+        )
+        self.session.commit()
 
     @staticmethod
     def candidate_hash(candidate: dict[str, Any]) -> str:
@@ -1667,6 +1842,7 @@ class DataRuleRepository:
         input_schema_refs: list[str] | None = None,
         output_schema_ref: str | None = None,
         version_id: str | None = None,
+        schema_resolver: Any | None = None,
     ) -> dict[str, Any]:
         """Return only immutable, trusted rule and standard versions."""
         if not isinstance(query, str) or len(query) > 200:
@@ -1722,6 +1898,8 @@ class DataRuleRepository:
                     "version_id, rv.rule_uid::text AS asset_uid, r.name, "
                     "rv.version_no, r.owner_uid::text AS owner_uid, "
                     "rv.status, rv.rule_spec AS schema_compatibility, "
+                    "lp.id::text AS logical_plan_id, lp.schema_hashes, "
+                    "'[]'::jsonb AS bound_rules, 1::integer AS binding_count, "
                     "((SELECT COUNT(*) FROM "
                     "public.dataflow_component_bindings cb "
                     "WHERE cb.rule_version_id = rv.id) + "
@@ -1768,6 +1946,43 @@ class DataRuleRepository:
                     "sv.standard_uid::text AS asset_uid, s.name, "
                     "sv.version_no, s.owner_uid::text AS owner_uid, "
                     "sv.status, sv.scope AS schema_compatibility, "
+                    "NULL::text AS logical_plan_id, NULL::jsonb AS schema_hashes, "
+                    "COALESCE((SELECT jsonb_agg(jsonb_build_object("
+                    "'rule_version_id', brv.id::text, "
+                    "'rule_spec', brv.rule_spec, "
+                    "'logical_plan_id', blp.id::text, "
+                    "'schema_hashes', blp.schema_hashes, "
+                    "'compile_evidence_id', bce.id::text, "
+                    "'test_evidence_id', bte.id::text) ORDER BY srb.clause_id) "
+                    "FROM public.standard_rule_bindings srb "
+                    "JOIN public.data_rule_versions brv "
+                    "ON brv.id = srb.rule_version_id "
+                    "JOIN LATERAL (SELECT value.* FROM "
+                    "public.rule_logical_plans value "
+                    "WHERE value.rule_version_id = brv.id "
+                    "AND value.status = 'published' "
+                    "ORDER BY value.created_at DESC LIMIT 1) blp ON TRUE "
+                    "JOIN LATERAL (SELECT value.id FROM "
+                    "public.rule_logical_compile_evidence value "
+                    "WHERE value.logical_plan_id = blp.id "
+                    "AND value.status = 'success' "
+                    "AND value.compiler_version = blp.compiler_version "
+                    "AND value.plan_hash = blp.plan_hash "
+                    "AND value.schema_hashes = blp.schema_hashes "
+                    "AND value.capabilities = blp.capabilities "
+                    "ORDER BY value.created_at DESC LIMIT 1) bce ON TRUE "
+                    "JOIN LATERAL (SELECT value.id FROM "
+                    "public.rule_logical_test_evidence value "
+                    "WHERE value.logical_plan_id = blp.id "
+                    "AND value.status = 'success' "
+                    "AND value.plan_hash = blp.plan_hash "
+                    "AND value.schema_hashes = blp.schema_hashes "
+                    "ORDER BY value.created_at DESC LIMIT 1) bte ON TRUE "
+                    "WHERE srb.standard_version_id = sv.id "
+                    "AND brv.status = 'published'), '[]'::jsonb) AS bound_rules, "
+                    "(SELECT COUNT(*) FROM public.standard_rule_bindings srb "
+                    "WHERE srb.standard_version_id = sv.id)::integer "
+                    "AS binding_count, "
                     "(SELECT COUNT(*) FROM "
                     "public.dataflow_component_bindings cb "
                     "WHERE cb.standard_version_id = sv.id)::integer "
@@ -1830,9 +2045,96 @@ class DataRuleRepository:
                 result["id"] = str(evidence_id)
             return result
 
-        def compatibility_summary(
-            value: Any, row_asset_type: str
+        def resolve_ref(schema_ref: Any) -> dict[str, Any] | None:
+            if schema_resolver is None or not isinstance(schema_ref, str):
+                return None
+            try:
+                snapshot = schema_resolver.resolve(schema_ref)
+            except Exception:
+                return None
+            if not isinstance(snapshot, dict):
+                return None
+            schema_hash = snapshot.get("schema_hash")
+            if not isinstance(schema_hash, str) or not re.fullmatch(
+                r"[0-9a-f]{64}", schema_hash
+            ):
+                return None
+            return {
+                "snapshot_id": (
+                    str(snapshot["id"]) if snapshot.get("id") is not None else None
+                ),
+                "schema_hash": schema_hash,
+                "source_revision": (
+                    str(snapshot["source_revision"])
+                    if snapshot.get("source_revision") is not None
+                    else None
+                ),
+            }
+
+        def rule_compatibility(
+            binding: dict[str, Any],
+            published_hashes: Any,
+            logical_plan_id: Any,
+            compile_evidence_id: Any,
+            test_evidence_id: Any,
         ) -> dict[str, Any]:
+            asset_input = binding.get("input_schema_ref")
+            asset_output = binding.get("output_schema_ref")
+            hashes = (
+                _object(published_hashes, "schema hashes")
+                if published_hashes is not None
+                else {}
+            )
+            input_snapshot = resolve_ref(asset_input)
+            output_snapshot = resolve_ref(asset_output)
+            refs_match = (
+                asset_input in input_schema_refs
+                and asset_output == output_schema_ref
+            )
+            if input_snapshot is None or output_snapshot is None:
+                status, reason = "unknown", "schema_resolution_unavailable"
+            else:
+                hashes_match = (
+                    input_snapshot["schema_hash"] == hashes.get("input")
+                    and output_snapshot["schema_hash"] == hashes.get("output")
+                )
+                status = (
+                    "compatible" if refs_match and hashes_match else "incompatible"
+                )
+                reason = (
+                    "trusted_schema_hashes_match"
+                    if status == "compatible"
+                    else "schema_ref_or_hash_drift"
+                )
+            return {
+                "status": status,
+                "reason": reason,
+                "evidence": {
+                    "logical_plan_id": (
+                        str(logical_plan_id)
+                        if logical_plan_id is not None
+                        else None
+                    ),
+                    "compile_evidence_id": (
+                        str(compile_evidence_id)
+                        if compile_evidence_id is not None
+                        else None
+                    ),
+                    "test_evidence_id": (
+                        str(test_evidence_id)
+                        if test_evidence_id is not None
+                        else None
+                    ),
+                    "input": input_snapshot,
+                    "output": output_snapshot,
+                    "published_schema_hashes": {
+                        "input": hashes.get("input"),
+                        "output": hashes.get("output"),
+                    },
+                },
+            }
+
+        def compatibility_summary(value: Any, row: Any) -> dict[str, Any]:
             binding = _object(value, "schema compatibility")
             if not has_context:
                 return {
@@ -1840,38 +2142,61 @@ class DataRuleRepository:
                     "reason": "production_line_context_required",
                     "evidence": None,
                 }
-            if row_asset_type == "rule":
-                asset_input = binding.get("input_schema_ref")
-                asset_output = binding.get("output_schema_ref")
-                compatible = (
-                    asset_input in input_schema_refs
-                    and asset_output == output_schema_ref
+            if str(row["asset_type"]) == "rule":
+                return rule_compatibility(
+                    binding,
+                    row.get("schema_hashes"),
+                    row.get("logical_plan_id"),
+                    row.get("compile_evidence_id"),
+                    row.get("test_evidence_id"),
                 )
-                evidence = {
-                    "asset_input_schema_ref": asset_input,
-                    "asset_output_schema_ref": asset_output,
-                    "flow_input_schema_refs": input_schema_refs,
-                    "flow_output_schema_ref": output_schema_ref,
-                }
-            else:
-                scope = binding.get("schema_ref")
-                compatible = scope in {
-                    *input_schema_refs,
-                    output_schema_ref,
-                }
-                evidence = {
-                    "asset_scope_schema_ref": scope,
-                    "flow_input_schema_refs": input_schema_refs,
-                    "flow_output_schema_ref": output_schema_ref,
+            bound_rules = _array(
+                row.get("bound_rules", []), "standard bound rules"
+            )
+            if not bound_rules or len(bound_rules) != int(
+                row.get("binding_count", 0)
+            ):
+                return {
+                    "status": "unknown",
+                    "reason": "standard_trusted_rule_evidence_incomplete",
+                    "evidence": {"bound_rules": []},
                 }
+            summaries = [
+                rule_compatibility(
+                    _object(item.get("rule_spec"), "bound rule spec"),
+                    item.get("schema_hashes"),
+                    item.get("logical_plan_id"),
+                    item.get("compile_evidence_id"),
+                    item.get("test_evidence_id"),
+                )
+                for item in bound_rules
+            ]
+            statuses = {summary["status"] for summary in summaries}
+            status = (
+                "unknown"
+                if "unknown" in statuses
+                else "incompatible"
+                if "incompatible" in statuses
+                else "compatible"
+            )
             return {
-                "status": "compatible" if compatible else "incompatible",
+                "status": status,
                 "reason": (
-                    "exact_schema_refs_match"
-                    if compatible
-                    else "exact_schema_refs_do_not_match"
+                    "all_standard_rule_hashes_match"
+                    if status == "compatible"
+                    else "standard_bound_rule_schema_drift"
                 ),
-                "evidence": evidence,
+                "evidence": {
+                    "bound_rules": [
+                        {
+                            "rule_version_id": str(item["rule_version_id"]),
+                            **summary["evidence"],
+                        }
+                        for item, summary in zip(
+                            bound_rules, summaries, strict=True
+                        )
+                    ]
+                },
             }
 
         items = []
@@ -1893,7 +2218,7 @@ class DataRuleRepository:
                     "status": str(row["status"]),
                     "schema_compatibility": compatibility_summary(
                         row["schema_compatibility"],
-                        str(row["asset_type"]),
+                        row,
                     ),
                     "impact_count": int(row["impact_count"]),
                     "backend": str(row["backend"]),
@@ -1918,12 +2243,14 @@ class DataRuleRepository:
         version_id: str,
         input_schema_refs: list[str] | None = None,
         output_schema_ref: str | None = None,
+        schema_resolver: Any | None = None,
     ) -> dict[str, Any]:
         page = self.search_published_assets(
             asset_type=asset_type,
             version_id=version_id,
             input_schema_refs=input_schema_refs,
             output_schema_ref=output_schema_ref,
+            schema_resolver=schema_resolver,
             limit=1,
             offset=0,
         )

frontend/src/utils/request.js

@@ -72,7 +72,7 @@ service.interceptors.response.use(
         router.push({ path: '/login-local', query: { redirect } })
       }
     }
-    return Promise.reject(error.message || error)
+    return Promise.reject(error.response?.data || error.message || error)
   }
 )
 

frontend/src/views/dataGovernance/dataProcess/components/edit.vue

@@ -183,6 +183,8 @@ export default {
       saving: false,
       governedDataflowUid: null,
       governedDraftReservation: null,
+      governedDraftExpiresAt: null,
+      governedDraftReplayRequired: false,
       scriptContent: '',
       businessDomain: [],
       filteredBusinessDomain: {
@@ -243,7 +245,7 @@ export default {
     this.loading = false
   },
   methods: {
-    async ensureDataflowUid () {
+    async ensureDataflowUid (force = false) {
       const existing = this.itemData.dataflow_uid ||
         this.itemData.data_flow_uid ||
         this.itemData.uid
@@ -251,9 +253,14 @@ export default {
         this.governedDataflowUid = existing
         return
       }
+      if (!force && this.governedDataflowUid && this.governedDraftReservation) {
+        return
+      }
       try {
         const { data } = await createProductionLineDraftIdentity()
         this.governedDataflowUid = data.dataflow_uid
+        this.governedDraftExpiresAt = data.expires_at
+        this.governedDraftReplayRequired = false
         this.governedDraftReservation = {
           reservation_id: data.reservation_id,
           dataflow_uid: data.dataflow_uid,
@@ -263,6 +270,45 @@ export default {
         this.$snackbar.error(error || '无法创建受治理数据流草稿标识')
       }
     },
+    async ensureFreshDraftReservation () {
+      if (Object.keys(this.itemData).length) return
+      // An ambiguous network failure may have lost a completed response.
+      // Replay the original receipt before considering a new UID.
+      if (this.governedDraftReplayRequired) return
+      const expiresAt = Date.parse(this.governedDraftExpiresAt || '')
+      const nearExpiry = !Number.isFinite(expiresAt) ||
+        expiresAt - Date.now() <= 60000
+      if (nearExpiry) await this.ensureDataflowUid(true)
+    },
+    isSafeReservationRefreshError (error) {
+      const message = typeof error === 'string'
+        ? error
+        : error?.message || error?.msg || ''
+      return message.includes('draft_reservation_refresh_required')
+    },
+    buildSavePayload (base) {
+      const payload = {
+        ...base,
+        script_type: 'governed',
+        script_path: '',
+        script_requirement: {
+          dataflow_spec: {
+            ...this.productionLineSpec,
+            dataflow_uid: this.dataflowUid,
+            name: base.name_zh || this.productionLineSpec.name
+          },
+          dataset_edges: {
+            source_table: [...this.datasetEdges.source_table],
+            target_table: this.normalizedTarget
+          },
+          migration_metadata: this.migrationMetadata
+        }
+      }
+      if (!Object.keys(this.itemData).length) {
+        payload.draft_reservation = this.governedDraftReservation
+      }
+      return payload
+    },
     async getList () {
       try {
         const { data } = await api.getBusinessDomainList2()
@@ -337,36 +383,28 @@ export default {
       const base = this.$refs.editBaseRefs.getValue()
       if (!base) return
       if (!this.dataflowUid) await this.ensureDataflowUid()
+      await this.ensureFreshDraftReservation()
       if (!this.dataflowUid) {
         this.$snackbar.error('无法取得服务端生产线草稿标识，暂不能保存')
         return
       }
-      const payload = {
-        ...base,
-        script_type: 'governed',
-        script_path: '',
-        script_requirement: {
-          dataflow_spec: {
-            ...this.productionLineSpec,
-            dataflow_uid: this.dataflowUid,
-            name: base.name_zh || this.productionLineSpec.name
-          },
-          dataset_edges: {
-            source_table: [...this.datasetEdges.source_table],
-            target_table: this.normalizedTarget
-          },
-          migration_metadata: this.migrationMetadata
-        }
-      }
-      if (!Object.keys(this.itemData).length) {
-        payload.draft_reservation = this.governedDraftReservation
-      }
+      let payload = this.buildSavePayload(base)
       this.saving = true
       try {
         if (Object.keys(this.itemData).length) {
           await api.updateDataFlow(this.itemData.id, payload)
         } else {
-          await api.addDataFlow(payload)
+          try {
+            await api.addDataFlow(payload)
+          } catch (error) {
+            if (!this.isSafeReservationRefreshError(error)) {
+              this.governedDraftReplayRequired = true
+              throw error
+            }
+            await this.ensureDataflowUid(true)
+            payload = this.buildSavePayload(base)
+            await api.addDataFlow(payload)
+          }
         }
         this.$snackbar.success('数据生产线定义已保存')
         this.$emit('success')

migrations/versions/20260724_220_dataflow_create_saga.py

@@ -0,0 +1,61 @@
+"""Make governed DataFlow creation recoverable and idempotent."""
+
+from alembic import op
+
+revision = "20260724_220"
+down_revision = "20260723_210"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE public.dataflow_draft_reservations
+            ADD COLUMN state VARCHAR(20) NOT NULL DEFAULT 'reserved',
+            ADD COLUMN lease_token UUID,
+            ADD COLUMN lease_expires_at TIMESTAMPTZ,
+            ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0,
+            ADD COLUMN dataflow_node_id BIGINT,
+            ADD COLUMN result JSONB,
+            ADD COLUMN result_digest CHAR(64),
+            ADD COLUMN error_code VARCHAR(100),
+            ADD COLUMN create_started_at TIMESTAMPTZ,
+            ADD COLUMN completed_at TIMESTAMPTZ,
+            ADD COLUMN failed_at TIMESTAMPTZ,
+            ADD COLUMN updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP;
+
+        UPDATE public.dataflow_draft_reservations
+        SET state = 'failed',
+            error_code = 'legacy_consumed_without_result',
+            failed_at = COALESCE(consumed_at, CURRENT_TIMESTAMP)
+        WHERE consumed_at IS NOT NULL;
+
+        ALTER TABLE public.dataflow_draft_reservations
+            ADD CONSTRAINT ck_dataflow_draft_saga_state
+            CHECK (state IN ('reserved', 'creating', 'completed', 'failed')),
+            ADD CONSTRAINT ck_dataflow_draft_saga_attempts
+            CHECK (attempt_count >= 0),
+            ADD CONSTRAINT ck_dataflow_draft_completed_result
+            CHECK (
+                state <> 'completed'
+                OR (
+                    result IS NOT NULL
+                    AND result_digest IS NOT NULL
+                    AND dataflow_node_id IS NOT NULL
+                    AND completed_at IS NOT NULL
+                )
+            );
+
+        CREATE INDEX idx_dataflow_draft_saga_claim
+            ON public.dataflow_draft_reservations(
+                state, lease_expires_at, expires_at
+            );
+        """
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError(
+        "DataFlow create saga reservations are forward-only and cannot downgrade"
+    )

tests/core/data_rules/test_data_rule_repository.py

@@ -1,9 +1,6 @@
 from __future__ import annotations
 
-import hashlib
 import json
-from concurrent.futures import ThreadPoolExecutor
-from threading import Lock
 
 import pytest
 
@@ -183,92 +180,82 @@ def test_create_rule_version_is_draft_immutable_and_idempotent():
     )
 
 
-def test_dataflow_draft_receipt_is_actor_bound_expiring_and_single_use():
+def test_dataflow_create_saga_claim_replay_and_integrity_are_closed():
     from app.core.data_rules.repository import DataRuleRepository
 
     actor = new_governance_uid()
-    other_actor = new_governance_uid()
     receipt = {
         "reservation_id": new_governance_uid(),
         "dataflow_uid": new_governance_uid(),
         "nonce": "unguessable-test-nonce-value",
     }
 
-    class ScalarResult:
-        def __init__(self, value):
-            self.value = value
-
-        def scalar_one_or_none(self):
-            return self.value
-
     class AtomicSession:
-        def __init__(self, *, expired=False):
-            self.expired = expired
-            self.consumed = False
-            self.lock = Lock()
+        def __init__(self):
+            self.state = "reserved"
+            self.result = None
+            self.digest = None
             self.calls = []
 
         def execute(self, statement, params=None):
             sql = str(statement)
             values = params or {}
             self.calls.append((sql, values))
-            with self.lock:
-                valid = (
-                    not self.expired
-                    and not self.consumed
-                    and values["id"] == receipt["reservation_id"]
-                    and values["dataflow_uid"] == receipt["dataflow_uid"]
-                    and values["actor_uid"] == actor
-                    and values["nonce_hash"]
-                    == hashlib.sha256(
-                        receipt["nonce"].encode("utf-8")
-                    ).hexdigest()
+            if "begin_dataflow_create" in sql and self.state == "reserved":
+                self.state = "creating"
+                return FakeResult(
+                    rows=[{"dataflow_uid": receipt["dataflow_uid"], "attempt_count": 1}]
                 )
-                if valid:
-                    self.consumed = True
-                    return ScalarResult(receipt["dataflow_uid"])
-                return ScalarResult(None)
+            if "begin_dataflow_create" in sql:
+                return FakeResult()
+            if "inspect_dataflow_create" in sql:
+                return FakeResult(
+                    rows=[
+                        {
+                            "state": self.state,
+                            "dataflow_uid": receipt["dataflow_uid"],
+                            "result": self.result,
+                            "result_digest": self.digest,
+                            "lease_expires_at": None,
+                            "expires_at": None,
+                        }
+                    ]
+                )
+            if "complete_dataflow_create" in sql and self.state == "creating":
+                self.state = "completed"
+                self.result = json.loads(values["result"])
+                self.digest = values["result_digest"]
+                return FakeResult(rows=[{"result": self.result}])
+            return FakeResult()
+
+        def commit(self):
+            return None
 
     session = AtomicSession()
     repository = DataRuleRepository(session)
-    with ThreadPoolExecutor(max_workers=2) as pool:
-        outcomes = list(
-            pool.map(
-                lambda _index: _consume_outcome(
-                    repository, receipt, actor
-                ),
-                range(2),
-            )
-        )
-    assert sorted(outcomes) == ["accepted", "rejected"]
+    claim = repository.begin_dataflow_create(receipt, actor_uid=actor)
+    assert claim["status"] == "claimed"
+    result = {"id": 31, "uid": receipt["dataflow_uid"]}
+    assert repository.complete_dataflow_create(
+        reservation_id=receipt["reservation_id"],
+        dataflow_uid=receipt["dataflow_uid"],
+        lease_token=claim["lease_token"],
+        dataflow_node_id=31,
+        result=result,
+    ) == result
+    assert repository.begin_dataflow_create(
+        receipt, actor_uid=actor
+    ) == {
+        "status": "completed",
+        "dataflow_uid": receipt["dataflow_uid"],
+        "result": result,
+    }
     sql = session.calls[0][0]
     assert "consumed_at IS NULL" in sql
-    assert "expires_at > CURRENT_TIMESTAMP" in sql
+    assert "lease_expires_at <= CURRENT_TIMESTAMP" in sql
     assert "actor_uid = CAST(:actor_uid AS uuid)" in sql
     assert "nonce_hash = :nonce_hash" in sql
 
-    with pytest.raises(ValueError):
-        DataRuleRepository(AtomicSession()).consume_dataflow_draft(
-            receipt, actor_uid=other_actor
-        )
-    with pytest.raises(ValueError):
-        DataRuleRepository(AtomicSession(expired=True)).consume_dataflow_draft(
-            receipt, actor_uid=actor
-        )
-    forged = {**receipt, "nonce": "forged"}
-    with pytest.raises(ValueError):
-        DataRuleRepository(AtomicSession()).consume_dataflow_draft(
-            forged, actor_uid=actor
-        )
-
-
-def _consume_outcome(repository, receipt, actor):
-    try:
-        repository.consume_dataflow_draft(receipt, actor_uid=actor)
-        return "accepted"
-    except ValueError:
-        return "rejected"
-
 
 def test_create_rule_version_rejects_legacy_v1_rule_specs():
     from app.core.data_rules.repository import DataRuleRepository
@@ -564,6 +551,10 @@ def test_catalog_compatibility_uses_exact_flow_refs_and_exact_version_hydration(
         "owner_uid": None,
         "status": "published",
         "schema_compatibility": spec,
+        "logical_plan_id": new_governance_uid(),
+        "schema_hashes": {"input": "a" * 64, "output": "b" * 64},
+        "bound_rules": [],
+        "binding_count": 1,
         "impact_count": 0,
         "backend": "polars_batch",
         "compile_evidence_id": new_governance_uid(),
@@ -576,31 +567,145 @@ def test_catalog_compatibility_uses_exact_flow_refs_and_exact_version_hydration(
         FakeSession(unified_catalog_rows=[row])
     )
 
+    class Resolver:
+        hashes = {
+            spec["input_schema_ref"]: "a" * 64,
+            spec["output_schema_ref"]: "b" * 64,
+        }
+
+        def resolve(self, schema_ref):
+            return {
+                "id": new_governance_uid(),
+                "schema_ref": schema_ref,
+                "schema_hash": self.hashes[schema_ref],
+                "source_revision": "neo4j:trusted:v1",
+            }
+
+    resolver = Resolver()
+
     compatible = repository.get_published_asset(
         asset_type="rule",
         version_id=version_id,
         input_schema_refs=[spec["input_schema_ref"]],
         output_schema_ref=spec["output_schema_ref"],
+        schema_resolver=resolver,
     )
     assert compatible["version_id"] == version_id
     assert compatible["schema_compatibility"]["status"] == "compatible"
     assert (
         compatible["schema_compatibility"]["reason"]
-        == "exact_schema_refs_match"
+        == "trusted_schema_hashes_match"
     )
-    assert compatible["schema_compatibility"]["evidence"][
-        "asset_input_schema_ref"
-    ] == spec["input_schema_ref"]
+    assert compatible["schema_compatibility"]["evidence"]["input"][
+        "schema_hash"
+    ] == "a" * 64
 
     incompatible = repository.search_published_assets(
         asset_type="rule",
         input_schema_refs=["bd:other:v1"],
         output_schema_ref=spec["output_schema_ref"],
+        schema_resolver=resolver,
     )
     assert (
         incompatible["items"][0]["schema_compatibility"]["status"]
         == "incompatible"
     )
+    resolver.hashes[spec["input_schema_ref"]] = "c" * 64
+    drifted = repository.search_published_assets(
+        asset_type="rule",
+        input_schema_refs=[spec["input_schema_ref"]],
+        output_schema_ref=spec["output_schema_ref"],
+        schema_resolver=resolver,
+    )
+    assert drifted["items"][0]["schema_compatibility"]["status"] == "incompatible"
+    assert (
+        drifted["items"][0]["schema_compatibility"]["reason"]
+        == "schema_ref_or_hash_drift"
+    )
+
+
+def test_standard_catalog_expands_every_fixed_rule_and_detects_bound_rule_drift():
+    from app.core.data_rules.repository import DataRuleRepository
+
+    spec = valid_rule_spec()
+    standard_version_id = new_governance_uid()
+    row = {
+        "asset_type": "standard",
+        "version_id": standard_version_id,
+        "asset_uid": new_governance_uid(),
+        "name": "客户标准",
+        "version_no": 1,
+        "owner_uid": None,
+        "status": "published",
+        "schema_compatibility": {"schema_ref": spec["input_schema_ref"]},
+        "logical_plan_id": None,
+        "schema_hashes": None,
+        "bound_rules": [
+            {
+                "rule_version_id": new_governance_uid(),
+                "rule_spec": spec,
+                "logical_plan_id": new_governance_uid(),
+                "schema_hashes": {
+                    "input": "a" * 64,
+                    "output": "b" * 64,
+                },
+                "compile_evidence_id": new_governance_uid(),
+                "test_evidence_id": new_governance_uid(),
+            }
+        ],
+        "binding_count": 1,
+        "impact_count": 0,
+        "backend": "composite",
+        "compile_evidence_id": None,
+        "compile_status": "success",
+        "test_evidence_id": None,
+        "test_status": "success",
+        "total_count": 1,
+    }
+
+    class Resolver:
+        output_hash = "b" * 64
+
+        def resolve(self, schema_ref):
+            return {
+                "id": new_governance_uid(),
+                "schema_ref": schema_ref,
+                "schema_hash": (
+                    "a" * 64
+                    if schema_ref == spec["input_schema_ref"]
+                    else self.output_hash
+                ),
+                "source_revision": "neo4j:trusted:v1",
+            }
+
+    repository = DataRuleRepository(FakeSession(unified_catalog_rows=[row]))
+    resolver = Resolver()
+    matching = repository.search_published_assets(
+        asset_type="standard",
+        input_schema_refs=[spec["input_schema_ref"]],
+        output_schema_ref=spec["output_schema_ref"],
+        schema_resolver=resolver,
+    )
+    assert matching["items"][0]["schema_compatibility"]["status"] == "compatible"
+    evidence = matching["items"][0]["schema_compatibility"]["evidence"]
+    assert set(evidence["bound_rules"][0]) == {
+        "rule_version_id",
+        "logical_plan_id",
+        "compile_evidence_id",
+        "test_evidence_id",
+        "input",
+        "output",
+        "published_schema_hashes",
+    }
+
+    resolver.output_hash = "c" * 64
+    drifted = repository.search_published_assets(
+        asset_type="standard",
+        input_schema_refs=[spec["input_schema_ref"]],
+        output_schema_ref=spec["output_schema_ref"],
+        schema_resolver=resolver,
+    )
+    assert drifted["items"][0]["schema_compatibility"]["status"] == "incompatible"
 
 
 @pytest.mark.parametrize(

tests/integration/test_dataflow_create_saga_neo4j.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import os
+
+import pytest
+from neo4j import GraphDatabase
+
+from app import create_app
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_flow.dataflows import DataFlowService
+
+
+def test_real_neo4j_uid_constraint_merge_replay_and_conflict():
+    uri = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_URI")
+    password = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_PASSWORD")
+    if not uri or not password:
+        pytest.skip("real Neo4j acceptance connection is not configured")
+    user = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_USER", "neo4j")
+    app = create_app()
+    app.config.update(
+        TESTING=True,
+        NEO4J_URI=uri,
+        NEO4J_USER=user,
+        NEO4J_PASSWORD=password,
+        NEO4J_ENCRYPTED=False,
+    )
+    uid = new_governance_uid()
+    node = {
+        "uid": uid,
+        "name_zh": f"Saga验收-{uid}",
+        "name_en": f"saga-{uid}",
+        "script_type": "governed",
+        "script_requirement": '{"dataflow_spec":{"schema_version":"2.0"}}',
+        "script_path": "",
+    }
+    driver = GraphDatabase.driver(uri, auth=(user, password), encrypted=False)
+    try:
+        with app.app_context():
+            first_id, first = DataFlowService._merge_governed_dataflow(node)
+            second_id, second = DataFlowService._merge_governed_dataflow(node)
+            assert first_id == second_id
+            assert first == second
+            with pytest.raises(ValueError, match="dataflow_uid_conflict"):
+                DataFlowService._merge_governed_dataflow(
+                    {**node, "name_zh": f"篡改-{uid}"}
+                )
+        with driver.session() as session:
+            count = session.run(
+                "MATCH (n:DataFlow {uid: $uid}) RETURN count(n) AS count",
+                {"uid": uid},
+            ).single()["count"]
+            constraints = [
+                record["name"]
+                for record in session.run(
+                    "SHOW CONSTRAINTS YIELD name WHERE name = 'data_flow_uid' "
+                    "RETURN name"
+                )
+            ]
+        assert count == 1
+        assert constraints == ["data_flow_uid"]
+    finally:
+        with driver.session() as session:
+            session.run("MATCH (n:DataFlow {uid: $uid}) DETACH DELETE n", {"uid": uid})
+        driver.close()

tests/integration/test_dataflow_draft_reservation_postgres.py

@@ -12,24 +12,24 @@ from app.core.common.identifiers import new_governance_uid
 from app.core.data_rules.repository import DataRuleRepository
 
 
-def _consume(url, receipt, actor):
+def _claim(url, receipt, actor):
     engine = create_engine(url)
     try:
         with Session(engine) as session:
             try:
-                DataRuleRepository(session).consume_dataflow_draft(
+                value = DataRuleRepository(session).begin_dataflow_create(
                     receipt, actor_uid=actor
                 )
                 session.commit()
-                return "accepted"
-            except ValueError:
+                return value
+            except ValueError as exc:
                 session.rollback()
-                return "rejected"
+                return {"status": str(exc)}
     finally:
         engine.dispose()
 
 
-def test_real_postgres_reservation_fk_expiry_and_concurrent_consume():
+def test_real_postgres_saga_fk_expiry_concurrent_claim_replay_and_lease_recovery():
     url = os.environ.get("DATA_RULE_POSTGRES_ACCEPTANCE_URL")
     if not url:
         pytest.skip("real PostgreSQL acceptance URL is not configured")
@@ -43,12 +43,9 @@ def test_real_postgres_reservation_fk_expiry_and_concurrent_consume():
                     "INSERT INTO public.users "
                     "(id, username, display_name, password_hash, status) "
                     "VALUES (CAST(:id AS uuid), :username, "
-                    "'Reservation Acceptance', 'not-a-login-hash', 'active')"
+                    "'Saga Acceptance', 'not-a-login-hash', 'active')"
                 ),
-                {
-                    "id": actor,
-                    "username": f"reservation-{actor}",
-                },
+                {"id": actor, "username": f"saga-{actor}"},
             )
             session.commit()
             receipt = DataRuleRepository(session).reserve_dataflow_draft(
@@ -57,19 +54,35 @@ def test_real_postgres_reservation_fk_expiry_and_concurrent_consume():
             created_ids.append(receipt["reservation_id"])
             session.commit()
 
-        closed_receipt = {
+        closed = {
             key: receipt[key]
             for key in ("reservation_id", "dataflow_uid", "nonce")
         }
         with ThreadPoolExecutor(max_workers=2) as pool:
             outcomes = list(
-                pool.map(
-                    lambda _index: _consume(url, closed_receipt, actor),
-                    range(2),
-                )
+                pool.map(lambda _index: _claim(url, closed, actor), range(2))
             )
-        assert sorted(outcomes) == ["accepted", "rejected"]
-        assert _consume(url, closed_receipt, new_governance_uid()) == "rejected"
+        claimed = [value for value in outcomes if value["status"] == "claimed"]
+        assert len(claimed) == 1
+        assert sorted(value["status"] for value in outcomes) == [
+            "claimed",
+            "dataflow_create_in_progress",
+        ]
+
+        result = {"id": 73, "uid": receipt["dataflow_uid"]}
+        with Session(engine) as session:
+            repository = DataRuleRepository(session)
+            assert repository.complete_dataflow_create(
+                reservation_id=receipt["reservation_id"],
+                dataflow_uid=receipt["dataflow_uid"],
+                lease_token=claimed[0]["lease_token"],
+                dataflow_node_id=73,
+                result=result,
+            ) == result
+            session.commit()
+        replay = _claim(url, closed, actor)
+        assert replay["status"] == "completed"
+        assert replay["result"] == result
 
         with Session(engine) as session:
             expired = DataRuleRepository(session).reserve_dataflow_draft(
@@ -86,11 +99,49 @@ def test_real_postgres_reservation_fk_expiry_and_concurrent_consume():
                 {"id": expired["reservation_id"]},
             )
             session.commit()
-        expired_receipt = {
+        expired_closed = {
             key: expired[key]
             for key in ("reservation_id", "dataflow_uid", "nonce")
         }
-        assert _consume(url, expired_receipt, actor) == "rejected"
+        assert _claim(url, expired_closed, actor)["status"] == (
+            "draft_reservation_refresh_required"
+        )
+
+        with Session(engine) as session:
+            recoverable = DataRuleRepository(session).reserve_dataflow_draft(
+                actor_uid=actor
+            )
+            created_ids.append(recoverable["reservation_id"])
+            session.commit()
+            first = DataRuleRepository(session).begin_dataflow_create(
+                {
+                    key: recoverable[key]
+                    for key in ("reservation_id", "dataflow_uid", "nonce")
+                },
+                actor_uid=actor,
+                lease_seconds=10,
+            )
+            session.commit()
+            session.execute(
+                text(
+                    "UPDATE public.dataflow_draft_reservations "
+                    "SET lease_expires_at = CURRENT_TIMESTAMP - INTERVAL '1 second' "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": recoverable["reservation_id"]},
+            )
+            session.commit()
+        recovered = _claim(
+            url,
+            {
+                key: recoverable[key]
+                for key in ("reservation_id", "dataflow_uid", "nonce")
+            },
+            actor,
+        )
+        assert first["attempt"] == 1
+        assert recovered["status"] == "claimed"
+        assert recovered["attempt"] == 2
 
         with Session(engine) as session:
             with pytest.raises(IntegrityError):

tests/test_data_rule_api.py

@@ -632,15 +632,12 @@ def test_catalog_passes_flow_context_and_hydrates_exact_off_page_version(
     assert exact.status_code == 200
     assert repository.calls[0][1]["input_schema_refs"] == inputs
     assert repository.calls[0][1]["output_schema_ref"] == output
-    assert repository.calls[1] == (
-        "get_published_asset",
-        {
-            "asset_type": "rule",
-            "version_id": version_id,
-            "input_schema_refs": inputs,
-            "output_schema_ref": output,
-        },
-    )
+    assert repository.calls[1][0] == "get_published_asset"
+    assert repository.calls[1][1]["asset_type"] == "rule"
+    assert repository.calls[1][1]["version_id"] == version_id
+    assert repository.calls[1][1]["input_schema_refs"] == inputs
+    assert repository.calls[1][1]["output_schema_ref"] == output
+    assert repository.calls[1][1]["schema_resolver"] is not None
 
 
 def test_catalog_and_evidence_queries_are_closed_bounded_and_rules_read_only(

tests/test_data_rule_frontend_contract.py

@@ -216,6 +216,22 @@ def test_catalog_never_guesses_schema_compatibility_from_arbitrary_objects():
     assert "outputSchemaRef" in source
 
 
+def test_production_line_editor_renews_expiring_receipts_and_retries_only_safe_errors():
+    source = DATAFLOW.read_text(encoding="utf-8")
+    request = Path("frontend/src/utils/request.js").read_text(encoding="utf-8")
+
+    assert "governedDraftExpiresAt" in source
+    assert "data.expires_at" in source
+    assert "expiresAt - Date.now() <= 60000" in source
+    assert "ensureFreshDraftReservation" in source
+    assert "draft_reservation_refresh_required" in source
+    assert "governedDraftReplayRequired" in source
+    assert "if (this.governedDraftReplayRequired) return" in source
+    assert "await this.ensureDataflowUid(true)" in source
+    assert source.count("await api.addDataFlow(payload)") == 2
+    assert "error.response?.data" in request
+
+
 def test_new_dataflow_gets_server_governed_uid_before_assembly():
     api = API.read_text(encoding="utf-8")
     dataflow = DATAFLOW.read_text(encoding="utf-8")

tests/test_dataflow_create_saga.py

@@ -0,0 +1,245 @@
+from __future__ import annotations
+
+import copy
+
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_flow.dataflows import DataFlowService
+
+
+class GraphResult:
+    def __init__(self, record=None):
+        self.record = record
+
+    def single(self):
+        return self.record
+
+
+class GraphSession:
+    def __init__(self, *, existing=None, conflict=False):
+        self.existing = existing
+        self.conflict = conflict
+        self.calls = []
+
+    def run(self, query, parameters=None, **kwargs):
+        values = parameters or kwargs
+        self.calls.append((query, values))
+        if query.startswith("CREATE CONSTRAINT"):
+            return GraphResult()
+        if "WHERE n.uid IS NULL OR n.uid <> $uid" in query:
+            return GraphResult({"uid": "other"}) if self.conflict else GraphResult()
+        if query.startswith("MERGE"):
+            node = copy.deepcopy(self.existing or values["properties"])
+            self.existing = node
+            return GraphResult({"n": node, "node_id": 73})
+        raise AssertionError(query)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return None
+
+
+class GraphDriver:
+    def __init__(self, session):
+        self.graph_session = session
+
+    def session(self):
+        return self.graph_session
+
+    def close(self):
+        return None
+
+
+def governed_node():
+    return {
+        "uid": new_governance_uid(),
+        "name_zh": "客户治理生产线",
+        "name_en": "customer_line",
+        "script_type": "governed",
+        "script_requirement": '{"dataflow_spec":{"schema_version":"2.0"}}',
+        "script_path": "",
+    }
+
+
+def test_governed_graph_create_installs_constraint_and_reconciles_same_uid(
+    monkeypatch,
+):
+    node = governed_node()
+    graph = GraphSession()
+    monkeypatch.setattr(
+        "app.core.data_flow.dataflows.connect_graph",
+        lambda: GraphDriver(graph),
+    )
+
+    first_id, first = DataFlowService._merge_governed_dataflow(node)
+    second_id, second = DataFlowService._merge_governed_dataflow(node)
+
+    assert first_id == second_id == 73
+    assert first == second
+    assert any(
+        call[0]
+        == "CREATE CONSTRAINT data_flow_uid IF NOT EXISTS "
+        "FOR (n:DataFlow) REQUIRE n.uid IS UNIQUE"
+        for call in graph.calls
+    )
+    assert sum(call[0].startswith("MERGE") for call in graph.calls) == 2
+
+
+def test_governed_graph_reconcile_fails_closed_on_uid_or_name_conflict(
+    monkeypatch,
+):
+    node = governed_node()
+    graph = GraphSession(existing={**node, "name_zh": "被篡改的名称"})
+    monkeypatch.setattr(
+        "app.core.data_flow.dataflows.connect_graph",
+        lambda: GraphDriver(graph),
+    )
+    with pytest.raises(ValueError, match="dataflow_uid_conflict"):
+        DataFlowService._merge_governed_dataflow(node)
+
+    name_conflict = GraphSession(conflict=True)
+    monkeypatch.setattr(
+        "app.core.data_flow.dataflows.connect_graph",
+        lambda: GraphDriver(name_conflict),
+    )
+    with pytest.raises(ValueError, match="dataflow_uid_conflict"):
+        DataFlowService._merge_governed_dataflow(node)
+
+
+def test_completed_saga_replays_result_without_another_neo4j_write(monkeypatch):
+    expected = {"id": 73, **governed_node()}
+
+    class Repository:
+        def load_published_assets(self, _flow):
+            return {}, {}
+
+        def begin_dataflow_create(self, _receipt, *, actor_uid):
+            assert actor_uid
+            return {
+                "status": "completed",
+                "dataflow_uid": expected["uid"],
+                "result": expected,
+            }
+
+    flow = {
+        "schema_version": "2.0",
+        "dataflow_uid": expected["uid"],
+        "name": "客户治理生产线",
+        "input_schema_refs": ["bd:customer:v1"],
+        "output_schema_ref": "bd:customer_clean:v1",
+        "components": [],
+    }
+    # Use the project's validator fixture shape through the repository-level
+    # contract; the replay must happen before any graph write.
+    from tests.core.data_rules.test_contracts import valid_dataflow_spec
+
+    flow = valid_dataflow_spec()
+    flow["dataflow_uid"] = expected["uid"]
+    envelope = {
+        "dataflow_spec": flow,
+        "dataset_edges": {
+            "source_table": flow["input_schema_refs"],
+            "target_table": flow["output_schema_ref"],
+        },
+        "migration_metadata": {
+            "status": "migrated",
+            "legacy_fields_present": False,
+            "preserved_for_read_only": True,
+            "governed_semantics": "dataflow_spec",
+        },
+    }
+    monkeypatch.setattr(
+        DataFlowService,
+        "_merge_governed_dataflow",
+        lambda _node: pytest.fail("Neo4j was called during completed replay"),
+    )
+
+    result = DataFlowService.create_dataflow(
+        {
+            "name_zh": "客户治理生产线",
+            "describe": "响应丢失重试",
+            "script_type": "governed",
+            "script_requirement": envelope,
+            "draft_reservation": {
+                "reservation_id": new_governance_uid(),
+                "dataflow_uid": expected["uid"],
+                "nonce": "response-loss",
+            },
+        },
+        repository=Repository(),
+        actor_uid=new_governance_uid(),
+    )
+    assert result == expected
+
+
+def test_graph_failure_is_persisted_but_finalize_failure_leaves_reconcilable_lease(
+    monkeypatch,
+):
+    from tests.core.data_rules.test_contracts import valid_dataflow_spec
+    from tests.test_legacy_governance_cutover import PublishedAssetRepository
+
+    flow = valid_dataflow_spec()
+    actor = new_governance_uid()
+
+    def payload(repository):
+        receipt = repository.reserve_dataflow_draft(actor_uid=actor)
+        receipt["dataflow_uid"] = flow["dataflow_uid"]
+        return {
+            "name_zh": "故障注入生产线",
+            "describe": "故障注入",
+            "script_type": "governed",
+            "script_requirement": {
+                "dataflow_spec": flow,
+                "dataset_edges": {
+                    "source_table": flow["input_schema_refs"],
+                    "target_table": flow["output_schema_ref"],
+                },
+                "migration_metadata": {
+                    "status": "migrated",
+                    "legacy_fields_present": False,
+                    "preserved_for_read_only": True,
+                    "governed_semantics": "dataflow_spec",
+                },
+            },
+            "draft_reservation": {
+                key: receipt[key]
+                for key in ("reservation_id", "dataflow_uid", "nonce")
+            },
+        }
+
+    graph_failure = PublishedAssetRepository()
+    failures = []
+    graph_failure.commit_dataflow_create_failure = (
+        lambda **kwargs: failures.append(kwargs)
+    )
+    monkeypatch.setattr(
+        DataFlowService,
+        "_merge_governed_dataflow",
+        lambda _node: (_ for _ in ()).throw(RuntimeError("neo4j unavailable")),
+    )
+    with pytest.raises(RuntimeError, match="neo4j unavailable"):
+        DataFlowService.create_dataflow(
+            payload(graph_failure), repository=graph_failure, actor_uid=actor
+        )
+    assert failures[0]["error_code"] == "neo4j_create_failed"
+
+    finalize_failure = PublishedAssetRepository()
+    monkeypatch.setattr(
+        DataFlowService,
+        "_merge_governed_dataflow",
+        lambda node: (73, {**node, "id": 73}),
+    )
+    finalize_failure.complete_dataflow_create = (
+        lambda **_kwargs: (_ for _ in ()).throw(
+            RuntimeError("postgres finalize failed")
+        )
+    )
+    with pytest.raises(RuntimeError, match="postgres finalize failed"):
+        DataFlowService.create_dataflow(
+            payload(finalize_failure),
+            repository=finalize_failure,
+            actor_uid=actor,
+        )

tests/test_legacy_governance_cutover.py

@@ -17,6 +17,7 @@ class PublishedAssetRepository:
         self.rule_calls = []
         self.dataflow_calls = []
         self.consumed = set()
+        self.completed = {}
 
     def require_published_rule_version(self, rule_version_id):
         self.rule_calls.append(rule_version_id)
@@ -38,12 +39,32 @@ class PublishedAssetRepository:
             "expires_at": "2099-01-01T00:00:00+00:00",
         }
 
-    def consume_dataflow_draft(self, receipt, *, actor_uid):
+    def begin_dataflow_create(self, receipt, *, actor_uid):
         key = (receipt["reservation_id"], actor_uid)
-        if key in self.consumed:
-            raise ValueError("already used")
+        if key in self.completed:
+            return {
+                "status": "completed",
+                "dataflow_uid": receipt["dataflow_uid"],
+                "result": self.completed[key],
+            }
         self.consumed.add(key)
-        return receipt["dataflow_uid"]
+        return {
+            "status": "claimed",
+            "dataflow_uid": receipt["dataflow_uid"],
+            "lease_token": new_governance_uid(),
+            "attempt": 1,
+        }
+
+    def commit_dataflow_create_claim(self):
+        return None
+
+    def commit_dataflow_create_failure(self, **_kwargs):
+        return None
+
+    def complete_dataflow_create(self, *, reservation_id, result, **_kwargs):
+        key = next(key for key in self.consumed if key[0] == reservation_id)
+        self.completed[key] = dict(result)
+        return dict(result)
 
 
 def _headers(app, role="editor"):
@@ -319,39 +340,20 @@ def test_governed_dataflow_creation_never_generates_legacy_task_or_workflow(
         key: receipt[key]
         for key in ("reservation_id", "dataflow_uid", "nonce")
     }
-    created = {}
-
-    class Result:
-        def single(self):
-            return None
-
-    class Session:
-        def run(self, *_args, **_kwargs):
-            return Result()
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, *_args):
-            return None
-
-    class Driver:
-        def session(self):
-            return Session()
-
     monkeypatch.setattr(
         "app.core.data_flow.dataflows.translate_and_parse",
         lambda _name: ["customer_governed_line"],
     )
+    created = {}
     monkeypatch.setattr(
-        "app.core.data_flow.dataflows.get_node", lambda *_args, **_kwargs: None
-    )
-    monkeypatch.setattr(
-        "app.core.data_flow.dataflows.create_or_get_node",
-        lambda _label, **properties: created.update(properties) or 31,
-    )
-    monkeypatch.setattr(
-        "app.core.data_flow.dataflows.connect_graph", lambda: Driver()
+        DataFlowService,
+        "_merge_governed_dataflow",
+        lambda properties: (
+            31,
+            {**created, **properties, "id": 31}
+            if not created.update(properties)
+            else {},
+        ),
     )
     monkeypatch.setattr(
         DataFlowService,