fix: keep task4 plans compiled only

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

@@ -335,3 +335,67 @@ no output
 - 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.
+
+---
+
+# Task 4 final review correction — compiled-only production boundary
+
+Date: 2026-07-23
+
+This section supersedes the earlier statement in this report that Task 4
+provided a production `compiled -> tested -> published` transition.
+
+## Boundary correction
+
+- Removed `record_bound_plan_test` and `publish_tested_bound_plan` from
+  `DataRuleRepository`. Task 4 exposes no production API that accepts
+  caller-authored execution evidence or promotes a plan.
+- Physical bound plans still persist only as `compiled`.
+- The production Runner remains published-only and rejects compiled plans.
+- The real-database integration test publishes only inside its explicitly
+  trusted in-memory test repository after a real preflight. This is not a
+  production publication surface.
+- Task 7 remains responsible for opaque server-created execution evidence and
+  the authoritative tested/publication state machine.
+
+## Additional fail-closed corrections
+
+- Integer division (`/`) is removed from PostgreSQL and MySQL V1 capability
+  support. Both dialect compilers reject a rule containing division; Polars
+  reference capability remains available for its defined decimal semantics.
+- Deterministic deduplication rejects a plan if any explicit or appended
+  total-order field has an unproven ordering type. JSON and binary regression
+  cases now fail compilation.
+- The already-applied `20260723_120` migration remains unchanged.
+  `20260723_130` pins the database status constraint to
+  `compiled`, `published`, and `revoked`; it does not add `tested`.
+  Its downgrade raises a clear `RuntimeError` because persisted compiled plans
+  make this status-contract migration intentionally forward-only.
+
+## Final TDD and verification evidence
+
+The RED run produced six expected failures:
+
+- production promotion API still present: 1
+- PostgreSQL/MySQL division still advertised: 2
+- JSON/binary deduplication accepted: 2
+- missing explicit forward-only migration downgrade: 1
+
+After the migration-chain correction, final verification was:
+
+```text
+Focused expressions/compiler/repository/release/runner/schema:
+89 passed
+
+Real PostgreSQL/MySQL integration:
+3 passed
+
+Full repository suite:
+501 passed, 26 skipped, 59 subtests passed
+
+Ruff on all changed Python and migration files:
+All checks passed!
+
+git diff --check:
+no output
+```

app/core/data_rules/compilers/sql.py

@@ -71,6 +71,17 @@ RESULT_CONTRACT = {
 _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"}
+_ORDERABLE_TYPES = {
+    "boolean",
+    "date",
+    "decimal",
+    "double",
+    "float",
+    "integer",
+    "string",
+    "timestamp",
+    "timestamptz",
+}
 _TYPE_NAMES = {
     "binary": "BINARY",
     "boolean": "BOOLEAN",
@@ -882,6 +893,13 @@ class SqlGlotRuleCompiler(RuleCompiler):
                         if name not in order_by
                     ],
                 ]
+                if any(
+                    field_types[item] not in _ORDERABLE_TYPES
+                    for item in total_order
+                ):
+                    raise ValueError(
+                        "deduplicate total order contains a non-orderable field"
+                    )
                 row_number = exp.Window(
                     this=exp.RowNumber(),
                     partition_by=[_column(item) for item in keys],

app/core/data_rules/expressions.py

@@ -7,15 +7,14 @@ runtime.  Compilers consume the resulting AST in a later release.
 
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass
 from datetime import date as Date
 from datetime import datetime
-from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
-import re
+from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
 from typing import Any
 from zoneinfo import ZoneInfo
 
-
 MAX_SOURCE_LENGTH = 20_000
 MAX_AST_DEPTH = 40
 MAX_AST_NODES = 500
@@ -406,11 +405,11 @@ def _validate_ast(ast: Any) -> None:
                 and len(arguments) >= 2
                 and arguments[1].get("kind") == "literal"
                 and arguments[1].get("type") == "string"
+                and len(arguments[1]["value"]) > MAX_REGEX_LENGTH
             ):
-                if len(arguments[1]["value"]) > MAX_REGEX_LENGTH:
-                    raise ValueError(
-                        f"regex pattern exceeds {MAX_REGEX_LENGTH} characters"
-                    )
+                raise ValueError(
+                    f"regex pattern exceeds {MAX_REGEX_LENGTH} characters"
+                )
             return
         raise ValueError("expression AST node kind is unsupported")
 
@@ -630,7 +629,10 @@ def backend_support(ast: dict) -> frozenset[str]:
         if kind == "unary":
             return support(node["operand"])
         if kind == "binary":
-            return support(node["left"]) & support(node["right"])
+            supported = support(node["left"]) & support(node["right"])
+            if node["operator"] == "/":
+                return supported & frozenset({"polars"})
+            return supported
         function = node["function"]
         supported = _FUNCTION_BACKENDS[function]
         for argument in node["arguments"]:

app/core/data_rules/repository.py

@@ -1193,112 +1193,6 @@ 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,
         *,

migrations/versions/20260723_130_bound_plan_lifecycle.py

@@ -1,4 +1,4 @@
-"""Add an explicit tested state to physical execution plans."""
+"""Pin the Task 4 physical-plan lifecycle to compiled-only creation."""
 
 from alembic import op
 
@@ -15,11 +15,13 @@ def upgrade() -> None:
             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'));
+            CHECK (status IN ('compiled','published','revoked'));
         """
     )
 
 
 def downgrade() -> None:
-    # Immutable runtime evidence and tested plans are retained on rollback.
-    pass
+    raise RuntimeError(
+        "bound plan status migration is forward-only and cannot downgrade "
+        "without rewriting persisted compiled plans"
+    )

tests/core/data_rules/test_sql_compiler.py

@@ -781,62 +781,64 @@ def test_logical_release_plan_is_persisted_compiled_not_published():
     assert "'published'" not in plan_insert
 
 
-def test_bound_plan_lifecycle_requires_successful_integration_evidence():
+def test_task4_repository_exposes_no_evidence_or_plan_promotion_api():
     from app.core.data_rules.repository import DataRuleRepository
 
-    plan_id = new_governance_uid()
+    assert not hasattr(DataRuleRepository, "record_bound_plan_test")
+    assert not hasattr(DataRuleRepository, "publish_tested_bound_plan")
 
-    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
+@pytest.mark.parametrize("dialect", ["postgresql", "mysql"])
+def test_sql_v1_rejects_integer_division_for_each_sql_dialect(dialect):
+    from app.core.data_rules.expressions import backend_support, parse_expression
 
-        def mappings(self):
-            return Mappings(self.row)
+    ast = parse_expression("customer_id / 2 > 0")
+    assert dialect not in backend_support(ast)
+    with pytest.raises(ValueError, match="unsupported|division"):
+        _compile(
+            dialect,
+            steps=[
+                {
+                    "id": "division",
+                    "op": "filter",
+                    "expression": "customer_id / 2 > 0",
+                }
+            ],
+        )
 
-    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)
+@pytest.mark.parametrize("field_type", ["json", "binary"])
+def test_deduplicate_rejects_non_orderable_projected_tie_breakers(field_type):
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
 
-    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"
+    fields = [
+        ("customer_id", "integer", False),
+        ("payload", field_type, True),
+    ]
+    input_schema = _schema("bd:customer:raw", fields)
+    output_schema = _schema("bd:customer:clean", fields)
+    rule = published_rule(
+        [
+            {
+                "id": "deduplicate",
+                "op": "deduplicate",
+                "keys": ["customer_id"],
+                "order_by": ["customer_id"],
+                "keep": "first",
+            }
+        ]
     )
-    with pytest.raises(ValueError, match="evidence"):
-        repository.record_bound_plan_test(
-            plan_id=plan_id,
-            evidence={**evidence, "commit_outcome": "unknown"},
+
+    with pytest.raises(ValueError, match="orderable"):
+        SqlGlotRuleCompiler("postgresql").compile(
+            rule_version=rule,
+            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(),
         )

tests/integration/test_data_rule_sql_execution.py

@@ -106,13 +106,12 @@ class PlanRepository:
             "plan_hash": compiled["plan_hash"],
         }
 
-    def publish_with_evidence(self, plan_hash, evidence):
+    def trust_test_only_preflight_and_publish(self, plan_hash, evidence):
         assert self.record is not None
         assert self.record["plan_status"] == "compiled"
         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):
@@ -305,7 +304,7 @@ def test_bound_rule_compiles_publishes_executes_and_rejects_tampering(
         )
         with engine.begin() as connection:
             connection.execute(text(f"DELETE FROM {target_name}"))
-        repository.publish_with_evidence(
+        repository.trust_test_only_preflight_and_publish(
             compiled["plan_hash"],
             preflight_evidence,
         )

tests/test_data_rule_schema.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import importlib.util
 from pathlib import Path
 
+import pytest
 
 ROOT = Path(__file__).resolve().parents[1]
 MIGRATION = (
@@ -16,6 +18,12 @@ RUNTIME_MIGRATION = (
     / "versions"
     / "20260723_120_rule_execution_runtime.py"
 )
+PLAN_STATUS_MIGRATION = (
+    ROOT
+    / "migrations"
+    / "versions"
+    / "20260723_130_bound_plan_lifecycle.py"
+)
 
 EXPECTED_TABLES = {
     "data_rules",
@@ -94,9 +102,28 @@ def test_rule_execution_runtime_migration_adds_pinned_runtime_evidence():
         assert expected in source
 
 
-def test_rule_execution_runtime_migration_is_forward_preserving():
+def test_rule_execution_runtime_migration_remains_forward_preserving():
     source = RUNTIME_MIGRATION.read_text(encoding="utf-8")
     downgrade = source.split("def downgrade()", 1)[1]
 
     assert "DROP TABLE" not in downgrade.upper()
     assert "pass" in downgrade
+
+
+def test_bound_plan_status_migration_is_forward_only_and_keeps_compiled_valid():
+    source = PLAN_STATUS_MIGRATION.read_text(encoding="utf-8")
+
+    assert 'revision = "20260723_130"' in source
+    assert 'down_revision = "20260723_120"' in source
+    assert "'compiled','published','revoked'" in source
+    assert "'tested'" not in source
+    assert "raise RuntimeError" in source.split("def downgrade()", 1)[1]
+    spec = importlib.util.spec_from_file_location(
+        "bound_plan_status_migration",
+        PLAN_STATUS_MIGRATION,
+    )
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    with pytest.raises(RuntimeError, match="forward-only|cannot downgrade"):
+        module.downgrade()