fix: harden governed SQL plan provenance

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

@@ -0,0 +1,337 @@
+# Task 4 report — SQLGlot executable vertical slice
+
+## Status
+
+Implemented the deployment-bound SQL vertical slice for PostgreSQL and MySQL.
+Logical DataFlow release remains a semantic, binding-free reference. Physical
+dataset bindings are consumed only by `BoundSqlPlanService`, which selects a
+registered concrete compiler and persists the resulting plan as `compiled`.
+The Runner still requires the exact bound plan and RuleVersion to be
+`published`; full test-evidence publication lifecycle hardening remains Task 7.
+
+## RED / GREEN evidence
+
+Initial RED command:
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/core/data_rules/test_sql_compiler.py \
+tests/runner/test_rule_sql.py
+```
+
+Initial RED result:
+
+```text
+18 failed in 1.38s
+```
+
+Representative expected failures:
+
+```text
+ModuleNotFoundError: No module named 'app.core.data_rules.compilers'
+ModuleNotFoundError: No module named 'app.runner.rule_sql'
+```
+
+Additional RED checks were added before each hardening fix:
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/core/data_rules/test_sql_compiler.py::test_repository_persists_bound_plan_only_for_one_deployment_linkage
+1 failed in 1.05s
+AttributeError: 'DataRuleRepository' object has no attribute 'persist_bound_component_plan'
+```
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/runner/test_rule_sql.py::test_sqlglot_rule_adapter_rejects_unimplemented_idempotency_strategy \
+tests/runner/test_rule_sql.py::test_rule_executor_matches_node_idempotency_to_persisted_component
+2 failed in 1.09s
+```
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/core/data_rules/test_sql_compiler.py::test_repository_persists_bound_plan_only_for_one_deployment_linkage
+1 failed in 0.88s
+Failed: DID NOT RAISE ValueError
+```
+
+This proved a compiled plan was not yet checked against the physical source
+and target relations stored for the deployment.
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/core/data_rules/test_sql_compiler.py::test_sql_compiler_fails_closed_for_unsupported_operations_and_binding_mismatch
+1 failed in 0.86s
+Failed: DID NOT RAISE ValueError
+```
+
+This proved `assert.on_failure=quarantine` was being accepted without a
+quarantine destination, so the compiler was tightened to support only
+`reject`.
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/runner/test_rule_sql.py::test_sqlglot_rule_adapter_verifies_and_executes_one_transaction
+1 failed in 0.98s
+```
+
+This proved affected-row counts were not a stable `rows_out` metric for
+idempotent upserts. The adapter now counts the compiled accepted-row query
+inside the same transaction before executing the write.
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/core/data_rules/test_sql_compiler.py::test_sql_compiler_rejects_unproven_operator_type_semantics
+5 failed in 0.90s
+```
+
+This proved text, fill, derive, map, and final output types required additional
+fail-closed compiler checks.
+
+Focused GREEN after the implementation and hardening:
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/core/data_rules/test_sql_compiler.py \
+tests/runner/test_rule_sql.py \
+tests/runner/test_rules.py
+33 passed in 0.84s
+```
+
+Related release/repository/bootstrap regression GREEN:
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/core/data_rules/test_release.py \
+tests/core/data_rules/test_data_rule_repository.py \
+tests/runner/test_bootstrap.py \
+tests/runner/test_rules.py \
+tests/core/data_rules/test_execution_contracts.py
+49 passed in 0.89s
+```
+
+## Real PostgreSQL and MySQL integration
+
+The source containers were already healthy on PostgreSQL port `25432` and
+MySQL port `23306`. The integration test:
+
+- creates only `task4_rule_source` and `task4_rule_target`;
+- inserts one accepted and one rejected customer;
+- compiles and persists a bound plan as `compiled`;
+- performs a real dialect preflight and records its successful evidence before
+  transitioning the in-test plan record to `published`;
+- executes through `RulePlanExecutor`;
+- verifies transformed target rows and exact `rows_in`, `rows_out`, and
+  `rows_rejected`;
+- repeats the upsert to prove idempotency for both dialects;
+- mutates the stored plan body without changing its hash and verifies
+  fail-closed execution;
+- drops only the two Task 4 tables in `finally`.
+
+Direct dialect integration result:
+
+```text
+TEST_DATABASE_URL=postgresql://dataops:dataops-test-password@127.0.0.1:15432/dataops \
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/integration/test_data_rule_sql_execution.py
+2 passed in 0.93s
+```
+
+Required vertical-slice command:
+
+```text
+TEST_DATABASE_URL=postgresql://dataops:dataops-test-password@127.0.0.1:15432/dataops \
+PYTHONPATH=. .venv/bin/pytest -q \
+tests/core/data_rules/test_sql_compiler.py \
+tests/runner/test_rule_sql.py \
+tests/integration/test_data_rule_sql_execution.py
+28 passed in 0.97s
+```
+
+## Full verification
+
+```text
+PYTHONPATH=. .venv/bin/pytest -q
+477 passed, 26 skipped, 59 subtests passed in 4.16s
+```
+
+```text
+.venv/bin/ruff check \
+app/core/data_rules/compilers \
+app/core/data_rules/release.py \
+app/core/data_rules/repository.py \
+app/runner/rule_sql.py \
+app/runner/rules.py \
+app/runner/bootstrap.py \
+tests/core/data_rules/test_sql_compiler.py \
+tests/runner/test_rule_sql.py \
+tests/runner/test_rules.py \
+tests/integration/test_data_rule_sql_execution.py
+All checks passed!
+```
+
+`git diff --check` produced no output.
+
+## Dependency and license
+
+- Added and installed `sqlglot==30.13.0`.
+- Installed package metadata reports `License-Expression: MIT`.
+- Installed package metadata reports `Requires-Python: >=3.9`.
+- The compiler provenance stored with bound plans is
+  `dataops-sqlglot-30.13.0`.
+
+## Files
+
+- `requirements.txt`
+- `app/core/data_rules/compilers/__init__.py`
+- `app/core/data_rules/compilers/base.py`
+- `app/core/data_rules/compilers/sql.py`
+- `app/core/data_rules/release.py`
+- `app/core/data_rules/repository.py`
+- `app/runner/rule_sql.py`
+- `app/runner/rules.py`
+- `app/runner/bootstrap.py`
+- `tests/core/data_rules/test_sql_compiler.py`
+- `tests/runner/test_rule_sql.py`
+- `tests/runner/test_rules.py`
+- `tests/integration/test_data_rule_sql_execution.py`
+- `.superpowers/sdd/task-4-report.md`
+
+## Self-review
+
+- All user identifiers are converted to quoted SQLGlot `Identifier` nodes only
+  after the closed identifier check. RuleSpec values become compiler-owned
+  named placeholders and never enter SQL text.
+- The compiler validates the published RuleVersion hash, server-owned input
+  and output snapshots, binding IDs, same datasource, relation kinds, access
+  modes, dialect, concrete timezone/collation/rounding/regex capabilities, and
+  final output field types.
+- The first executable operator slice covers `cast`, trim-only
+  `normalize_text`, portable `regex_replace`, type-compatible `fill_null`,
+  typed `filter`, typed `derive`, reject-only `assert`, deterministic
+  `deduplicate`, and string `map_values`. Unsupported or semantically
+  unprovable variants fail closed.
+- Bound-plan persistence re-parses the plan and proves its source relation,
+  target relation, datasource UID, dialect, binding IDs, and schema hashes
+  match one deployment owning the logical component.
+- Runner execution re-validates the closed plan and SQL AST, exact hash,
+  RuleVersion ID, datasource dialect/capabilities, write authorization, and
+  persisted component idempotency. Only AST-built `upsert` is implemented;
+  the other declared orchestration strategies fail closed at this adapter.
+- Raw input count, accepted output count, and the AST-built upsert execute
+  within one datasource transaction. Known commit failures return
+  `not_committed`; commit ambiguity returns `unknown`.
+- The incorrect `quality_check -> SQL write adapter` alias was removed.
+  `SqlGlotQualityPlanAdapter` explicitly fails closed until a read-only
+  quality plan contract is implemented.
+
+## Concerns / deferred work
+
+- Production publication and revocation of compiled bound plans, including
+  durable test evidence and approval transitions, remain Task 7. This task
+  deliberately persists only `compiled` and does not claim tested/published
+  production state.
+- Only same-datasource table/view-to-table SQL pushdown is supported. Cross
+  source, lookup joins, aggregate, mask, lower/upper Unicode normalization,
+  quarantine writes, and non-append output bindings fail closed.
+- `partition_replace` and `deduplication_key` remain valid orchestration
+  contract values but are not executable by this SQL adapter; only an exact
+  persisted `upsert` binding is accepted.
+- The datasource definition must carry server-owned
+  `sql_rule_capabilities`. Existing deployments without those concrete
+  capabilities will fail closed until Task 9 binds and validates them.
+- The explicit quality adapter is intentionally non-executable. A separate
+  read-only quality plan and evidence contract is needed before quality nodes
+  can run.
+
+---
+
+# Task 4 review-fix pass — canonical provenance and publication safety
+
+Date: 2026-07-23
+
+## Outcome
+
+All critical, important, and identifier-limit review findings were corrected.
+The bound SQL compiler now accepts only canonical identifiers at its public
+service boundary. RuleVersion, schema snapshots, deployment bindings, and the
+server SQL capability profile are loaded through one repository join before
+compilation. Caller-provided RuleSpec, schema, binding, or capability objects
+are no longer accepted.
+
+The plan attests the exact compiler version, RuleSpec hash, input/output schema
+snapshot IDs and hashes, and deployment binding IDs. Persistence compares
+those attestations against the canonical database rows. The runner loads and
+rechecks the same canonical RuleVersion and snapshots before dispatch.
+
+Logical release plans now persist as `compiled`, and released package plan
+references explicitly identify themselves as `plan_kind=semantic` and
+`status=compiled`. Physical plans use the explicit lifecycle
+`compiled -> tested -> published`; only successful `integration_preflight`
+evidence can produce `tested`, and only an attested tested plan whose
+RuleVersion remains published can be published. The runner rejects both
+compiled and tested plans.
+
+## Correctness and security changes
+
+- PostgreSQL upsert now proves that the requested idempotency column is an
+  exact single-column PRIMARY KEY or UNIQUE constraint using live server
+  metadata.
+- MySQL upsert proves the same exact key and rejects any alternate unique
+  index, preventing `ON DUPLICATE KEY UPDATE` from merging a different
+  logical row through another unique collision path.
+- SQL casts accept only `on_error=fail`; reject, quarantine, and warn fail
+  closed.
+- Final schema compatibility is exact. Numeric type changes require an
+  explicit cast.
+- Deduplication extends the requested ordering with every remaining projected
+  field and explicit null ordering, yielding a deterministic total value
+  order; fully identical rows remain equivalent.
+- The runner requires the exact supported `dataops-sqlglot-30.13.0`
+  attestation and validates a closed INSERT-SELECT SQLGlot AST subset.
+  Anonymous and non-allowlisted functions, joins, CTEs, set operations, and
+  other out-of-compiler nodes are rejected even if a malicious publisher
+  recomputes the plan hash.
+- PostgreSQL identifiers are bounded to 63 UTF-8 bytes and MySQL identifiers
+  to 64 UTF-8 bytes for schemas, tables, schema fields, expression fields,
+  step columns, and derived/deduplication fields.
+- A new migration adds `tested` to the durable execution-plan status
+  constraint.
+
+## TDD evidence
+
+The initial RED run produced the expected failures for the old caller-object
+API, canonical component-to-rule mismatch, and function-bearing plan.
+Additional regression tests cover non-fail cast actions, exact numeric types,
+total deduplication order, UTF-8 identifier byte limits, logical compiled
+status, registry availability before release allocation, canonical runner
+attestation tampering, and lifecycle evidence.
+
+Final verification:
+
+```text
+Focused compiler/repository/release/runner plus real database integration:
+63 passed
+
+Real PostgreSQL/MySQL execution and MySQL uniqueness failure cases:
+3 passed
+
+Full repository suite:
+496 passed, 26 skipped, 59 subtests passed
+
+Ruff on all changed Python and migration files:
+All checks passed!
+
+git diff --check:
+no output
+```
+
+## Remaining bounded concerns
+
+- The server capability profile is intentionally a closed policy profile for
+  the two supported dialects (`C`/POSIX for PostgreSQL and
+  `utf8mb4_0900_bin`/ICU for MySQL). A deployment with different concrete
+  datasource capabilities fails closed at runner dispatch.
+- Only same-datasource table/view-to-table SQL pushdown and exact-key upsert
+  are executable in this slice. Other write strategies and the read-only
+  quality executor remain fail-closed.

app/core/data_rules/compilers/sql.py

@@ -48,10 +48,16 @@ CAPABILITY_KEYS = {
 }
 PLAN_KEYS = {
     "schema_version",
+    "compiler_version",
     "dialect",
     "capabilities",
     "data_source_uid",
     "rule_version_id",
+    "rule_spec_hash",
+    "input_schema_snapshot_id",
+    "input_schema_hash",
+    "output_schema_snapshot_id",
+    "output_schema_hash",
     "input_binding_id",
     "output_binding_id",
     "statements",
@@ -62,7 +68,7 @@ RESULT_CONTRACT = {
     "rows_out": "counted",
     "rows_rejected": "counted",
 }
-_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,127}$")
+_PARAMETER_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,99}$")
 _PORTABLE_REGEX_FORBIDDEN = re.compile(r"\(\?|\\[1-9]|\\[pP]")
 _NUMERIC_TYPES = {"integer", "decimal", "float", "double"}
 _TYPE_NAMES = {
@@ -81,10 +87,7 @@ _TYPE_NAMES = {
 
 
 def _types_compatible(source_type: str, target_type: str) -> bool:
-    return source_type == target_type or {
-        source_type,
-        target_type,
-    } <= _NUMERIC_TYPES
+    return source_type == target_type
 
 
 def _value_matches_type(value: Any, field_type: str) -> bool:
@@ -136,10 +139,15 @@ def _dialect(value: Any) -> str:
     return "postgresql" if normalized == "postgres" else normalized
 
 
-def _identifier(value: Any, label: str) -> str:
+def _identifier(value: Any, label: str, dialect: str) -> str:
     normalized = str(value or "")
-    if _IDENTIFIER.fullmatch(normalized) is None:
+    if not normalized.isidentifier():
         raise ValueError(f"{label} must be a SQL identifier")
+    limit = 63 if dialect == "postgresql" else 64
+    if len(normalized.encode("utf-8")) > limit:
+        raise ValueError(
+            f"{label} exceeds the {dialect} {limit}-byte identifier limit"
+        )
     return normalized
 
 
@@ -151,12 +159,12 @@ def _column(value: str) -> exp.Column:
     return exp.Column(this=_quoted_identifier(value))
 
 
-def _table(object_ref: Any) -> exp.Table:
+def _table(object_ref: Any, dialect: str) -> exp.Table:
     parts = str(object_ref or "").split(".")
     if len(parts) != 2:
         raise ValueError("table object_ref must be schema.name")
-    schema_name = _identifier(parts[0], "table schema")
-    table_name = _identifier(parts[1], "table name")
+    schema_name = _identifier(parts[0], "table schema", dialect)
+    table_name = _identifier(parts[1], "table name", dialect)
     return exp.Table(
         this=_quoted_identifier(table_name),
         db=_quoted_identifier(schema_name),
@@ -248,7 +256,9 @@ class _ExpressionCompiler:
     def _compile(self, node: dict[str, Any]) -> exp.Expression:
         kind = node["kind"]
         if kind == "identifier":
-            return _column(_identifier(node["name"], "expression field"))
+            return _column(
+                _identifier(node["name"], "expression field", self.dialect)
+            )
         if kind == "literal":
             literal_type = node["type"]
             if literal_type == "null":
@@ -392,6 +402,8 @@ def validate_bound_sql_plan(value: Any) -> dict[str, Any]:
     plan = copy.deepcopy(value)
     if plan["schema_version"] != PLAN_SCHEMA_VERSION:
         raise ValueError("unsupported bound SQL plan schema version")
+    if plan["compiler_version"] != COMPILER_VERSION:
+        raise ValueError("unsupported bound SQL compiler attestation")
     dialect = _dialect(plan["dialect"])
     if dialect not in SUPPORTED_DIALECTS:
         raise ValueError("unsupported bound SQL plan dialect")
@@ -404,10 +416,20 @@ def validate_bound_sql_plan(value: Any) -> dict[str, Any]:
     for key in (
         "data_source_uid",
         "rule_version_id",
+        "input_schema_snapshot_id",
+        "output_schema_snapshot_id",
         "input_binding_id",
         "output_binding_id",
     ):
         plan[key] = _uid(plan[key], key)
+    for key in (
+        "rule_spec_hash",
+        "input_schema_hash",
+        "output_schema_hash",
+    ):
+        value = str(plan[key] or "")
+        if re.fullmatch(r"[0-9a-f]{64}", value) is None:
+            raise ValueError(f"{key} must be a sha256 digest")
     statements = plan["statements"]
     if not isinstance(statements, list) or len(statements) != 1:
         raise ValueError("bound SQL plan must contain one statement")
@@ -436,11 +458,111 @@ def validate_bound_sql_plan(value: Any) -> dict[str, Any]:
     if len(expressions) != 1 or not isinstance(expressions[0], exp.Insert):
         raise ValueError("bound SQL statement must be one INSERT")
     parsed = expressions[0]
+    if parsed.args.get("conflict") is not None or parsed.args.get(
+        "returning"
+    ) is not None:
+        raise ValueError("bound SQL statement contains unsupported INSERT clauses")
+    query = parsed.expression
+    if not isinstance(query, exp.Select):
+        raise ValueError("bound SQL INSERT source must be a SELECT")
     if any(
-        isinstance(node, (exp.Command, exp.Delete, exp.Update))
+        isinstance(
+            node,
+            (
+                exp.Anonymous,
+                exp.Command,
+                exp.Delete,
+                exp.Update,
+                exp.Join,
+                exp.Union,
+                exp.Intersect,
+                exp.Except,
+                exp.CTE,
+            ),
+        )
         for node in parsed.walk()
     ):
         raise ValueError("bound SQL statement contains unsupported operations")
+    allowed_functions = (
+        exp.Add,
+        exp.Abs,
+        exp.And,
+        exp.Cast,
+        exp.Case,
+        exp.Coalesce,
+        exp.If,
+        exp.Length,
+        exp.Mod,
+        exp.Mul,
+        exp.Neg,
+        exp.Not,
+        exp.Or,
+        exp.RegexpLike,
+        exp.RegexpReplace,
+        exp.Round,
+        exp.RowNumber,
+        exp.Trim,
+        exp.Div,
+        exp.Sub,
+    )
+    if any(
+        isinstance(node, exp.Func) and not isinstance(node, allowed_functions)
+        for node in parsed.walk()
+    ):
+        raise ValueError("bound SQL statement contains an unsupported function")
+    allowed_ast_nodes = (
+        exp.Abs,
+        exp.Add,
+        exp.Alias,
+        exp.And,
+        exp.Case,
+        exp.Cast,
+        exp.Coalesce,
+        exp.Column,
+        exp.DataType,
+        exp.Div,
+        exp.EQ,
+        exp.From,
+        exp.GT,
+        exp.GTE,
+        exp.Identifier,
+        exp.If,
+        exp.Insert,
+        exp.Is,
+        exp.Length,
+        exp.Literal,
+        exp.LT,
+        exp.LTE,
+        exp.Mod,
+        exp.Mul,
+        exp.NEQ,
+        exp.Neg,
+        exp.Not,
+        exp.Null,
+        exp.Or,
+        exp.Order,
+        exp.Ordered,
+        exp.Placeholder,
+        exp.RegexpLike,
+        exp.RegexpReplace,
+        exp.Round,
+        exp.RowNumber,
+        exp.Schema,
+        exp.Select,
+        exp.Sub,
+        exp.Subquery,
+        exp.Table,
+        exp.TableAlias,
+        exp.Trim,
+        exp.Where,
+        exp.Window,
+    )
+    if any(
+        not isinstance(node, allowed_ast_nodes) for node in parsed.walk()
+    ):
+        raise ValueError(
+            "bound SQL statement is outside the compiler AST subset"
+        )
     source_tables = list(parsed.expression.find_all(exp.Table))
     if len(source_tables) != 1:
         raise ValueError("bound SQL plan must read exactly one source table")
@@ -448,7 +570,7 @@ def validate_bound_sql_plan(value: Any) -> dict[str, Any]:
     if not isinstance(parameters, dict) or len(parameters) > 500:
         raise ValueError("bound SQL parameters must be a bounded object")
     for name, item in parameters.items():
-        if _IDENTIFIER.fullmatch(str(name)) is None or not isinstance(
+        if _PARAMETER_IDENTIFIER.fullmatch(str(name)) is None or not isinstance(
             item, (type(None), bool, int, float, str)
         ):
             raise ValueError("bound SQL parameters are invalid")
@@ -562,13 +684,17 @@ class SqlGlotRuleCompiler(RuleCompiler):
         field_types = {
             field["name"]: field["type"] for field in source_schema["fields"]
         }
+        for field in (*source_schema["fields"], *target_schema["fields"]):
+            _identifier(field["name"], "schema field name", self.dialect)
         supported = validate_rule_expressions(spec, field_types)
         if self.dialect not in supported:
             raise ValueError("rule expressions are unsupported by the bound dialect")
         input_fields = [field["name"] for field in source_schema["fields"]]
         output_fields = [field["name"] for field in target_schema["fields"]]
         current_fields = list(input_fields)
-        source: exp.Expression = _table(source_binding["object_ref"])
+        source: exp.Expression = _table(
+            source_binding["object_ref"], self.dialect
+        )
         parameters = _ParameterStore()
         expression_compiler = _ExpressionCompiler(
             dialect=self.dialect,
@@ -605,10 +731,16 @@ class SqlGlotRuleCompiler(RuleCompiler):
                 "normalize_text",
                 "regex_replace",
             }:
-                column_name = _identifier(step.get("column"), f"{operation} column")
+                column_name = _identifier(
+                    step.get("column"), f"{operation} column", self.dialect
+                )
                 if column_name not in current_fields:
                     raise ValueError(f"unknown SQL rule column: {column_name}")
             if operation == "cast":
+                if step.get("on_error", "fail") != "fail":
+                    raise ValueError(
+                        "SQL cast supports only fail on_error semantics"
+                    )
                 target_type = str(step.get("to") or "").strip().lower()
                 type_name = _TYPE_NAMES.get(target_type)
                 if type_name is None:
@@ -670,7 +802,9 @@ class SqlGlotRuleCompiler(RuleCompiler):
                 predicate = expression_compiler.compile(step["expression_ast"])
                 replace({}, where=predicate)
             elif operation == "derive":
-                target = _identifier(step.get("target"), "derive target")
+                target = _identifier(
+                    step.get("target"), "derive target", self.dialect
+                )
                 if target not in current_fields and target not in output_fields:
                     raise ValueError("derive target is not in the output schema")
                 result_type = type_check_expression(
@@ -720,11 +854,15 @@ class SqlGlotRuleCompiler(RuleCompiler):
                 replace({column_name: case})
             elif operation == "deduplicate":
                 keys = [
-                    _identifier(item, "deduplicate key")
+                    _identifier(item, "deduplicate key", self.dialect)
                     for item in step.get("keys", [])
                 ]
                 order_by = [
-                    _identifier(item, "deduplicate order field")
+                    _identifier(
+                        item,
+                        "deduplicate order field",
+                        self.dialect,
+                    )
                     for item in step.get("order_by", [])
                 ]
                 if (
@@ -736,6 +874,14 @@ class SqlGlotRuleCompiler(RuleCompiler):
                         "deduplicate requires known keys and deterministic order"
                     )
                 descending = step.get("keep", "first") == "last"
+                total_order = [
+                    *order_by,
+                    *[
+                        name
+                        for name in current_fields
+                        if name not in order_by
+                    ],
+                ]
                 row_number = exp.Window(
                     this=exp.RowNumber(),
                     partition_by=[_column(item) for item in keys],
@@ -744,8 +890,9 @@ class SqlGlotRuleCompiler(RuleCompiler):
                             exp.Ordered(
                                 this=_column(item),
                                 desc=descending,
+                                nulls_first=False,
                             )
-                            for item in order_by
+                            for item in total_order
                         ]
                     ),
                 ).as_("_dataops_row_number")
@@ -787,7 +934,7 @@ class SqlGlotRuleCompiler(RuleCompiler):
             expressions=[_column(name) for name in output_fields]
         ).from_(source)
         target = exp.Schema(
-            this=_table(target_binding["object_ref"]),
+            this=_table(target_binding["object_ref"], self.dialect),
             expressions=[_quoted_identifier(name) for name in output_fields],
         )
         statement_ast = exp.Insert(this=target, expression=final_select)
@@ -795,10 +942,16 @@ class SqlGlotRuleCompiler(RuleCompiler):
         plan = validate_bound_sql_plan(
             {
                 "schema_version": PLAN_SCHEMA_VERSION,
+                "compiler_version": COMPILER_VERSION,
                 "dialect": self.dialect,
                 "capabilities": capabilities,
                 "data_source_uid": source_binding["data_source_uid"],
                 "rule_version_id": rule_version_id,
+                "rule_spec_hash": rule_version["spec_hash"],
+                "input_schema_snapshot_id": source_schema["id"],
+                "input_schema_hash": source_schema["schema_hash"],
+                "output_schema_snapshot_id": target_schema["id"],
+                "output_schema_hash": target_schema["schema_hash"],
                 "input_binding_id": source_binding["id"],
                 "output_binding_id": target_binding["id"],
                 "statements": [

app/core/data_rules/production_line.py

@@ -12,13 +12,12 @@ from typing import Any
 from app.core.common.identifiers import ensure_governance_uid
 from app.core.data_rules.contracts import (
     dataflow_spec_hash,
+    read_rule_spec,
     rule_spec_hash,
     validate_dataflow_spec,
-    read_rule_spec,
 )
 from app.core.orchestration.spec import validate_workflow_spec
 
-
 PLAN_KEYS = {"backend", "plan_hash"}
 PLAN_BACKENDS = {
     "sql_pushdown",
@@ -72,7 +71,12 @@ def _execution_plan(rule_version: dict[str, Any]) -> dict[str, str]:
     plan_hash = str(plan.get("plan_hash") or "")
     if not re.fullmatch(r"[0-9a-f]{64}", plan_hash):
         raise ValueError("execution plan hash must be a sha256 hex digest")
-    return {"backend": backend, "plan_hash": plan_hash}
+    return {
+        "backend": backend,
+        "plan_hash": plan_hash,
+        "plan_kind": "semantic",
+        "status": "compiled",
+    }
 
 
 def _validated_rule_version(
@@ -218,6 +222,8 @@ def resolve_production_line(
                         "rule_version_id": rule_id,
                         "plan_hash": plan["plan_hash"],
                         "backend": plan["backend"],
+                        "plan_kind": plan["plan_kind"],
+                        "status": plan["status"],
                     }
                 )
             continue
@@ -242,6 +248,8 @@ def resolve_production_line(
                 "rule_version_id": rule_id,
                 "plan_hash": plan["plan_hash"],
                 "backend": plan["backend"],
+                "plan_kind": plan["plan_kind"],
+                "status": plan["status"],
             }
         )
 
@@ -250,7 +258,7 @@ def resolve_production_line(
         raise ValueError("resolved production-line node ids must be unique")
     edges = [
         {"from": source, "to": target}
-        for source, target in zip(node_ids, node_ids[1:])
+        for source, target in zip(node_ids, node_ids[1:], strict=False)
     ]
     workflow_spec = {
         "schema_version": "1.0",

app/core/data_rules/release.py

@@ -28,9 +28,20 @@ def _source(value: Any) -> str:
 
 
 class ProductionLineReleaseService:
-    def __init__(self, repository, *, schema_resolver=None):
+    def __init__(
+        self,
+        repository,
+        *,
+        schema_resolver=None,
+        available_plan_backends=None,
+    ):
         self.repository = repository
         self.schema_resolver = schema_resolver
+        self.available_plan_backends = frozenset(
+            available_plan_backends
+            if available_plan_backends is not None
+            else {"sql_pushdown", "polars_batch", "quality_check"}
+        )
 
     def release(
         self,
@@ -105,6 +116,20 @@ class ProductionLineReleaseService:
             else:
                 validate_published_rule(component["rule_version_id"])
 
+        compiled: dict[str, dict[str, Any]] = {}
+        for rule_version_id, supported_backends in sorted(
+            rule_backend_support.items()
+        ):
+            plan = compile_rule_plan(
+                rules[rule_version_id],
+                supported_backends=supported_backends,
+            )
+            if plan["backend"] not in self.available_plan_backends:
+                raise ValueError(
+                    f"plan backend {plan['backend']} is not registered"
+                )
+            compiled[rule_version_id] = plan
+
         version = self.repository.begin_dataflow_release(
             dataflow_spec=flow,
             source_text=source,
@@ -113,7 +138,6 @@ class ProductionLineReleaseService:
             created_by=actor,
         )
         version_id = _uid(version.get("id"), "dataflow_version_id")
-        compiled: dict[str, dict[str, Any]] = {}
         binding_ids: dict[str, str] = {}
 
         def add_binding(
@@ -132,11 +156,6 @@ class ProductionLineReleaseService:
                 raise ValueError(
                     f"published rule version {rule_version_id} was not found"
                 )
-            if rule_version_id not in compiled:
-                compiled[rule_version_id] = compile_rule_plan(
-                    rule,
-                    supported_backends=rule_backend_support[rule_version_id],
-                )
             plan = compiled[rule_version_id]
             binding_id = new_governance_uid()
             binding_ids[binding_key] = binding_id
@@ -232,16 +251,55 @@ class BoundSqlPlanService:
         self,
         *,
         component_binding_id: str,
-        rule_version: dict[str, Any],
-        input_schema: dict[str, Any],
-        output_schema: dict[str, Any],
-        input_binding: dict[str, Any],
-        output_binding: dict[str, Any],
-        backend: dict[str, Any],
+        rule_version_id: str,
+        input_schema_snapshot_id: str,
+        output_schema_snapshot_id: str,
+        input_binding_id: str,
+        output_binding_id: str,
     ) -> dict[str, Any]:
-        component_id = _uid(
-            component_binding_id, "component_binding_id"
-        )
+        ids = {
+            "component_binding_id": _uid(
+                component_binding_id, "component_binding_id"
+            ),
+            "rule_version_id": _uid(rule_version_id, "rule_version_id"),
+            "input_schema_snapshot_id": _uid(
+                input_schema_snapshot_id, "input_schema_snapshot_id"
+            ),
+            "output_schema_snapshot_id": _uid(
+                output_schema_snapshot_id, "output_schema_snapshot_id"
+            ),
+            "input_binding_id": _uid(input_binding_id, "input_binding_id"),
+            "output_binding_id": _uid(output_binding_id, "output_binding_id"),
+        }
+        context = self.repository.load_bound_compile_context(**ids)
+        if not isinstance(context, dict):
+            raise ValueError("canonical bound compile context was not found")
+        try:
+            component = context["component_binding"]
+            rule_version = context["rule_version"]
+            input_schema = context["input_schema"]
+            output_schema = context["output_schema"]
+            input_binding = context["input_binding"]
+            output_binding = context["output_binding"]
+            backend = context["backend"]
+        except KeyError as exc:
+            raise ValueError(
+                "canonical bound compile context is incomplete"
+            ) from exc
+        canonical_ids = {
+            "component_binding_id": component.get("id"),
+            "rule_version_id": rule_version.get("id"),
+            "input_schema_snapshot_id": input_schema.get("id"),
+            "output_schema_snapshot_id": output_schema.get("id"),
+            "input_binding_id": input_binding.get("id"),
+            "output_binding_id": output_binding.get("id"),
+        }
+        if canonical_ids != ids or component.get(
+            "rule_version_id"
+        ) != ids["rule_version_id"]:
+            raise ValueError(
+                "canonical bound compile context identifiers do not match"
+            )
         compiler = self.compiler_registry.select(
             rule_version.get("rule_spec"),
             input_binding,
@@ -256,16 +314,10 @@ class BoundSqlPlanService:
             backend=backend,
         )
         return self.repository.persist_bound_component_plan(
-            component_binding_id=component_id,
-            rule_version_id=_uid(
-                rule_version.get("id"), "rule_version_id"
-            ),
-            input_binding_id=_uid(
-                input_binding.get("id"), "input_binding_id"
-            ),
-            output_binding_id=_uid(
-                output_binding.get("id"), "output_binding_id"
-            ),
+            component_binding_id=ids["component_binding_id"],
+            rule_version_id=ids["rule_version_id"],
+            input_binding_id=ids["input_binding_id"],
+            output_binding_id=ids["output_binding_id"],
             compiled=compiled,
             status="compiled",
         )

app/core/data_rules/repository.py

@@ -865,7 +865,7 @@ class DataRuleRepository:
                 "plan_hash, schema_hashes, status) "
                 "VALUES (CAST(:id AS uuid), CAST(:binding_id AS uuid), "
                 ":backend, :compiler_version, CAST(:plan AS jsonb), "
-                ":plan_hash, CAST(:schema_hashes AS jsonb), 'published')"
+                ":plan_hash, CAST(:schema_hashes AS jsonb), 'compiled')"
             ),
             {
                 "id": new_governance_uid(),
@@ -878,6 +878,157 @@ class DataRuleRepository:
             },
         )
 
+    def load_bound_compile_context(
+        self,
+        *,
+        component_binding_id: str,
+        rule_version_id: str,
+        input_schema_snapshot_id: str,
+        output_schema_snapshot_id: str,
+        input_binding_id: str,
+        output_binding_id: str,
+    ) -> dict[str, Any] | None:
+        """Load all physical compilation inputs through one canonical join."""
+
+        ids = {
+            "component_binding_id": _uid(
+                component_binding_id, "component_binding_id"
+            ),
+            "rule_version_id": _uid(rule_version_id, "rule_version_id"),
+            "input_schema_snapshot_id": _uid(
+                input_schema_snapshot_id, "input_schema_snapshot_id"
+            ),
+            "output_schema_snapshot_id": _uid(
+                output_schema_snapshot_id, "output_schema_snapshot_id"
+            ),
+            "input_binding_id": _uid(input_binding_id, "input_binding_id"),
+            "output_binding_id": _uid(output_binding_id, "output_binding_id"),
+        }
+        row = (
+            self.session.execute(
+                text(
+                    "SELECT cb.id::text AS component_binding_id, "
+                    "cb.rule_version_id::text AS component_rule_version_id, "
+                    "cb.component_kind, cb.idempotency, "
+                    "rv.id::text AS rule_version_id, rv.rule_spec, "
+                    "rv.spec_hash, rv.status AS rule_status, "
+                    "ins.id::text AS input_schema_snapshot_id, "
+                    "ins.schema_ref AS input_schema_ref, "
+                    "ins.schema_hash AS input_schema_hash, "
+                    "ins.fields AS input_schema_fields, "
+                    "ins.source_revision AS input_source_revision, "
+                    "outs.id::text AS output_schema_snapshot_id, "
+                    "outs.schema_ref AS output_schema_ref, "
+                    "outs.schema_hash AS output_schema_hash, "
+                    "outs.fields AS output_schema_fields, "
+                    "outs.source_revision AS output_source_revision, "
+                    "ib.id::text AS input_binding_id, "
+                    "ib.data_source_uid::text AS input_data_source_uid, "
+                    "ib.object_kind AS input_object_kind, "
+                    "ib.object_ref AS input_object_ref, "
+                    "ib.access_mode AS input_access_mode, "
+                    "ib.dialect AS input_dialect, "
+                    "ib.write_mode AS input_write_mode, "
+                    "ob.id::text AS output_binding_id, "
+                    "ob.data_source_uid::text AS output_data_source_uid, "
+                    "ob.object_kind AS output_object_kind, "
+                    "ob.object_ref AS output_object_ref, "
+                    "ob.access_mode AS output_access_mode, "
+                    "ob.dialect AS output_dialect, "
+                    "ob.write_mode AS output_write_mode "
+                    "FROM public.dataflow_component_bindings cb "
+                    "JOIN public.data_rule_versions rv "
+                    "ON rv.id = cb.rule_version_id "
+                    "JOIN public.dataflow_deployments d "
+                    "ON d.dataflow_version_id = cb.dataflow_version_id "
+                    "JOIN public.dataflow_dataset_bindings ib "
+                    "ON ib.dataflow_deployment_id = d.id "
+                    "JOIN public.data_schema_snapshots ins "
+                    "ON ins.id = ib.schema_snapshot_id "
+                    "JOIN public.dataflow_dataset_bindings ob "
+                    "ON ob.dataflow_deployment_id = d.id "
+                    "JOIN public.data_schema_snapshots outs "
+                    "ON outs.id = ob.schema_snapshot_id "
+                    "WHERE cb.id = CAST(:component_binding_id AS uuid) "
+                    "AND rv.id = CAST(:rule_version_id AS uuid) "
+                    "AND ins.id = CAST(:input_schema_snapshot_id AS uuid) "
+                    "AND outs.id = CAST(:output_schema_snapshot_id AS uuid) "
+                    "AND ib.id = CAST(:input_binding_id AS uuid) "
+                    "AND ob.id = CAST(:output_binding_id AS uuid) "
+                    "/* canonical_bound_compile_context */"
+                ),
+                ids,
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if row is None:
+            return None
+
+        def binding(prefix: str, snapshot_id: str) -> dict[str, Any]:
+            return {
+                "id": str(row[f"{prefix}_binding_id"]),
+                "data_source_uid": str(row[f"{prefix}_data_source_uid"]),
+                "object_kind": str(row[f"{prefix}_object_kind"]),
+                "object_ref": str(row[f"{prefix}_object_ref"]),
+                "schema_snapshot_id": snapshot_id,
+                "access_mode": str(row[f"{prefix}_access_mode"]),
+                "dialect": str(row[f"{prefix}_dialect"]),
+                "write_mode": row[f"{prefix}_write_mode"],
+            }
+
+        input_snapshot_id = str(row["input_schema_snapshot_id"])
+        output_snapshot_id = str(row["output_schema_snapshot_id"])
+        rule_spec = _object(row["rule_spec"], "rule_spec")
+        dialect = str(row["input_dialect"]).strip().lower()
+        dialect = "postgresql" if dialect == "postgres" else dialect
+        if dialect not in {"postgresql", "mysql"}:
+            raise ValueError("canonical SQL binding dialect is unsupported")
+        backend = {
+            "dialect": dialect,
+            "timezone": str(rule_spec.get("timezone") or ""),
+            "collation": (
+                "C" if dialect == "postgresql" else "utf8mb4_0900_bin"
+            ),
+            "rounding_mode": "half_away_from_zero",
+            "regex_engine": "posix" if dialect == "postgresql" else "icu",
+        }
+        return {
+            "component_binding": {
+                "id": str(row["component_binding_id"]),
+                "rule_version_id": str(row["component_rule_version_id"]),
+                "component_kind": str(row["component_kind"]),
+                "idempotency": row["idempotency"],
+            },
+            "rule_version": {
+                "id": str(row["rule_version_id"]),
+                "rule_spec": rule_spec,
+                "spec_hash": str(row["spec_hash"]),
+                "status": str(row["rule_status"]),
+            },
+            "input_schema": {
+                "id": input_snapshot_id,
+                "schema_ref": str(row["input_schema_ref"]),
+                "schema_hash": str(row["input_schema_hash"]),
+                "fields": _array(
+                    row["input_schema_fields"], "input schema fields"
+                ),
+                "source_revision": str(row["input_source_revision"]),
+            },
+            "output_schema": {
+                "id": output_snapshot_id,
+                "schema_ref": str(row["output_schema_ref"]),
+                "schema_hash": str(row["output_schema_hash"]),
+                "fields": _array(
+                    row["output_schema_fields"], "output schema fields"
+                ),
+                "source_revision": str(row["output_source_revision"]),
+            },
+            "input_binding": binding("input", input_snapshot_id),
+            "output_binding": binding("output", output_snapshot_id),
+            "backend": backend,
+        }
+
     def persist_bound_component_plan(
         self,
         *,
@@ -930,6 +1081,7 @@ class DataRuleRepository:
             plan["rule_version_id"] != rule_id
             or plan["input_binding_id"] != input_id
             or plan["output_binding_id"] != output_id
+            or plan["compiler_version"] != compiler_version
         ):
             raise ValueError("compiled bound SQL plan identifiers do not match")
 
@@ -937,7 +1089,10 @@ class DataRuleRepository:
             self.session.execute(
                 text(
                     "SELECT cb.id::text AS component_binding_id, "
+                    "rv.spec_hash AS rule_spec_hash, "
+                    "ins.id::text AS input_schema_snapshot_id, "
                     "ins.schema_hash AS input_schema_hash, "
+                    "outs.id::text AS output_schema_snapshot_id, "
                     "outs.schema_hash AS output_schema_hash, "
                     "ib.object_ref AS input_object_ref, "
                     "ob.object_ref AS output_object_ref, "
@@ -945,6 +1100,8 @@ class DataRuleRepository:
                     "ib.dialect AS input_dialect, "
                     "ob.dialect AS output_dialect "
                     "FROM public.dataflow_component_bindings cb "
+                    "JOIN public.data_rule_versions rv "
+                    "ON rv.id = cb.rule_version_id "
                     "JOIN public.dataflow_deployments d "
                     "ON d.dataflow_version_id = cb.dataflow_version_id "
                     "JOIN public.dataflow_dataset_bindings ib "
@@ -982,6 +1139,13 @@ class DataRuleRepository:
             or str(linkage["output_object_ref"]) != relations["output_object_ref"]
             or str(linkage["input_dialect"]) != plan["dialect"]
             or str(linkage["output_dialect"]) != plan["dialect"]
+            or str(linkage["rule_spec_hash"]) != plan["rule_spec_hash"]
+            or str(linkage["input_schema_snapshot_id"])
+            != plan["input_schema_snapshot_id"]
+            or str(linkage["input_schema_hash"]) != plan["input_schema_hash"]
+            or str(linkage["output_schema_snapshot_id"])
+            != plan["output_schema_snapshot_id"]
+            or str(linkage["output_schema_hash"]) != plan["output_schema_hash"]
         ):
             raise ValueError(
                 "bound SQL plan does not match its physical dataset bindings"
@@ -1005,14 +1169,15 @@ class DataRuleRepository:
                     "plan_hash": plan_hash,
                     "schema_hashes": _json(
                         {
-                            "input": _digest(
-                                linkage["input_schema_hash"],
-                                "input_schema_hash",
-                            ),
-                            "output": _digest(
-                                linkage["output_schema_hash"],
-                                "output_schema_hash",
-                            ),
+                            "rule_spec_hash": plan["rule_spec_hash"],
+                            "input_schema_snapshot_id": plan[
+                                "input_schema_snapshot_id"
+                            ],
+                            "input_schema_hash": plan["input_schema_hash"],
+                            "output_schema_snapshot_id": plan[
+                                "output_schema_snapshot_id"
+                            ],
+                            "output_schema_hash": plan["output_schema_hash"],
                         }
                     ),
                     "status": status,
@@ -1028,6 +1193,112 @@ class DataRuleRepository:
             "plan_hash": plan_hash,
         }
 
+    def record_bound_plan_test(
+        self,
+        *,
+        plan_id: str,
+        evidence: dict[str, Any],
+        test_kind: str = "integration_preflight",
+    ) -> dict[str, Any]:
+        """Attest integration evidence and transition compiled to tested."""
+
+        execution_plan_id = _uid(plan_id, "plan_id")
+        kind = _text(test_kind, "test_kind", 40)
+        if not isinstance(evidence, dict):
+            raise ValueError("test evidence must be an object")
+        rows_in = evidence.get("rows_in")
+        rows_out = evidence.get("rows_out")
+        rows_rejected = evidence.get("rows_rejected")
+        counts = (rows_in, rows_out, rows_rejected)
+        if (
+            evidence.get("commit_outcome") != "committed"
+            or any(
+                isinstance(value, bool)
+                or not isinstance(value, int)
+                or value < 0
+                for value in counts
+            )
+            or rows_out > rows_in
+            or rows_rejected != rows_in - rows_out
+        ):
+            raise ValueError("successful integration preflight evidence is required")
+        evidence_hash = _canonical_hash(evidence)
+        row = (
+            self.session.execute(
+                text(
+                    "WITH eligible AS ("
+                    "SELECT id FROM public.rule_execution_plans "
+                    "WHERE id = CAST(:plan_id AS uuid) "
+                    "AND backend = 'sql_pushdown' AND status = 'compiled' "
+                    "FOR UPDATE"
+                    "), inserted AS ("
+                    "INSERT INTO public.rule_test_evidence "
+                    "(id, rule_execution_plan_id, test_kind, evidence_hash, "
+                    "status, evidence) "
+                    "SELECT CAST(:evidence_id AS uuid), id, :test_kind, "
+                    ":evidence_hash, 'success', CAST(:evidence AS jsonb) "
+                    "FROM eligible RETURNING rule_execution_plan_id"
+                    ") "
+                    "UPDATE public.rule_execution_plans p SET status = 'tested' "
+                    "FROM inserted i WHERE p.id = i.rule_execution_plan_id "
+                    "RETURNING p.id::text AS id, p.status"
+                ),
+                {
+                    "plan_id": execution_plan_id,
+                    "evidence_id": new_governance_uid(),
+                    "test_kind": kind,
+                    "evidence_hash": evidence_hash,
+                    "evidence": _json(evidence),
+                },
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if row is None:
+            raise ValueError("only a compiled SQL plan may be tested")
+        return {
+            "id": str(row["id"]),
+            "status": str(row["status"]),
+            "evidence_hash": evidence_hash,
+        }
+
+    def publish_tested_bound_plan(self, *, plan_id: str) -> dict[str, Any]:
+        """Publish only a tested plan with canonical successful evidence."""
+
+        execution_plan_id = _uid(plan_id, "plan_id")
+        row = (
+            self.session.execute(
+                text(
+                    "UPDATE public.rule_execution_plans p "
+                    "SET status = 'published' "
+                    "FROM public.dataflow_component_bindings cb, "
+                    "public.data_rule_versions rv "
+                    "WHERE p.id = CAST(:plan_id AS uuid) "
+                    "AND p.status = 'tested' "
+                    "AND cb.id = p.component_binding_id "
+                    "AND rv.id = cb.rule_version_id "
+                    "AND rv.status = 'published' "
+                    "AND EXISTS (SELECT 1 FROM public.rule_test_evidence e "
+                    "WHERE e.rule_execution_plan_id = p.id "
+                    "AND e.test_kind = 'integration_preflight' "
+                    "AND e.status = 'success') "
+                    "RETURNING p.id::text AS id, p.status, p.plan_hash"
+                ),
+                {"plan_id": execution_plan_id},
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if row is None:
+            raise ValueError(
+                "only a successfully tested canonical plan may be published"
+            )
+        return {
+            "id": str(row["id"]),
+            "status": str(row["status"]),
+            "plan_hash": str(row["plan_hash"]),
+        }
+
     def complete_dataflow_release(
         self,
         *,

app/runner/rule_sql.py

@@ -135,6 +135,81 @@ def _upsert_statement(plan, key):
     return statement
 
 
+def _target_relation(plan):
+    dialect = "postgres" if plan["dialect"] == "postgresql" else plan["dialect"]
+    insert = parse_one(plan["statements"][0]["sql"], read=dialect)
+    target = insert.this
+    if not isinstance(target, exp.Schema) or not isinstance(
+        target.this, exp.Table
+    ):
+        raise NodeExecutionError("published SQL rule target schema is invalid")
+    table = target.this
+    if not table.db or not table.name:
+        raise NodeExecutionError(
+            "published SQL rule target must be schema-qualified"
+        )
+    return table.db, table.name
+
+
+def _attest_upsert_key(connection, plan, key):
+    """Prove the exact server-side unique-key contract before any write."""
+
+    schema_name, table_name = _target_relation(plan)
+    if plan["dialect"] == "postgresql":
+        query = text(
+            "SELECT tc.constraint_name, tc.constraint_type, "
+            "string_agg(kcu.column_name, ',' ORDER BY kcu.ordinal_position) "
+            "AS columns "
+            "FROM information_schema.table_constraints tc "
+            "JOIN information_schema.key_column_usage kcu "
+            "ON kcu.constraint_catalog = tc.constraint_catalog "
+            "AND kcu.constraint_schema = tc.constraint_schema "
+            "AND kcu.constraint_name = tc.constraint_name "
+            "WHERE tc.table_schema = :schema_name "
+            "AND tc.table_name = :table_name "
+            "AND tc.constraint_type IN ('PRIMARY KEY', 'UNIQUE') "
+            "GROUP BY tc.constraint_name, tc.constraint_type"
+        )
+    else:
+        query = text(
+            "SELECT index_name AS constraint_name, "
+            "CASE WHEN index_name = 'PRIMARY' THEN 'PRIMARY KEY' ELSE 'UNIQUE' END "
+            "AS constraint_type, "
+            "GROUP_CONCAT(column_name ORDER BY seq_in_index SEPARATOR ',') "
+            "AS columns "
+            "FROM information_schema.statistics "
+            "WHERE table_schema = :schema_name "
+            "AND table_name = :table_name AND non_unique = 0 "
+            "GROUP BY index_name"
+        )
+    rows = (
+        connection.execute(
+            query,
+            {"schema_name": schema_name, "table_name": table_name},
+        )
+        .mappings()
+        .all()
+    )
+
+    def columns(row):
+        value = row["columns"]
+        if isinstance(value, str):
+            return value.split(",") if value else []
+        return list(value or [])
+
+    unique_keys = [columns(row) for row in rows]
+    if [key] not in unique_keys:
+        raise NodeExecutionError(
+            "governed SQL rule idempotency key is not an exact unique key"
+        )
+    if plan["dialect"] == "mysql" and any(
+        unique_key != [key] for unique_key in unique_keys
+    ):
+        raise NodeExecutionError(
+            "MySQL upsert target has an alternate unique collision path"
+        )
+
+
 class SqlGlotRulePlanAdapter:
     """Execute a validated INSERT plan and count its source in one transaction."""
 
@@ -213,6 +288,11 @@ class SqlGlotRulePlanAdapter:
                     ).scalar_one()
                     or 0
                 )
+                _attest_upsert_key(
+                    connection,
+                    normalized,
+                    idempotency["key"],
+                )
                 result = connection.execute(
                     text(
                         _upsert_statement(

app/runner/rules.py

@@ -11,6 +11,12 @@ from typing import Any
 from sqlalchemy import text
 
 from app.core.common.identifiers import ensure_governance_uid
+from app.core.data_rules.compilers.sql import (
+    COMPILER_VERSION as SQL_COMPILER_VERSION,
+)
+from app.core.data_rules.compilers.sql import (
+    validate_bound_sql_plan,
+)
 from app.runner.nodes import NodeExecutionError
 
 CONFIG_KEYS = {
@@ -67,10 +73,17 @@ class PostgresRulePlanRepository:
                 p.component_binding_id::text AS component_binding_id,
                 b.rule_version_id::text AS rule_version_id,
                 p.backend,
+                p.compiler_version,
                 p.plan,
                 p.plan_hash,
+                p.schema_hashes,
                 p.status AS plan_status,
                 r.status AS rule_status,
+                r.spec_hash AS canonical_rule_spec_hash,
+                ins.id::text AS canonical_input_schema_snapshot_id,
+                ins.schema_hash AS canonical_input_schema_hash,
+                outs.id::text AS canonical_output_schema_snapshot_id,
+                outs.schema_hash AS canonical_output_schema_hash,
                 b.component_kind,
                 b.idempotency AS binding_idempotency
             FROM public.rule_execution_plans p
@@ -78,6 +91,14 @@ class PostgresRulePlanRepository:
               ON b.id = p.component_binding_id
             JOIN public.data_rule_versions r
               ON r.id = b.rule_version_id
+            LEFT JOIN public.dataflow_dataset_bindings ib
+              ON ib.id = CAST(p.plan->>'input_binding_id' AS uuid)
+            LEFT JOIN public.data_schema_snapshots ins
+              ON ins.id = ib.schema_snapshot_id
+            LEFT JOIN public.dataflow_dataset_bindings ob
+              ON ob.id = CAST(p.plan->>'output_binding_id' AS uuid)
+            LEFT JOIN public.data_schema_snapshots outs
+              ON outs.id = ob.schema_snapshot_id
             WHERE p.component_binding_id = CAST(:component_binding_id AS uuid)
               AND b.rule_version_id = CAST(:rule_version_id AS uuid)
               AND p.plan_hash = :plan_hash
@@ -168,6 +189,42 @@ class RulePlanExecutor:
                 "governed rule idempotency does not match its binding"
             )
         backend = record.get("backend")
+        if backend == "sql_pushdown":
+            try:
+                plan = validate_bound_sql_plan(record.get("plan"))
+            except ValueError as exc:
+                raise NodeExecutionError(
+                    "published rule plan is not executable"
+                ) from exc
+            expected_schema_hashes = {
+                "rule_spec_hash": plan["rule_spec_hash"],
+                "input_schema_snapshot_id": plan[
+                    "input_schema_snapshot_id"
+                ],
+                "input_schema_hash": plan["input_schema_hash"],
+                "output_schema_snapshot_id": plan[
+                    "output_schema_snapshot_id"
+                ],
+                "output_schema_hash": plan["output_schema_hash"],
+            }
+            if (
+                record.get("compiler_version") != SQL_COMPILER_VERSION
+                or plan["compiler_version"] != SQL_COMPILER_VERSION
+                or record.get("schema_hashes") != expected_schema_hashes
+                or record.get("canonical_rule_spec_hash")
+                != plan["rule_spec_hash"]
+                or record.get("canonical_input_schema_snapshot_id")
+                != plan["input_schema_snapshot_id"]
+                or record.get("canonical_input_schema_hash")
+                != plan["input_schema_hash"]
+                or record.get("canonical_output_schema_snapshot_id")
+                != plan["output_schema_snapshot_id"]
+                or record.get("canonical_output_schema_hash")
+                != plan["output_schema_hash"]
+            ):
+                raise NodeExecutionError(
+                    "published rule plan canonical attestation does not match"
+                )
         adapter = self.adapters.get(backend)
         if adapter is None or not callable(getattr(adapter, "execute", None)):
             raise NodeExecutionError("rule plan backend is not registered")

migrations/versions/20260723_130_bound_plan_lifecycle.py

@@ -0,0 +1,25 @@
+"""Add an explicit tested state to physical execution plans."""
+
+from alembic import op
+
+revision = "20260723_130"
+down_revision = "20260723_120"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE public.rule_execution_plans
+            DROP CONSTRAINT rule_execution_plans_status_check;
+        ALTER TABLE public.rule_execution_plans
+            ADD CONSTRAINT rule_execution_plans_status_check
+            CHECK (status IN ('compiled','tested','published','revoked'));
+        """
+    )
+
+
+def downgrade() -> None:
+    # Immutable runtime evidence and tested plans are retained on rollback.
+    pass

tests/core/data_rules/test_release.py

@@ -201,6 +201,36 @@ def test_release_expands_standard_and_persists_fixed_bindings_and_plans():
     ]
 
 
+def test_release_rejects_unregistered_semantic_backend_before_version_write():
+    from app.core.data_rules.release import ProductionLineReleaseService
+
+    rule_id = new_governance_uid()
+    spec = valid_rule_spec()
+    repository = ReleaseRepository(
+        standards={},
+        rules={rule_id: _published_rule(rule_id, spec)},
+    )
+    flow = valid_dataflow_spec(rule_version_id=rule_id)
+    flow["components"] = [flow["components"][0]]
+
+    with pytest.raises(ValueError, match="not registered"):
+        ProductionLineReleaseService(
+            repository,
+            schema_resolver=FakeSchemaResolver(),
+            available_plan_backends=frozenset(),
+        ).release(
+            dataflow_uid=flow["dataflow_uid"],
+            dataflow_spec=flow,
+            source_text="release registry check",
+            created_by=new_governance_uid(),
+        )
+
+    assert not any(
+        method == "begin_dataflow_release"
+        for method, _payload in repository.calls
+    )
+
+
 def test_release_rejects_path_uid_mismatch_before_database_writes():
     import pytest
 

tests/core/data_rules/test_sql_compiler.py

@@ -272,6 +272,151 @@ def test_sql_compiler_rejects_unproven_operator_type_semantics(step):
         _compile(steps=[step])
 
 
+@pytest.mark.parametrize("on_error", ["reject", "quarantine", "warn"])
+def test_sql_cast_rejects_unimplemented_non_fail_error_semantics(on_error):
+    with pytest.raises(ValueError, match="fail on_error"):
+        _compile(
+            steps=[
+                {
+                    "id": "cast",
+                    "op": "cast",
+                    "column": "balance",
+                    "to": "decimal",
+                    "on_error": on_error,
+                }
+            ]
+        )
+
+
+def test_deduplicate_extends_order_to_total_projected_value_order():
+    from sqlglot import exp, parse_one
+
+    compiled = _compile(
+        steps=[
+            {
+                "id": "deduplicate",
+                "op": "deduplicate",
+                "keys": ["customer_id"],
+                "order_by": ["customer_id"],
+                "keep": "first",
+            }
+        ]
+    )
+    parsed = parse_one(
+        compiled["plan"]["statements"][0]["sql"],
+        read="postgres",
+    )
+    window = next(parsed.find_all(exp.Window))
+    ordered = window.args["order"].expressions
+
+    assert [item.this.name for item in ordered] == [
+        "customer_id",
+        "balance",
+        "mobile",
+        "name",
+    ]
+    assert all(item.args["nulls_first"] is False for item in ordered)
+
+
+def test_numeric_types_require_an_explicit_cast():
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+
+    input_schema = _schema(
+        "bd:customer:raw",
+        [("balance", "integer", True)],
+    )
+    output_schema = _schema(
+        "bd:customer:clean",
+        [("balance", "decimal", True)],
+    )
+    common = {
+        "input_schema": input_schema,
+        "output_schema": output_schema,
+        "input_binding": table_binding(input_schema, "raw.customer"),
+        "output_binding": table_binding(
+            output_schema, "clean.customer", access_mode="write"
+        ),
+        "backend": backend(),
+    }
+    with pytest.raises(ValueError, match="field types"):
+        SqlGlotRuleCompiler("postgresql").compile(
+            rule_version=published_rule(
+                [
+                    {
+                        "id": "fill",
+                        "op": "fill_null",
+                        "column": "balance",
+                        "value": 0,
+                    }
+                ]
+            ),
+            **common,
+        )
+    compiled = SqlGlotRuleCompiler("postgresql").compile(
+        rule_version=published_rule(
+            [
+                {
+                    "id": "cast",
+                    "op": "cast",
+                    "column": "balance",
+                    "to": "decimal",
+                    "on_error": "fail",
+                }
+            ]
+        ),
+        **common,
+    )
+    assert "CAST" in compiled["plan"]["statements"][0]["sql"]
+
+
+@pytest.mark.parametrize(
+    ("dialect", "field_name"),
+    [
+        ("postgresql", "x" * 64),
+        ("mysql", "x" * 65),
+        ("postgresql", "界" * 22),
+    ],
+)
+def test_sql_compiler_enforces_dialect_identifier_utf8_byte_limits(
+    dialect, field_name
+):
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+
+    input_schema = _schema(
+        "bd:customer:raw",
+        [(field_name, "integer", False)],
+    )
+    output_schema = _schema(
+        "bd:customer:clean",
+        [(field_name, "integer", False)],
+    )
+    with pytest.raises(ValueError, match="byte identifier limit"):
+        SqlGlotRuleCompiler(dialect).compile(
+            rule_version=published_rule(
+                [
+                    {
+                        "id": "cast",
+                        "op": "cast",
+                        "column": field_name,
+                        "to": "integer",
+                    }
+                ]
+            ),
+            input_schema=input_schema,
+            output_schema=output_schema,
+            input_binding=table_binding(
+                input_schema, "raw.customer", dialect=dialect
+            ),
+            output_binding=table_binding(
+                output_schema,
+                "clean.customer",
+                dialect=dialect,
+                access_mode="write",
+            ),
+            backend=backend(dialect),
+        )
+
+
 @pytest.mark.parametrize(
     "step",
     [
@@ -385,22 +530,44 @@ def test_bound_plan_service_persists_compiled_state_without_claiming_publication
         def __init__(self):
             self.persisted = None
 
+        def load_bound_compile_context(self, **ids):
+            assert ids == {
+                "component_binding_id": component_binding_id,
+                "rule_version_id": rule["id"],
+                "input_schema_snapshot_id": input_schema["id"],
+                "output_schema_snapshot_id": output_schema["id"],
+                "input_binding_id": source["id"],
+                "output_binding_id": target["id"],
+            }
+            return {
+                "component_binding": {
+                    "id": component_binding_id,
+                    "rule_version_id": rule["id"],
+                },
+                "rule_version": rule,
+                "input_schema": input_schema,
+                "output_schema": output_schema,
+                "input_binding": source,
+                "output_binding": target,
+                "backend": backend(),
+            }
+
         def persist_bound_component_plan(self, **kwargs):
             self.persisted = kwargs
             return {"id": new_governance_uid(), "status": kwargs["status"]}
 
     repository = Repository()
+    component_binding_id = new_governance_uid()
     result = BoundSqlPlanService(
         repository,
         CompilerRegistry({"postgresql": SqlGlotRuleCompiler("postgresql")}),
     ).compile_and_persist(
-        component_binding_id=new_governance_uid(),
-        rule_version=rule,
-        input_schema=input_schema,
-        output_schema=output_schema,
-        input_binding=source,
-        output_binding=target,
-        backend=backend(),
+        component_binding_id=component_binding_id,
+        rule_version_id=rule["id"],
+        input_schema_snapshot_id=input_schema["id"],
+        output_schema_snapshot_id=output_schema["id"],
+        input_binding_id=source["id"],
+        output_binding_id=target["id"],
     )
 
     assert result["status"] == "compiled"
@@ -408,6 +575,52 @@ def test_bound_plan_service_persists_compiled_state_without_claiming_publication
     assert repository.persisted["compiled"]["plan"]["input_binding_id"] == source["id"]
 
 
+def test_bound_plan_service_rejects_noncanonical_component_rule_linkage():
+    from app.core.data_rules.compilers import CompilerRegistry
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+    from app.core.data_rules.release import BoundSqlPlanService
+
+    input_schema, output_schema = customer_schemas()
+    source = table_binding(input_schema, "raw.customer")
+    target = table_binding(output_schema, "clean.customer", access_mode="write")
+    rule = published_rule(
+        [{"id": "fill", "op": "fill_null", "column": "name", "value": "x"}]
+    )
+    requested_rule_id = rule["id"]
+    component_binding_id = new_governance_uid()
+
+    class Repository:
+        def load_bound_compile_context(self, **_ids):
+            return {
+                "component_binding": {
+                    "id": component_binding_id,
+                    "rule_version_id": new_governance_uid(),
+                },
+                "rule_version": rule,
+                "input_schema": input_schema,
+                "output_schema": output_schema,
+                "input_binding": source,
+                "output_binding": target,
+                "backend": backend(),
+            }
+
+        def persist_bound_component_plan(self, **_kwargs):
+            raise AssertionError("tampered canonical linkage must not persist")
+
+    with pytest.raises(ValueError, match="canonical"):
+        BoundSqlPlanService(
+            Repository(),
+            CompilerRegistry({"postgresql": SqlGlotRuleCompiler("postgresql")}),
+        ).compile_and_persist(
+            component_binding_id=component_binding_id,
+            rule_version_id=requested_rule_id,
+            input_schema_snapshot_id=input_schema["id"],
+            output_schema_snapshot_id=output_schema["id"],
+            input_binding_id=source["id"],
+            output_binding_id=target["id"],
+        )
+
+
 def test_repository_persists_bound_plan_only_for_one_deployment_linkage():
     from app.core.data_rules.repository import DataRuleRepository
 
@@ -432,9 +645,14 @@ def test_repository_persists_bound_plan_only_for_one_deployment_linkage():
             return Mappings(self.row)
 
     class Session:
-        def __init__(self, target_object_ref="clean.customer"):
+        def __init__(
+            self,
+            target_object_ref="clean.customer",
+            rule_spec_hash=None,
+        ):
             self.calls = []
             self.target_object_ref = target_object_ref
+            self.rule_spec_hash = rule_spec_hash
 
         def execute(self, statement, params=None):
             sql = str(statement)
@@ -442,10 +660,22 @@ def test_repository_persists_bound_plan_only_for_one_deployment_linkage():
             self.calls.append((sql, values))
             if "bound_plan_linkage" in sql:
                 return Result(
-                    {
-                        "component_binding_id": component_binding_id,
-                        "input_schema_hash": "a" * 64,
-                        "output_schema_hash": "b" * 64,
+                        {
+                            "component_binding_id": component_binding_id,
+                            "rule_spec_hash": self.rule_spec_hash
+                            or compiled["plan"]["rule_spec_hash"],
+                            "input_schema_snapshot_id": compiled["plan"][
+                                "input_schema_snapshot_id"
+                            ],
+                            "input_schema_hash": compiled["plan"][
+                                "input_schema_hash"
+                            ],
+                            "output_schema_snapshot_id": compiled["plan"][
+                                "output_schema_snapshot_id"
+                            ],
+                            "output_schema_hash": compiled["plan"][
+                                "output_schema_hash"
+                            ],
                         "input_object_ref": "raw.customer",
                         "output_object_ref": self.target_object_ref,
                         "data_source_uid": compiled["plan"]["data_source_uid"],
@@ -501,3 +731,112 @@ def test_repository_persists_bound_plan_only_for_one_deployment_linkage():
             compiled=compiled,
             status="compiled",
         )
+    with pytest.raises(ValueError, match="physical dataset"):
+        DataRuleRepository(
+            Session(rule_spec_hash="f" * 64)
+        ).persist_bound_component_plan(
+            component_binding_id=component_binding_id,
+            rule_version_id=compiled["plan"]["rule_version_id"],
+            input_binding_id=compiled["plan"]["input_binding_id"],
+            output_binding_id=compiled["plan"]["output_binding_id"],
+            compiled=compiled,
+            status="compiled",
+        )
+
+
+def test_logical_release_plan_is_persisted_compiled_not_published():
+    from app.core.data_rules.compiler import compile_rule_plan
+    from app.core.data_rules.repository import DataRuleRepository
+
+    class Session:
+        def __init__(self):
+            self.calls = []
+
+        def execute(self, statement, params=None):
+            self.calls.append((str(statement), params or {}))
+
+    rule = published_rule(
+        [{"id": "fill", "op": "fill_null", "column": "name", "value": "x"}]
+    )
+    session = Session()
+    DataRuleRepository(session).persist_component_plan(
+        dataflow_version_id=new_governance_uid(),
+        component_binding_id=new_governance_uid(),
+        component_id="logical_rule",
+        component_kind="rule.apply",
+        rule_version_id=rule["id"],
+        stage="transform",
+        order_no=1,
+        idempotency={"strategy": "upsert", "key": "customer_id"},
+        provenance={},
+        plan=compile_rule_plan(rule),
+        schema_hashes={"inputs": {}, "output": "a" * 64},
+    )
+    plan_insert = next(
+        sql
+        for sql, _params in session.calls
+        if "INSERT INTO public.rule_execution_plans" in sql
+    )
+    assert "'compiled'" in plan_insert
+    assert "'published'" not in plan_insert
+
+
+def test_bound_plan_lifecycle_requires_successful_integration_evidence():
+    from app.core.data_rules.repository import DataRuleRepository
+
+    plan_id = new_governance_uid()
+
+    class Mappings:
+        def __init__(self, row):
+            self.row = row
+
+        def one_or_none(self):
+            return self.row
+
+    class Result:
+        def __init__(self, row):
+            self.row = row
+
+        def mappings(self):
+            return Mappings(self.row)
+
+    class Session:
+        def __init__(self):
+            self.calls = []
+
+        def execute(self, statement, params=None):
+            sql = str(statement)
+            self.calls.append((sql, params or {}))
+            if "rule_test_evidence" in sql and "SET status = 'tested'" in sql:
+                return Result({"id": plan_id, "status": "tested"})
+            if "SET status = 'published'" in sql:
+                return Result(
+                    {
+                        "id": plan_id,
+                        "status": "published",
+                        "plan_hash": "a" * 64,
+                    }
+                )
+            return Result(None)
+
+    session = Session()
+    repository = DataRuleRepository(session)
+    evidence = {
+        "rows_in": 3,
+        "rows_out": 2,
+        "rows_rejected": 1,
+        "commit_outcome": "committed",
+    }
+    tested = repository.record_bound_plan_test(
+        plan_id=plan_id,
+        evidence=evidence,
+    )
+    assert tested["status"] == "tested"
+    assert repository.publish_tested_bound_plan(plan_id=plan_id)["status"] == (
+        "published"
+    )
+    with pytest.raises(ValueError, match="evidence"):
+        repository.record_bound_plan_test(
+            plan_id=plan_id,
+            evidence={**evidence, "commit_outcome": "unknown"},
+        )

tests/integration/test_data_rule_sql_execution.py

@@ -61,18 +61,40 @@ class DirectManager:
 
 
 class PlanRepository:
-    def __init__(self, idempotency):
+    def __init__(self, idempotency, context):
         self.idempotency = idempotency
+        self.context = context
         self.record = None
 
+    def load_bound_compile_context(self, **_ids):
+        return self.context
+
     def persist_bound_component_plan(self, **kwargs):
         compiled = kwargs["compiled"]
+        plan = compiled["plan"]
         self.record = {
             "component_binding_id": kwargs["component_binding_id"],
             "rule_version_id": kwargs["rule_version_id"],
             "backend": compiled["backend"],
+            "compiler_version": compiled["compiler_version"],
             "plan": compiled["plan"],
             "plan_hash": compiled["plan_hash"],
+            "schema_hashes": {
+                "rule_spec_hash": plan["rule_spec_hash"],
+                "input_schema_snapshot_id": plan["input_schema_snapshot_id"],
+                "input_schema_hash": plan["input_schema_hash"],
+                "output_schema_snapshot_id": plan["output_schema_snapshot_id"],
+                "output_schema_hash": plan["output_schema_hash"],
+            },
+            "canonical_rule_spec_hash": plan["rule_spec_hash"],
+            "canonical_input_schema_snapshot_id": plan[
+                "input_schema_snapshot_id"
+            ],
+            "canonical_input_schema_hash": plan["input_schema_hash"],
+            "canonical_output_schema_snapshot_id": plan[
+                "output_schema_snapshot_id"
+            ],
+            "canonical_output_schema_hash": plan["output_schema_hash"],
             "plan_status": kwargs["status"],
             "rule_status": "published",
             "component_kind": "rule.apply",
@@ -90,6 +112,7 @@ class PlanRepository:
         assert self.record["plan_hash"] == plan_hash
         assert evidence["commit_outcome"] == "committed"
         assert evidence["rows_in"] >= evidence["rows_out"]
+        self.record["plan_status"] = "tested"
         self.record["plan_status"] = "published"
 
     def load(self, **_kwargs):
@@ -223,7 +246,21 @@ def test_bound_rule_compiles_publishes_executes_and_rejects_tampering(
             "strategy": "upsert",
             "key": "customer_id",
         }
-        repository = PlanRepository(idempotency)
+        repository = PlanRepository(
+            idempotency,
+            {
+                "component_binding": {
+                    "id": component_binding_id,
+                    "rule_version_id": rule["id"],
+                },
+                "rule_version": rule,
+                "input_schema": input_schema,
+                "output_schema": output_schema,
+                "input_binding": input_binding,
+                "output_binding": output_binding,
+                "backend": capabilities,
+            },
+        )
         BoundSqlPlanService(
             repository,
             CompilerRegistry(
@@ -231,12 +268,11 @@ def test_bound_rule_compiles_publishes_executes_and_rejects_tampering(
             ),
         ).compile_and_persist(
             component_binding_id=component_binding_id,
-            rule_version=rule,
-            input_schema=input_schema,
-            output_schema=output_schema,
-            input_binding=input_binding,
-            output_binding=output_binding,
-            backend=capabilities,
+            rule_version_id=rule["id"],
+            input_schema_snapshot_id=input_schema["id"],
+            output_schema_snapshot_id=output_schema["id"],
+            input_binding_id=input_binding["id"],
+            output_binding_id=output_binding["id"],
         )
         record = repository.record
         assert record["plan_status"] == "compiled"
@@ -314,3 +350,147 @@ def test_bound_rule_compiles_publishes_executes_and_rejects_tampering(
             connection.execute(text(f"DROP TABLE IF EXISTS {target_name}"))
             connection.execute(text(f"DROP TABLE IF EXISTS {source_name}"))
         engine.dispose()
+
+
+def test_mysql_upsert_rejects_real_nonunique_and_alternate_unique_targets():
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+    from app.runner.rule_sql import SqlGlotRulePlanAdapter
+
+    dialect, url, schema_name, collation, regex_engine = CASES[1]
+    engine = create_engine(url, pool_pre_ping=True)
+    source_name = "task4_unique_source"
+    nonunique_name = "task4_nonunique_target"
+    alternate_name = "task4_alternate_target"
+    capabilities = {
+        "dialect": dialect,
+        "timezone": "Asia/Shanghai",
+        "collation": collation,
+        "rounding_mode": "half_away_from_zero",
+        "regex_engine": regex_engine,
+    }
+    datasource_uid = new_governance_uid()
+    input_schema = _snapshot("bd:task4:unique:raw")
+    output_schema = _snapshot("bd:task4:unique:clean")
+    spec = validate_rule_spec(
+        {
+            "schema_version": "2.0",
+            "rule_uid": new_governance_uid(),
+            "name": "task4_unique_attestation",
+            "input_schema_ref": input_schema["schema_ref"],
+            "output_schema_ref": output_schema["schema_ref"],
+            "steps": [
+                {
+                    "id": "trim_name",
+                    "op": "normalize_text",
+                    "column": "name",
+                    "trim": True,
+                }
+            ],
+            "null_policy": "explicit",
+            "timezone": "Asia/Shanghai",
+        }
+    )
+    rule = {
+        "id": new_governance_uid(),
+        "status": "published",
+        "rule_spec": spec,
+        "spec_hash": rule_spec_hash(spec),
+    }
+    input_binding = {
+        "id": new_governance_uid(),
+        "data_source_uid": datasource_uid,
+        "object_kind": "table",
+        "object_ref": f"{schema_name}.{source_name}",
+        "schema_snapshot_id": input_schema["id"],
+        "access_mode": "read",
+        "dialect": dialect,
+        "write_mode": "append",
+    }
+
+    try:
+        with engine.begin() as connection:
+            for name in (alternate_name, nonunique_name, source_name):
+                connection.execute(text(f"DROP TABLE IF EXISTS {name}"))
+            connection.execute(
+                text(
+                    f"CREATE TABLE {source_name} ("
+                    "customer_id BIGINT PRIMARY KEY, "
+                    "name VARCHAR(100), mobile VARCHAR(30))"
+                )
+            )
+            connection.execute(
+                text(
+                    f"CREATE TABLE {nonunique_name} ("
+                    "customer_id BIGINT, name VARCHAR(100), mobile VARCHAR(30))"
+                )
+            )
+            connection.execute(
+                text(
+                    f"CREATE TABLE {alternate_name} ("
+                    "customer_id BIGINT PRIMARY KEY, "
+                    "name VARCHAR(100), mobile VARCHAR(30) UNIQUE)"
+                )
+            )
+            connection.execute(
+                text(
+                    f"INSERT INTO {source_name} "
+                    "(customer_id, name, mobile) "
+                    "VALUES (1, ' Alice ', '13800138000')"
+                )
+            )
+
+        adapter = SqlGlotRulePlanAdapter(
+            DirectManager(engine, Definition(dialect, capabilities))
+        )
+        for target_name, error in (
+            (nonunique_name, "exact unique key"),
+            (alternate_name, "alternate unique"),
+        ):
+            output_binding = {
+                "id": new_governance_uid(),
+                "data_source_uid": datasource_uid,
+                "object_kind": "table",
+                "object_ref": f"{schema_name}.{target_name}",
+                "schema_snapshot_id": output_schema["id"],
+                "access_mode": "write",
+                "dialect": dialect,
+                "write_mode": "append",
+            }
+            compiled = SqlGlotRuleCompiler(dialect).compile(
+                rule_version=rule,
+                input_schema=input_schema,
+                output_schema=output_schema,
+                input_binding=input_binding,
+                output_binding=output_binding,
+                backend=capabilities,
+            )
+            node = {
+                "id": "task4_unique_rule",
+                "type": "rule.apply",
+                "purpose": "write",
+                "idempotency": {
+                    "strategy": "upsert",
+                    "key": "customer_id",
+                },
+                "config": {
+                    "component_binding_id": new_governance_uid(),
+                    "rule_version_id": rule["id"],
+                    "execution_plan_hash": compiled["plan_hash"],
+                },
+            }
+            with pytest.raises(NodeExecutionError, match=error):
+                adapter.execute(
+                    plan=compiled["plan"],
+                    node=node,
+                    parameters={},
+                    write_authorized=True,
+                )
+            with engine.connect() as connection:
+                assert connection.execute(
+                    text(f"SELECT COUNT(*) FROM {target_name}")
+                ).scalar_one() == 0
+    finally:
+        with engine.begin() as connection:
+            for name in (alternate_name, nonunique_name, source_name):
+                connection.execute(text(f"DROP TABLE IF EXISTS {name}"))
+        engine.dispose()

tests/runner/test_rule_sql.py

@@ -32,17 +32,25 @@ class Definitions:
 
 
 class Result:
-    def __init__(self, *, scalar=None, rowcount=0):
+    def __init__(self, *, scalar=None, rowcount=0, rows=None):
         self._scalar = scalar
         self.rowcount = rowcount
+        self._rows = rows or []
 
     def scalar_one(self):
         return self._scalar
 
+    def mappings(self):
+        return self
+
+    def all(self):
+        return self._rows
+
 
 class Connection:
-    def __init__(self):
+    def __init__(self, unique_rows=None):
         self.calls = []
+        self.unique_rows = unique_rows
 
     def execute(self, statement, parameters):
         self.calls.append((str(statement), parameters))
@@ -50,13 +58,31 @@ class Connection:
             return Result(scalar=3)
         if len(self.calls) == 2:
             return Result(scalar=2)
+        if "information_schema" in str(statement):
+            return Result(
+                rows=self.unique_rows
+                if self.unique_rows is not None
+                else [
+                    {
+                        "constraint_name": "customer_pkey",
+                        "constraint_type": "PRIMARY KEY",
+                        "columns": ["id"],
+                    }
+                ]
+            )
         return Result(rowcount=2)
 
 
 class Manager:
-    def __init__(self, dialect="postgresql", *, unknown_commit=False):
+    def __init__(
+        self,
+        dialect="postgresql",
+        *,
+        unknown_commit=False,
+        unique_rows=None,
+    ):
         self.definitions = Definitions(Definition(dialect))
-        self.connection = Connection()
+        self.connection = Connection(unique_rows)
         self.unknown_commit = unknown_commit
         self.calls = []
 
@@ -69,17 +95,26 @@ class Manager:
 
 
 def sql_plan(dialect="postgresql"):
-    from app.core.data_rules.compilers.sql import bound_sql_plan_hash
+    from app.core.data_rules.compilers.sql import (
+        COMPILER_VERSION,
+        bound_sql_plan_hash,
+    )
 
     uid = new_governance_uid()
     capabilities = Definition(dialect).extra_properties["sql_rule_capabilities"]
     quote = '"' if dialect == "postgresql" else "`"
     plan = {
         "schema_version": "1.0",
+        "compiler_version": COMPILER_VERSION,
         "dialect": dialect,
         "capabilities": capabilities,
         "data_source_uid": uid,
         "rule_version_id": new_governance_uid(),
+        "rule_spec_hash": "a" * 64,
+        "input_schema_snapshot_id": new_governance_uid(),
+        "input_schema_hash": "b" * 64,
+        "output_schema_snapshot_id": new_governance_uid(),
+        "output_schema_hash": "c" * 64,
         "input_binding_id": new_governance_uid(),
         "output_binding_id": new_governance_uid(),
         "statements": [
@@ -137,7 +172,7 @@ def test_sqlglot_rule_adapter_verifies_and_executes_one_transaction():
         "commit_outcome": "committed",
     }
     assert manager.calls == [(plan["data_source_uid"], "dataflow_write")]
-    assert len(manager.connection.calls) == 3
+    assert len(manager.connection.calls) == 4
 
 
 def test_sqlglot_rule_adapter_fails_closed_for_dialect_hash_and_authorization():
@@ -226,6 +261,73 @@ def test_rule_executor_matches_node_idempotency_to_persisted_component():
         ).execute(node, {}, write_authorized=True)
 
 
+def test_rule_executor_rechecks_canonical_rule_schema_and_compiler_attestations():
+    from app.core.data_rules.compilers.sql import COMPILER_VERSION
+    from app.runner.rules import RulePlanExecutor
+
+    plan, plan_hash = sql_plan()
+    node = node_for(plan, plan_hash)
+
+    def record(**overrides):
+        value = {
+            "component_binding_id": node["config"]["component_binding_id"],
+            "rule_version_id": plan["rule_version_id"],
+            "backend": "sql_pushdown",
+            "compiler_version": COMPILER_VERSION,
+            "plan": plan,
+            "plan_hash": plan_hash,
+            "schema_hashes": {
+                "rule_spec_hash": plan["rule_spec_hash"],
+                "input_schema_snapshot_id": plan[
+                    "input_schema_snapshot_id"
+                ],
+                "input_schema_hash": plan["input_schema_hash"],
+                "output_schema_snapshot_id": plan[
+                    "output_schema_snapshot_id"
+                ],
+                "output_schema_hash": plan["output_schema_hash"],
+            },
+            "canonical_rule_spec_hash": plan["rule_spec_hash"],
+            "canonical_input_schema_snapshot_id": plan[
+                "input_schema_snapshot_id"
+            ],
+            "canonical_input_schema_hash": plan["input_schema_hash"],
+            "canonical_output_schema_snapshot_id": plan[
+                "output_schema_snapshot_id"
+            ],
+            "canonical_output_schema_hash": plan["output_schema_hash"],
+            "plan_status": "published",
+            "rule_status": "published",
+            "component_kind": "rule.apply",
+            "binding_idempotency": node["idempotency"],
+        }
+        value.update(overrides)
+        return value
+
+    class Repository:
+        def __init__(self, value):
+            self.value = value
+
+        def load(self, **_kwargs):
+            return self.value
+
+    class Adapter:
+        def execute(self, **_kwargs):
+            return {"rows_in": 0, "rows_out": 0, "rows_rejected": 0}
+
+    for tampered in (
+        {"compiler_version": "dataops-sqlglot-99.0.0"},
+        {"canonical_rule_spec_hash": "f" * 64},
+        {"canonical_input_schema_hash": "e" * 64},
+        {"canonical_output_schema_snapshot_id": new_governance_uid()},
+    ):
+        with pytest.raises(NodeExecutionError, match="attestation"):
+            RulePlanExecutor(
+                Repository(record(**tampered)),
+                adapters={"sql_pushdown": Adapter()},
+            ).execute(node, {}, write_authorized=True)
+
+
 def test_sqlglot_rule_adapter_reports_unknown_commit_outcome():
     from app.runner.rule_sql import SqlGlotRulePlanAdapter
 
@@ -240,3 +342,61 @@ def test_sqlglot_rule_adapter_reports_unknown_commit_outcome():
         )
 
     assert error.value.commit_outcome == "unknown"
+
+
+def test_sqlglot_rule_adapter_rejects_function_injected_into_attested_plan():
+    from app.core.data_rules.compilers.sql import bound_sql_plan_hash
+    from app.runner.rule_sql import SqlGlotRulePlanAdapter
+
+    plan, _ = sql_plan()
+    plan["statements"][0]["sql"] = plan["statements"][0]["sql"].replace(
+        'SELECT "id"', 'SELECT pg_sleep(1)'
+    )
+    with pytest.raises(ValueError, match="unsupported"):
+        bound_sql_plan_hash(plan)
+
+    # A malicious publisher cannot bypass the AST allowlist by recomputing a
+    # hash because the adapter independently validates the compiler subset.
+    node = node_for(plan, "0" * 64)
+    with pytest.raises(NodeExecutionError, match="invalid"):
+        SqlGlotRulePlanAdapter(Manager()).execute(
+            plan=plan,
+            node=node,
+            parameters={},
+            write_authorized=True,
+        )
+
+
+@pytest.mark.parametrize(
+    "unique_rows",
+    [
+        [],
+        [
+            {
+                "constraint_name": "customer_id_idx",
+                "constraint_type": "UNIQUE",
+                "columns": "id",
+            },
+            {
+                "constraint_name": "mobile_idx",
+                "constraint_type": "UNIQUE",
+                "columns": "mobile",
+            },
+        ],
+    ],
+)
+def test_mysql_upsert_requires_exact_key_and_no_alternate_unique_path(
+    unique_rows,
+):
+    from app.runner.rule_sql import SqlGlotRulePlanAdapter
+
+    plan, plan_hash = sql_plan("mysql")
+    with pytest.raises(NodeExecutionError, match="unique"):
+        SqlGlotRulePlanAdapter(
+            Manager("mysql", unique_rows=unique_rows)
+        ).execute(
+            plan=plan,
+            node=node_for(plan, plan_hash),
+            parameters={},
+            write_authorized=True,
+        )

tests/runner/test_rules.py

@@ -110,6 +110,8 @@ def test_rule_executor_loads_only_published_plan_by_fixed_identifiers():
     "record",
     [
         None,
+        {"plan_status": "compiled"},
+        {"plan_status": "tested"},
         {"plan_status": "revoked"},
         {"rule_status": "deprecated"},
         {"plan_hash": "b" * 64},