fix: enforce final rule publication trust boundaries

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

@@ -25,9 +25,10 @@ The delivered chain is:
    `draft` RuleVersion plus an immutable validation profile.
 5. Logical validation compiles a closed, validation-only Polars plan against
    the separately pinned input and output schema snapshots.
-6. Logical testing executes that exact plan in the isolated Polars worker
-   against the pinned MinIO sample artifact. Digest, schema, resource, row,
-   expiry, and ownership checks are enforced by `ArtifactStore`.
+6. Logical testing executes that exact plan and any golden comparison in the
+   same isolated Polars worker against pinned MinIO artifacts. Digest, schema,
+   resource, row, expiry, and ownership checks are enforced by
+   `ArtifactStore` and the bounded worker.
 7. Successful evidence advances RuleVersion `draft -> validated` and logical
    plan `compiled -> tested`.
 8. Publication requires exact successful compile and test evidence and advances
@@ -61,11 +62,19 @@ clock skew are checked cryptographically and against current database time
 during consumption.
 
 Draft creation performs `SELECT ... FOR UPDATE` on the generation record and a
-compare-and-set consumption update. A receipt is rejected if it is expired,
+compare-and-set consumption update. A new consumption is rejected if it is expired,
 tampered, belongs to another actor, has a different source/RuleSpec/candidate
 or validation context, has already been consumed, or is linked to another
 version.
 
+An exact same-actor receipt replay after an unknown commit outcome returns the
+already-linked canonical draft even after the short-lived receipt expires.
+This replay path still requires the same signed receipt hash and every canonical
+claim to match; expiry is bypassed only after the consumed row is proven exact.
+New consumption uses PostgreSQL `clock_timestamp()` rather than the
+transaction-start timestamp, so a long-running transaction cannot extend a
+receipt's usable lifetime.
+
 Publication locks the rule/plan/profile/schema or
 plan/component/rule/binding/schema rows in one transaction and re-reads the
 current compiler, RuleSpec, plan, schema, binding, data source, object,
@@ -129,7 +138,12 @@ physical backends are rejected on current schema, binding, dialect, plan-hash,
 RuleVersion, or logical-evidence drift.
 
 Logical dry-run supports a distinct output schema and optional golden Parquet
-result. Golden comparison uses canonical output-field order. The real
+result. The server stages both artifacts and passes their local paths and
+digests to one spawned worker governed by the plan's single RLIMIT/RSS/timeout
+budget. That worker verifies the staged golden digest and compares exact
+schema, row count, row order, null positions, and field values. The Flask
+process does not materialize either Parquet result. Concurrent large-golden
+tests fail closed inside the same resource boundary. The real
 cross-schema acceptance reads `name/mobile`, derives `name_copy`, removes one
 quarantined row, and compares the output to the pinned golden artifact.
 
@@ -148,9 +162,11 @@ Migration `20260723_200` is forward-only and adds:
 - validation failures and publication audits;
 - the published-rule catalog index.
 
-The migration refuses to run when revision 190 contains any published
-RuleVersion or execution plan, because pre-Task-7 rows cannot be silently
-upgraded into trusted evidence. Physical evidence carries an explicit
+The migration refuses to run when revision 190 contains any RuleVersion or
+execution plan, including draft and validated rows, because no pre-Task-7 row
+can possess the required generation, compile, test, and publication provenance.
+Operators must export and remove all legacy rows, apply migration 200, and then
+rebuild each rule through the Task 7 chain. Physical evidence carries an explicit
 `legacy_untrusted` marker, and publication/Runner/DataFlow gates require it to
 be false. Downgrade is deliberately rejected because deleting publication and
 generation evidence would break audit and replay guarantees.
@@ -176,6 +192,18 @@ validation/testing while publication keeps the existing administrative
 permission boundary. The documented canonical catalog path is implemented;
 the prior nested path remains a compatibility alias.
 
+`/interpret` preflights the dedicated receipt signer before resolving context
+or invoking the model, so missing configuration cannot incur model cost.
+Local and production environment templates and deployment runbooks document
+generation and coordinated rotation of `RULE_GENERATION_RECEIPT_SECRET`
+without containing a real secret.
+
+DataFlow release no longer trusts RuleVersion status alone. Its production
+asset loader requires the exact published logical plan, successful compile
+evidence, successful test evidence, matching plan/schema/capability hashes,
+and the matching publication audit. Revoked or drifted logical evidence makes
+the referenced rule unavailable and the release fails closed.
+
 ## Acceptance evidence
 
 Fail-first evidence included:
@@ -191,19 +219,24 @@ Fail-first evidence included:
 
 Final verification:
 
-- focused data-rule/API/migration suite: `183 passed`;
+- focused second-round trust-boundary suite: `39 passed`;
+- focused data-rule/API/migration suite from the initial delivery:
+  `183 passed`;
 - Runner plus real PostgreSQL/MySQL/Polars regression: `126 passed`;
 - real PostgreSQL + MinIO cross-schema receipt/logical lifecycle:
   `1 passed`;
 - real `PhysicalPlanPublicationService` PostgreSQL/MySQL lifecycle, drift,
   evidence replay, and PostgreSQL concurrency: `2 passed`;
 - full repository suite:
-  `626 passed, 29 skipped, 59 subtests passed`;
+  `632 passed, 29 skipped, 59 subtests passed`;
 - changed-file Ruff: `All checks passed!`;
 - `git diff --check`: passed;
 - empty local PostgreSQL full Alembic rebuild:
   base through `20260723_200 (head)`;
-- real revision-190 legacy publication upgrade guard: `1 passed`.
+- real revision-190 draft/validated/published upgrade guards: `3 passed`.
+- rebuilt local Docker backend: container reported `healthy`, and
+  `GET /api/system/health` returned HTTP/application code `200` with database
+  and Neo4j healthy.
 
 Skipped tests are environment-gated integration suites. The required real
 PostgreSQL, MySQL, and MinIO acceptance tests above were run explicitly against
@@ -217,9 +250,10 @@ for logical validation because those rows require a real deployment and
 component binding. Production execution evidence remains owned by the Task 6
 Runner path after a physical plan is published.
 
-Task 7 now supports separate input/output snapshots and cross-schema
-transformations. The remaining scope is production operational rollout:
-secret provisioning/rotation, multi-replica load and failure testing,
-monitoring/alert thresholds, and production canary/rollback evidence. Those
+Task 7 now supports separate input/output snapshots, cross-schema
+transformations, deployable secret templates, and a documented rotation
+procedure. The remaining scope is production operational rollout:
+multi-replica load and failure testing, monitoring/alert thresholds, and
+production canary/rollback evidence. Those
 operational gates are outside this local Docker acceptance and must be closed
 before a production environment is declared ready.

app/api/data_rules/routes.py

@@ -247,6 +247,7 @@ def interpret_rule():
         body = _closed_body(
             {"source_text", "authoring_surface", "context"}
         )
+        receipt_signer = _receipt_signer()
         repository = _repository()
         validation_context = repository.resolve_validation_context(
             body.get("context", {})
@@ -292,7 +293,7 @@ def interpret_rule():
                 expires_at=datetime.now(UTC)
                 + timedelta(minutes=10),
             )
-            result["generation_receipt"] = _receipt_signer().issue(claims)
+            result["generation_receipt"] = receipt_signer.issue(claims)
         db.session.commit()
         return jsonify(success(result))
     except (TypeError, ValueError):

app/core/data_rules/publication.py

@@ -10,7 +10,7 @@ import json
 import os
 import re
 import tempfile
-from contextlib import suppress
+from contextlib import ExitStack, suppress
 from datetime import UTC, datetime
 from typing import Any
 
@@ -204,6 +204,7 @@ class GenerationReceiptSigner:
         source_text: str,
         rule_spec: dict[str, Any],
         now: datetime | None = None,
+        allow_expired: bool = False,
     ) -> dict[str, Any]:
         if not isinstance(receipt, str) or len(receipt) > 4096:
             raise ValueError("generation receipt is invalid")
@@ -225,7 +226,11 @@ class GenerationReceiptSigner:
         current = now or self._clock()
         if current.tzinfo is None:
             raise ValueError("generation receipt clock must be timezone aware")
-        self._validate_time(claims, current)
+        self._validate_time(
+            claims,
+            current,
+            allow_expired=allow_expired,
+        )
         if claims["actor_uid"] != _uid(actor_uid, "actor_uid"):
             raise ValueError("generation receipt actor does not match")
         if claims["source_text_hash"] != _hash_text(
@@ -239,12 +244,16 @@ class GenerationReceiptSigner:
         return claims
 
     def _validate_time(
-        self, claims: dict[str, Any], current: datetime
+        self,
+        claims: dict[str, Any],
+        current: datetime,
+        *,
+        allow_expired: bool = False,
     ) -> None:
         if not isinstance(current, datetime) or current.tzinfo is None:
             raise ValueError("generation receipt clock must be timezone aware")
         now = int(current.timestamp())
-        if now >= claims["expires_at"]:
+        if now >= claims["expires_at"] and not allow_expired:
             raise ValueError("generation receipt has expired")
         if claims["issued_at"] > now + self.clock_skew_seconds:
             raise ValueError("generation receipt issue time is in the future")
@@ -380,6 +389,7 @@ class RulePublicationService:
             actor_uid=actor,
             source_text=source_text,
             rule_spec=spec,
+            allow_expired=True,
         )
         receipt_hash = hashlib.sha256(
             generation_receipt.encode("utf-8")
@@ -697,14 +707,38 @@ class ServerOwnedLogicalDryRunRunner:
             or described["schema_hash"] != plan["input_schema_hash"]
         ):
             raise ValueError("sample artifact schema has drifted")
+        golden_ref = context.get("golden_output_artifact_ref")
+        golden_digest = context.get("golden_output_artifact_digest")
+        if (golden_ref is None) != (golden_digest is None):
+            raise ValueError("golden output artifact is incomplete")
+        if golden_ref is not None:
+            if not isinstance(golden_ref, str) or not golden_ref:
+                raise ValueError("golden output artifact is invalid")
+            golden_digest = _digest(
+                golden_digest,
+                "golden_output_artifact_digest",
+            )
         output_path = None
         try:
-            with self.artifact_store.stage(
-                sample_ref,
-                sample_digest,
-                expected_schema_fields=plan["input_fields"],
-                limits=plan["resource_limits"],
-            ) as input_path:
+            with ExitStack() as stack:
+                input_path = stack.enter_context(
+                    self.artifact_store.stage(
+                        sample_ref,
+                        sample_digest,
+                        expected_schema_fields=plan["input_fields"],
+                        limits=plan["resource_limits"],
+                    )
+                )
+                golden_path = None
+                if golden_ref is not None:
+                    golden_path = stack.enter_context(
+                        self.artifact_store.stage(
+                            golden_ref,
+                            golden_digest,
+                            expected_schema_fields=plan["output_fields"],
+                            limits=plan["resource_limits"],
+                        )
+                    )
                 with tempfile.NamedTemporaryFile(
                     prefix="dataops-logical-dry-run-",
                     suffix=".parquet",
@@ -717,6 +751,8 @@ class ServerOwnedLogicalDryRunRunner:
                         "input_path": input_path,
                         "lookup_paths": {},
                         "output_path": output_path,
+                        "golden_path": golden_path,
+                        "golden_digest": golden_digest,
                         "masking_policies": {
                             "customer_mobile_last4": "preserve_last_4",
                             "redact": "redact",
@@ -726,27 +762,6 @@ class ServerOwnedLogicalDryRunRunner:
                         "memory_limit_bytes"
                     ],
                 )
-            golden_ref = context.get("golden_output_artifact_ref")
-            golden_digest = context.get("golden_output_artifact_digest")
-            if golden_ref is not None:
-                import polars as pl
-
-                expected = self.artifact_store.read(
-                    golden_ref,
-                    _digest(
-                        golden_digest,
-                        "golden_output_artifact_digest",
-                    ),
-                    expected_schema_fields=plan["output_fields"],
-                    limits=plan["resource_limits"],
-                ).select(
-                    [field["name"] for field in plan["output_fields"]]
-                ).collect()
-                actual = pl.read_parquet(output_path)
-                if not actual.equals(expected, null_equal=True):
-                    raise ValueError(
-                        "logical dry-run does not match golden output"
-                    )
             prepared = self.artifact_store.prepare_path(
                 output_path,
                 new_governance_uid(),
@@ -754,6 +769,14 @@ class ServerOwnedLogicalDryRunRunner:
                 schema_fields=plan["output_fields"],
                 limits=plan["resource_limits"],
             )
+            compared_output_digest = result.get("output_file_digest")
+            if (
+                compared_output_digest is not None
+                and prepared["digest"] != compared_output_digest
+            ):
+                raise ValueError(
+                    "golden-compared output artifact has drifted"
+                )
         finally:
             if output_path is not None:
                 with suppress(FileNotFoundError):
@@ -784,7 +807,12 @@ class ServerOwnedLogicalDryRunRunner:
                 "violation_digest": hashlib.sha256(
                     _canonical(result.get("violations", []))
                 ).hexdigest(),
-                "golden_output_digest": golden_digest,
+                "golden_output_digest": result.get(
+                    "golden_output_digest"
+                ),
+                "golden_rows_compared": result.get(
+                    "golden_rows_compared"
+                ),
             },
         }
 

app/core/data_rules/repository.py

@@ -564,7 +564,7 @@ class DataRuleRepository:
                     "rule_version_id, created_by::text AS created_by, "
                     "source_text_hash, candidate_hash, context_hash, "
                     "model_hash, prompt_hash, candidate, validation_context, "
-                    "decision, receipt_hash, CURRENT_TIMESTAMP AS db_now "
+                    "decision, receipt_hash, clock_timestamp() AS db_now "
                     "FROM public.rule_generation_runs "
                     "WHERE id = CAST(:id AS uuid) FOR UPDATE "
                     "/* consume_generation_receipt */"
@@ -597,13 +597,6 @@ class DataRuleRepository:
             for key, value in expected.items()
         ):
             raise ValueError("generation receipt claims do not match audit")
-        db_now = generation["db_now"]
-        if (
-            not hasattr(db_now, "timestamp")
-            or int(db_now.timestamp())
-            >= int(receipt_claims.get("expires_at") or 0)
-        ):
-            raise ValueError("generation receipt has expired")
         candidate = validate_rule_candidate(generation["candidate"])
         if (
             generation["rule_version_id"] is not None
@@ -645,6 +638,13 @@ class DataRuleRepository:
                     replay["validation_profile_id"]
                 ),
             }
+        db_now = generation["db_now"]
+        if (
+            not hasattr(db_now, "timestamp")
+            or int(db_now.timestamp())
+            >= int(receipt_claims.get("expires_at") or 0)
+        ):
+            raise ValueError("generation receipt has expired")
         if (
             generation["decision"] != "ready"
             or str(generation["created_by"] or "") != actor
@@ -1827,12 +1827,37 @@ class DataRuleRepository:
         if rule_ids:
             rule_rows = self.session.execute(
                 text(
-                    "SELECT rv.id::text AS id, rv.status, rv.rule_spec, "
-                    "rv.spec_hash "
+                    "SELECT DISTINCT ON (rv.id) rv.id::text AS id, "
+                    "rv.status, rv.rule_spec, rv.spec_hash "
                     "FROM public.data_rule_versions rv "
+                    "JOIN public.rule_logical_plans lp "
+                    "ON lp.rule_version_id = rv.id "
+                    "JOIN public.rule_logical_compile_evidence lce "
+                    "ON lce.logical_plan_id = lp.id "
+                    "JOIN public.rule_logical_test_evidence lte "
+                    "ON lte.logical_plan_id = lp.id "
                     "WHERE rv.id = ANY(CAST(:rule_ids AS uuid[])) "
                     "AND rv.status = 'published' "
-                    "/* published_rule_assets */"
+                    "AND lp.status = 'published' "
+                    "AND lp.plan->>'rule_version_id' = rv.id::text "
+                    "AND lp.plan->>'rule_spec_hash' = rv.spec_hash "
+                    "AND lce.status = 'success' "
+                    "AND lce.compiler_version = lp.compiler_version "
+                    "AND lce.plan_hash = lp.plan_hash "
+                    "AND lce.schema_hashes = lp.schema_hashes "
+                    "AND lce.capabilities = lp.capabilities "
+                    "AND lte.status = 'success' "
+                    "AND lte.plan_hash = lp.plan_hash "
+                    "AND lte.schema_hashes = lp.schema_hashes "
+                    "AND EXISTS (SELECT 1 FROM "
+                    "public.rule_publication_audits pa "
+                    "WHERE pa.rule_version_id = rv.id "
+                    "AND pa.rule_execution_plan_id IS NULL "
+                    "AND pa.action = 'published' "
+                    "AND pa.to_status = 'published' "
+                    "AND pa.evidence_hash = lp.plan_hash) "
+                    "/* published_rule_assets */ "
+                    "ORDER BY rv.id, lp.created_at DESC"
                 ),
                 {"rule_ids": sorted(rule_ids)},
             ).mappings().all()

app/runner/polars_worker.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import multiprocessing
 import os
 import time
@@ -198,6 +199,60 @@ def _validate_lazy_schema(
             raise ValueError("artifact nullable contract does not match")
 
 
+def _file_digest(path: str) -> str:
+    digest = hashlib.sha256()
+    with open(path, "rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _compare_golden_output(
+    *,
+    output_path: str,
+    golden_path: str,
+    golden_digest: str,
+    output_fields: list[dict[str, Any]],
+) -> dict[str, Any]:
+    if _file_digest(golden_path) != golden_digest:
+        raise ValueError("golden output artifact digest has drifted")
+    columns = [field["name"] for field in output_fields]
+    actual = pl.scan_parquet(output_path).select(columns)
+    expected = pl.scan_parquet(golden_path).select(columns)
+    _validate_lazy_schema(actual, output_fields)
+    _validate_lazy_schema(expected, output_fields)
+    actual_rows = _count(actual)
+    expected_rows = _count(expected)
+    if actual_rows != expected_rows:
+        raise ValueError("logical dry-run does not match golden output")
+
+    expected_names = {
+        name: f"__expected_{index}" for index, name in enumerate(columns)
+    }
+    paired = actual.with_row_index("__row_index").join(
+        expected.rename(expected_names).with_row_index("__row_index"),
+        on="__row_index",
+        how="inner",
+    )
+    mismatches = [
+        (
+            (pl.col(name) != pl.col(expected_names[name])).fill_null(False)
+            | (
+                pl.col(name).is_null()
+                != pl.col(expected_names[name]).is_null()
+            )
+        )
+        for name in columns
+    ]
+    if _count(paired.filter(pl.any_horizontal(mismatches))):
+        raise ValueError("logical dry-run does not match golden output")
+    return {
+        "golden_output_digest": golden_digest,
+        "output_file_digest": _file_digest(output_path),
+        "golden_rows_compared": actual_rows,
+    }
+
+
 def _execute_polars_job(job: dict[str, Any]) -> dict[str, Any]:
     plan = validate_bound_polars_plan(job["plan"])
     frame = pl.scan_parquet(job["input_path"])
@@ -429,6 +484,23 @@ def _execute_polars_job(job: dict[str, Any]) -> dict[str, Any]:
     if rows_out > plan["resource_limits"]["max_rows"]:
         raise ValueError("output row limit exceeded")
     frame.sink_parquet(job["output_path"], engine="streaming")
+    golden_path = job.get("golden_path")
+    golden_digest = job.get("golden_digest")
+    if (golden_path is None) != (golden_digest is None):
+        raise ValueError("golden output artifact is incomplete")
+    golden_result = {}
+    if golden_path is not None:
+        if (
+            not isinstance(golden_path, str)
+            or not isinstance(golden_digest, str)
+        ):
+            raise ValueError("golden output artifact is invalid")
+        golden_result = _compare_golden_output(
+            output_path=job["output_path"],
+            golden_path=golden_path,
+            golden_digest=golden_digest,
+            output_fields=plan["output_fields"],
+        )
     return {
         "worker_pid": os.getpid(),
         "rows_in": rows_in,
@@ -437,6 +509,7 @@ def _execute_polars_job(job: dict[str, Any]) -> dict[str, Any]:
         "violation_count": sum(item["count"] for item in violations),
         "violations": violations,
         "_violation_sample": violation_sample,
+        **golden_result,
     }
 
 

deploy/docker/.env.example

@@ -1,6 +1,10 @@
 # Optional external generative model. Leave empty for infrastructure-only tests.
 DEEPSEEK_API_KEY=
 
+# Required before using AI rule interpretation or creating rule drafts.
+# Generate a dedicated value with: openssl rand -hex 32
+RULE_GENERATION_RECEIPT_SECRET=replace-with-dedicated-64-hex-random-value
+
 # Create this key in the local n8n UI after owner setup, then restart backend.
 N8N_API_KEY=
 

deploy/docker/README.md

@@ -8,12 +8,20 @@ Docker 卷，不连接生产服务。
 
 ```bash
 cp deploy/docker/.env.example deploy/docker/.env.local
+# AI 规则解析为必填；生成后写入 .env.local，不要提交真实值
+openssl rand -hex 32
 docker compose --env-file deploy/docker/.env.local \
   -f deploy/docker/docker-compose.yml up -d --build
 docker compose -f deploy/docker/docker-compose.yml ps
 ```
 
 不需要验证 DeepSeek 或 n8n 管理 API 时，可以不创建 `.env.local`，直接启动。基础设施和页面仍可使用。
+自然语言规则解析及规则草稿创建必须同时配置 `DEEPSEEK_API_KEY` 和独立的
+`RULE_GENERATION_RECEIPT_SECRET`；后者不得复用 Flask `SECRET_KEY`。
+
+回执密钥轮换会让尚未消费的十分钟短时回执失效。生产轮换时应先暂停新的规则解析，
+等待现有回执消费或过期，再一次性更新所有后端实例并重启；紧急轮换可立即执行，但
+需要用户重新解析尚未提交的规则。
 
 ## 本地地址
 

deployment/.env.production.example

@@ -3,6 +3,9 @@
 
 FLASK_ENV=production
 SECRET_KEY=replace-with-a-long-random-secret
+# Dedicated HMAC key for short-lived AI rule generation receipts.
+# Generate independently with: openssl rand -hex 32
+RULE_GENERATION_RECEIPT_SECRET=replace-with-dedicated-64-hex-random-value
 DEBUG=False
 
 # Gunicorn / Flask 监听端口（保持一致，默认 5500）

deployment/README.md

@@ -63,12 +63,24 @@ sudo bash deploy_dataops.sh
 
 ```bash
 sudo vim /etc/dataops-platform/dataops.env
-# 必改: SECRET_KEY, DEEPSEEK_API_KEY, N8N_API_KEY
+# 必改: SECRET_KEY, RULE_GENERATION_RECEIPT_SECRET, DEEPSEEK_API_KEY, N8N_API_KEY
 # 核对: DATABASE_URL, NEO4J_*, MINIO_*, API_BASE_URL
 
 sudo bash deploy_dataops.sh
 ```
 
+`RULE_GENERATION_RECEIPT_SECRET` 是 AI 规则解析回执的专用 HMAC 密钥，必须
+独立于 `SECRET_KEY` 生成：
+
+```bash
+openssl rand -hex 32
+```
+
+将输出只写入 `/etc/dataops-platform/dataops.env` 或密钥管理系统，不要提交到
+仓库。缺少该值时 `/api/rules/interpret` 会在调用大模型前失败，避免产生无效模型
+费用。轮换会使未消费的十分钟回执失效：常规轮换应暂停新解析、等待回执消费或过期，
+再统一更新并重启全部后端实例；紧急轮换后需重新解析未提交规则。
+
 ### 可选：同时配置 Nginx
 
 ```bash

deployment/dataops.env

@@ -4,6 +4,9 @@
 
 FLASK_ENV=production
 SECRET_KEY=replace-with-a-long-random-secret
+# Dedicated HMAC key for short-lived AI rule generation receipts.
+# Generate independently with: openssl rand -hex 32
+RULE_GENERATION_RECEIPT_SECRET=replace-with-dedicated-64-hex-random-value
 
 # 平台 PostgreSQL（可与平台同机；部署前替换密码）
 DATABASE_URL=postgresql://dataops_user:replace-password@127.0.0.1:5432/dataops

env.example

@@ -15,6 +15,9 @@ DATABASE_URL=postgresql://username:password@localhost:5432/database_name
 
 # 安全配置
 SECRET_KEY=your-secret-key-here
+# AI 数据规则回执专用密钥，必须与 SECRET_KEY 分开生成。
+# openssl rand -hex 32
+RULE_GENERATION_RECEIPT_SECRET=replace-with-dedicated-64-hex-random-value
 ALGORITHM=HS256
 ACCESS_TOKEN_EXPIRE_MINUTES=30
 
@@ -60,4 +63,3 @@ SMTP_PORT=587
 SMTP_USERNAME=your-email@gmail.com
 SMTP_PASSWORD=your-app-password
 
-

migrations/versions/20260723_200_rule_publication_gates.py

@@ -15,22 +15,15 @@ def upgrade() -> None:
         BEGIN
             IF EXISTS (
                 SELECT 1 FROM public.data_rule_versions
-                WHERE status = 'published'
             ) OR EXISTS (
                 SELECT 1 FROM public.rule_execution_plans
-                WHERE status = 'published'
             ) THEN
                 RAISE EXCEPTION
-                    'legacy published rules or plans cannot be trusted by '
-                    'migration 200; revoke or rebuild them through the '
-                    'Task7 compile-test-publication chain';
-            END IF;
-            IF EXISTS (
-                SELECT 1 FROM public.data_rule_versions
-                WHERE status NOT IN ('draft','validated','published')
-            ) THEN
-                RAISE EXCEPTION
-                    'legacy rule version statuses must be resolved before 200';
+                    'pre-Task7 rule versions or execution plans lack trusted '
+                    'generation, compile, test, and publication provenance; '
+                    'export and remove every legacy row, run migration 200, '
+                    'then rebuild each rule through the Task7 publication '
+                    'chain';
             END IF;
         END $$;
 

tests/core/data_rules/test_data_rule_repository.py

@@ -53,6 +53,8 @@ class FakeSession:
         published_rule_ids=None,
         catalog_rule_rows=None,
         catalog_standard_rows=None,
+        published_asset_rule_rows=None,
+        published_asset_standard_rows=None,
     ):
         self.calls = []
         self.duplicate = duplicate
@@ -61,6 +63,12 @@ class FakeSession:
         self.published_rule_ids = set(published_rule_ids or [])
         self.catalog_rule_rows = list(catalog_rule_rows or [])
         self.catalog_standard_rows = list(catalog_standard_rows or [])
+        self.published_asset_rule_rows = list(
+            published_asset_rule_rows or []
+        )
+        self.published_asset_standard_rows = list(
+            published_asset_standard_rows or []
+        )
 
     def execute(self, statement, params=None):
         sql = str(statement)
@@ -91,6 +99,10 @@ class FakeSession:
             return FakeResult(rows=self.catalog_rule_rows)
         if "catalog_standard_versions" in sql:
             return FakeResult(rows=self.catalog_standard_rows)
+        if "published_rule_assets" in sql:
+            return FakeResult(rows=self.published_asset_rule_rows)
+        if "published_standard_assets" in sql:
+            return FakeResult(rows=self.published_asset_standard_rows)
         return FakeResult()
 
 
@@ -264,6 +276,53 @@ def test_server_catalog_loader_uses_only_published_database_versions():
     assert "rule_test_evidence" in _sql(session)
 
 
+def test_dataflow_asset_loader_requires_exact_task7_logical_chain():
+    from app.core.data_rules.repository import DataRuleRepository
+
+    rule_id = new_governance_uid()
+    standard_id = new_governance_uid()
+    flow = valid_dataflow_spec(standard_id, rule_id)
+    session = FakeSession(
+        published_asset_standard_rows=[
+            {
+                "id": standard_id,
+                "status": "published",
+                "clauses": [
+                    {
+                        "clause_id": "mobile_format",
+                        "rule_version_id": rule_id,
+                        "severity": "error",
+                        "exception_policy": "quarantine",
+                    }
+                ],
+            }
+        ],
+        published_asset_rule_rows=[
+            {
+                "id": rule_id,
+                "status": "published",
+                "rule_spec": valid_rule_spec(),
+                "spec_hash": "a" * 64,
+            }
+        ]
+    )
+
+    _standards, rules = DataRuleRepository(session).load_published_assets(flow)
+
+    assert rule_id in rules
+    sql = _sql(session)
+    assert "JOIN public.rule_logical_plans lp" in sql
+    assert "lp.status = 'published'" in sql
+    assert "lce.compiler_version = lp.compiler_version" in sql
+    assert "lce.plan_hash = lp.plan_hash" in sql
+    assert "lce.schema_hashes = lp.schema_hashes" in sql
+    assert "lce.capabilities = lp.capabilities" in sql
+    assert "lte.plan_hash = lp.plan_hash" in sql
+    assert "lte.schema_hashes = lp.schema_hashes" in sql
+    assert "rule_publication_audits" in sql
+    assert "pa.evidence_hash = lp.plan_hash" in sql
+
+
 def test_generation_run_persists_model_hashes_uncertainty_and_decision():
     from app.core.data_rules.repository import DataRuleRepository
 

tests/integration/test_rule_publication_lifecycle.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import hashlib
 import json
 import os
+import time
 from datetime import UTC, datetime, timedelta
 
 import polars as pl
@@ -22,9 +23,13 @@ from app.core.data_rules.publication import (
     ServerOwnedPhysicalPreflightRunner,
     generation_receipt_claims,
 )
+from app.core.data_rules.release import ProductionLineReleaseService
 from app.core.data_rules.repository import DataRuleRepository
 from app.runner.artifacts import ArtifactStore
-from tests.core.data_rules.test_contracts import valid_rule_spec
+from tests.core.data_rules.test_contracts import (
+    valid_dataflow_spec,
+    valid_rule_spec,
+)
 from tests.integration.test_data_rule_polars_execution import _compose_value
 
 pytestmark = pytest.mark.integration
@@ -213,7 +218,7 @@ def test_real_postgres_receipt_to_logical_compile_test_publish(database_url):
                 model_hash=evidence["model_hash"],
                 prompt_hash=evidence["prompt_hash"],
                 context_hash=evidence["context_hash"],
-                expires_at=datetime.now(UTC) + timedelta(minutes=5),
+                expires_at=datetime.now(UTC) + timedelta(seconds=2),
             )
             receipt = signer.issue(claims)
             service = RulePublicationService(
@@ -242,6 +247,49 @@ def test_real_postgres_receipt_to_logical_compile_test_publish(database_url):
                 source_language="zh-CN",
                 generated_kind="rulespec",
             ) == draft
+            while int(datetime.now(UTC).timestamp()) < claims["expires_at"]:
+                time.sleep(0.05)
+            assert service.create_draft(
+                rule_spec=spec,
+                source_text=evidence["source_text"],
+                actor_uid=actor,
+                generation_receipt=receipt,
+                category="standard_clause",
+                source_language="zh-CN",
+                generated_kind="rulespec",
+            ) == draft
+            next_generation = repository.record_generation_run(
+                evidence=evidence,
+                created_by=actor,
+                validation_context=validation_context,
+            )
+            next_claims = generation_receipt_claims(
+                generation_run_id=next_generation["id"],
+                actor_uid=actor,
+                source_text=evidence["source_text"],
+                candidate_hash=evidence["candidate_hash"],
+                rule_spec=spec,
+                model_hash=evidence["model_hash"],
+                prompt_hash=evidence["prompt_hash"],
+                context_hash=evidence["context_hash"],
+                expires_at=datetime.now(UTC) + timedelta(seconds=2),
+            )
+            next_receipt = signer.issue(next_claims)
+            while (
+                int(datetime.now(UTC).timestamp())
+                < next_claims["expires_at"]
+            ):
+                time.sleep(0.05)
+            with pytest.raises(ValueError, match="expired"):
+                service.create_draft(
+                    rule_spec=spec,
+                    source_text=evidence["source_text"],
+                    actor_uid=actor,
+                    generation_receipt=next_receipt,
+                    category="standard_clause",
+                    source_language="zh-CN",
+                    generated_kind="rulespec",
+                )
 
             compiled = service.validate(draft["id"], actor)
             assert compiled["plan_status"] == "compiled"
@@ -318,6 +366,76 @@ def test_real_postgres_receipt_to_logical_compile_test_publish(database_url):
                 "id"
             ] == draft["id"]
 
+            flow = valid_dataflow_spec(rule_version_id=draft["id"])
+            flow["input_schema_refs"] = [spec["input_schema_ref"]]
+            flow["output_schema_ref"] = spec["output_schema_ref"]
+            flow["components"] = [
+                component
+                for component in flow["components"]
+                if component["type"] == "rule.apply"
+            ]
+
+            class PinnedResolver:
+                def resolve(self, schema_ref):
+                    if schema_ref == input_snapshot["schema_ref"]:
+                        return input_snapshot
+                    if schema_ref == output_snapshot["schema_ref"]:
+                        return output_snapshot
+                    raise ValueError("unexpected schema ref")
+
+            release_service = ProductionLineReleaseService(
+                repository,
+                schema_resolver=PinnedResolver(),
+            )
+            logical_test_id = tested["test_evidence_id"]
+            connection.execute(
+                text(
+                    "UPDATE public.rule_logical_test_evidence "
+                    "SET status = 'failed' "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": logical_test_id},
+            )
+            with pytest.raises(ValueError, match="published rule"):
+                release_service.release(
+                    dataflow_uid=flow["dataflow_uid"],
+                    dataflow_spec=flow,
+                    source_text="跨模式数据生产线",
+                    created_by=actor,
+                )
+            connection.execute(
+                text(
+                    "UPDATE public.rule_logical_test_evidence "
+                    "SET status = 'success', schema_hashes = '{}'::jsonb "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": logical_test_id},
+            )
+            with pytest.raises(ValueError, match="published rule"):
+                release_service.release(
+                    dataflow_uid=flow["dataflow_uid"],
+                    dataflow_spec=flow,
+                    source_text="跨模式数据生产线",
+                    created_by=actor,
+                )
+            connection.execute(
+                text(
+                    "UPDATE public.rule_logical_test_evidence te "
+                    "SET schema_hashes = lp.schema_hashes "
+                    "FROM public.rule_logical_plans lp "
+                    "WHERE te.logical_plan_id = lp.id "
+                    "AND te.id = CAST(:id AS uuid)"
+                ),
+                {"id": logical_test_id},
+            )
+            released = release_service.release(
+                dataflow_uid=flow["dataflow_uid"],
+                dataflow_spec=flow,
+                source_text="跨模式数据生产线",
+                created_by=actor,
+            )
+            assert released["status"] == "released"
+
             rows = connection.execute(
                 text(
                     "SELECT rv.status, lp.status, "

tests/integration/test_rule_publication_migration_upgrade.py

@@ -12,7 +12,8 @@ from tests.integration.test_rule_artifact_migration_upgrade import (
 )
 
 
-def test_legacy_published_rule_blocks_real_190_to_200_upgrade():
+@pytest.mark.parametrize("legacy_status", ["draft", "validated", "published"])
+def test_any_legacy_rule_blocks_real_190_to_200_upgrade(legacy_status):
     platform_user = _compose_value(
         r"\n  postgres:.*?POSTGRES_USER:\s*([^\s]+)"
     )
@@ -45,7 +46,7 @@ def test_legacy_published_rule_blocks_real_190_to_200_upgrade():
                     "INSERT INTO public.data_rules "
                     "(id, rule_uid, name, category, status) VALUES "
                     "(CAST(:id AS uuid), CAST(:rule_uid AS uuid), "
-                    "'legacy published', 'legacy', 'active')"
+                    "'legacy rule', 'legacy', 'active')"
                 ),
                 {"id": new_governance_uid(), "rule_uid": rule_uid},
             )
@@ -55,18 +56,19 @@ def test_legacy_published_rule_blocks_real_190_to_200_upgrade():
                     "(id, rule_uid, version_no, source_text, rule_spec, "
                     "spec_hash, status) VALUES "
                     "(CAST(:id AS uuid), CAST(:rule_uid AS uuid), 1, "
-                    "'legacy', '{}'::jsonb, :spec_hash, 'published')"
+                    "'legacy', '{}'::jsonb, :spec_hash, :legacy_status)"
                 ),
                 {
                     "id": new_governance_uid(),
                     "rule_uid": rule_uid,
                     "spec_hash": "a" * 64,
+                    "legacy_status": legacy_status,
                 },
             )
         engine.dispose()
         engine = None
 
-        with pytest.raises(Exception, match="legacy published"):
+        with pytest.raises(Exception, match="pre-Task7"):
             _upgrade(database_url, "20260723_200")
 
         engine = create_engine(database_url, pool_pre_ping=True)

tests/runner/test_polars_worker.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+from concurrent.futures import ThreadPoolExecutor
 
 import polars as pl
 import pytest
@@ -161,6 +162,115 @@ def test_worker_start_failure_closes_pipe_endpoints_and_maps_safe_error(
     assert process.closed is True
 
 
+def test_worker_rejects_golden_row_drift_inside_isolated_boundary(tmp_path):
+    from app.runner.polars_worker import (
+        PolarsWorkerError,
+        execute_isolated_polars_plan,
+    )
+
+    schema = _schema("bd:golden:input", [("value", "string", False)])
+    plan = _compile_worker_plan(
+        input_schema=schema,
+        output_schema={
+            **schema,
+            "id": _schema(
+                "bd:golden:output", [("value", "string", False)]
+            )["id"],
+            "schema_ref": "bd:golden:output",
+        },
+        steps=[
+            {
+                "id": "trim_value",
+                "op": "normalize_text",
+                "column": "value",
+                "trim": True,
+            }
+        ],
+    )
+    plan["resource_limits"]["memory_limit_bytes"] = 128 * 1024 * 1024
+    input_path = tmp_path / "golden-input.parquet"
+    golden_path = tmp_path / "golden-expected.parquet"
+    pl.DataFrame({"value": ["actual"]}).write_parquet(input_path)
+    pl.DataFrame({"value": ["expected"]}).write_parquet(golden_path)
+
+    with pytest.raises(PolarsWorkerError, match="execution failed"):
+        execute_isolated_polars_plan(
+            {
+                "plan": plan,
+                "input_path": str(input_path),
+                "lookup_paths": {},
+                "output_path": str(tmp_path / "golden-output.parquet"),
+                "golden_path": str(golden_path),
+                "golden_digest": "a" * 64,
+                "masking_policies": {},
+            },
+            memory_limit_bytes=plan["resource_limits"][
+                "memory_limit_bytes"
+            ],
+        )
+
+
+def test_concurrent_large_golden_comparisons_fail_closed(tmp_path):
+    from app.runner.polars_worker import (
+        PolarsWorkerError,
+        execute_isolated_polars_plan,
+    )
+
+    schema = _schema("bd:golden:large", [("value", "string", False)])
+    plan = _compile_worker_plan(
+        input_schema=schema,
+        output_schema={
+            **schema,
+            "id": _schema(
+                "bd:golden:large-output",
+                [("value", "string", False)],
+            )["id"],
+            "schema_ref": "bd:golden:large-output",
+        },
+        steps=[
+            {
+                "id": "trim_value",
+                "op": "normalize_text",
+                "column": "value",
+                "trim": True,
+            }
+        ],
+    )
+    plan["resource_limits"]["memory_limit_bytes"] = 128 * 1024 * 1024
+    input_path = tmp_path / "large-input.parquet"
+    golden_path = tmp_path / "large-golden.parquet"
+    pl.DataFrame({"value": ["actual"]}).write_parquet(input_path)
+    pl.DataFrame(
+        {"value": [f"expected-{index:08d}" for index in range(200_000)]}
+    ).write_parquet(golden_path)
+
+    def compare(index):
+        try:
+            execute_isolated_polars_plan(
+                {
+                    "plan": plan,
+                    "input_path": str(input_path),
+                    "lookup_paths": {},
+                    "output_path": str(
+                        tmp_path / f"large-output-{index}.parquet"
+                    ),
+                    "golden_path": str(golden_path),
+                    "golden_digest": "b" * 64,
+                    "masking_policies": {},
+                },
+                memory_limit_bytes=plan["resource_limits"][
+                    "memory_limit_bytes"
+                ],
+            )
+        except PolarsWorkerError:
+            return "rejected"
+        return "unsafe-success"
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = list(executor.map(compare, range(2)))
+    assert results == ["rejected", "rejected"]
+
+
 def test_regex_peak_allocation_fails_inside_isolated_worker(tmp_path):
     schema = _schema(
         "bd:regex:raw",

tests/test_data_rule_api.py

@@ -348,6 +348,41 @@ def test_rule_interpret_uses_configured_agent_and_preserves_surface(monkeypatch)
     assert repository.calls[1][0] == "record_generation_run"
 
 
+def test_rule_interpret_preflights_receipt_signer_before_model_call(monkeypatch):
+    from app import create_app
+
+    app = create_app()
+    _use_token_identity(monkeypatch)
+    app.config["TESTING"] = True
+    app.config["RULE_GENERATION_RECEIPT_SECRET"] = None
+    agent = FakeAuthoringAgent()
+    repository = FakeRuleRepository()
+    app.extensions["data_rule_authoring_agent"] = agent
+    app.extensions["data_rule_repository"] = repository
+    client = app.test_client()
+
+    response = client.post(
+        "/api/rules/interpret",
+        json={
+            "source_text": "手机号去空格后必须为11位数字",
+            "authoring_surface": "data_standard",
+            "context": {
+                "input_schema_snapshot_id": new_governance_uid(),
+                "output_schema_snapshot_id": new_governance_uid(),
+                "input_sample_artifact_ref": (
+                    "minio://trusted/input.parquet"
+                ),
+                "golden_output_artifact_ref": None,
+            },
+        },
+        headers=_headers(app, "editor"),
+    )
+
+    assert response.status_code == 503
+    assert agent.calls == []
+    assert repository.calls == []
+
+
 def test_rule_interpret_and_validate_reject_unknown_fields(monkeypatch):
     from app import create_app