fix: close governed rule assembly cutover gaps

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

@@ -0,0 +1,184 @@
+# Task 8 Report — Standard / Data Flow product convergence
+
+Date: 2026-07-24
+Branch: `codex/data-rule-execution-m3a-m5`
+
+## Outcome
+
+Task 8 removes the user-facing parallel rule implementations from Data
+Standard and Data Flow authoring.
+
+The governed path is now:
+
+1. describe a constraint in natural language against fixed input/output
+   schema context;
+2. review AI assumptions, ambiguities, model/prompt provenance and the signed
+   generation receipt;
+3. create an immutable draft RuleVersion;
+4. run server-owned logical compilation and isolated sample testing;
+5. publish only after trusted evidence succeeds;
+6. select a published fixed RuleVersion or StandardVersion from one catalog;
+7. assemble those fixed version IDs into ordered production-line stations;
+8. hand the resulting DataFlowSpec to the release/Data Factory boundary.
+
+Generated source code is never editable in the new path. Legacy code and
+free-text flow rules remain visible only in explicit migration sections and
+are excluded from new execution semantics.
+
+## Trusted catalog and evidence API
+
+The control plane now exposes:
+
+- `GET /api/rules/catalog`
+  - closed `query`, `asset_type`, `limit`, and `offset` parameters;
+  - rules and standards in one paginated contract;
+  - human-readable name, stable asset UID, fixed version, owner, status,
+    schema binding/scope, impact count, compiler backend, and latest
+    compile/test evidence;
+  - only published rules with an exact published logical plan, matching
+    compile evidence, matching test evidence, and a publication audit;
+  - only published standards whose fixed rule bindings are still published.
+- `GET /api/rules/catalog/assets/<rule|standard>/<version_id>/evidence`
+  - safe generation, logical compilation, dry-run, publication and physical
+    plan stages;
+  - IDs, hashes, lifecycle states, test kinds, and run IDs only;
+  - no source text, generated candidate body, artifacts, samples, or
+    unbounded evidence documents.
+
+Compatibility aliases remain available at
+`/api/rules/catalog/rule-versions` and
+`/api/rules/rule-versions/<version_id>/evidence`. Global API policy keeps all
+GET paths behind `rules:read`.
+
+## AI authoring lifecycle
+
+`RuleAuthoringPanel` now:
+
+- blocks interpretation until input and output context are present;
+- displays clarification requirements rather than guessing;
+- shows assumptions, unresolved ambiguities, confidence, model/provider,
+  prompt revision, repair count, and generation receipt status;
+- creates a draft with the signed generation receipt;
+- exposes explicit compile, isolated test and publish actions;
+- enforces `rules:edit` and `rules:publish` UI gates;
+- displays the trusted evidence timeline;
+- never renders or edits generated code or a serialized candidate body.
+
+## Data Standard convergence
+
+Data Standard editing now links a published RuleVersion through
+`RuleCatalogPicker`. New saves contain the fixed `rule_version_id` and bounded
+`migration_metadata`; they do not require or write an operation-code field.
+
+Existing operation code is shown in a read-only migration area with clear
+`unmigrated` / `linked` state and a path back to natural-language
+re-authoring. It cannot become execution semantics merely by saving the
+legacy record.
+
+## Data Flow production-line assembly
+
+Data Flow no longer edits free-text rules or embeds rule definitions.
+`ProductionLineAssembler` models the flow as ordered factory stations:
+
+- `standard.enforce` fixes a `standard_version_id`;
+- `rule.apply` and `quality.check` fix a `rule_version_id`;
+- input/output selections are shown as dataset edges;
+- schema-compatibility and latest-evidence summaries are shown per station;
+- missing UID, dataset edges, fixed assets, compatibility, or evidence become
+  explicit release-readiness blockers;
+- reorder, remove and keyboard-accessible catalog selection are supported.
+
+The new saved `script_requirement` envelope contains only:
+
+- `dataflow_spec`;
+- `dataset_edges`;
+- bounded `migration_metadata`.
+
+It contains no inline rule definition or generated source. Existing legacy
+requirements remain read-only and are not copied into that envelope.
+
+## UX and accessibility
+
+The implementation reuses Vue 2, Vuetify, Material Design Icons, and the
+existing request client. No dependency or design system was added.
+
+The interface uses a low-motion, medium/high-density governance-workbench
+style. Catalog loading, empty and error states are explicit. Catalog items
+support focus, arrow navigation, Enter/Space selection, listbox semantics,
+ARIA labels, and skeleton loading. Responsive layout collapses the Data Flow
+dataset grid on narrow screens.
+
+## Acceptance evidence
+
+Fail-first:
+
+- the initial frontend contract run produced `5 failed, 3 passed` because the
+  three governed components and convergence behavior did not exist.
+
+Final:
+
+- Task 8 API/repository/frontend and legacy-cutover contracts:
+  `61 passed, 59 subtests passed`;
+- full repository suite:
+  `659 passed, 29 skipped, 59 subtests passed`;
+- changed Python Ruff: `All checks passed!`;
+- `git diff --check`: passed;
+- frontend production build: completed with `0 errors`;
+- build retained 20 pre-existing `no-console` warnings plus existing CSS
+  ordering and bundle-size warnings;
+- unified catalog SQL executed against the migrated local PostgreSQL service
+  and returned a valid empty page;
+- rebuilt local Docker backend and frontend images successfully;
+- backend, frontend, PostgreSQL, Neo4j, MinIO and Runner containers reported
+  healthy;
+- `/api/system/health` returned application code `200`;
+- unauthenticated `/api/rules/catalog` returned HTTP `401`.
+
+Browser automation opened the rebuilt deployment and verified the local login
+surface. The documented local sample administrator password did not match the
+persisted administrator in this long-lived Docker volume, so authenticated
+Standard/Data Flow navigation was not forced by overwriting or resetting
+credentials. The build, component contracts, authenticated API tests, real
+PostgreSQL query, and container health checks provide the Task 8 acceptance
+evidence; an authenticated visual walkthrough can be repeated during the
+phase-level M4 acceptance with the environment owner’s current local
+credential.
+
+## Review closeout
+
+The post-implementation review found eight convergence gaps. All eight are
+closed:
+
+1. idempotency is emitted only for `rule.apply`; `standard.enforce` and
+   `quality.check` now match the closed DataFlow contract;
+2. catalog compatibility is never inferred from an arbitrary non-empty
+   object; absent production-line context remains explicitly `unknown`;
+3. `POST /api/interface/data/standard/code` is closed with HTTP 410 and
+   read-only-migration semantics;
+4. Standard add/update rejects executable code and server-attests any
+   `rule_version_id` against the complete trusted publication evidence chain;
+5. DataFlow drafts receive a server-owned UUIDv7 before assembly and saving
+   fails closed if no identity can be obtained;
+6. governed DataFlow create/update validates a closed envelope, verifies all
+   referenced versions are published, preserves the immutable identity, and
+   never invokes task-list, generated-script, n8n-workflow, or legacy data
+   product side effects;
+7. Standard authoring no longer fabricates a client-side `validated` state;
+   new Standard clauses must bind a server-published RuleVersion;
+8. generation evidence state `ready` is rendered as successful and async
+   station values are reloaded by the existing deep watcher.
+
+The cutover deliberately preserves legacy non-governed records and
+read-only migration displays, while preventing those fields from becoming
+new governed execution semantics.
+
+## Residual scope
+
+- Production-line cross-station compatibility remains ultimately authoritative
+  at server-side release resolution; the UI presents current bound schema
+  evidence and blocks on missing evidence rather than claiming runtime success.
+- Data Factory activation, canary, rollback, deployment permissions and
+  environment operations remain Task 9 scope.
+- The repository has no published catalog fixtures in the current persisted
+  local database, so catalog empty state was the live deployment state during
+  this acceptance.

app/api/data_flow/routes.py

@@ -1,21 +1,34 @@
 import json
 import logging
 
-from flask import g, jsonify, request
+from flask import current_app, g, jsonify, request
 
 from app import db
 from app.api.data_flow import bp
-from app.core.data_flow.dataflows import DataFlowService
-from app.core.graph.graph_operations import MyEncoder
-from app.models.result import failed, success
 from app.core.data_factory.n8n_client import N8nClient, N8nClientError
+from app.core.data_flow.dataflows import DataFlowService
 from app.core.data_flow.workflow_activation import activate_version
 from app.core.data_flow.workflow_repository import create_version, list_versions
-from app.core.system.permissions import ACTIVATE_WORKFLOW, EDIT_GOVERNANCE, READ_GOVERNANCE, require_permissions
+from app.core.data_rules.repository import DataRuleRepository
+from app.core.graph.graph_operations import MyEncoder
+from app.core.system.permissions import (
+    ACTIVATE_WORKFLOW,
+    EDIT_GOVERNANCE,
+    READ_GOVERNANCE,
+    require_permissions,
+)
+from app.models.result import failed, success
 
 logger = logging.getLogger(__name__)
 
 
+def _repository():
+    configured = current_app.extensions.get("data_rule_repository")
+    if configured is not None:
+        return configured
+    return DataRuleRepository(db.session)
+
+
 @bp.route("/<dataflow_uid>/workflow-versions", methods=["GET"])
 @require_permissions(READ_GOVERNANCE)
 def get_workflow_versions(dataflow_uid):
@@ -114,15 +127,17 @@ def create_dataflow():
         data = request.get_json()
         if not data:
             res = failed("请求数据不能为空", code=400)
-            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
+            return jsonify(res), 400
 
-        result = DataFlowService.create_dataflow(data)
+        result = DataFlowService.create_dataflow(
+            data, repository=_repository()
+        )
         res = success(result, "数据流创建成功")
         return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
     except ValueError as ve:
         logger.error(f"创建数据流参数错误: {str(ve)}")
         res = failed(f"参数错误: {str(ve)}", code=400)
-        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
+        return jsonify(res), 400
     except Exception as e:
         logger.error(f"创建数据流失败: {str(e)}")
         res = failed(f"创建数据流失败: {str(e)}")
@@ -136,15 +151,21 @@ def update_dataflow(dataflow_id):
         data = request.get_json()
         if not data:
             res = failed("请求数据不能为空", code=400)
-            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
+            return jsonify(res), 400
 
-        result = DataFlowService.update_dataflow(dataflow_id, data)
+        result = DataFlowService.update_dataflow(
+            dataflow_id, data, repository=_repository()
+        )
         if result:
             res = success(result, "数据流更新成功")
             return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
         else:
             res = failed("数据流不存在", code=404)
             return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
+    except ValueError as ve:
+        logger.error(f"更新数据流参数错误: {str(ve)}")
+        res = failed(f"参数错误: {str(ve)}", code=400)
+        return jsonify(res), 400
     except Exception as e:
         logger.error(f"更新数据流失败: {str(e)}")
         res = failed(f"更新数据流失败: {str(e)}")

app/api/data_interface/routes.py

@@ -1,23 +1,46 @@
 import json
 
-from flask import Response, jsonify, request
+from flask import Response, current_app, jsonify, request
 
+from app import db
 from app.api.data_interface import bp
 from app.core.data_interface import interface
+from app.core.data_rules.repository import DataRuleRepository
 from app.core.graph.graph_operations import (
     MyEncoder,
     connect_graph,
     create_or_get_node,
 )
-from app.core.llm import code_generate_standard
 from app.core.meta_data import get_formatted_time, translate_and_parse
 from app.models.result import failed, success
 
 
+def _repository():
+    configured = current_app.extensions.get("data_rule_repository")
+    if configured is not None:
+        return configured
+    return DataRuleRepository(db.session)
+
+
+def _prepare_standard_write(value):
+    if not isinstance(value, dict):
+        raise ValueError("request body must be an object")
+    receiver = dict(value)
+    if "code" in receiver:
+        raise ValueError(
+            "data standard executable code is read-only migration data"
+        )
+    rule_version_id = receiver.get("rule_version_id")
+    if rule_version_id is not None:
+        _repository().require_published_rule_version(rule_version_id)
+    receiver.pop("rule_version_status", None)
+    return receiver
+
+
 @bp.route("/data/standard/add", methods=["POST"])
 def data_standard_add():
     try:
-        receiver = request.get_json()
+        receiver = _prepare_standard_write(request.get_json(silent=True))
         name_zh = receiver["name_zh"]
         name_en = translate_and_parse(name_zh)
         receiver["name_en"] = name_en[0]
@@ -29,6 +52,9 @@ def data_standard_add():
         res = success("", "success")
         return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
 
+    except (KeyError, TypeError, ValueError) as e:
+        res = failed(str(e), 400, {})
+        return jsonify(res), 400
     except Exception as e:
         res = failed(str(e), 500, {})
         return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
@@ -62,29 +88,22 @@ def data_standard_detail():
 
 @bp.route("/data/standard/code", methods=["POST"])
 def data_standard_code():
-    try:
-        receiver = request.get_json()
-        input = receiver["input"]
-        describe = receiver["describe"]
-        output = receiver["output"]
-        relation = {
-            "input_params": input,
-            "output_params": output,
-        }
-        result = code_generate_standard(describe, relation)
-
-        res = success(result, "success")
-        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
-
-    except Exception as e:
-        res = failed(str(e), 500, {})
-        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
+    return (
+        jsonify(
+            failed(
+                "旧数据标准代码生成入口已关闭，请发布 RuleVersion 后绑定",
+                410,
+                {"semantics": "read_only_migration"},
+            )
+        ),
+        410,
+    )
 
 
 @bp.route("/data/standard/update", methods=["POST"])
 def data_standard_update():
     try:
-        receiver = request.get_json()
+        receiver = _prepare_standard_write(request.get_json(silent=True))
         name_zh = receiver["name_zh"]
         name_en = translate_and_parse(name_zh)
         receiver["name_en"] = name_en[0]
@@ -95,6 +114,9 @@ def data_standard_update():
         res = success("", "success")
         return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
 
+    except (KeyError, TypeError, ValueError) as e:
+        res = failed(str(e), 400, {})
+        return jsonify(res), 400
     except Exception as e:
         res = failed(str(e), 500, {})
         return json.dumps(res, ensure_ascii=False, cls=MyEncoder)

app/api/data_rules/routes.py

@@ -11,6 +11,7 @@ from flask import current_app, g, jsonify, request
 
 from app import db
 from app.api.data_rules import bp
+from app.core.common.identifiers import new_governance_uid
 from app.core.data_rules.authoring import (
     OpenAICompatibleRuleModel,
     RuleAuthoringAgent,
@@ -207,6 +208,25 @@ def capabilities():
     )
 
 
+@bp.post("/production-lines/draft-identity")
+def create_production_line_draft_identity():
+    """Issue the immutable server-owned identity used by a DataFlow draft."""
+    try:
+        _closed_body(set())
+        return (
+            jsonify(
+                success(
+                    {"dataflow_uid": new_governance_uid()},
+                    "生产线草稿身份已创建",
+                    code=201,
+                )
+            ),
+            201,
+        )
+    except (TypeError, ValueError):
+        return _bad_request("生产线草稿身份请求无效")
+
+
 @bp.post("/validate")
 def validate_asset():
     try:

app/core/data_flow/dataflows.py

@@ -1,4 +1,5 @@
 import contextlib
+import copy
 import json
 import logging
 import os
@@ -11,6 +12,7 @@ from sqlalchemy import text
 
 from app import db
 from app.core.common.identifiers import ensure_governance_uid
+from app.core.data_rules.contracts import validate_dataflow_spec
 from app.core.data_service.data_product_service import DataProductService
 from app.core.graph.graph_operations import (
     connect_graph,
@@ -29,6 +31,78 @@ PROJECT_ROOT = Path(__file__).parent.parent.parent.parent
 class DataFlowService:
     """数据流服务类，处理数据流相关的业务逻辑"""
 
+    _GOVERNED_REQUIREMENT_KEYS = {
+        "dataflow_spec",
+        "dataset_edges",
+        "migration_metadata",
+    }
+    _MIGRATION_METADATA_KEYS = {
+        "status",
+        "legacy_fields_present",
+        "preserved_for_read_only",
+        "governed_semantics",
+    }
+
+    @staticmethod
+    def _decode_script_requirement(value: Any) -> Any:
+        if not isinstance(value, str):
+            return copy.deepcopy(value)
+        try:
+            return json.loads(value)
+        except (TypeError, json.JSONDecodeError):
+            return value
+
+    @classmethod
+    def validate_governed_requirement(
+        cls, value: Any, *, repository
+    ) -> Dict[str, Any]:
+        """Validate the closed, published-asset DataFlow save envelope."""
+        requirement = cls._decode_script_requirement(value)
+        if not isinstance(requirement, dict):
+            raise ValueError("governed script_requirement must be an object")
+        if set(requirement) != cls._GOVERNED_REQUIREMENT_KEYS:
+            raise ValueError(
+                "governed script_requirement contains unsupported fields"
+            )
+
+        flow = validate_dataflow_spec(requirement.get("dataflow_spec"))
+        edges = requirement.get("dataset_edges")
+        if not isinstance(edges, dict) or set(edges) != {
+            "source_table",
+            "target_table",
+        }:
+            raise ValueError("dataset edges must be a closed object")
+        if (
+            edges.get("source_table") != flow["input_schema_refs"]
+            or edges.get("target_table") != flow["output_schema_ref"]
+        ):
+            raise ValueError("dataset edges must match the DataFlow schema refs")
+
+        metadata = requirement.get("migration_metadata")
+        if (
+            not isinstance(metadata, dict)
+            or set(metadata) != cls._MIGRATION_METADATA_KEYS
+        ):
+            raise ValueError("migration metadata must be a closed object")
+        if metadata.get("status") not in {"migrated", "unmigrated"}:
+            raise ValueError("unsupported migration status")
+        if type(metadata.get("legacy_fields_present")) is not bool:
+            raise ValueError("legacy_fields_present must be a boolean")
+        if metadata.get("preserved_for_read_only") is not True:
+            raise ValueError("legacy fields must be preserved for read-only use")
+        if metadata.get("governed_semantics") != "dataflow_spec":
+            raise ValueError("unsupported governed semantics")
+
+        repository.load_published_assets(flow)
+        return {
+            "dataflow_spec": flow,
+            "dataset_edges": {
+                "source_table": list(flow["input_schema_refs"]),
+                "target_table": flow["output_schema_ref"],
+            },
+            "migration_metadata": copy.deepcopy(metadata),
+        }
+
     @staticmethod
     def get_dataflows(
         page: int = 1,
@@ -204,7 +278,9 @@ class DataFlowService:
             raise e
 
     @staticmethod
-    def create_dataflow(data: Dict[str, Any]) -> Dict[str, Any]:
+    def create_dataflow(
+        data: Dict[str, Any], *, repository=None
+    ) -> Dict[str, Any]:
         """
         创建新的数据流
 
@@ -235,8 +311,26 @@ class DataFlowService:
                 logger.warning(f"翻译失败，使用默认英文名: {str(e)}")
                 name_en = dataflow_name.lower().replace(" ", "_")
 
-            # 处理 script_requirement，将其转换为 JSON 字符串
             script_requirement = data.get("script_requirement")
+            decoded_requirement = DataFlowService._decode_script_requirement(
+                script_requirement
+            )
+            governed = isinstance(decoded_requirement, dict) and (
+                "dataflow_spec" in decoded_requirement
+            )
+            if governed:
+                if repository is None:
+                    from app.core.data_rules.repository import DataRuleRepository
+
+                    repository = DataRuleRepository(db.session)
+                decoded_requirement = (
+                    DataFlowService.validate_governed_requirement(
+                        decoded_requirement, repository=repository
+                    )
+                )
+                script_requirement = decoded_requirement
+
+            # 处理 script_requirement，将其转换为 JSON 字符串
             if script_requirement is not None:
                 # 如果是字典或列表，转换为 JSON 字符串
                 if isinstance(script_requirement, (dict, list)):
@@ -266,6 +360,11 @@ class DataFlowService:
                 "created_at": get_formatted_time(),
                 "updated_at": get_formatted_time(),
             }
+            if governed:
+                node_data["uid"] = decoded_requirement["dataflow_spec"][
+                    "dataflow_uid"
+                ]
+                node_data["script_type"] = "governed"
             ensure_governance_uid(node_data)
 
             # 创建或获取数据流节点
@@ -283,24 +382,29 @@ class DataFlowService:
                 except Exception as e:
                     logger.warning(f"处理标签关系时出错: {str(e)}")
 
-            # 成功创建图数据库节点后，写入PG数据库
-            try:
-                DataFlowService._save_to_pg_database(data, dataflow_name, name_en)
-                logger.info(f"数据流信息已写入PG数据库: {dataflow_name}")
-
-                # PG数据库记录成功写入后，在neo4j图数据库中创建script关系
+            # Governed definitions are control-plane assets. Data Factory owns
+            # all later task, code, workflow and product deployment side effects.
+            if not governed:
                 try:
-                    DataFlowService._handle_script_relationships(
+                    DataFlowService._save_to_pg_database(
                         data, dataflow_name, name_en
                     )
-                    logger.info(f"脚本关系创建成功: {dataflow_name}")
-                except Exception as script_error:
-                    logger.warning(f"创建脚本关系失败: {str(script_error)}")
+                    logger.info(
+                        f"数据流信息已写入PG数据库: {dataflow_name}"
+                    )
 
-            except Exception as pg_error:
-                logger.error(f"写入PG数据库失败: {str(pg_error)}")
-                # 注意：这里可以选择回滚图数据库操作，但目前保持图数据库数据
-                # 在实际应用中，可能需要考虑分布式事务
+                    try:
+                        DataFlowService._handle_script_relationships(
+                            data, dataflow_name, name_en
+                        )
+                        logger.info(f"脚本关系创建成功: {dataflow_name}")
+                    except Exception as script_error:
+                        logger.warning(
+                            f"创建脚本关系失败: {str(script_error)}"
+                        )
+
+                except Exception as pg_error:
+                    logger.error(f"写入PG数据库失败: {str(pg_error)}")
 
             # 返回创建的数据流信息
             # 查询创建的节点获取完整信息
@@ -318,23 +422,23 @@ class DataFlowService:
                     # 如果查询失败，返回基本信息
                     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(),
                     }
 
-            # 注册数据产品到数据服务
-            try:
-                DataFlowService._register_data_product(
-                    data=data,
-                    dataflow_name=dataflow_name,
-                    name_en=name_en,
-                    dataflow_id=result.get("id"),
-                )
-                logger.info(f"数据产品注册成功: {dataflow_name}")
-            except Exception as product_error:
-                logger.warning(f"注册数据产品失败: {str(product_error)}")
-                # 不影响主流程，仅记录警告
+            if not governed:
+                try:
+                    DataFlowService._register_data_product(
+                        data=data,
+                        dataflow_name=dataflow_name,
+                        name_en=name_en,
+                        dataflow_id=result.get("id"),
+                    )
+                    logger.info(f"数据产品注册成功: {dataflow_name}")
+                except Exception as product_error:
+                    logger.warning(f"注册数据产品失败: {str(product_error)}")
 
             logger.info(f"创建数据流成功: {dataflow_name}")
             return result
@@ -1132,6 +1236,8 @@ class DataFlowService:
     def update_dataflow(
         dataflow_id: int,
         data: Dict[str, Any],
+        *,
+        repository=None,
     ) -> Optional[Dict[str, Any]]:
         """
         更新数据流
@@ -1144,6 +1250,26 @@ class DataFlowService:
             更新后的数据流信息，如果不存在则返回None
         """
         try:
+            data = copy.deepcopy(data)
+            requirement = DataFlowService._decode_script_requirement(
+                data.get("script_requirement")
+            )
+            governed = isinstance(requirement, dict) and (
+                "dataflow_spec" in requirement
+            )
+            if governed:
+                if repository is None:
+                    from app.core.data_rules.repository import DataRuleRepository
+
+                    repository = DataRuleRepository(db.session)
+                requirement = DataFlowService.validate_governed_requirement(
+                    requirement, repository=repository
+                )
+                data["script_requirement"] = requirement
+                data["script_type"] = "governed"
+                data["script_path"] = ""
+                data["uid"] = requirement["dataflow_spec"]["dataflow_uid"]
+
             # 提取 tag 数组（不作为节点属性存储）
             tag_list = data.pop("tag", None)
 
@@ -1154,6 +1280,13 @@ class DataFlowService:
 
                 if not result:
                     return None
+                if governed:
+                    existing_uid = dict(result[0]["n"]).get("uid")
+                    requested_uid = data["uid"]
+                    if existing_uid and str(existing_uid) != requested_uid:
+                        raise ValueError(
+                            "dataflow_uid cannot replace an existing identity"
+                        )
 
                 # 更新节点属性
                 update_fields = []

app/core/data_rules/repository.py

@@ -1705,6 +1705,17 @@ class DataRuleRepository:
                 result["id"] = str(evidence_id)
             return result
 
+        def compatibility_summary(value: Any) -> dict[str, Any]:
+            binding = _object(value, "schema compatibility")
+            status = binding.get("status")
+            if status in {"unknown", "compatible", "incompatible"}:
+                return binding
+            return {
+                "status": "unknown",
+                "reason": "production_line_context_required",
+                "binding": binding,
+            }
+
         items = []
         for row in rows:
             if row["version_id"] is None:
@@ -1722,9 +1733,8 @@ class DataRuleRepository:
                         else None
                     ),
                     "status": str(row["status"]),
-                    "schema_compatibility": _object(
-                        row["schema_compatibility"],
-                        "schema compatibility",
+                    "schema_compatibility": compatibility_summary(
+                        row["schema_compatibility"]
                     ),
                     "impact_count": int(row["impact_count"]),
                     "backend": str(row["backend"]),
@@ -1742,6 +1752,59 @@ class DataRuleRepository:
             "offset": offset,
         }
 
+    def require_published_rule_version(
+        self, rule_version_id: str
+    ) -> dict[str, Any]:
+        """Attest a published RuleVersion and its exact trusted evidence."""
+        version = _uid(rule_version_id, "rule_version_id")
+        row = (
+            self.session.execute(
+                text(
+                    "SELECT rv.id::text AS id, rv.rule_uid::text AS rule_uid, "
+                    "rv.version_no, rv.status "
+                    "FROM public.data_rule_versions rv "
+                    "JOIN public.rule_logical_plans lp "
+                    "ON lp.rule_version_id = rv.id "
+                    "WHERE rv.id = CAST(:id AS uuid) "
+                    "AND rv.status = 'published' "
+                    "AND lp.status = 'published' "
+                    "AND EXISTS (SELECT 1 FROM "
+                    "public.rule_logical_compile_evidence ce "
+                    "WHERE ce.logical_plan_id = lp.id "
+                    "AND ce.status = 'success' "
+                    "AND ce.compiler_version = lp.compiler_version "
+                    "AND ce.plan_hash = lp.plan_hash "
+                    "AND ce.schema_hashes = lp.schema_hashes "
+                    "AND ce.capabilities = lp.capabilities) "
+                    "AND EXISTS (SELECT 1 FROM "
+                    "public.rule_logical_test_evidence te "
+                    "WHERE te.logical_plan_id = lp.id "
+                    "AND te.status = 'success' "
+                    "AND te.plan_hash = lp.plan_hash "
+                    "AND te.schema_hashes = lp.schema_hashes) "
+                    "AND EXISTS (SELECT 1 FROM "
+                    "public.rule_publication_audits pa "
+                    "WHERE pa.rule_version_id = rv.id "
+                    "AND pa.action = 'published' "
+                    "AND pa.to_status = 'published' "
+                    "AND pa.evidence_hash = lp.plan_hash) "
+                    "ORDER BY lp.created_at DESC LIMIT 1 "
+                    "/* require_published_rule_version */"
+                ),
+                {"id": version},
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if row is None:
+            raise ValueError("rule version is not published and trusted")
+        return {
+            "id": str(row["id"]),
+            "rule_uid": str(row["rule_uid"]),
+            "version_no": int(row["version_no"]),
+            "status": str(row["status"]),
+        }
+
     def get_asset_evidence(
         self, *, asset_type: str, version_id: str
     ) -> dict[str, Any]:

frontend/src/api/dataRules.js

@@ -16,6 +16,10 @@ export function resolveProductionLinePreview (payload) {
   return http.post('/rules/production-lines/resolve', payload)
 }
 
+export function createProductionLineDraftIdentity () {
+  return http.post('/rules/production-lines/draft-identity', {})
+}
+
 export function createRuleVersion (payload) {
   return http.post('/rules/rule-versions', payload)
 }

frontend/src/components/DataRules/CompilationEvidence.vue

@@ -103,7 +103,7 @@ export default {
     },
     stage (key, label, value = null) {
       const state = value?.status || 'pending'
-      const successful = ['success', 'published', 'tested', 'compiled'].includes(state)
+      const successful = ['ready', 'success', 'published', 'tested', 'compiled'].includes(state)
       return {
         key,
         label,

frontend/src/components/DataRules/ProductionLineAssembler.vue

@@ -169,7 +169,7 @@ export default {
           blockers.push(`工位 ${index + 1} 尚未确认目录兼容性与证据`)
         }
         if (station.selected_asset && station.selected_asset.schema_compatibility !== 'compatible') {
-          blockers.push(`工位 ${index + 1} 的 schema_compatibility 未通过`)
+          blockers.push(`工位 ${index + 1} 的 Schema 兼容性必须由服务端发布预检确认`)
         }
         if (station.selected_asset && station.selected_asset.latest_evidence?.test_status !== 'success') {
           blockers.push(`工位 ${index + 1} 缺少可信测试证据`)
@@ -263,14 +263,19 @@ export default {
           id: `station_${index + 1}`,
           type: station.component_kind,
           stage: station.stage,
-          order: index + 1,
-          idempotency: { strategy: 'partition_replace', key: 'run_partition' }
+          order: index + 1
         }
         if (station.component_kind === 'standard.enforce') {
           component.standard_version_id = station.standard_version_id
         } else {
           component.rule_version_id = station.rule_version_id
         }
+        if (station.component_kind === 'rule.apply') {
+          component.idempotency = {
+            strategy: 'partition_replace',
+            key: 'run_partition'
+          }
+        }
         return component
       })
       this.$emit('input', {

frontend/src/components/DataRules/RuleAuthoringPanel.vue

@@ -124,7 +124,7 @@
           采用候选
         </v-btn>
         <v-btn
-          v-if="candidateAccepted && !versionAsset"
+          v-if="candidateAccepted && !versionAsset && isRuleCandidate"
           class="ml-3"
           small
           color="primary"
@@ -135,6 +135,15 @@
           创建草稿版本
         </v-btn>
       </div>
+      <v-alert
+        v-if="!isRuleCandidate"
+        class="mt-3 mb-0"
+        type="info"
+        text
+        dense
+      >
+        当前标准编辑流程需先生成并发布 RuleVersion，再通过目录选择器绑定到标准条款。
+      </v-alert>
     </div>
 
     <div v-if="versionAsset" class="rule-authoring__lifecycle mt-4">
@@ -178,7 +187,7 @@
         ref="evidence"
         class="mt-4"
         :version-id="versionAsset.id"
-        :asset-type="candidate.candidate_type === 'standard' ? 'standard' : 'rule'"
+        asset-type="rule"
       />
     </div>
   </section>
@@ -187,10 +196,8 @@
 <script>
 import {
   createRuleVersion,
-  createStandardVersion,
   interpretRule,
   publishRuleVersion,
-  publishStandardVersion,
   testRuleVersion,
   validateRuleVersion
 } from '@/api/dataRules'
@@ -248,6 +255,9 @@ export default {
         ? '数据标准候选已生成'
         : '可执行规则候选已生成'
     },
+    isRuleCandidate () {
+      return this.candidate?.candidate_type === 'rule'
+    },
     governedName () {
       return this.candidate?.rule_spec?.name || this.candidate?.standard_spec?.name || '未命名候选'
     },
@@ -318,6 +328,10 @@ export default {
       }
     },
     acceptCandidate () {
+      if (!this.isRuleCandidate) {
+        this.$snackbar.info('请先发布规则，再在标准条款中绑定')
+        return
+      }
       this.candidateAccepted = true
       this.$emit('candidate', this.candidate, this.sourceText)
       this.$snackbar.success('候选已确认，请创建受治理草稿')
@@ -325,20 +339,14 @@ export default {
     async createGovernedVersion () {
       this.versionLoading = true
       try {
-        const isStandard = this.candidate.candidate_type === 'standard'
-        const response = isStandard
-          ? await createStandardVersion({
-            source_text: this.sourceText,
-            standard_spec: this.candidate.standard_spec
-          })
-          : await createRuleVersion({
-            source_text: this.sourceText,
-            rule_spec: this.candidate.rule_spec,
-            generation_receipt: this.result.generation_receipt,
-            category: this.authoringSurface === 'data_standard'
-              ? 'standard_clause'
-              : 'flow_scoped'
-          })
+        const response = await createRuleVersion({
+          source_text: this.sourceText,
+          rule_spec: this.candidate.rule_spec,
+          generation_receipt: this.result.generation_receipt,
+          category: this.authoringSurface === 'data_standard'
+            ? 'standard_clause'
+            : 'flow_scoped'
+        })
         this.versionAsset = response.data
         this.$emit('version', this.versionAsset)
         this.$snackbar.success('草稿已创建，尚不可进入生产线')
@@ -349,10 +357,6 @@ export default {
       }
     },
     async validateVersion () {
-      if (this.candidate.candidate_type === 'standard') {
-        this.versionAsset = { ...this.versionAsset, status: 'validated' }
-        return
-      }
       this.validateLoading = true
       try {
         const { data } = await validateRuleVersion(this.versionAsset.id)
@@ -381,9 +385,7 @@ export default {
     async publishVersion () {
       this.publishLoading = true
       try {
-        const response = this.candidate.candidate_type === 'standard'
-          ? await publishStandardVersion(this.versionAsset.id)
-          : await publishRuleVersion(this.versionAsset.id)
+        const response = await publishRuleVersion(this.versionAsset.id)
         this.versionAsset = response.data
         this.$emit('version', this.versionAsset)
         this.refreshEvidence()

frontend/src/components/DataRules/RuleCatalogPicker.vue

@@ -170,10 +170,9 @@ export default {
             rule_uid: item.asset_type === 'rule' ? item.asset_uid : null,
             standard_uid: item.asset_type === 'standard' ? item.asset_uid : null,
             schema_context: item.schema_compatibility,
-            schema_compatibility: item.schema_compatibility?.status ||
-              (item.schema_compatibility && Object.keys(item.schema_compatibility).length
-                ? 'compatible'
-                : 'unknown'),
+            schema_compatibility: this.normalizeCompatibility(
+              item.schema_compatibility
+            ),
             latest_evidence: {
               compile_status: item.latest_evidence?.compile?.status || null,
               test_status: item.latest_evidence?.test?.status || null
@@ -223,6 +222,12 @@ export default {
     compatibilityColor (value) {
       return value === 'compatible' ? 'success' : value === 'incompatible' ? 'error' : 'warning'
     },
+    normalizeCompatibility (value) {
+      const status = value && value.status
+      return ['unknown', 'compatible', 'incompatible'].includes(status)
+        ? status
+        : 'unknown'
+    },
     evidenceLabel (value = {}) {
       if (value.test_status === 'success') return '编译与测试证据有效'
       if (value.compile_status === 'success') return '已编译，待样本测试'

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

@@ -162,6 +162,7 @@
 import EditBase from './editBase.vue'
 import { api } from '@/api/dataGovernance'
 import ProductionLineAssembler from '@/components/DataRules/ProductionLineAssembler'
+import { createProductionLineDraftIdentity } from '@/api/dataRules'
 
 export default {
   name: 'editPage',
@@ -180,6 +181,7 @@ export default {
       drawer: false,
       loading: false,
       saving: false,
+      governedDataflowUid: null,
       scriptContent: '',
       businessDomain: [],
       filteredBusinessDomain: {
@@ -202,7 +204,11 @@ export default {
   },
   computed: {
     dataflowUid () {
-      return this.itemData.dataflow_uid || this.itemData.data_flow_uid || this.itemData.uid || null
+      return this.governedDataflowUid ||
+        this.itemData.dataflow_uid ||
+        this.itemData.data_flow_uid ||
+        this.itemData.uid ||
+        null
     },
     dataflowName () {
       return this.itemData.name_zh || this.$refs.editBaseRefs?.formValues?.name_zh || '未命名数据生产线'
@@ -230,11 +236,27 @@ export default {
   },
   async created () {
     this.loading = true
+    await this.ensureDataflowUid()
     await this.getList()
     if (Object.keys(this.itemData).length) await this.getDetails()
     this.loading = false
   },
   methods: {
+    async ensureDataflowUid () {
+      const existing = this.itemData.dataflow_uid ||
+        this.itemData.data_flow_uid ||
+        this.itemData.uid
+      if (existing) {
+        this.governedDataflowUid = existing
+        return
+      }
+      try {
+        const { data } = await createProductionLineDraftIdentity()
+        this.governedDataflowUid = data.dataflow_uid
+      } catch (error) {
+        this.$snackbar.error(error || '无法创建受治理数据流草稿标识')
+      }
+    },
     async getList () {
       try {
         const { data } = await api.getBusinessDomainList2()
@@ -270,6 +292,11 @@ export default {
         const requirement = data.script_requirement && typeof data.script_requirement === 'object'
           ? data.script_requirement
           : {}
+        const persistedUid = requirement.dataflow_spec?.dataflow_uid ||
+          data.dataflow_uid ||
+          data.data_flow_uid ||
+          data.uid
+        if (persistedUid) this.governedDataflowUid = persistedUid
         this.productionLineSpec = requirement.dataflow_spec || {}
         const edges = requirement.dataset_edges || {}
         this.datasetEdges = {
@@ -303,11 +330,17 @@ export default {
     async handleSubmit () {
       const base = this.$refs.editBaseRefs.getValue()
       if (!base) return
+      if (!this.dataflowUid) await this.ensureDataflowUid()
+      if (!this.dataflowUid) {
+        this.$snackbar.error('无法取得服务端生产线草稿标识，暂不能保存')
+        return
+      }
       const payload = {
         ...base,
         script_requirement: {
           dataflow_spec: {
             ...this.productionLineSpec,
+            dataflow_uid: this.dataflowUid,
             name: base.name_zh || this.productionLineSpec.name
           },
           dataset_edges: {

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

@@ -282,7 +282,6 @@ export default {
       return {
         ...values,
         rule_version_id: this.linkedRuleVersionId,
-        rule_version_status: 'published',
         migration_metadata: this.legacyMigration
       }
     }

tests/core/data_rules/test_data_rule_repository.py

@@ -58,6 +58,7 @@ class FakeSession:
         unified_catalog_rows=None,
         rule_asset_evidence_rows=None,
         standard_asset_evidence_rows=None,
+        published_rule_attestation=None,
     ):
         self.calls = []
         self.duplicate = duplicate
@@ -75,6 +76,7 @@ class FakeSession:
         self.standard_asset_evidence_rows = list(
             standard_asset_evidence_rows or []
         )
+        self.published_rule_attestation = published_rule_attestation
 
     def execute(self, statement, params=None):
         sql = str(statement)
@@ -115,6 +117,14 @@ class FakeSession:
             return FakeResult(rows=self.rule_asset_evidence_rows)
         if "safe_standard_asset_evidence" in sql:
             return FakeResult(rows=self.standard_asset_evidence_rows)
+        if "require_published_rule_version" in sql:
+            return FakeResult(
+                rows=(
+                    [self.published_rule_attestation]
+                    if self.published_rule_attestation
+                    else []
+                )
+            )
         return FakeResult()
 
 
@@ -198,6 +208,33 @@ def test_publish_rule_version_cannot_bypass_compile_and_test_gate():
     assert "publication_gate_state" in _sql(session)
 
 
+def test_standard_link_attests_the_exact_trusted_published_rule_version():
+    from app.core.data_rules.repository import DataRuleRepository
+
+    version_id = new_governance_uid()
+    row = {
+        "id": version_id,
+        "rule_uid": new_governance_uid(),
+        "version_no": 7,
+        "status": "published",
+    }
+    repository = DataRuleRepository(
+        FakeSession(published_rule_attestation=row)
+    )
+
+    assert repository.require_published_rule_version(version_id) == row
+    sql = _sql(repository.session)
+    assert "lp.status = 'published'" in sql
+    assert "rule_logical_compile_evidence" in sql
+    assert "rule_logical_test_evidence" in sql
+    assert "rule_publication_audits" in sql
+
+    with pytest.raises(ValueError, match="not published and trusted"):
+        DataRuleRepository(FakeSession()).require_published_rule_version(
+            version_id
+        )
+
+
 def test_standard_version_requires_published_rule_versions_and_fixed_bindings():
     from app.core.data_rules.repository import DataRuleRepository
 
@@ -386,8 +423,12 @@ def test_unified_published_catalog_is_paginated_searchable_and_trusted():
                 "owner": owner_uid,
                 "status": "published",
                 "schema_compatibility": {
-                    "input": "a" * 64,
-                    "output": "b" * 64,
+                    "status": "unknown",
+                    "reason": "production_line_context_required",
+                    "binding": {
+                        "input": "a" * 64,
+                        "output": "b" * 64,
+                    },
                 },
                 "impact_count": 4,
                 "backend": "polars_batch",

tests/test_data_rule_frontend_contract.py

@@ -1,5 +1,8 @@
 from pathlib import Path
 
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_rules.contracts import validate_dataflow_spec
+
 API = Path("frontend/src/api/dataRules.js")
 AUTHORING = Path("frontend/src/components/DataRules/RuleAuthoringPanel.vue")
 CATALOG = Path("frontend/src/components/DataRules/RuleCatalogPicker.vue")
@@ -46,8 +49,8 @@ def test_reusable_authoring_panel_preserves_human_review_gate():
     assert "clarification_required" in source
     assert "$emit('candidate'" in source
     assert "createRuleVersion" in source
-    assert "createStandardVersion" in source
     assert "publishRuleVersion" in source
+    assert "先生成并发布 RuleVersion" in source
     assert "rules:publish" in source
     assert "$emit('version'" in source
     assert "自动执行" not in source
@@ -105,6 +108,7 @@ def test_compilation_evidence_shows_trusted_chain_without_sensitive_payloads():
         assert token in source
     for forbidden in ["raw_sample", "secret", "password", "token"]:
         assert forbidden not in source.lower()
+    assert "'ready'" in source
 
 
 def test_standard_links_published_rules_and_keeps_legacy_code_read_only():
@@ -119,6 +123,7 @@ def test_standard_links_published_rules_and_keeps_legacy_code_read_only():
     assert "迁移" in standard
     assert "dataStandardCodeGenerate" not in standard
     assert "handleCodeGenerate" not in standard
+    assert "rule_version_status" not in standard
 
 
 def test_dataflow_assembler_uses_published_catalog_ids_without_inline_rules():
@@ -142,6 +147,91 @@ def test_dataflow_assembler_uses_published_catalog_ids_without_inline_rules():
     assert "handleRuleCandidate" not in dataflow
 
 
+def test_assembler_emits_a_real_closed_dataflow_spec_for_every_station_kind():
+    standard_version_id = new_governance_uid()
+    apply_rule_version_id = new_governance_uid()
+    quality_rule_version_id = new_governance_uid()
+    normalized = validate_dataflow_spec(
+        {
+            "schema_version": "1.0",
+            "dataflow_uid": new_governance_uid(),
+            "name": "客户数据生产线",
+            "description": "固定受治理资产版本",
+            "input_schema_refs": ["bd:customer_raw:v1"],
+            "output_schema_ref": "bd:customer_clean:v1",
+            "components": [
+                {
+                    "id": "station_1",
+                    "type": "standard.enforce",
+                    "standard_version_id": standard_version_id,
+                    "stage": "quality_gate",
+                    "order": 1,
+                },
+                {
+                    "id": "station_2",
+                    "type": "rule.apply",
+                    "rule_version_id": apply_rule_version_id,
+                    "stage": "transform",
+                    "order": 2,
+                    "idempotency": {
+                        "strategy": "partition_replace",
+                        "key": "run_partition",
+                    },
+                },
+                {
+                    "id": "station_3",
+                    "type": "quality.check",
+                    "rule_version_id": quality_rule_version_id,
+                    "stage": "quality_gate",
+                    "order": 3,
+                },
+            ],
+            "parameters": {},
+        }
+    )
+
+    assert "idempotency" not in normalized["components"][0]
+    assert "idempotency" in normalized["components"][1]
+    assert "idempotency" not in normalized["components"][2]
+
+    source = ASSEMBLER.read_text(encoding="utf-8")
+    assert "station.component_kind === 'rule.apply'" in source
+    assert "component.idempotency" in source
+    assert "idempotency: {" not in source
+
+
+def test_catalog_never_guesses_schema_compatibility_from_arbitrary_objects():
+    source = CATALOG.read_text(encoding="utf-8")
+    assembler = ASSEMBLER.read_text(encoding="utf-8")
+
+    assert "normalizeCompatibility" in source
+    assert "Object.keys(item.schema_compatibility).length" not in source
+    assert "unknown" in source
+    assert "schema_compatibility !== 'compatible'" in assembler
+    assert "服务端发布预检" in assembler
+
+
+def test_new_dataflow_gets_server_governed_uid_before_assembly():
+    api = API.read_text(encoding="utf-8")
+    dataflow = DATAFLOW.read_text(encoding="utf-8")
+
+    assert "createProductionLineDraftIdentity" in api
+    assert "/rules/production-lines/draft-identity" in api
+    assert "ensureDataflowUid" in dataflow
+    assert "if (!this.dataflowUid) await this.ensureDataflowUid()" in dataflow
+    assert "persistedUid" in dataflow
+    assert "dataflow_uid: null" not in dataflow
+
+
+def test_standard_authoring_does_not_forge_validated_state_client_side():
+    source = AUTHORING.read_text(encoding="utf-8")
+
+    assert "candidate.candidate_type === 'standard'" not in source or (
+        "status: 'validated'" not in source
+    )
+    assert "先发布规则，再在标准条款中绑定" in source
+
+
 def test_data_factory_exposes_release_ready_but_no_fake_activation():
     page = FACTORY.read_text(encoding="utf-8")
     index = FACTORY_INDEX.read_text(encoding="utf-8")

tests/test_legacy_governance_cutover.py

@@ -0,0 +1,408 @@
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.system.tokens import decode_access_token, issue_access_token
+from tests.core.data_rules.test_contracts import valid_dataflow_spec
+
+
+class PublishedAssetRepository:
+    def __init__(self, *, published=True):
+        self.published = published
+        self.rule_calls = []
+        self.dataflow_calls = []
+
+    def require_published_rule_version(self, rule_version_id):
+        self.rule_calls.append(rule_version_id)
+        if not self.published:
+            raise ValueError("rule version is not published")
+        return {"id": rule_version_id, "status": "published"}
+
+    def load_published_assets(self, dataflow_spec):
+        self.dataflow_calls.append(dataflow_spec)
+        if not self.published:
+            raise ValueError("dataflow references unpublished assets")
+        return {}, {}
+
+
+def _headers(app, role="editor"):
+    token = issue_access_token(
+        user_id=new_governance_uid(),
+        roles=[role],
+        secret=app.config["SECRET_KEY"],
+        now=datetime.now(UTC),
+        lifetime=timedelta(minutes=10),
+    )
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _use_token_identity(monkeypatch):
+    def load(token, *, secret):
+        claims = decode_access_token(token, secret=secret)
+        return {
+            "id": claims["sub"],
+            "username": "cutover-test",
+            "display_name": "Cutover Test",
+            "roles": claims["roles"],
+        }
+
+    monkeypatch.setattr("app.core.system.auth.load_identity_from_token", load)
+
+
+def test_production_line_draft_identity_is_server_owned_closed_and_governed(
+    monkeypatch,
+):
+    from app import create_app
+
+    app = create_app()
+    app.config["TESTING"] = True
+    _use_token_identity(monkeypatch)
+    client = app.test_client()
+
+    response = client.post(
+        "/api/rules/production-lines/draft-identity",
+        json={},
+        headers=_headers(app),
+    )
+
+    assert response.status_code == 201
+    value = response.get_json()["data"]["dataflow_uid"]
+    parsed = uuid.UUID(value)
+    assert parsed.version == 7
+    assert parsed.variant == uuid.RFC_4122
+    assert (
+        client.post(
+            "/api/rules/production-lines/draft-identity",
+            json={"dataflow_uid": value},
+            headers=_headers(app),
+        ).status_code
+        == 400
+    )
+    assert (
+        client.post(
+            "/api/rules/production-lines/draft-identity",
+            json={},
+            headers=_headers(app, "viewer"),
+        ).status_code
+        == 403
+    )
+
+
+def test_legacy_standard_code_generation_is_closed_and_code_cannot_be_written(
+    monkeypatch,
+):
+    from app import create_app
+
+    app = create_app()
+    app.config["TESTING"] = True
+    _use_token_identity(monkeypatch)
+    client = app.test_client()
+
+    monkeypatch.setattr(
+        "app.api.data_interface.routes.create_or_get_node",
+        lambda *_args, **_kwargs: pytest.fail("legacy code was persisted"),
+    )
+
+    generated = client.post(
+        "/api/interface/data/standard/code",
+        json={"input": [], "describe": "生成代码", "output": []},
+        headers=_headers(app),
+    )
+    assert generated.status_code == 410
+    assert generated.get_json()["data"]["semantics"] == "read_only_migration"
+
+    added = client.post(
+        "/api/interface/data/standard/add",
+        json={
+            "name_zh": "旧代码标准",
+            "tag": [],
+            "code": "print('must not persist')",
+        },
+        headers=_headers(app),
+    )
+    assert added.status_code == 400
+    updated = client.post(
+        "/api/interface/data/standard/update",
+        json={
+            "name_zh": "旧代码标准",
+            "tag": [],
+            "code": "print('must not persist')",
+        },
+        headers=_headers(app),
+    )
+    assert updated.status_code == 400
+
+
+def test_governed_legacy_standard_link_is_attested_server_side(monkeypatch):
+    from app import create_app
+
+    app = create_app()
+    app.config["TESTING"] = True
+    _use_token_identity(monkeypatch)
+    repository = PublishedAssetRepository()
+    app.extensions["data_rule_repository"] = repository
+    captured = {}
+
+    monkeypatch.setattr(
+        "app.api.data_interface.routes.translate_and_parse",
+        lambda _value: ["published_standard"],
+    )
+    monkeypatch.setattr(
+        "app.api.data_interface.routes.create_or_get_node",
+        lambda _label, **properties: captured.update(properties) or 17,
+    )
+    client = app.test_client()
+    rule_version_id = new_governance_uid()
+
+    response = client.post(
+        "/api/interface/data/standard/add",
+        json={
+            "name_zh": "已治理标准",
+            "tag": [],
+            "rule_version_id": rule_version_id,
+            "rule_version_status": "draft",
+        },
+        headers=_headers(app),
+    )
+
+    assert response.status_code == 200
+    assert repository.rule_calls == [rule_version_id]
+    assert captured["rule_version_id"] == rule_version_id
+    assert "rule_version_status" not in captured
+
+    repository.published = False
+    rejected = client.post(
+        "/api/interface/data/standard/add",
+        json={
+            "name_zh": "伪造发布状态",
+            "tag": [],
+            "rule_version_id": new_governance_uid(),
+            "rule_version_status": "published",
+        },
+        headers=_headers(app),
+    )
+    assert rejected.status_code == 400
+
+
+def test_governed_dataflow_envelope_is_closed_and_uses_published_assets():
+    from app.core.data_flow.dataflows import DataFlowService
+
+    repository = PublishedAssetRepository()
+    flow = valid_dataflow_spec()
+    envelope = {
+        "dataflow_spec": flow,
+        "dataset_edges": {
+            "source_table": list(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",
+        },
+    }
+
+    normalized = DataFlowService.validate_governed_requirement(
+        envelope, repository=repository
+    )
+
+    assert normalized["dataflow_spec"]["dataflow_uid"] == flow["dataflow_uid"]
+    assert repository.dataflow_calls == [normalized["dataflow_spec"]]
+
+    invalid = dict(envelope)
+    invalid["task_list"] = []
+    with pytest.raises(ValueError, match="unsupported fields"):
+        DataFlowService.validate_governed_requirement(
+            invalid, repository=repository
+        )
+
+    mismatched = json.loads(json.dumps(envelope))
+    mismatched["dataset_edges"]["target_table"] = "bd:other:v1"
+    with pytest.raises(ValueError, match="dataset edges"):
+        DataFlowService.validate_governed_requirement(
+            mismatched, repository=repository
+        )
+
+
+def test_governed_dataflow_creation_never_generates_legacy_task_or_workflow(
+    monkeypatch,
+):
+    from app.core.data_flow.dataflows import DataFlowService
+
+    repository = PublishedAssetRepository()
+    flow = valid_dataflow_spec()
+    data = {
+        "name_zh": "客户治理生产线",
+        "describe": "固定发布版本的数据生产线",
+        "script_type": "python",
+        "script_requirement": {
+            "dataflow_spec": flow,
+            "dataset_edges": {
+                "source_table": list(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",
+            },
+        },
+    }
+    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"],
+    )
+    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()
+    )
+    monkeypatch.setattr(
+        DataFlowService,
+        "_save_to_pg_database",
+        lambda *_args, **_kwargs: pytest.fail("task_list write was invoked"),
+    )
+    monkeypatch.setattr(
+        DataFlowService,
+        "_handle_script_relationships",
+        lambda *_args, **_kwargs: pytest.fail("legacy script path was invoked"),
+    )
+    monkeypatch.setattr(
+        DataFlowService,
+        "_register_data_product",
+        lambda *_args, **_kwargs: None,
+    )
+
+    result = DataFlowService.create_dataflow(data, repository=repository)
+
+    assert result["id"] == 31
+    assert created["uid"] == flow["dataflow_uid"]
+    assert json.loads(created["script_requirement"])["dataflow_spec"] == flow
+
+
+def test_governed_dataflow_update_preserves_identity_and_closed_envelope(
+    monkeypatch,
+):
+    from app.core.data_flow.dataflows import DataFlowService
+
+    repository = PublishedAssetRepository()
+    flow = valid_dataflow_spec()
+    envelope = {
+        "dataflow_spec": flow,
+        "dataset_edges": {
+            "source_table": list(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",
+        },
+    }
+    updated = {}
+
+    class Result:
+        def __init__(self, data=None, single=None):
+            self._data = data or []
+            self._single = single
+
+        def data(self):
+            return self._data
+
+        def single(self):
+            return self._single
+
+    class Session:
+        def run(self, query, params=None, **kwargs):
+            values = params or kwargs
+            if "RETURN n" in query and "SET " not in query:
+                return Result(data=[{"n": {"uid": flow["dataflow_uid"]}}])
+            if "SET " in query:
+                updated.update(values)
+                return Result(
+                    data=[
+                        {
+                            "n": {
+                                "uid": values["uid"],
+                                "script_type": values["script_type"],
+                                "script_requirement": values[
+                                    "script_requirement"
+                                ],
+                            },
+                            "node_id": 44,
+                        }
+                    ]
+                )
+            return Result(single={"tags": []})
+
+        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.connect_graph", lambda: Driver()
+    )
+
+    result = DataFlowService.update_dataflow(
+        44,
+        {
+            "script_type": "python",
+            "script_path": "/tmp/forged.py",
+            "script_requirement": envelope,
+        },
+        repository=repository,
+    )
+
+    assert result["id"] == 44
+    assert updated["uid"] == flow["dataflow_uid"]
+    assert updated["script_type"] == "governed"
+    assert updated["script_path"] == ""
+    assert json.loads(updated["script_requirement"]) == envelope
+
+    mismatched = json.loads(json.dumps(envelope))
+    mismatched["dataflow_spec"]["dataflow_uid"] = new_governance_uid()
+    with pytest.raises(ValueError, match="cannot replace"):
+        DataFlowService.update_dataflow(
+            44,
+            {"script_requirement": mismatched},
+            repository=repository,
+        )