fix: harden Polars artifact execution

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

@@ -2,11 +2,18 @@
 
 Date: 2026-07-23
 Branch: `codex/data-rule-execution-m3a-m5`
-Lifecycle boundary: physical Polars plans are persisted as `compiled`; this task does not publish them.
+Lifecycle boundary: physical Polars plans are created as `compiled`; this task
+does not add a production promotion API. The real execution test inserts a
+published fixture only to exercise the pre-existing Runner publication gate.
 
 ## Outcome
 
-Implemented a closed JSON Polars compiler, allowlisted LazyFrame reconstruction, server-owned digest-bound Parquet artifact storage, PostgreSQL-backed artifact binding resolution, Runner registration, canonical plan attestations, and a real PostgreSQL + MySQL + MinIO integration.
+Implemented a closed JSON Polars compiler, shared compile/runtime semantic
+validation, allowlisted LazyFrame reconstruction, server-owned digest-bound
+Parquet artifact storage, correlation-scoped PostgreSQL artifact handoff,
+current output-binding re-attestation, separated operation metrics, Runner
+registration, canonical plan attestations, and a real PostgreSQL + MySQL +
+MinIO integration through the production repository/resolver/executor path.
 
 The plan contains canonical RuleVersion, SchemaSnapshot, and DatasetBinding hashes plus `dataops-polars-1.42.1` provenance. It contains no Python source, pickle, callable, module, client path, arbitrary URL, secret, or Polars internal serialized plan.
 
@@ -25,12 +32,30 @@ RED evidence was captured before each production slice:
 - Closed operation semantic revalidation: `3 failed` because tampered identifiers/flags were not rejected.
 - Strict MinIO artifact binding contract: `1 failed` because all URI-shaped Parquet refs were previously rejected.
 - Repository-attested binding hashes: `1 failed` because compiler input rejected the canonical `binding_hash`.
+- Artifact remediation slice: `4 failed, 6 passed` before per-plan limits,
+  upload rollback, full schema contracts, and scoped expiry cleanup existed.
+- Shared semantic remediation slice: `3 failed, 11 deselected` before timezone,
+  expression type replay, and lookup target collision checks existed.
+- Runtime remediation slice: `3 failed, 3 deselected` before output-binding
+  attestation, exact decimal/timestamptz casts, and separated metrics existed.
+- Artifact catalog migration: `1 failed, 6 deselected` before migration
+  `20260723_140` existed.
+- PostgreSQL artifact resolver: `2 failed, 10 deselected` before catalog
+  resolution, re-attestation, and registration existed.
+- Dependency SBOM: `1 failed` before the CycloneDX JSON document existed.
+- Incomplete runtime type contracts: `2 failed, 14 deselected` before decimal
+  precision/scale and timestamptz timezone were mandatory.
+- Correlation re-attestation: `1 failed, 11 deselected` before a catalog row's
+  object key was checked against the requested correlation prefix.
 
 GREEN:
 
-- Final required focused command:
-  - `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_polars_compiler.py tests/runner/test_rule_polars.py tests/runner/test_artifacts.py tests/integration/test_data_rule_polars_execution.py`
-  - Result: `23 passed in 0.95s`.
+- Final focused unit/schema/SBOM command:
+  - `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_polars_compiler.py tests/runner/test_rule_polars.py tests/runner/test_artifacts.py tests/test_data_rule_schema.py tests/test_data_rule_runtime_sbom.py`
+  - Result: `42 passed in 1.02s`.
+- Final real integration command:
+  - `PYTHONPATH=. .venv/bin/pytest -q tests/integration/test_data_rule_polars_execution.py --show-capture=no -o log_cli=false`
+  - Result: `1 passed in 1.01s`.
 - Expanded Task 5 / SQL regression slice during implementation:
   - Result: `197 passed`, then `75 passed` after repository hardening.
 
@@ -41,14 +66,25 @@ The integration used runtime values discovered from `deploy/docker/docker-compos
 - PostgreSQL: live source at `127.0.0.1:25432`.
 - MySQL: live lookup source at `127.0.0.1:23306`.
 - MinIO: live object store at `127.0.0.1:19000`.
-- Flow: read PostgreSQL customers, read MySQL segments, write both as bounded Parquet artifacts, compile and execute normalize → lookup join → assert → deduplicate, write and reread the output artifact.
-- Evidence: 4 input rows → 2 output rows, 2 rejected rows, 1 assertion violation; output values and segment enrichment were verified.
-- Cleanup: only `rules/<test-correlation-id>/` objects were removed. The test asserted the exact prefix was empty afterward. Test-owned PostgreSQL/MySQL tables were also dropped.
+- Platform PostgreSQL: live canonical plans, bindings, snapshots, and artifact
+  catalog at `127.0.0.1:15432`.
+- Flow: read PostgreSQL customers, read MySQL segments, write both as bounded
+  Parquet artifacts, register them through `PostgresArtifactResolver`, load the
+  published fixture through `PostgresRulePlanRepository`, execute through
+  `RulePlanExecutor`, normalize → lookup join → assert → deduplicate, register
+  and reread the output artifact.
+- Evidence: 4 input rows → 2 output rows, 1 assertion reject, 1 deduplicated
+  row; output values and segment enrichment were verified. A second execution
+  with the same correlation succeeded and retained its identical output digest
+  idempotently in the stable artifact catalog.
+- Cleanup: only `rules/<test-correlation-id>/` objects were removed. The test
+  asserted the exact prefix was empty afterward. Test-owned source tables and
+  canonical platform rows were also removed.
 
 ## Full verification
 
 - `PYTHONPATH=. .venv/bin/pytest -q`
-  - `527 passed, 26 skipped, 59 subtests passed in 4.69s`.
+  - `541 passed, 26 skipped, 59 subtests passed in 4.52s`.
 - `docker compose -f deploy/docker/docker-compose.yml config --quiet`
   - Passed.
 - Ruff over all Task 5 production and test files
@@ -58,9 +94,11 @@ The integration used runtime values discovered from `deploy/docker/docker-compos
 
 ## Dependency and license
 
-- Added exact pin: `polars==1.42.1`.
+- Exact pins: `polars==1.42.1`, `minio==7.2.10`.
 - Installed into `.venv`: Polars `1.42.1` and its matching `polars-runtime-32==1.42.1`.
 - Installed metadata: Python `>=3.10`; MIT license text.
+- Added `docs/security/data-rule-runtime-sbom.json` with versions, licenses,
+  official sources, package URLs, and runtime purposes.
 - Plans remain independent of Polars internal serialization formats.
 
 ## Files
@@ -74,6 +112,9 @@ Created:
 - `tests/runner/test_artifacts.py`
 - `tests/runner/test_rule_polars.py`
 - `tests/integration/test_data_rule_polars_execution.py`
+- `migrations/versions/20260723_140_rule_run_artifacts.py`
+- `docs/security/data-rule-runtime-sbom.json`
+- `tests/test_data_rule_runtime_sbom.py`
 
 Modified:
 
@@ -97,7 +138,17 @@ Modified:
 - Dataset bindings and lookup contexts are resolved by canonical IDs; repository binding hashes are preserved and re-attested at execution.
 - Artifact keys are generated only by the server as `rules/<correlation-id>/<artifact-id>.parquet`.
 - Artifact reads validate store ownership, strict reference shape, content type, size, TTL, digest, row count, schema digest, and memory bounds.
-- Artifact writes reread the server object before returning a ref, catching same-size corruption.
+- Artifact reads/writes enforce plan limits before and after transfer,
+  decompression/materialization, and serialization. Full field contracts cover
+  nullability, exact decimal precision/scale, and timestamptz timezone.
+- Artifact writes reread the server object before returning a ref, catching
+  same-size corruption, and delete the generated object on validation failure.
+- High-risk runtime intermediates are materialized through the same bounded
+  checkpoint; assertion rejects, filters, deduplication, joins, and aggregation
+  have separate counters.
+- Artifact catalog resolution is exact-correlation scoped, current binding
+  hashes are re-attested, and expired cleanup is limited to the configured
+  bucket's exact `rules/<correlation>/` prefix.
 - MinIO credentials stay in Runner settings with `repr=False` and never enter refs, logs, or plans.
 
 ## Concerns / deferred work

app/core/data_rules/compilers/polars.py

@@ -8,6 +8,7 @@ import json
 import re
 from decimal import Decimal, InvalidOperation
 from typing import Any
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from app.core.common.identifiers import ensure_governance_uid
 from app.core.data_rules.compilers.base import RuleCompiler
@@ -29,6 +30,10 @@ RESULT_CONTRACT = {
     "rows_in": "counted",
     "rows_out": "counted",
     "rows_rejected": "counted",
+    "rows_filtered": "counted",
+    "rows_deduplicated": "counted",
+    "rows_join_dropped": "counted",
+    "rows_aggregated": "counted",
     "violations": "counted",
 }
 SUPPORTED_OPERATIONS = {
@@ -50,6 +55,7 @@ SUPPORTED_MASK_POLICIES = {"preserve_last_4", "redact"}
 _PLAN_KEYS = {
     "schema_version",
     "compiler_version",
+    "timezone",
     "rule_version_id",
     "rule_spec_hash",
     "input_schema_snapshot_id",
@@ -264,6 +270,69 @@ def _fields(snapshot: dict[str, Any]) -> dict[str, str]:
     return {field["name"]: field["type"] for field in snapshot["fields"]}
 
 
+def _field_contracts(fields: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    contracts = {}
+    for raw_field in fields:
+        field = copy.deepcopy(raw_field)
+        field_type = field["type"]
+        if field_type == "decimal":
+            precision = field.get("precision")
+            scale = field.get("scale")
+            if (
+                isinstance(precision, bool)
+                or not isinstance(precision, int)
+                or precision < 1
+            ):
+                raise ValueError("decimal field precision is required")
+            if (
+                isinstance(scale, bool)
+                or not isinstance(scale, int)
+                or scale < 0
+                or scale > precision
+            ):
+                raise ValueError("decimal field scale is invalid")
+        elif field_type == "timestamptz":
+            field["timezone"] = _timezone(field.get("timezone"))
+        elif field_type == "timestamp" and field.get("timezone") is not None:
+            raise ValueError("timestamp field must not carry a timezone")
+        contracts[field["name"]] = field
+    return contracts
+
+
+def _semantic_type(field: dict[str, Any]) -> tuple[Any, ...]:
+    field_type = field["type"]
+    if field_type == "decimal":
+        return (
+            field_type,
+            field.get("precision"),
+            field.get("scale"),
+        )
+    if field_type == "timestamptz":
+        return (field_type, field.get("timezone"))
+    return (field_type,)
+
+
+def _field_for_type(
+    field_type: str,
+    *,
+    output_field: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    if output_field is not None and output_field["type"] == field_type:
+        return copy.deepcopy(output_field)
+    return {"name": "", "type": field_type, "nullable": True}
+
+
+def _timezone(value: Any) -> str:
+    timezone = str(value or "")
+    if not timezone or len(timezone) > 100:
+        raise ValueError("bound Polars timezone is invalid")
+    try:
+        ZoneInfo(timezone)
+    except (ValueError, ZoneInfoNotFoundError) as exc:
+        raise ValueError("bound Polars timezone is invalid") from exc
+    return timezone
+
+
 def _field_list(snapshot: dict[str, Any]) -> list[dict[str, Any]]:
     return copy.deepcopy(snapshot["fields"])
 
@@ -439,6 +508,7 @@ def validate_bound_polars_plan(value: Any) -> dict[str, Any]:
         raise ValueError("unsupported bound Polars plan schema version")
     if plan["compiler_version"] != COMPILER_VERSION:
         raise ValueError("unsupported bound Polars compiler version")
+    plan["timezone"] = _timezone(plan["timezone"])
     for key in (
         "rule_version_id",
         "input_schema_snapshot_id",
@@ -469,6 +539,127 @@ def validate_bound_polars_plan(value: Any) -> dict[str, Any]:
     ):
         raise ValueError("Polars operations must be a non-empty bounded array")
     plan["operations"] = [_operation(item) for item in raw_operations]
+    input_fields = _field_contracts(plan["input_fields"])
+    output_fields = _field_contracts(plan["output_fields"])
+    current_fields = copy.deepcopy(input_fields)
+    current_types = {
+        name: field["type"] for name, field in current_fields.items()
+    }
+    for operation in plan["operations"]:
+        op = operation["op"]
+        if op in {"assert", "derive", "filter"}:
+            result_type = type_check_expression(
+                operation["expression_ast"], current_types
+            )
+            if "polars" not in backend_support(operation["expression_ast"]):
+                raise ValueError("expression is unsupported by Polars")
+            if op in {"assert", "filter"} and result_type != "boolean":
+                raise ValueError(f"{op} expression must return boolean")
+            if op == "derive":
+                target = operation["target"]
+                target_field = output_fields.get(target)
+                if target_field is None or target_field["type"] != result_type:
+                    raise ValueError(
+                        "derive expression type does not match its target type"
+                    )
+                current_fields[target] = copy.deepcopy(target_field)
+                current_types[target] = result_type
+        elif op in {
+            "fill_null",
+            "map_values",
+            "mask",
+            "normalize_text",
+            "regex_replace",
+        }:
+            column = operation["column"]
+            if column not in current_fields:
+                raise ValueError(f"{op} references an unknown field")
+            if op in {
+                "map_values",
+                "mask",
+                "normalize_text",
+                "regex_replace",
+            } and current_types[column] != "string":
+                raise ValueError(f"{op} requires a string field type")
+            if op == "fill_null" and not _value_matches_type(
+                operation["value"], current_types[column]
+            ):
+                raise ValueError("fill_null value does not match the field type")
+        elif op == "cast":
+            column = operation["column"]
+            if column not in current_fields:
+                raise ValueError("cast references an unknown field")
+            target = output_fields.get(column)
+            current_fields[column] = _field_for_type(
+                operation["to"], output_field=target
+            )
+            current_fields[column]["name"] = column
+            current_types[column] = operation["to"]
+        elif op == "deduplicate":
+            if not set(operation["keys"] + operation["order_by"]) <= set(
+                current_fields
+            ):
+                raise ValueError("deduplicate references unknown fields")
+        elif op == "aggregate":
+            if not set(operation["group_by"]) <= set(current_fields):
+                raise ValueError("aggregate references unknown group fields")
+            next_fields = {
+                name: copy.deepcopy(current_fields[name])
+                for name in operation["group_by"]
+            }
+            for aggregate in operation["aggregations"]:
+                source = aggregate["column"]
+                target = aggregate["target"]
+                function = aggregate["function"]
+                if source not in current_fields or target not in output_fields:
+                    raise ValueError("aggregate field is unknown")
+                source_type = current_types[source]
+                if function in {"mean", "sum"} and source_type not in _NUMERIC_TYPES:
+                    raise ValueError("aggregate function requires a numeric field")
+                result_type = (
+                    "integer"
+                    if function == "count"
+                    else "double" if function == "mean" else source_type
+                )
+                if output_fields[target]["type"] != result_type:
+                    raise ValueError(
+                        "aggregate result type does not match output schema"
+                    )
+                next_fields[target] = copy.deepcopy(output_fields[target])
+            current_fields = next_fields
+            current_types = {
+                name: field["type"] for name, field in current_fields.items()
+            }
+        elif op == "lookup_join":
+            lookup_fields = _field_contracts(operation["lookup_fields"])
+            for left, right in zip(
+                operation["left_on"], operation["right_on"], strict=True
+            ):
+                if left not in current_fields or right not in lookup_fields:
+                    raise ValueError("lookup join key is unknown")
+                if _semantic_type(current_fields[left]) != _semantic_type(
+                    lookup_fields[right]
+                ):
+                    raise ValueError("lookup join key types do not match")
+            for target, source in operation["select"].items():
+                if target in current_fields:
+                    raise ValueError("lookup selected target collision")
+                if source not in lookup_fields or target not in output_fields:
+                    raise ValueError("lookup select field is unknown")
+                if _semantic_type(lookup_fields[source]) != _semantic_type(
+                    output_fields[target]
+                ):
+                    raise ValueError(
+                        "lookup selected type does not match output schema"
+                    )
+                current_fields[target] = copy.deepcopy(output_fields[target])
+                current_types[target] = output_fields[target]["type"]
+    for name, expected in output_fields.items():
+        actual = current_fields.get(name)
+        if actual is None or _semantic_type(actual) != _semantic_type(expected):
+            raise ValueError(
+                "compiled Polars field types do not match the output schema"
+            )
     if plan["result_contract"] != RESULT_CONTRACT:
         raise ValueError("bound Polars result contract is unsupported")
     _reject_forbidden(plan)
@@ -821,6 +1012,8 @@ class PolarsRuleCompiler(RuleCompiler):
                 for target, source in sorted(select.items()):
                     target = _identifier(target, "lookup target")
                     source = _identifier(source, "lookup source")
+                    if target in current_fields:
+                        raise ValueError("lookup selected target collision")
                     if source not in lookup_types or target not in output_types:
                         raise ValueError("lookup select field is unknown")
                     if lookup_types[source] != output_types[target]:
@@ -862,6 +1055,7 @@ class PolarsRuleCompiler(RuleCompiler):
             {
                 "schema_version": PLAN_SCHEMA_VERSION,
                 "compiler_version": COMPILER_VERSION,
+                "timezone": spec["timezone"],
                 "rule_version_id": rule_version_id,
                 "rule_spec_hash": rule_version["spec_hash"],
                 "input_schema_snapshot_id": source_schema["id"],

app/runner/artifacts.py

@@ -8,6 +8,7 @@ import json
 import os
 import re
 import tempfile
+from base64 import urlsafe_b64decode, urlsafe_b64encode
 from collections.abc import Mapping
 from contextlib import suppress
 from datetime import UTC, datetime, timedelta
@@ -20,6 +21,7 @@ from app.core.common.identifiers import (
     ensure_governance_uid,
     new_governance_uid,
 )
+from app.core.data_rules.execution_contracts import canonical_schema_hash
 
 PARQUET_CONTENT_TYPE = "application/x-parquet"
 _DIGEST = re.compile(r"^[0-9a-f]{64}$")
@@ -55,23 +57,133 @@ def _uid(value: Any, label: str) -> str:
         raise ValueError(f"{label} must be a valid UUIDv7") from exc
 
 
-def _schema_hash(frame: pl.DataFrame | pl.LazyFrame) -> str:
+def _inferred_schema_fields(
+    frame: pl.DataFrame | pl.LazyFrame,
+) -> list[dict[str, Any]]:
     schema = (
         frame.collect_schema()
         if isinstance(frame, pl.LazyFrame)
         else frame.schema
     )
-    canonical = [
-        {"name": name, "dtype": str(dtype)}
-        for name, dtype in schema.items()
-    ]
+    fields = []
+    for name, dtype in schema.items():
+        field: dict[str, Any] = {
+            "name": name,
+            "nullable": True,
+        }
+        if dtype == pl.Boolean:
+            field["type"] = "boolean"
+        elif dtype == pl.Date:
+            field["type"] = "date"
+        elif dtype == pl.String:
+            field["type"] = "string"
+        elif dtype.is_integer():
+            field["type"] = "integer"
+        elif dtype == pl.Float32:
+            field["type"] = "float"
+        elif dtype == pl.Float64:
+            field["type"] = "double"
+        elif dtype.is_decimal():
+            field.update(
+                {
+                    "type": "decimal",
+                    "precision": dtype.precision,
+                    "scale": dtype.scale,
+                }
+            )
+        elif isinstance(dtype, pl.Datetime):
+            field["type"] = (
+                "timestamptz" if dtype.time_zone else "timestamp"
+            )
+            if dtype.time_zone:
+                field["timezone"] = dtype.time_zone
+        else:
+            raise ValueError(f"unsupported artifact dtype for {name}")
+        fields.append(field)
+    return sorted(fields, key=lambda item: item["name"])
+
+
+def _normalized_schema_fields(value: Any) -> list[dict[str, Any]]:
+    canonical_schema_hash(value)
+    return sorted(
+        [dict(field) for field in value],
+        key=lambda item: item["name"],
+    )
+
+
+def _schema_contract(value: Any) -> tuple[list[dict[str, Any]], str, str]:
+    fields = _normalized_schema_fields(value)
     encoded = json.dumps(
-        canonical,
+        fields,
         sort_keys=True,
         separators=(",", ":"),
         ensure_ascii=False,
+    ).encode("utf-8")
+    return (
+        fields,
+        canonical_schema_hash(fields),
+        urlsafe_b64encode(encoded).decode("ascii"),
     )
-    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+def _decode_schema_contract(value: str) -> list[dict[str, Any]]:
+    try:
+        decoded = urlsafe_b64decode(value.encode("ascii"))
+        fields = json.loads(decoded.decode("utf-8"))
+    except Exception as exc:
+        raise ValueError("artifact schema contract metadata is invalid") from exc
+    return _normalized_schema_fields(fields)
+
+
+def _validate_frame_schema(
+    frame: pl.DataFrame | pl.LazyFrame,
+    fields: list[dict[str, Any]],
+) -> None:
+    schema = (
+        frame.collect_schema()
+        if isinstance(frame, pl.LazyFrame)
+        else frame.schema
+    )
+    expected_names = {field["name"] for field in fields}
+    if set(schema.names()) != expected_names:
+        raise ValueError("artifact schema fields do not match")
+    for field in fields:
+        dtype = schema[field["name"]]
+        field_type = field["type"]
+        matches = (
+            (field_type == "boolean" and dtype == pl.Boolean)
+            or (field_type == "date" and dtype == pl.Date)
+            or (field_type == "string" and dtype == pl.String)
+            or (field_type == "integer" and dtype.is_integer())
+            or (field_type == "float" and dtype == pl.Float32)
+            or (field_type == "double" and dtype == pl.Float64)
+            or (
+                field_type == "decimal"
+                and dtype.is_decimal()
+                and dtype.precision == field.get("precision")
+                and dtype.scale == field.get("scale")
+            )
+            or (
+                field_type == "timestamp"
+                and isinstance(dtype, pl.Datetime)
+                and dtype.time_zone is None
+            )
+            or (
+                field_type == "timestamptz"
+                and isinstance(dtype, pl.Datetime)
+                and dtype.time_zone == field.get("timezone")
+            )
+        )
+        if not matches:
+            raise ValueError(
+                f"artifact schema type for {field['name']} does not match"
+            )
+    if isinstance(frame, pl.DataFrame):
+        for field in fields:
+            if not field["nullable"] and frame[field["name"]].null_count():
+                raise ValueError(
+                    f"artifact nullable contract for {field['name']} does not match"
+                )
 
 
 def _metadata(value: Any) -> dict[str, str]:
@@ -88,6 +200,7 @@ def _metadata(value: Any) -> dict[str, str]:
             "schema-sha256",
             "expires-at",
             "artifact-bytes",
+            "schema-contract",
         }:
             normalized[name] = str(item)
     required = {
@@ -96,6 +209,7 @@ def _metadata(value: Any) -> dict[str, str]:
         "schema-sha256",
         "expires-at",
         "artifact-bytes",
+        "schema-contract",
     }
     if set(normalized) != required:
         raise ValueError("artifact content metadata is incomplete")
@@ -135,6 +249,31 @@ class ArtifactStore:
         if not self.client.bucket_exists(self.bucket):
             raise ValueError("artifact bucket does not exist")
 
+    def _limits(self, value: Any = None) -> dict[str, int]:
+        configured = {
+            "max_rows": self.max_rows,
+            "max_artifact_bytes": self.max_artifact_bytes,
+            "memory_limit_bytes": self.memory_limit_bytes,
+        }
+        if value is None:
+            return configured
+        if not isinstance(value, dict) or set(value) != set(configured):
+            raise ValueError("artifact limits must have a closed shape")
+        result = {}
+        for key, ceiling in configured.items():
+            item = value[key]
+            if (
+                isinstance(item, bool)
+                or not isinstance(item, int)
+                or item < 1
+                or item > ceiling
+            ):
+                raise ValueError(
+                    f"artifact {key} exceeds the configured ceiling"
+                )
+            result[key] = item
+        return result
+
     def _parse_ref(self, ref: Any) -> str:
         prefix = f"minio://{self.bucket}/"
         if not isinstance(ref, str) or not ref.startswith(prefix):
@@ -155,11 +294,15 @@ class ArtifactStore:
         key: str,
         *,
         expected_digest: str | None = None,
+        limits: dict[str, int] | None = None,
     ) -> tuple[Any, dict[str, str]]:
+        effective = self._limits(limits)
         stat = self.client.stat_object(self.bucket, key)
         size = int(getattr(stat, "size", -1))
-        if size < 1 or size > self.max_artifact_bytes:
+        if size < 1 or size > effective["max_artifact_bytes"]:
             raise ValueError("artifact size exceeds the configured limit")
+        if size > effective["memory_limit_bytes"]:
+            raise ValueError("artifact download exceeds the memory limit")
         if str(getattr(stat, "content_type", "")).lower() != PARQUET_CONTENT_TYPE:
             raise ValueError("artifact content type is invalid")
         metadata = _metadata(getattr(stat, "metadata", None))
@@ -173,7 +316,7 @@ class ArtifactStore:
             metadata_size = int(metadata["artifact-bytes"])
         except (TypeError, ValueError) as exc:
             raise ValueError("artifact count metadata is invalid") from exc
-        if row_count < 0 or row_count > self.max_rows:
+        if row_count < 0 or row_count > effective["max_rows"]:
             raise ValueError("artifact row count exceeds the configured limit")
         if metadata_size != size:
             raise ValueError("artifact size metadata does not match")
@@ -188,7 +331,11 @@ class ArtifactStore:
         frame: pl.LazyFrame | pl.DataFrame,
         correlation_id: str,
         ttl_seconds: int,
+        *,
+        schema_fields: list[dict[str, Any]] | None = None,
+        limits: dict[str, int] | None = None,
     ) -> dict[str, Any]:
+        effective = self._limits(limits)
         correlation = _uid(correlation_id, "correlation_id")
         if (
             isinstance(ttl_seconds, bool)
@@ -203,18 +350,24 @@ class ArtifactStore:
             lazy = frame
         else:
             raise ValueError("artifact frame must be a Polars frame")
-        collected = lazy.head(self.max_rows + 1).collect(engine="streaming")
-        if collected.height > self.max_rows:
+        collected = lazy.head(effective["max_rows"] + 1).collect(
+            engine="streaming"
+        )
+        if collected.height > effective["max_rows"]:
             raise ValueError("artifact row count exceeds the configured limit")
-        if collected.estimated_size() > self.memory_limit_bytes:
+        if collected.estimated_size() > effective["memory_limit_bytes"]:
             raise ValueError("artifact frame exceeds the configured memory limit")
-        schema_digest = _schema_hash(collected)
+        fields, schema_digest, schema_encoded = _schema_contract(
+            schema_fields or _inferred_schema_fields(collected)
+        )
+        _validate_frame_schema(collected, fields)
         expires_at = _timestamp(
             _now_utc(self.clock) + timedelta(seconds=ttl_seconds)
         )
         artifact_id = new_governance_uid()
         key = f"rules/{correlation}/{artifact_id}.parquet"
         path = None
+        uploaded = False
         try:
             with tempfile.NamedTemporaryFile(
                 prefix="dataops-rule-artifact-",
@@ -224,8 +377,17 @@ class ArtifactStore:
                 path = handle.name
             collected.write_parquet(path)
             size = os.path.getsize(path)
-            if size < 1 or size > self.max_artifact_bytes:
+            if size < 1 or size > effective["max_artifact_bytes"]:
                 raise ValueError("artifact size exceeds the configured limit")
+            if size > effective["memory_limit_bytes"]:
+                raise ValueError("serialized artifact exceeds the memory limit")
+            if (
+                size + collected.estimated_size()
+                > effective["memory_limit_bytes"]
+            ):
+                raise ValueError(
+                    "artifact serialization exceeds the memory limit"
+                )
             digest = hashlib.sha256()
             with open(path, "rb") as handle:
                 while chunk := handle.read(1024 * 1024):
@@ -244,11 +406,27 @@ class ArtifactStore:
                         "schema-sha256": schema_digest,
                         "expires-at": expires_at,
                         "artifact-bytes": str(size),
+                        "schema-contract": schema_encoded,
                     },
                 )
-            self._validated_stat(key, expected_digest=digest_hex)
+            uploaded = True
+            self._validated_stat(
+                key,
+                expected_digest=digest_hex,
+                limits=effective,
+            )
             artifact_ref = f"minio://{self.bucket}/{key}"
-            self.read(artifact_ref, digest_hex)
+            self.read(
+                artifact_ref,
+                digest_hex,
+                expected_schema_fields=fields,
+                limits=effective,
+            )
+        except Exception:
+            if uploaded:
+                with suppress(Exception):
+                    self.client.remove_object(self.bucket, key)
+            raise
         finally:
             if path is not None:
                 with suppress(FileNotFoundError):
@@ -258,6 +436,7 @@ class ArtifactStore:
             "digest": digest_hex,
             "row_count": collected.height,
             "schema_hash": schema_digest,
+            "schema_fields": fields,
             "expires_at": expires_at,
         }
 
@@ -266,21 +445,42 @@ class ArtifactStore:
 
         key = self._parse_ref(ref)
         _stat, metadata = self._validated_stat(key)
+        fields = _decode_schema_contract(metadata["schema-contract"])
+        if canonical_schema_hash(fields) != metadata["schema-sha256"]:
+            raise ValueError("artifact schema contract does not match")
         return {
             "artifact_ref": ref,
             "digest": metadata["sha256"],
             "row_count": int(metadata["row-count"]),
             "schema_hash": metadata["schema-sha256"],
+            "schema_fields": fields,
             "expires_at": metadata["expires-at"],
         }
 
-    def read(self, ref: str, expected_digest: str) -> pl.LazyFrame:
+    def read(
+        self,
+        ref: str,
+        expected_digest: str,
+        *,
+        expected_schema_fields: list[dict[str, Any]] | None = None,
+        limits: dict[str, int] | None = None,
+    ) -> pl.LazyFrame:
+        effective = self._limits(limits)
         if _DIGEST.fullmatch(str(expected_digest or "")) is None:
             raise ValueError("expected artifact digest is invalid")
         key = self._parse_ref(ref)
         _stat, metadata = self._validated_stat(
-            key, expected_digest=expected_digest
+            key,
+            expected_digest=expected_digest,
+            limits=effective,
         )
+        fields = _decode_schema_contract(metadata["schema-contract"])
+        if canonical_schema_hash(fields) != metadata["schema-sha256"]:
+            raise ValueError("artifact schema contract does not match")
+        if expected_schema_fields is not None:
+            expected = _normalized_schema_fields(expected_schema_fields)
+            if expected != fields:
+                raise ValueError("artifact schema contract is not expected")
         response = self.client.get_object(self.bucket, key)
         digest = hashlib.sha256()
         payload = io.BytesIO()
@@ -288,10 +488,14 @@ class ArtifactStore:
         try:
             while chunk := response.read(1024 * 1024):
                 size += len(chunk)
-                if size > self.max_artifact_bytes:
+                if size > effective["max_artifact_bytes"]:
                     raise ValueError(
                         "artifact size exceeds the configured limit"
                     )
+                if size > effective["memory_limit_bytes"]:
+                    raise ValueError(
+                        "artifact download exceeds the memory limit"
+                    )
                 digest.update(chunk)
                 payload.write(chunk)
         finally:
@@ -303,19 +507,56 @@ class ArtifactStore:
             raise ValueError("artifact digest does not match content")
         payload.seek(0)
         try:
-            frame = pl.read_parquet(payload)
+            frame = pl.read_parquet(
+                payload,
+                n_rows=effective["max_rows"] + 1,
+                memory_map=False,
+            )
         except Exception as exc:
             raise ValueError("artifact is not valid Parquet") from exc
         if frame.height != int(metadata["row-count"]):
             raise ValueError("artifact row count does not match metadata")
-        if frame.height > self.max_rows:
+        if frame.height > effective["max_rows"]:
             raise ValueError("artifact row count exceeds the configured limit")
-        if frame.estimated_size() > self.memory_limit_bytes:
+        if frame.estimated_size() > effective["memory_limit_bytes"]:
             raise ValueError("artifact frame exceeds the configured memory limit")
-        if _schema_hash(frame) != metadata["schema-sha256"]:
-            raise ValueError("artifact schema does not match metadata")
+        if size + frame.estimated_size() > effective["memory_limit_bytes"]:
+            raise ValueError("artifact decompression exceeds the memory limit")
+        _validate_frame_schema(frame, fields)
         return frame.lazy()
 
+    def cleanup_expired(self, correlation_id: str) -> int:
+        correlation = _uid(correlation_id, "correlation_id")
+        prefix = f"rules/{correlation}/"
+        removed = 0
+        for item in self.client.list_objects(
+            self.bucket,
+            prefix=prefix,
+            recursive=True,
+        ):
+            key = str(getattr(item, "object_name", ""))
+            if not key.startswith(prefix):
+                continue
+            try:
+                self._parse_ref(f"minio://{self.bucket}/{key}")
+                stat = self.client.stat_object(self.bucket, key)
+                metadata = _metadata(getattr(stat, "metadata", None))
+                expired = _parse_timestamp(
+                    metadata["expires-at"]
+                ) <= _now_utc(self.clock)
+            except ValueError:
+                continue
+            if expired:
+                self.client.remove_object(self.bucket, key)
+                removed += 1
+        return removed
+
+    def delete(self, ref: str) -> None:
+        """Delete one exact store-owned artifact after validating its key."""
+
+        key = self._parse_ref(ref)
+        self.client.remove_object(self.bucket, key)
+
 
 class PostgresArtifactResolver:
     """Resolve a canonical artifact binding without accepting caller paths."""
@@ -329,25 +570,166 @@ class PostgresArtifactResolver:
         correlation = _uid(correlation_id, "artifact correlation id")
         statement = text(
             """
-            SELECT object_ref, binding_hash
-            FROM public.dataflow_dataset_bindings
-            WHERE id = CAST(:binding_id AS uuid)
-              AND object_kind = 'parquet_artifact'
-              AND access_mode IN ('read', 'read_write')
+            SELECT
+                a.artifact_ref,
+                a.artifact_digest,
+                a.row_count,
+                a.schema_hash,
+                a.schema_fields,
+                a.expires_at,
+                b.binding_hash
+            FROM public.rule_run_artifacts a
+            JOIN public.dataflow_dataset_bindings b
+              ON b.id = a.binding_id
+            WHERE a.binding_id = CAST(:binding_id AS uuid)
+              AND a.correlation_id = CAST(:correlation_id AS uuid)
+              AND a.expires_at > CURRENT_TIMESTAMP
+              AND b.object_kind = 'parquet_artifact'
+              AND b.access_mode IN ('read', 'read_write')
+            ORDER BY a.created_at DESC
+            LIMIT 1
             """
         )
         with self.engine.connect() as connection:
             row = connection.execute(
-                statement, {"binding_id": binding}
+                statement,
+                {
+                    "binding_id": binding,
+                    "correlation_id": correlation,
+                },
             ).mappings().one_or_none()
         if row is None:
             raise ValueError("canonical artifact binding was not found")
-        artifact_ref = str(row["object_ref"])
-        if f"/rules/{correlation}/" not in artifact_ref:
+        artifact_ref = str(row["artifact_ref"])
+        key = self.artifact_store._parse_ref(artifact_ref)
+        if not key.startswith(f"rules/{correlation}/"):
             raise ValueError(
-                "artifact binding does not match the execution correlation"
+                "catalog artifact does not match the execution correlation"
             )
+        described = self.artifact_store.describe(artifact_ref)
+        row_fields = row["schema_fields"]
+        if isinstance(row_fields, str):
+            row_fields = json.loads(row_fields)
+        if (
+            described["digest"] != str(row["artifact_digest"])
+            or described["row_count"] != int(row["row_count"])
+            or described["schema_hash"] != str(row["schema_hash"])
+            or described["schema_fields"]
+            != _normalized_schema_fields(row_fields)
+        ):
+            raise ValueError("catalog artifact metadata does not match storage")
         return {
-            **self.artifact_store.describe(artifact_ref),
+            **described,
             "binding_hash": str(row["binding_hash"]),
         }
+
+    def attest_binding(
+        self,
+        *,
+        binding_id: str,
+        binding_hash: str,
+        access_mode: str,
+    ) -> dict[str, str]:
+        binding = _uid(binding_id, "artifact binding id")
+        if _DIGEST.fullmatch(str(binding_hash or "")) is None:
+            raise ValueError("artifact binding hash is invalid")
+        allowed = {
+            "read": {"read", "read_write"},
+            "write": {"write", "read_write"},
+        }.get(access_mode)
+        if allowed is None:
+            raise ValueError("artifact access mode is invalid")
+        with self.engine.connect() as connection:
+            row = connection.execute(
+                text(
+                    """
+                    SELECT binding_hash, access_mode, object_kind
+                    FROM public.dataflow_dataset_bindings
+                    WHERE id = CAST(:binding_id AS uuid)
+                    """
+                ),
+                {"binding_id": binding},
+            ).mappings().one_or_none()
+        if (
+            row is None
+            or row["object_kind"] != "parquet_artifact"
+            or row["access_mode"] not in allowed
+            or str(row["binding_hash"]) != binding_hash
+        ):
+            raise ValueError("canonical artifact binding no longer matches")
+        return {"binding_hash": str(row["binding_hash"])}
+
+    def register(
+        self,
+        *,
+        binding_id: str,
+        correlation_id: str,
+        artifact: dict[str, Any],
+        kind: str,
+        binding_hash: str,
+    ) -> None:
+        binding = _uid(binding_id, "artifact binding id")
+        correlation = _uid(correlation_id, "artifact correlation id")
+        if kind not in {"input", "lookup", "output"}:
+            raise ValueError("artifact kind is invalid")
+        self.attest_binding(
+            binding_id=binding,
+            binding_hash=binding_hash,
+            access_mode="write" if kind == "output" else "read",
+        )
+        if not isinstance(artifact, dict):
+            raise ValueError("artifact metadata is invalid")
+        artifact_ref = artifact.get("artifact_ref")
+        key = self.artifact_store._parse_ref(artifact_ref)
+        if not key.startswith(f"rules/{correlation}/"):
+            raise ValueError(
+                "artifact does not match the execution correlation"
+            )
+        described = self.artifact_store.describe(artifact_ref)
+        for key in (
+            "artifact_ref",
+            "digest",
+            "row_count",
+            "schema_hash",
+            "schema_fields",
+            "expires_at",
+        ):
+            if described[key] != artifact.get(key):
+                raise ValueError("artifact metadata does not match storage")
+        with self.engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_run_artifacts (
+                        id, correlation_id, binding_id, artifact_ref,
+                        artifact_digest, row_count, schema_hash, schema_fields,
+                        artifact_kind, expires_at
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:correlation_id AS uuid),
+                        CAST(:binding_id AS uuid), :artifact_ref,
+                        :artifact_digest, :row_count, :schema_hash,
+                        CAST(:schema_fields AS jsonb), :artifact_kind,
+                        CAST(:expires_at AS timestamptz)
+                    )
+                    ON CONFLICT (
+                        correlation_id, binding_id, artifact_digest
+                    ) DO NOTHING
+                    """
+                ),
+                {
+                    "id": new_governance_uid(),
+                    "correlation_id": correlation,
+                    "binding_id": binding,
+                    "artifact_ref": described["artifact_ref"],
+                    "artifact_digest": described["digest"],
+                    "row_count": described["row_count"],
+                    "schema_hash": described["schema_hash"],
+                    "schema_fields": json.dumps(
+                        described["schema_fields"],
+                        sort_keys=True,
+                        separators=(",", ":"),
+                    ),
+                    "artifact_kind": kind,
+                    "expires_at": described["expires_at"],
+                },
+            )

app/runner/rule_polars.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from contextlib import suppress
 from decimal import Decimal
 from typing import Any
 
@@ -22,7 +23,6 @@ _TYPE_MAP = {
     "integer": pl.Int64,
     "string": pl.String,
     "timestamp": pl.Datetime,
-    "timestamptz": pl.Datetime,
 }
 _IDEMPOTENCY = {
     "deduplication_key",
@@ -38,40 +38,6 @@ def _uid(value: Any, label: str) -> str:
         raise NodeExecutionError(f"{label} is invalid") from exc
 
 
-def _dtype_name(dtype: pl.DataType) -> str:
-    if dtype == pl.Boolean:
-        return "boolean"
-    if dtype == pl.Date:
-        return "date"
-    if dtype == pl.String:
-        return "string"
-    if dtype.is_integer():
-        return "integer"
-    if dtype == pl.Float32:
-        return "float"
-    if dtype == pl.Float64:
-        return "double"
-    if dtype.is_decimal():
-        return "decimal"
-    if isinstance(dtype, pl.Datetime):
-        return "timestamptz" if dtype.time_zone else "timestamp"
-    return str(dtype).lower()
-
-
-def _validate_schema(
-    frame: pl.LazyFrame,
-    expected_fields: list[dict[str, Any]],
-    label: str,
-) -> None:
-    schema = frame.collect_schema()
-    expected = {field["name"]: field["type"] for field in expected_fields}
-    actual = {name: _dtype_name(dtype) for name, dtype in schema.items()}
-    if actual != expected:
-        raise NodeExecutionError(
-            f"{label} artifact schema does not match the published plan"
-        )
-
-
 class _ExpressionCompiler:
     def compile(self, ast: dict[str, Any]) -> pl.Expr:
         kind = ast["kind"]
@@ -127,15 +93,53 @@ class _ExpressionCompiler:
         raise NodeExecutionError("published Polars expression is unsupported")
 
 
-def _cast_type(name: str) -> pl.DataType:
+def _cast_type(field: dict[str, Any]) -> pl.DataType:
+    name = field["type"]
     if name == "decimal":
-        return pl.Decimal(38, 9)
+        precision = field.get("precision")
+        scale = field.get("scale")
+        if precision is None or scale is None:
+            raise NodeExecutionError(
+                "published decimal cast contract is incomplete"
+            )
+        return pl.Decimal(precision=precision, scale=scale)
+    if name == "timestamptz":
+        timezone = field.get("timezone")
+        if not timezone:
+            raise NodeExecutionError(
+                "published timestamptz cast contract is incomplete"
+            )
+        return pl.Datetime(time_zone=timezone)
     dtype = _TYPE_MAP.get(name)
     if dtype is None:
         raise NodeExecutionError("published Polars cast type is unsupported")
     return dtype
 
 
+def _bounded_materialize(
+    frame: pl.LazyFrame | pl.DataFrame,
+    limits: dict[str, int],
+    label: str,
+) -> pl.DataFrame:
+    try:
+        collected = (
+            frame.lazy() if isinstance(frame, pl.DataFrame) else frame
+        ).head(limits["max_rows"] + 1).collect(engine="streaming")
+    except Exception as exc:
+        raise NodeExecutionError(
+            f"published Polars {label} materialization failed"
+        ) from exc
+    if collected.height > limits["max_rows"]:
+        raise NodeExecutionError(
+            f"published Polars {label} exceeds its row limit"
+        )
+    if collected.estimated_size() > limits["memory_limit_bytes"]:
+        raise NodeExecutionError(
+            f"published Polars {label} exceeds its memory limit"
+        )
+    return collected
+
+
 def _resolve_artifact(
     resolver,
     *,
@@ -219,6 +223,16 @@ class PolarsRulePlanAdapter:
                 "bound Polars rule plans do not accept runtime parameters"
             )
         correlation = _uid(correlation_id, "correlation_id")
+        try:
+            self.artifact_resolver.attest_binding(
+                binding_id=normalized["output_binding_id"],
+                binding_hash=normalized["output_binding_hash"],
+                access_mode="write",
+            )
+        except Exception as exc:
+            raise NodeExecutionError(
+                "published Polars output binding no longer matches"
+            ) from exc
         source = _resolve_artifact(
             self.artifact_resolver,
             binding_id=normalized["input_binding_id"],
@@ -227,30 +241,35 @@ class PolarsRulePlanAdapter:
         )
         try:
             frame = self.artifact_store.read(
-                source["artifact_ref"], source["digest"]
+                source["artifact_ref"],
+                source["digest"],
+                expected_schema_fields=normalized["input_fields"],
+                limits=normalized["resource_limits"],
             )
         except ValueError as exc:
             raise NodeExecutionError(
                 "published Polars input artifact is invalid"
             ) from exc
-        _validate_schema(frame, normalized["input_fields"], "input")
         limits = normalized["resource_limits"]
-        initial = frame.head(limits["max_rows"] + 1).collect(engine="streaming")
-        if initial.height > limits["max_rows"]:
-            raise NodeExecutionError(
-                "published Polars input exceeds its row limit"
-            )
-        if initial.estimated_size() > limits["memory_limit_bytes"]:
-            raise NodeExecutionError(
-                "published Polars input exceeds its memory limit"
-            )
+        initial = _bounded_materialize(frame, limits, "input")
         rows_in = initial.height
         frame = initial.lazy()
         expressions = _ExpressionCompiler()
         violations = []
+        rows_rejected = 0
+        rows_filtered = 0
+        rows_deduplicated = 0
+        rows_join_dropped = 0
+        rows_aggregated = 0
+        output_fields = {
+            field["name"]: field for field in normalized["output_fields"]
+        }
 
-        for operation in normalized["operations"]:
+        for index, operation in enumerate(normalized["operations"]):
             op = operation["op"]
+            before = _bounded_materialize(
+                frame, limits, f"{op} input {index}"
+            ).height
             if op == "normalize_text":
                 expression = pl.col(operation["column"])
                 if operation["trim"]:
@@ -283,6 +302,12 @@ class PolarsRulePlanAdapter:
                         False
                     )
                 )
+                after = _bounded_materialize(
+                    frame, limits, f"{op} output {index}"
+                )
+                rows_filtered += before - after.height
+                frame = after.lazy()
+                continue
             elif op == "assert":
                 predicate = expressions.compile(
                     operation["expression_ast"]
@@ -296,6 +321,7 @@ class PolarsRulePlanAdapter:
                 violations.append(
                     {"step_id": operation["step_id"], "count": invalid}
                 )
+                rows_rejected += invalid
                 frame = frame.filter(predicate)
             elif op == "derive":
                 frame = frame.with_columns(
@@ -312,10 +338,34 @@ class PolarsRulePlanAdapter:
                     ).alias(operation["column"])
                 )
             elif op == "cast":
+                target_field = output_fields.get(operation["column"])
+                if target_field is None:
+                    target_field = {
+                        "type": operation["to"],
+                    }
+                column_expression = pl.col(operation["column"])
+                source_dtype = frame.collect_schema()[operation["column"]]
+                if (
+                    source_dtype == pl.String
+                    and target_field["type"] == "timestamptz"
+                ):
+                    column_expression = column_expression.str.to_datetime(
+                        time_zone=target_field["timezone"],
+                        strict=True,
+                    )
+                elif (
+                    source_dtype == pl.String
+                    and target_field["type"] == "timestamp"
+                ):
+                    column_expression = column_expression.str.to_datetime(
+                        strict=True,
+                    )
+                else:
+                    column_expression = column_expression.cast(
+                        _cast_type(target_field), strict=True
+                    )
                 frame = frame.with_columns(
-                    pl.col(operation["column"])
-                    .cast(_cast_type(operation["to"]), strict=True)
-                    .alias(operation["column"])
+                    column_expression.alias(operation["column"])
                 )
             elif op == "deduplicate":
                 order = [
@@ -339,6 +389,12 @@ class PolarsRulePlanAdapter:
                         maintain_order=True,
                     )
                 )
+                after = _bounded_materialize(
+                    frame, limits, f"{op} output {index}"
+                )
+                rows_deduplicated += before - after.height
+                frame = after.lazy()
+                continue
             elif op == "mask":
                 if self.masking_policies.get(
                     operation["policy_id"]
@@ -381,6 +437,12 @@ class PolarsRulePlanAdapter:
                 frame = frame.group_by(
                     operation["group_by"], maintain_order=True
                 ).agg(aggregations)
+                after = _bounded_materialize(
+                    frame, limits, f"{op} output {index}"
+                )
+                rows_aggregated += before - after.height
+                frame = after.lazy()
+                continue
             elif op == "lookup_join":
                 lookup_artifact = _resolve_artifact(
                     self.artifact_resolver,
@@ -392,14 +454,16 @@ class PolarsRulePlanAdapter:
                     lookup = self.artifact_store.read(
                         lookup_artifact["artifact_ref"],
                         lookup_artifact["digest"],
+                        expected_schema_fields=operation["lookup_fields"],
+                        limits=limits,
                     )
                 except ValueError as exc:
                     raise NodeExecutionError(
                         "published Polars lookup artifact is invalid"
                     ) from exc
-                _validate_schema(
-                    lookup, operation["lookup_fields"], "lookup"
-                )
+                lookup = _bounded_materialize(
+                    lookup, limits, f"lookup input {index}"
+                ).lazy()
                 duplicate_count = (
                     lookup.group_by(operation["right_on"])
                     .len()
@@ -432,35 +496,62 @@ class PolarsRulePlanAdapter:
                     right_on=operation["right_on"],
                     how=operation["how"],
                 )
+                after = _bounded_materialize(
+                    frame, limits, f"{op} output {index}"
+                )
+                if after.height > before:
+                    raise NodeExecutionError(
+                        "published Polars lookup join expanded its input"
+                    )
+                rows_join_dropped += before - after.height
+                frame = after.lazy()
+                continue
+
+            frame = _bounded_materialize(
+                frame, limits, f"{op} output {index}"
+            ).lazy()
 
         output_names = [field["name"] for field in normalized["output_fields"]]
         frame = frame.select(output_names)
-        _validate_schema(frame, normalized["output_fields"], "output")
-        bounded = frame.head(limits["max_rows"] + 1).collect(engine="streaming")
-        if bounded.height > limits["max_rows"]:
-            raise NodeExecutionError(
-                "published Polars output exceeds its row limit"
-            )
-        if bounded.estimated_size() > limits["memory_limit_bytes"]:
-            raise NodeExecutionError(
-                "published Polars output exceeds its memory limit"
-            )
+        bounded = _bounded_materialize(frame, limits, "output")
         try:
             artifact = self.artifact_store.write(
                 bounded.lazy(),
                 correlation,
                 self.artifact_ttl_seconds,
+                schema_fields=normalized["output_fields"],
+                limits=limits,
             )
         except ValueError as exc:
             raise NodeExecutionError(
                 "published Polars output artifact write failed"
             ) from exc
+        try:
+            self.artifact_resolver.register(
+                binding_id=normalized["output_binding_id"],
+                correlation_id=correlation,
+                artifact=artifact,
+                kind="output",
+                binding_hash=normalized["output_binding_hash"],
+            )
+        except Exception as exc:
+            delete = getattr(self.artifact_store, "delete", None)
+            if callable(delete):
+                with suppress(Exception):
+                    delete(artifact["artifact_ref"])
+            raise NodeExecutionError(
+                "published Polars output artifact registration failed"
+            ) from exc
         violation_count = sum(item["count"] for item in violations)
         return {
             **artifact,
             "rows_in": rows_in,
             "rows_out": bounded.height,
-            "rows_rejected": max(0, rows_in - bounded.height),
+            "rows_rejected": rows_rejected,
+            "rows_filtered": rows_filtered,
+            "rows_deduplicated": rows_deduplicated,
+            "rows_join_dropped": rows_join_dropped,
+            "rows_aggregated": rows_aggregated,
             "violation_count": violation_count,
             "violations": violations,
             "commit_outcome": "committed",

docs/security/data-rule-runtime-sbom.json

@@ -0,0 +1,31 @@
+{
+  "bomFormat": "CycloneDX",
+  "specVersion": "1.5",
+  "version": 1,
+  "metadata": {
+    "component": {
+      "type": "application",
+      "name": "dataops-governed-rule-runtime"
+    }
+  },
+  "components": [
+    {
+      "type": "library",
+      "name": "polars",
+      "version": "1.42.1",
+      "purl": "pkg:pypi/polars@1.42.1",
+      "license": "MIT",
+      "officialSource": "https://github.com/pola-rs/polars",
+      "purpose": "Bounded lazy/dataframe execution for governed batch rules"
+    },
+    {
+      "type": "library",
+      "name": "minio",
+      "version": "7.2.10",
+      "purl": "pkg:pypi/minio@7.2.10",
+      "license": "Apache-2.0",
+      "officialSource": "https://github.com/minio/minio-py",
+      "purpose": "Digest-bound Parquet artifact transport and metadata"
+    }
+  ]
+}

migrations/versions/20260723_140_rule_run_artifacts.py

@@ -0,0 +1,40 @@
+"""Add a stable, correlation-scoped catalog for runner artifacts."""
+
+from alembic import op
+
+revision = "20260723_140"
+down_revision = "20260723_130"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE public.rule_run_artifacts (
+            id UUID PRIMARY KEY,
+            correlation_id UUID NOT NULL,
+            binding_id UUID NOT NULL
+                REFERENCES public.dataflow_dataset_bindings(id)
+                ON DELETE RESTRICT,
+            artifact_ref VARCHAR(1000) NOT NULL,
+            artifact_digest CHAR(64) NOT NULL,
+            row_count BIGINT NOT NULL CHECK (row_count >= 0),
+            schema_hash CHAR(64) NOT NULL,
+            schema_fields JSONB NOT NULL,
+            artifact_kind VARCHAR(20) NOT NULL
+                CHECK (artifact_kind IN ('input','lookup','output')),
+            expires_at TIMESTAMPTZ NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (correlation_id, binding_id, artifact_digest)
+        );
+        CREATE INDEX idx_rule_run_artifacts_resolve
+            ON public.rule_run_artifacts
+            (correlation_id, binding_id, created_at DESC);
+        """
+    )
+
+
+def downgrade() -> None:
+    # Runtime handoff evidence is immutable and intentionally retained.
+    pass

tests/core/data_rules/test_polars_compiler.py

@@ -529,3 +529,194 @@ def test_polars_compiler_preserves_repository_attested_binding_hashes():
 
     assert compiled["plan"]["input_binding_hash"] == "a" * 64
     assert compiled["plan"]["output_binding_hash"] == "b" * 64
+
+
+def test_polars_plan_carries_timezone_and_complete_decimal_temporal_semantics():
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    input_schema = _schema(
+        "bd:finance:raw",
+        [
+            ("id", "integer", False),
+            ("amount", "decimal", False),
+            ("occurred_at", "timestamptz", False),
+        ],
+    )
+    output_schema = _schema(
+        "bd:finance:clean",
+        [
+            ("id", "integer", False),
+            ("amount", "decimal", False),
+            ("occurred_at", "timestamptz", False),
+        ],
+    )
+    for schema in (input_schema, output_schema):
+        for field in schema["fields"]:
+            if field["name"] == "amount":
+                field.update({"precision": 12, "scale": 2})
+            if field["name"] == "occurred_at":
+                field["timezone"] = "Asia/Shanghai"
+        schema["schema_hash"] = canonical_schema_hash(schema["fields"])
+    source = _binding(input_schema, access_mode="read")
+    target = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "dedup",
+                "op": "deduplicate",
+                "keys": ["id"],
+                "order_by": ["occurred_at"],
+            }
+        ],
+    )
+
+    plan = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=source,
+        output_binding=target,
+        backend=_backend(),
+    )["plan"]
+
+    assert plan["timezone"] == "Asia/Shanghai"
+    assert next(
+        field for field in plan["output_fields"] if field["name"] == "amount"
+    )["precision"] == 12
+    assert next(
+        field
+        for field in plan["output_fields"]
+        if field["name"] == "occurred_at"
+    )["timezone"] == "Asia/Shanghai"
+
+
+def test_shared_polars_semantic_validator_rejects_tampered_expression_type():
+    from app.core.data_rules.compilers.polars import (
+        validate_bound_polars_plan,
+    )
+
+    compiled, _context = _compile(
+        [
+            {
+                "id": "derive_id",
+                "op": "derive",
+                "target": "customer_id",
+                "expression": "customer_id + 1",
+            }
+        ]
+    )
+    compiled["plan"]["operations"][0]["expression_ast"] = {
+        "kind": "identifier",
+        "name": "name",
+    }
+
+    with pytest.raises(ValueError, match="type|target"):
+        validate_bound_polars_plan(compiled["plan"])
+
+
+def test_lookup_join_rejects_selected_target_collision():
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    input_schema = _schema(
+        "bd:customer:raw",
+        [
+            ("customer_id", "integer", False),
+            ("segment_code", "string", True),
+        ],
+    )
+    lookup_schema = _schema(
+        "bd:segment:lookup",
+        [("code", "string", False)],
+    )
+    output_schema = _schema(
+        "bd:customer:enriched",
+        [
+            ("customer_id", "integer", False),
+            ("segment_code", "string", True),
+        ],
+    )
+    source = _binding(input_schema, access_mode="read")
+    lookup = _binding(lookup_schema, access_mode="read")
+    target = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "collision",
+                "op": "lookup_join",
+                "lookup": {
+                    "binding_id": lookup["id"],
+                    "left_on": ["segment_code"],
+                    "right_on": ["code"],
+                    "select": {"segment_code": "code"},
+                    "how": "left",
+                },
+            }
+        ],
+    )
+
+    with pytest.raises(ValueError, match="collision"):
+        PolarsRuleCompiler().compile(
+            rule_version=rule,
+            input_schema=input_schema,
+            output_schema=output_schema,
+            input_binding=source,
+            output_binding=target,
+            backend=_backend(
+                lookup_bindings={
+                    lookup["id"]: {
+                        "binding": lookup,
+                        "schema": lookup_schema,
+                    }
+                }
+            ),
+        )
+
+
+@pytest.mark.parametrize(
+    ("field_type", "message"),
+    [
+        ("decimal", "precision"),
+        ("timestamptz", "timezone"),
+    ],
+)
+def test_polars_plan_rejects_incomplete_runtime_type_contracts(
+    field_type, message
+):
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    input_schema = _schema(
+        "bd:semantic:raw",
+        [("id", "integer", False), ("value", field_type, False)],
+    )
+    output_schema = _schema(
+        "bd:semantic:clean",
+        [("id", "integer", False), ("value", field_type, False)],
+    )
+    source = _binding(input_schema, access_mode="read")
+    target = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "dedup",
+                "op": "deduplicate",
+                "keys": ["id"],
+                "order_by": ["id"],
+            }
+        ],
+    )
+
+    with pytest.raises(ValueError, match=message):
+        PolarsRuleCompiler().compile(
+            rule_version=rule,
+            input_schema=input_schema,
+            output_schema=output_schema,
+            input_binding=source,
+            output_binding=target,
+            backend=_backend(),
+        )

tests/integration/test_data_rule_polars_execution.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import re
 from pathlib import Path
 
@@ -53,20 +54,11 @@ def _binding(schema, *, source_uid, access_mode, object_ref):
     }
 
 
-class Resolver:
-    def __init__(self, artifacts):
-        self.artifacts = artifacts
-
-    def resolve(self, *, binding_id, correlation_id):
-        artifact = self.artifacts[binding_id]
-        assert f"/rules/{correlation_id}/" in artifact["artifact_ref"]
-        return artifact
-
-
 def test_real_postgres_mysql_minio_polars_cross_source_execution():
     from app.core.data_rules.compilers.polars import PolarsRuleCompiler
-    from app.runner.artifacts import ArtifactStore
+    from app.runner.artifacts import ArtifactStore, PostgresArtifactResolver
     from app.runner.rule_polars import PolarsRulePlanAdapter
+    from app.runner.rules import PostgresRulePlanRepository, RulePlanExecutor
 
     source_user = _compose_value(
         r"source-postgres:.*?POSTGRES_USER:\s*([^\s]+)"
@@ -74,11 +66,18 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
     source_password = _compose_value(
         r"source-postgres:.*?POSTGRES_PASSWORD:\s*([^\s]+)"
     )
+    platform_user = _compose_value(
+        r"\n  postgres:.*?POSTGRES_USER:\s*([^\s]+)"
+    )
+    platform_password = _compose_value(
+        r"\n  postgres:.*?POSTGRES_PASSWORD:\s*([^\s]+)"
+    )
     postgres_port = _compose_value(r'"(25432):5432"')
     mysql_port = _compose_value(r'"(23306):3306"')
     minio_user = _compose_value(r"MINIO_ROOT_USER:\s*([^\s]+)")
     minio_password = _compose_value(r"MINIO_ROOT_PASSWORD:\s*([^\s]+)")
     minio_port = _compose_value(r'"(19000):9000"')
+    platform_port = _compose_value(r'"(15432):5432"')
     bucket = _compose_value(r"mc mb --ignore-existing local/([^\s]+)")
     postgres = create_engine(
         f"postgresql+psycopg2://{source_user}:{source_password}"
@@ -90,6 +89,11 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
         f"@127.0.0.1:{mysql_port}/acceptance",
         pool_pre_ping=True,
     )
+    platform = create_engine(
+        f"postgresql+psycopg2://{platform_user}:{platform_password}"
+        f"@127.0.0.1:{platform_port}/dataops",
+        pool_pre_ping=True,
+    )
     minio = Minio(
         f"127.0.0.1:{minio_port}",
         access_key=minio_user,
@@ -108,6 +112,13 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
     prefix = f"rules/{correlation_id}/"
     customer_table = "task5_polars_customers"
     segment_table = "task5_polars_segments"
+    rule_uid = new_governance_uid()
+    rule_id = new_governance_uid()
+    dataflow_uid = new_governance_uid()
+    dataflow_version_id = new_governance_uid()
+    deployment_id = new_governance_uid()
+    component_binding_id = new_governance_uid()
+    plan_id = new_governance_uid()
 
     try:
         with postgres.begin() as connection:
@@ -165,13 +176,6 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
                     )
                 ).mappings()
             ]
-        customer_artifact = store.write(
-            pl.DataFrame(customer_rows).lazy(), correlation_id, 900
-        )
-        segment_artifact = store.write(
-            pl.DataFrame(segment_rows).lazy(), correlation_id, 900
-        )
-
         input_schema = _schema(
             "bd:task5:customer:raw",
             [
@@ -265,7 +269,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
             }
         )
         rule = {
-            "id": new_governance_uid(),
+            "id": rule_id,
             "status": "published",
             "rule_spec": spec,
             "spec_hash": rule_spec_hash(spec),
@@ -290,21 +294,221 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
             },
         )
         lookup_operation = compiled["plan"]["operations"][1]
-        resolver = Resolver(
-            {
-                input_binding["id"]: {
-                    **customer_artifact,
-                    "binding_hash": compiled["plan"][
-                        "input_binding_hash"
-                    ],
+        schema_hashes = {
+            "rule_spec_hash": compiled["plan"]["rule_spec_hash"],
+            "input_schema_snapshot_id": input_schema["id"],
+            "input_schema_hash": input_schema["schema_hash"],
+            "output_schema_snapshot_id": output_schema["id"],
+            "output_schema_hash": output_schema["schema_hash"],
+        }
+        with platform.begin() as connection:
+            for schema in (input_schema, lookup_schema, output_schema):
+                connection.execute(
+                    text(
+                        """
+                        INSERT INTO public.data_schema_snapshots
+                        (id, schema_ref, schema_hash, fields, source_revision)
+                        VALUES (CAST(:id AS uuid), :schema_ref, :schema_hash,
+                                CAST(:fields AS jsonb), :source_revision)
+                        """
+                    ),
+                    {**schema, "fields": json.dumps(schema["fields"])},
+                )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.data_rules
+                    (id, rule_uid, name, category, status)
+                    VALUES (CAST(:id AS uuid), CAST(:rule_uid AS uuid),
+                            :name, 'general', 'active')
+                    """
+                ),
+                {
+                    "id": new_governance_uid(),
+                    "rule_uid": rule_uid,
+                    "name": spec["name"],
+                },
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.data_rule_versions
+                    (id, rule_uid, version_no, source_text, source_language,
+                     rule_spec, spec_hash, generated_kind, status, published_at)
+                    VALUES (CAST(:id AS uuid), CAST(:rule_uid AS uuid), 1,
+                            :source_text, 'en', CAST(:rule_spec AS jsonb),
+                            :spec_hash, 'polars', 'published',
+                            CURRENT_TIMESTAMP)
+                    """
+                ),
+                {
+                    "id": rule_id,
+                    "rule_uid": rule_uid,
+                    "source_text": "Task 5 real cross-source integration",
+                    "rule_spec": json.dumps(spec),
+                    "spec_hash": rule["spec_hash"],
                 },
-                lookup_binding["id"]: {
-                    **segment_artifact,
-                    "binding_hash": lookup_operation[
-                        "lookup_binding_hash"
-                    ],
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.dataflow_versions
+                    (id, dataflow_uid, version_no, name, dataflow_spec,
+                     input_schema_hashes, output_schema_hash, status,
+                     released_at)
+                    VALUES (CAST(:id AS uuid), CAST(:dataflow_uid AS uuid), 1,
+                            :name, '{}'::jsonb,
+                            CAST(:input_schema_hashes AS jsonb),
+                            :output_schema_hash, 'released', CURRENT_TIMESTAMP)
+                    """
+                ),
+                {
+                    "id": dataflow_version_id,
+                    "dataflow_uid": dataflow_uid,
+                    "name": "Task 5 real cross-source integration",
+                    "input_schema_hashes": json.dumps(
+                        [
+                            input_schema["schema_hash"],
+                            lookup_schema["schema_hash"],
+                        ]
+                    ),
+                    "output_schema_hash": output_schema["schema_hash"],
                 },
-            }
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.dataflow_deployments
+                    (id, dataflow_version_id, environment, deployment_config,
+                     status, activated_at)
+                    VALUES (CAST(:id AS uuid),
+                            CAST(:dataflow_version_id AS uuid), 'test',
+                            '{}'::jsonb, 'active', CURRENT_TIMESTAMP)
+                    """
+                ),
+                {
+                    "id": deployment_id,
+                    "dataflow_version_id": dataflow_version_id,
+                },
+            )
+            for logical_ref, binding, binding_hash in (
+                (
+                    "customers",
+                    input_binding,
+                    compiled["plan"]["input_binding_hash"],
+                ),
+                (
+                    "segments",
+                    lookup_binding,
+                    lookup_operation["lookup_binding_hash"],
+                ),
+                (
+                    "enriched",
+                    output_binding,
+                    compiled["plan"]["output_binding_hash"],
+                ),
+            ):
+                connection.execute(
+                    text(
+                        """
+                        INSERT INTO public.dataflow_dataset_bindings
+                        (id, dataflow_deployment_id, logical_ref,
+                         data_source_uid, object_kind, object_ref,
+                         schema_snapshot_id, dialect, access_mode, write_mode,
+                         binding_hash)
+                        VALUES (CAST(:id AS uuid), CAST(:deployment_id AS uuid),
+                                :logical_ref, CAST(:source_uid AS uuid),
+                                'parquet_artifact', :object_ref,
+                                CAST(:schema_snapshot_id AS uuid), 'parquet',
+                                :access_mode, 'append', :binding_hash)
+                        """
+                    ),
+                    {
+                        "id": binding["id"],
+                        "deployment_id": deployment_id,
+                        "logical_ref": logical_ref,
+                        "source_uid": binding["data_source_uid"],
+                        "object_ref": binding["object_ref"],
+                        "schema_snapshot_id": binding["schema_snapshot_id"],
+                        "access_mode": binding["access_mode"],
+                        "binding_hash": binding_hash,
+                    },
+                )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.dataflow_component_bindings
+                    (id, dataflow_version_id, component_id, component_kind,
+                     rule_version_id, stage, order_no, idempotency, provenance)
+                    VALUES (CAST(:id AS uuid),
+                            CAST(:dataflow_version_id AS uuid),
+                            'task5_real_polars', 'rule.apply',
+                            CAST(:rule_version_id AS uuid), 'transform', 0,
+                            CAST(:idempotency AS jsonb), '{}'::jsonb)
+                    """
+                ),
+                {
+                    "id": component_binding_id,
+                    "dataflow_version_id": dataflow_version_id,
+                    "rule_version_id": rule_id,
+                    "idempotency": json.dumps(
+                        {
+                            "strategy": "deduplication_key",
+                            "key": "customer_id",
+                        }
+                    ),
+                },
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_execution_plans
+                    (id, component_binding_id, backend, compiler_version, plan,
+                     plan_hash, schema_hashes, status)
+                    VALUES (CAST(:id AS uuid),
+                            CAST(:component_binding_id AS uuid),
+                            'polars_batch', :compiler_version,
+                            CAST(:plan AS jsonb), :plan_hash,
+                            CAST(:schema_hashes AS jsonb), 'published')
+                    """
+                ),
+                {
+                    "id": plan_id,
+                    "component_binding_id": component_binding_id,
+                    "compiler_version": compiled["compiler_version"],
+                    "plan": json.dumps(compiled["plan"]),
+                    "plan_hash": compiled["plan_hash"],
+                    "schema_hashes": json.dumps(schema_hashes),
+                },
+            )
+        customer_artifact = store.write(
+            pl.DataFrame(customer_rows),
+            correlation_id,
+            900,
+            schema_fields=input_schema["fields"],
+            limits=compiled["plan"]["resource_limits"],
+        )
+        segment_artifact = store.write(
+            pl.DataFrame(segment_rows),
+            correlation_id,
+            900,
+            schema_fields=lookup_schema["fields"],
+            limits=compiled["plan"]["resource_limits"],
+        )
+        resolver = PostgresArtifactResolver(platform, store)
+        resolver.register(
+            binding_id=input_binding["id"],
+            binding_hash=compiled["plan"]["input_binding_hash"],
+            correlation_id=correlation_id,
+            artifact=customer_artifact,
+            kind="input",
+        )
+        resolver.register(
+            binding_id=lookup_binding["id"],
+            binding_hash=lookup_operation["lookup_binding_hash"],
+            correlation_id=correlation_id,
+            artifact=segment_artifact,
+            kind="lookup",
         )
         node = {
             "id": "task5_real_polars",
@@ -315,32 +519,50 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
                 "key": "customer_id",
             },
             "config": {
-                "component_binding_id": new_governance_uid(),
+                "component_binding_id": component_binding_id,
                 "rule_version_id": rule["id"],
                 "execution_plan_hash": compiled["plan_hash"],
             },
         }
-        result = PolarsRulePlanAdapter(
-            artifact_store=store,
-            artifact_resolver=resolver,
-            artifact_ttl_seconds=900,
-        ).execute(
-            plan=compiled["plan"],
-            node=node,
-            parameters={},
+        executor = RulePlanExecutor(
+            PostgresRulePlanRepository(platform),
+            adapters={
+                "polars_batch": PolarsRulePlanAdapter(
+                    artifact_store=store,
+                    artifact_resolver=resolver,
+                    artifact_ttl_seconds=900,
+                )
+            },
+        )
+        result = executor.execute(
+            node,
+            {},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+        repeated = executor.execute(
+            node,
+            {},
             write_authorized=True,
             correlation_id=correlation_id,
         )
 
         assert result["rows_in"] == 4
         assert result["rows_out"] == 2
-        assert result["rows_rejected"] == 2
+        assert result["rows_rejected"] == 1
+        assert result["rows_deduplicated"] == 1
+        assert result["rows_filtered"] == 0
+        assert result["rows_join_dropped"] == 0
+        assert result["rows_aggregated"] == 0
         assert result["violation_count"] == 1
         assert result["violations"] == [
             {"step_id": "valid_mobile", "count": 1}
         ]
         output = store.read(
-            result["artifact_ref"], result["digest"]
+            result["artifact_ref"],
+            result["digest"],
+            expected_schema_fields=output_schema["fields"],
+            limits=compiled["plan"]["resource_limits"],
         ).collect()
         assert output.sort("customer_id").to_dicts() == [
             {
@@ -366,6 +588,21 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
                 bucket, prefix=prefix, recursive=True
             )
         )
+        assert repeated["rows_out"] == 2
+        with platform.connect() as connection:
+            catalog_count = connection.execute(
+                text(
+                    """
+                    SELECT COUNT(*)
+                    FROM public.rule_run_artifacts
+                    WHERE correlation_id = CAST(:correlation_id AS uuid)
+                    """
+                ),
+                {"correlation_id": correlation_id},
+            ).scalar_one()
+        # The repeated deterministic output has the same digest and is
+        # idempotently retained as one stable catalog handoff.
+        assert catalog_count == 3
     finally:
         for item in list(
             minio.list_objects(bucket, prefix=prefix, recursive=True)
@@ -378,5 +615,71 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
             connection.execute(text(f"DROP TABLE IF EXISTS {customer_table}"))
         with mysql.begin() as connection:
             connection.execute(text(f"DROP TABLE IF EXISTS {segment_table}"))
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    "DELETE FROM public.rule_run_artifacts "
+                    "WHERE correlation_id = CAST(:correlation_id AS uuid)"
+                ),
+                {"correlation_id": correlation_id},
+            )
+            connection.execute(
+                text(
+                    "DELETE FROM public.rule_execution_plans "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": plan_id},
+            )
+            connection.execute(
+                text(
+                    "DELETE FROM public.dataflow_dataset_bindings "
+                    "WHERE dataflow_deployment_id = CAST(:id AS uuid)"
+                ),
+                {"id": deployment_id},
+            )
+            connection.execute(
+                text(
+                    "DELETE FROM public.dataflow_component_bindings "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": component_binding_id},
+            )
+            connection.execute(
+                text(
+                    "DELETE FROM public.dataflow_deployments "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": deployment_id},
+            )
+            connection.execute(
+                text(
+                    "DELETE FROM public.dataflow_versions "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": dataflow_version_id},
+            )
+            connection.execute(
+                text(
+                    "DELETE FROM public.data_rule_versions "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": rule_id},
+            )
+            connection.execute(
+                text(
+                    "DELETE FROM public.data_rules "
+                    "WHERE rule_uid = CAST(:rule_uid AS uuid)"
+                ),
+                {"rule_uid": rule_uid},
+            )
+            for schema in (input_schema, lookup_schema, output_schema):
+                connection.execute(
+                    text(
+                        "DELETE FROM public.data_schema_snapshots "
+                        "WHERE id = CAST(:id AS uuid)"
+                    ),
+                    {"id": schema["id"]},
+                )
         postgres.dispose()
         mysql.dispose()
+        platform.dispose()

tests/runner/test_artifacts.py

@@ -19,6 +19,8 @@ class FakeMinio:
     def __init__(self):
         self.buckets = {"dataops-rules"}
         self.objects = {}
+        self.get_calls = []
+        self.removed = []
 
     def bucket_exists(self, bucket):
         return bucket in self.buckets
@@ -55,8 +57,21 @@ class FakeMinio:
         )
 
     def get_object(self, bucket, key):
+        self.get_calls.append((bucket, key))
         return Response(self.objects[(bucket, key)]["payload"])
 
+    def remove_object(self, bucket, key):
+        self.removed.append((bucket, key))
+        self.objects.pop((bucket, key), None)
+
+    def list_objects(self, bucket, *, prefix, recursive):
+        assert recursive is True
+        return [
+            SimpleNamespace(object_name=key)
+            for object_bucket, key in sorted(self.objects)
+            if object_bucket == bucket and key.startswith(prefix)
+        ]
+
 
 def _store(client, *, clock=None, max_rows=100):
     from app.runner.artifacts import ArtifactStore
@@ -191,3 +206,263 @@ def test_artifact_store_does_not_return_ref_for_corrupted_server_content():
             new_governance_uid(),
             60,
         )
+    assert not store.client.objects
+
+
+def test_artifact_store_applies_plan_limits_before_download_and_serialization():
+    client = FakeMinio()
+    store = _store(client, max_rows=100)
+    correlation_id = new_governance_uid()
+    artifact = store.write(
+        pl.DataFrame({"id": [1, 2]}).lazy(),
+        correlation_id,
+        60,
+    )
+    calls_before = len(client.get_calls)
+
+    with pytest.raises(ValueError, match="row count"):
+        store.read(
+            artifact["artifact_ref"],
+            artifact["digest"],
+            limits={
+                "max_rows": 1,
+                "max_artifact_bytes": 1024 * 1024,
+                "memory_limit_bytes": 4 * 1024 * 1024,
+            },
+        )
+    assert len(client.get_calls) == calls_before
+
+    with pytest.raises(ValueError, match="row count"):
+        store.write(
+            pl.DataFrame({"id": [1, 2]}).lazy(),
+            correlation_id,
+            60,
+            limits={
+                "max_rows": 1,
+                "max_artifact_bytes": 1024 * 1024,
+                "memory_limit_bytes": 4 * 1024 * 1024,
+            },
+        )
+
+
+def test_artifact_schema_contract_covers_nullability_decimal_and_timezone():
+    client = FakeMinio()
+    store = _store(client)
+    correlation_id = new_governance_uid()
+    fields = [
+        {
+            "name": "amount",
+            "type": "decimal",
+            "nullable": False,
+            "precision": 12,
+            "scale": 2,
+        },
+        {
+            "name": "occurred_at",
+            "type": "timestamptz",
+            "nullable": False,
+            "timezone": "Asia/Shanghai",
+        },
+    ]
+    frame = pl.DataFrame(
+        {
+            "amount": ["12.34"],
+            "occurred_at": ["2026-07-23T10:00:00+08:00"],
+        }
+    ).with_columns(
+        pl.col("amount").cast(pl.Decimal(12, 2)),
+        pl.col("occurred_at")
+        .str.to_datetime(time_zone="Asia/Shanghai")
+        .alias("occurred_at"),
+    )
+    artifact = store.write(
+        frame.lazy(),
+        correlation_id,
+        60,
+        schema_fields=fields,
+    )
+
+    assert artifact["schema_fields"] == fields
+    store.read(
+        artifact["artifact_ref"],
+        artifact["digest"],
+        expected_schema_fields=fields,
+    ).collect()
+
+    copy_fields = [dict(field) for field in fields]
+    copy_fields[1] = {**copy_fields[1], "timezone": "UTC"}
+    with pytest.raises(ValueError, match="timezone|schema"):
+        store.read(
+            artifact["artifact_ref"],
+            artifact["digest"],
+            expected_schema_fields=copy_fields,
+        )
+
+    with pytest.raises(ValueError, match="nullable"):
+        store.write(
+            pl.DataFrame(
+                {
+                    "amount": [None],
+                    "occurred_at": [None],
+                },
+                schema={
+                    "amount": pl.Decimal(12, 2),
+                    "occurred_at": pl.Datetime(
+                        "us", "Asia/Shanghai"
+                    ),
+                },
+            ).lazy(),
+            correlation_id,
+            60,
+            schema_fields=fields,
+        )
+
+
+def test_artifact_cleanup_is_expired_and_correlation_scoped_only():
+    client = FakeMinio()
+    now = datetime(2026, 7, 23, 10, 0, tzinfo=UTC)
+    store = _store(client, clock=lambda: now)
+    first = new_governance_uid()
+    second = new_governance_uid()
+    expired = store.write(pl.DataFrame({"id": [1]}), first, 10)
+    active = store.write(pl.DataFrame({"id": [2]}), second, 300)
+    now = datetime(2026, 7, 23, 10, 0, 20, tzinfo=UTC)
+
+    assert store.cleanup_expired(first) == 1
+    assert not any(
+        key == expired["artifact_ref"].split("/", 3)[-1]
+        for _bucket, key in client.objects
+    )
+    assert store.describe(active["artifact_ref"])["row_count"] == 1
+
+
+class _Rows:
+    def __init__(self, row=None):
+        self.row = row
+
+    def mappings(self):
+        return self
+
+    def one_or_none(self):
+        return self.row
+
+
+class _Connection:
+    def __init__(self, engine):
+        self.engine = engine
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return None
+
+    def execute(self, statement, parameters):
+        sql = str(statement)
+        self.engine.calls.append((sql, dict(parameters)))
+        return _Rows(self.engine.handler(sql, parameters))
+
+
+class _Engine:
+    def __init__(self, handler):
+        self.handler = handler
+        self.calls = []
+
+    def connect(self):
+        return _Connection(self)
+
+    def begin(self):
+        return _Connection(self)
+
+
+def test_postgres_artifact_resolver_uses_catalog_and_rechecks_binding_hash():
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    store = _store(FakeMinio())
+    correlation_id = new_governance_uid()
+    binding_id = new_governance_uid()
+    binding_hash = "a" * 64
+    artifact = store.write(
+        pl.DataFrame({"id": [1]}),
+        correlation_id,
+        60,
+    )
+
+    def handler(sql, _parameters):
+        if "FROM public.rule_run_artifacts" in sql:
+            return {
+                **artifact,
+                "artifact_digest": artifact["digest"],
+                "binding_hash": binding_hash,
+            }
+        raise AssertionError(sql)
+
+    engine = _Engine(handler)
+    resolved = PostgresArtifactResolver(engine, store).resolve(
+        binding_id=binding_id,
+        correlation_id=correlation_id,
+    )
+
+    assert resolved["artifact_ref"] == artifact["artifact_ref"]
+    assert resolved["binding_hash"] == binding_hash
+    assert "dataflow_dataset_bindings b" in engine.calls[0][0]
+    assert "a.correlation_id = CAST(:correlation_id AS uuid)" in (
+        engine.calls[0][0]
+    )
+
+    with pytest.raises(ValueError, match="correlation"):
+        PostgresArtifactResolver(engine, store).resolve(
+            binding_id=binding_id,
+            correlation_id=new_governance_uid(),
+        )
+
+
+def test_postgres_artifact_resolver_attests_and_registers_stable_handoff():
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    store = _store(FakeMinio())
+    correlation_id = new_governance_uid()
+    binding_id = new_governance_uid()
+    binding_hash = "b" * 64
+    artifact = store.write(
+        pl.DataFrame({"id": [1]}),
+        correlation_id,
+        60,
+    )
+
+    def handler(sql, parameters):
+        if "FROM public.dataflow_dataset_bindings" in sql:
+            return {
+                "binding_hash": binding_hash,
+                "access_mode": "read_write",
+                "object_kind": "parquet_artifact",
+            }
+        if "INSERT INTO public.rule_run_artifacts" in sql:
+            assert parameters["artifact_digest"] == artifact["digest"]
+            assert parameters["schema_fields"]
+            return None
+        raise AssertionError(sql)
+
+    engine = _Engine(handler)
+    resolver = PostgresArtifactResolver(engine, store)
+    resolver.attest_binding(
+        binding_id=binding_id,
+        binding_hash=binding_hash,
+        access_mode="write",
+    )
+    resolver.register(
+        binding_id=binding_id,
+        correlation_id=correlation_id,
+        artifact=artifact,
+        kind="output",
+        binding_hash=binding_hash,
+    )
+
+    assert sum(
+        "FROM public.dataflow_dataset_bindings" in sql
+        for sql, _parameters in engine.calls
+    ) == 2
+    assert any(
+        "INSERT INTO public.rule_run_artifacts" in sql
+        for sql, _parameters in engine.calls
+    )

tests/runner/test_rule_polars.py

@@ -13,18 +13,61 @@ from tests.core.data_rules.test_polars_compiler import (
     _published_rule,
     _schema,
 )
-from tests.runner.test_artifacts import FakeMinio, _store
+from tests.runner.test_artifacts import FakeMinio
+
+
+def _plan_store(client):
+    from app.runner.artifacts import ArtifactStore
+
+    return ArtifactStore(
+        client,
+        bucket="dataops-rules",
+        max_artifact_bytes=8 * 1024 * 1024,
+        max_rows=10_000,
+        memory_limit_bytes=32 * 1024 * 1024,
+        max_ttl_seconds=3600,
+    )
 
 
 class Resolver:
     def __init__(self, values):
         self.values = values
         self.calls = []
+        self.events = []
+        self.attest_error = None
+        self.registrations = []
 
     def resolve(self, *, binding_id, correlation_id):
         self.calls.append((binding_id, correlation_id))
+        self.events.append(("resolve", binding_id))
         return self.values[binding_id]
 
+    def attest_binding(self, *, binding_id, binding_hash, access_mode):
+        self.events.append(("attest", binding_id, binding_hash, access_mode))
+        if self.attest_error is not None:
+            raise self.attest_error
+        return {"binding_hash": binding_hash}
+
+    def register(
+        self,
+        *,
+        binding_id,
+        correlation_id,
+        artifact,
+        kind,
+        binding_hash,
+    ):
+        self.events.append(("register", binding_id, kind))
+        self.registrations.append(
+            {
+                "binding_id": binding_id,
+                "correlation_id": correlation_id,
+                "artifact": artifact,
+                "kind": kind,
+                "binding_hash": binding_hash,
+            }
+        )
+
 
 def _compiled_plan(steps):
     from app.core.data_rules.compilers.polars import PolarsRuleCompiler
@@ -94,7 +137,7 @@ def test_polars_adapter_reconstructs_assert_and_deduplicate_and_writes_artifact(
             },
         ]
     )
-    store = _store(FakeMinio())
+    store = _plan_store(FakeMinio())
     correlation_id = new_governance_uid()
     source = store.write(
         pl.DataFrame(
@@ -106,6 +149,8 @@ def test_polars_adapter_reconstructs_assert_and_deduplicate_and_writes_artifact(
         ).lazy(),
         correlation_id,
         600,
+        schema_fields=compiled["plan"]["input_fields"],
+        limits=compiled["plan"]["resource_limits"],
     )
     resolver = Resolver(
         {
@@ -134,14 +179,28 @@ def test_polars_adapter_reconstructs_assert_and_deduplicate_and_writes_artifact(
 
     assert result["rows_in"] == 3
     assert result["rows_out"] == 1
-    assert result["rows_rejected"] == 2
+    assert result["rows_rejected"] == 1
+    assert result["rows_filtered"] == 0
+    assert result["rows_deduplicated"] == 1
+    assert result["rows_join_dropped"] == 0
+    assert result["rows_aggregated"] == 0
     assert result["violation_count"] == 1
     assert result["violations"] == [
         {"step_id": "mobile_format", "count": 1}
     ]
     assert result["commit_outcome"] == "committed"
+    assert resolver.events[0] == (
+        "attest",
+        compiled["plan"]["output_binding_id"],
+        compiled["plan"]["output_binding_hash"],
+        "write",
+    )
+    assert resolver.registrations[0]["kind"] == "output"
     assert store.read(
-        result["artifact_ref"], result["digest"]
+        result["artifact_ref"],
+        result["digest"],
+        expected_schema_fields=compiled["plan"]["output_fields"],
+        limits=compiled["plan"]["resource_limits"],
     ).collect().to_dicts() == [
         {
             "customer_id": 1,
@@ -164,7 +223,7 @@ def test_polars_adapter_fails_closed_for_plan_hash_binding_and_authorization():
             }
         ]
     )
-    store = _store(FakeMinio())
+    store = _plan_store(FakeMinio())
     correlation_id = new_governance_uid()
     source = store.write(
         pl.DataFrame(
@@ -172,6 +231,8 @@ def test_polars_adapter_fails_closed_for_plan_hash_binding_and_authorization():
         ).lazy(),
         correlation_id,
         600,
+        schema_fields=compiled["plan"]["input_fields"],
+        limits=compiled["plan"]["resource_limits"],
     )
     resolver = Resolver(
         {
@@ -229,6 +290,164 @@ def test_polars_adapter_fails_closed_for_plan_hash_binding_and_authorization():
         )
 
 
+def test_polars_adapter_attests_current_output_binding_before_reading_input():
+    from app.runner.rule_polars import PolarsRulePlanAdapter
+
+    compiled, input_binding = _compiled_plan(
+        [
+            {
+                "id": "trim_name",
+                "op": "normalize_text",
+                "column": "name",
+                "trim": True,
+            }
+        ]
+    )
+    store = _plan_store(FakeMinio())
+    correlation_id = new_governance_uid()
+    source = store.write(
+        pl.DataFrame(
+            {"customer_id": [1], "name": [" A "], "mobile": ["1"]}
+        ),
+        correlation_id,
+        600,
+        schema_fields=compiled["plan"]["input_fields"],
+        limits=compiled["plan"]["resource_limits"],
+    )
+    resolver = Resolver(
+        {
+            input_binding["id"]: {
+                **source,
+                "binding_hash": compiled["plan"]["input_binding_hash"],
+            }
+        }
+    )
+    resolver.attest_error = ValueError("binding changed")
+    reads_before_execute = list(store.client.get_calls)
+
+    with pytest.raises(NodeExecutionError, match="output binding"):
+        PolarsRulePlanAdapter(
+            artifact_store=store,
+            artifact_resolver=resolver,
+        ).execute(
+            plan=compiled["plan"],
+            node=_node(compiled),
+            parameters={},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+
+    assert resolver.events == [
+        (
+            "attest",
+            compiled["plan"]["output_binding_id"],
+            compiled["plan"]["output_binding_hash"],
+            "write",
+        )
+    ]
+    assert store.client.get_calls == reads_before_execute
+
+
+def test_polars_adapter_uses_exact_decimal_and_timestamptz_output_contracts():
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+    from app.core.data_rules.execution_contracts import canonical_schema_hash
+    from app.runner.rule_polars import PolarsRulePlanAdapter
+
+    input_schema = _schema(
+        "bd:payment:raw",
+        [
+            ("amount", "string", False),
+            ("occurred_at", "string", False),
+        ],
+    )
+    output_schema = _schema(
+        "bd:payment:clean",
+        [
+            ("amount", "decimal", False),
+            ("occurred_at", "timestamptz", False),
+        ],
+    )
+    output_schema["fields"][0].update({"precision": 12, "scale": 2})
+    output_schema["fields"][1]["timezone"] = "Asia/Shanghai"
+    output_schema["schema_hash"] = canonical_schema_hash(
+        output_schema["fields"]
+    )
+    source_binding = _binding(input_schema, access_mode="read")
+    output_binding = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "cast_amount",
+                "op": "cast",
+                "column": "amount",
+                "to": "decimal",
+                "on_error": "fail",
+            },
+            {
+                "id": "cast_time",
+                "op": "cast",
+                "column": "occurred_at",
+                "to": "timestamptz",
+                "on_error": "fail",
+            },
+        ],
+    )
+    compiled = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=source_binding,
+        output_binding=output_binding,
+        backend=_backend(),
+    )
+    store = _plan_store(FakeMinio())
+    correlation_id = new_governance_uid()
+    source = store.write(
+        pl.DataFrame(
+            {
+                "amount": ["12.34"],
+                "occurred_at": ["2026-07-23T12:30:00+08:00"],
+            }
+        ),
+        correlation_id,
+        600,
+        schema_fields=compiled["plan"]["input_fields"],
+        limits=compiled["plan"]["resource_limits"],
+    )
+    resolver = Resolver(
+        {
+            source_binding["id"]: {
+                **source,
+                "binding_hash": compiled["plan"]["input_binding_hash"],
+            }
+        }
+    )
+
+    result = PolarsRulePlanAdapter(
+        artifact_store=store,
+        artifact_resolver=resolver,
+    ).execute(
+        plan=compiled["plan"],
+        node=_node(compiled),
+        parameters={},
+        write_authorized=True,
+        correlation_id=correlation_id,
+    )
+
+    output = store.read(
+        result["artifact_ref"],
+        result["digest"],
+        expected_schema_fields=compiled["plan"]["output_fields"],
+        limits=compiled["plan"]["resource_limits"],
+    ).collect()
+    assert output.schema["amount"] == pl.Decimal(precision=12, scale=2)
+    assert output.schema["occurred_at"] == pl.Datetime(
+        time_zone="Asia/Shanghai"
+    )
+
+
 def test_rule_executor_attests_polars_canonical_hashes_and_forwards_correlation():
     from app.runner.rules import RulePlanExecutor
 

tests/test_data_rule_runtime_sbom.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SBOM = ROOT / "docs" / "security" / "data-rule-runtime-sbom.json"
+
+
+def test_data_rule_runtime_sbom_pins_official_oss_dependencies():
+    document = json.loads(SBOM.read_text(encoding="utf-8"))
+    components = {
+        component["name"]: component
+        for component in document["components"]
+    }
+
+    assert document["bomFormat"] == "CycloneDX"
+    assert document["specVersion"] == "1.5"
+    assert components["polars"] == {
+        "type": "library",
+        "name": "polars",
+        "version": "1.42.1",
+        "purl": "pkg:pypi/polars@1.42.1",
+        "license": "MIT",
+        "officialSource": "https://github.com/pola-rs/polars",
+        "purpose": "Bounded lazy/dataframe execution for governed batch rules",
+    }
+    assert components["minio"] == {
+        "type": "library",
+        "name": "minio",
+        "version": "7.2.10",
+        "purl": "pkg:pypi/minio@7.2.10",
+        "license": "Apache-2.0",
+        "officialSource": "https://github.com/minio/minio-py",
+        "purpose": "Digest-bound Parquet artifact transport and metadata",
+    }
+    requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8")
+    assert "polars==1.42.1" in requirements
+    assert "minio==7.2.10" in requirements

tests/test_data_rule_schema.py

@@ -24,6 +24,12 @@ PLAN_STATUS_MIGRATION = (
     / "versions"
     / "20260723_130_bound_plan_lifecycle.py"
 )
+ARTIFACT_CATALOG_MIGRATION = (
+    ROOT
+    / "migrations"
+    / "versions"
+    / "20260723_140_rule_run_artifacts.py"
+)
 
 EXPECTED_TABLES = {
     "data_rules",
@@ -127,3 +133,24 @@ def test_bound_plan_status_migration_is_forward_only_and_keeps_compiled_valid():
     spec.loader.exec_module(module)
     with pytest.raises(RuntimeError, match="forward-only|cannot downgrade"):
         module.downgrade()
+
+
+def test_rule_run_artifact_catalog_is_correlation_scoped_and_forward_preserving():
+    source = ARTIFACT_CATALOG_MIGRATION.read_text(encoding="utf-8")
+
+    assert 'revision = "20260723_140"' in source
+    assert 'down_revision = "20260723_130"' in source
+    assert "CREATE TABLE public.rule_run_artifacts" in source
+    for expected in (
+        "correlation_id UUID NOT NULL",
+        "binding_id UUID NOT NULL",
+        "artifact_ref VARCHAR(1000) NOT NULL",
+        "artifact_digest CHAR(64) NOT NULL",
+        "schema_fields JSONB NOT NULL",
+        "expires_at TIMESTAMPTZ NOT NULL",
+        "UNIQUE (correlation_id, binding_id, artifact_digest)",
+    ):
+        assert expected in source
+    downgrade = source.split("def downgrade()", 1)[1]
+    assert "DROP TABLE" not in downgrade.upper()
+    assert "pass" in downgrade