fix: enforce isolated Polars artifact invariants

app/core/data_rules/compilers/polars.py

@@ -102,7 +102,7 @@ _OPERATION_KEYS = {
         "severity",
         "step_id",
     },
-    "cast": {"op", "column", "to", "on_error"},
+    "cast": {"op", "column", "to", "on_error", "target_field"},
     "deduplicate": {"op", "keys", "order_by", "keep"},
     "derive": {"op", "target", "expression_ast"},
     "fill_null": {"op", "column", "value"},
@@ -312,16 +312,6 @@ def _semantic_type(field: dict[str, Any]) -> tuple[Any, ...]:
     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:
@@ -431,6 +421,13 @@ def _operation(value: Any) -> dict[str, Any]:
         result["to"] not in _CAST_TYPES or result["on_error"] != "fail"
     ):
         raise ValueError("cast operation is invalid")
+    if operation == "cast":
+        canonical_schema_hash([result["target_field"]])
+        target_field = _field_contracts([result["target_field"]])
+        target = target_field.get(result["column"])
+        if target is None or target["type"] != result["to"]:
+            raise ValueError("cast target_field does not match its column/type")
+        result["target_field"] = target
     if operation == "deduplicate":
         _string_list(result["keys"], "deduplicate keys")
         _string_list(result["order_by"], "deduplicate order_by")
@@ -589,11 +586,9 @@ def validate_bound_polars_plan(value: Any) -> dict[str, Any]:
             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] = copy.deepcopy(
+                operation["target_field"]
             )
-            current_fields[column]["name"] = column
             current_types[column] = operation["to"]
         elif op == "deduplicate":
             if not set(operation["keys"] + operation["order_by"]) <= set(
@@ -721,6 +716,10 @@ class PolarsRuleCompiler(RuleCompiler):
         if "polars" not in supported:
             raise ValueError("rule expressions are unsupported by polars")
         output_types = _fields(target_schema)
+        output_field_contracts = {
+            field["name"]: copy.deepcopy(field)
+            for field in target_schema["fields"]
+        }
         current_fields = list(field_types)
         operations = []
 
@@ -851,12 +850,21 @@ class PolarsRuleCompiler(RuleCompiler):
                 target_type = str(step.get("to") or "").lower()
                 if target_type not in _CAST_TYPES:
                     raise ValueError("unsupported Polars cast target")
+                target_field = output_field_contracts.get(column)
+                if (
+                    target_field is None
+                    or target_field["type"] != target_type
+                ):
+                    raise ValueError(
+                        "cast target does not match the output schema"
+                    )
                 operations.append(
                     {
                         "op": op,
                         "column": column,
                         "to": target_type,
                         "on_error": "fail",
+                        "target_field": target_field,
                     }
                 )
                 field_types[column] = target_type

app/runner/artifacts.py

@@ -3,18 +3,17 @@
 from __future__ import annotations
 
 import hashlib
-import io
 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 contextlib import contextmanager, suppress
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
 import polars as pl
+import pyarrow.parquet as pq
 from sqlalchemy import text
 
 from app.core.common.identifiers import (
@@ -27,6 +26,22 @@ PARQUET_CONTENT_TYPE = "application/x-parquet"
 _DIGEST = re.compile(r"^[0-9a-f]{64}$")
 
 
+def _parquet_footer_bounds(path: str) -> tuple[int, int]:
+    try:
+        metadata = pq.ParquetFile(path).metadata
+    except Exception as exc:
+        raise ValueError("artifact Parquet footer is invalid") from exc
+    if metadata is None or metadata.num_row_groups < 1:
+        raise ValueError("artifact Parquet footer is incomplete")
+    uncompressed = sum(
+        metadata.row_group(index).total_byte_size
+        for index in range(metadata.num_row_groups)
+    )
+    if metadata.num_rows < 0 or uncompressed < 0:
+        raise ValueError("artifact Parquet footer is invalid")
+    return metadata.num_rows, uncompressed
+
+
 def _now_utc(clock) -> datetime:
     value = clock()
     if not isinstance(value, datetime):
@@ -111,28 +126,9 @@ def _normalized_schema_fields(value: Any) -> list[dict[str, Any]]:
     )
 
 
-def _schema_contract(value: Any) -> tuple[list[dict[str, Any]], str, str]:
+def _schema_contract(value: Any) -> tuple[list[dict[str, Any]], str]:
     fields = _normalized_schema_fields(value)
-    encoded = json.dumps(
-        fields,
-        sort_keys=True,
-        separators=(",", ":"),
-        ensure_ascii=False,
-    ).encode("utf-8")
-    return (
-        fields,
-        canonical_schema_hash(fields),
-        urlsafe_b64encode(encoded).decode("ascii"),
-    )
-
-
-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)
+    return fields, canonical_schema_hash(fields)
 
 
 def _validate_frame_schema(
@@ -200,7 +196,6 @@ def _metadata(value: Any) -> dict[str, str]:
             "schema-sha256",
             "expires-at",
             "artifact-bytes",
-            "schema-contract",
         }:
             normalized[name] = str(item)
     required = {
@@ -209,10 +204,11 @@ 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")
+    if sum(len(key) + len(item) for key, item in normalized.items()) > 2_048:
+        raise ValueError("artifact content metadata exceeds the safe limit")
     return normalized
 
 
@@ -357,7 +353,7 @@ class ArtifactStore:
             raise ValueError("artifact row count exceeds the configured limit")
         if collected.estimated_size() > effective["memory_limit_bytes"]:
             raise ValueError("artifact frame exceeds the configured memory limit")
-        fields, schema_digest, schema_encoded = _schema_contract(
+        fields, schema_digest = _schema_contract(
             schema_fields or _inferred_schema_fields(collected)
         )
         _validate_frame_schema(collected, fields)
@@ -406,7 +402,6 @@ class ArtifactStore:
                         "schema-sha256": schema_digest,
                         "expires-at": expires_at,
                         "artifact-bytes": str(size),
-                        "schema-contract": schema_encoded,
                     },
                 )
             uploaded = True
@@ -416,12 +411,13 @@ class ArtifactStore:
                 limits=effective,
             )
             artifact_ref = f"minio://{self.bucket}/{key}"
-            self.read(
+            with self.stage(
                 artifact_ref,
                 digest_hex,
                 expected_schema_fields=fields,
                 limits=effective,
-            )
+            ):
+                pass
         except Exception:
             if uploaded:
                 with suppress(Exception):
@@ -440,20 +436,100 @@ class ArtifactStore:
             "expires_at": expires_at,
         }
 
+    def write_path(
+        self,
+        path: str,
+        correlation_id: str,
+        ttl_seconds: int,
+        *,
+        schema_fields: list[dict[str, Any]],
+        limits: dict[str, int] | None = None,
+    ) -> dict[str, Any]:
+        """Upload a worker-produced Parquet file without collecting it."""
+
+        effective = self._limits(limits)
+        correlation = _uid(correlation_id, "correlation_id")
+        if (
+            isinstance(ttl_seconds, bool)
+            or not isinstance(ttl_seconds, int)
+            or ttl_seconds < 1
+            or ttl_seconds > self.max_ttl_seconds
+        ):
+            raise ValueError("artifact TTL is outside the configured limit")
+        if not isinstance(path, str) or not os.path.isfile(path):
+            raise ValueError("worker artifact path is invalid")
+        size = os.path.getsize(path)
+        if size < 1 or size > effective["max_artifact_bytes"]:
+            raise ValueError("artifact size exceeds the configured limit")
+        rows, uncompressed = _parquet_footer_bounds(path)
+        if rows > effective["max_rows"]:
+            raise ValueError("artifact row count exceeds the configured limit")
+        if uncompressed > effective["memory_limit_bytes"]:
+            raise ValueError(
+                "Parquet footer uncompressed size exceeds the memory limit"
+            )
+        fields, schema_digest = _schema_contract(schema_fields)
+        _validate_frame_schema(pl.scan_parquet(path), fields)
+        digest = hashlib.sha256()
+        with open(path, "rb") as handle:
+            while chunk := handle.read(1024 * 1024):
+                digest.update(chunk)
+        digest_hex = digest.hexdigest()
+        expires_at = _timestamp(
+            _now_utc(self.clock) + timedelta(seconds=ttl_seconds)
+        )
+        artifact_id = new_governance_uid()
+        key = f"rules/{correlation}/{artifact_id}.parquet"
+        uploaded = False
+        try:
+            with open(path, "rb") as handle:
+                self.client.put_object(
+                    self.bucket,
+                    key,
+                    handle,
+                    size,
+                    content_type=PARQUET_CONTENT_TYPE,
+                    metadata={
+                        "sha256": digest_hex,
+                        "row-count": str(rows),
+                        "schema-sha256": schema_digest,
+                        "expires-at": expires_at,
+                        "artifact-bytes": str(size),
+                    },
+                )
+            uploaded = True
+            artifact_ref = f"minio://{self.bucket}/{key}"
+            with self.stage(
+                artifact_ref,
+                digest_hex,
+                expected_schema_fields=fields,
+                limits=effective,
+            ):
+                pass
+        except Exception:
+            if uploaded:
+                with suppress(Exception):
+                    self.client.remove_object(self.bucket, key)
+            raise
+        return {
+            "artifact_ref": artifact_ref,
+            "digest": digest_hex,
+            "row_count": rows,
+            "schema_hash": schema_digest,
+            "schema_fields": fields,
+            "expires_at": expires_at,
+        }
+
     def describe(self, ref: str) -> dict[str, Any]:
         """Return validated object metadata without exposing MinIO credentials."""
 
         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"],
         }
 
@@ -465,6 +541,37 @@ class ArtifactStore:
         expected_schema_fields: list[dict[str, Any]] | None = None,
         limits: dict[str, int] | None = None,
     ) -> pl.LazyFrame:
+        with self.stage(
+            ref,
+            expected_digest,
+            expected_schema_fields=expected_schema_fields,
+            limits=limits,
+        ) as path:
+            effective = self._limits(limits)
+            try:
+                frame = pl.read_parquet(
+                    path,
+                    n_rows=effective["max_rows"] + 1,
+                    memory_map=False,
+                )
+            except Exception as exc:
+                raise ValueError("artifact is not valid Parquet") from exc
+        if frame.estimated_size() > effective["memory_limit_bytes"]:
+            raise ValueError("artifact frame exceeds the configured memory limit")
+        _validate_frame_schema(
+            frame, _normalized_schema_fields(expected_schema_fields)
+        )
+        return frame.lazy()
+
+    @contextmanager
+    def stage(
+        self,
+        ref: str,
+        expected_digest: str,
+        *,
+        expected_schema_fields: list[dict[str, Any]] | None = None,
+        limits: dict[str, int] | None = None,
+    ):
         effective = self._limits(limits)
         if _DIGEST.fullmatch(str(expected_digest or "")) is None:
             raise ValueError("expected artifact digest is invalid")
@@ -474,56 +581,56 @@ class ArtifactStore:
             expected_digest=expected_digest,
             limits=effective,
         )
-        fields = _decode_schema_contract(metadata["schema-contract"])
+        if expected_schema_fields is None:
+            raise ValueError("expected artifact schema fields are required")
+        fields = _normalized_schema_fields(expected_schema_fields)
         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")
+            raise ValueError("artifact schema contract is not expected")
         response = self.client.get_object(self.bucket, key)
         digest = hashlib.sha256()
-        payload = io.BytesIO()
+        path = None
         size = 0
         try:
-            while chunk := response.read(1024 * 1024):
-                size += len(chunk)
-                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)
+            try:
+                with tempfile.NamedTemporaryFile(
+                    prefix="dataops-rule-stage-",
+                    suffix=".parquet",
+                    delete=False,
+                ) as handle:
+                    path = handle.name
+                    while chunk := response.read(1024 * 1024):
+                        size += len(chunk)
+                        if size > effective["max_artifact_bytes"]:
+                            raise ValueError(
+                                "artifact size exceeds the configured limit"
+                            )
+                        digest.update(chunk)
+                        handle.write(chunk)
+            finally:
+                response.close()
+                release = getattr(response, "release_conn", None)
+                if callable(release):
+                    release()
+            if digest.hexdigest() != expected_digest:
+                raise ValueError("artifact digest does not match content")
+            footer_rows, uncompressed = _parquet_footer_bounds(path)
+            if footer_rows != int(metadata["row-count"]):
+                raise ValueError("artifact row count does not match metadata")
+            if footer_rows > effective["max_rows"]:
+                raise ValueError(
+                    "artifact row count exceeds the configured limit"
+                )
+            if uncompressed > effective["memory_limit_bytes"]:
+                raise ValueError(
+                    "Parquet footer uncompressed size exceeds the memory limit"
+                )
+            lazy = pl.scan_parquet(path)
+            _validate_frame_schema(lazy, fields)
+            yield path
         finally:
-            response.close()
-            release = getattr(response, "release_conn", None)
-            if callable(release):
-                release()
-        if digest.hexdigest() != expected_digest:
-            raise ValueError("artifact digest does not match content")
-        payload.seek(0)
-        try:
-            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 > effective["max_rows"]:
-            raise ValueError("artifact row count exceeds the configured limit")
-        if frame.estimated_size() > effective["memory_limit_bytes"]:
-            raise ValueError("artifact frame exceeds the configured memory limit")
-        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()
+            if path is not None:
+                with suppress(FileNotFoundError):
+                    os.unlink(path)
 
     def cleanup_expired(self, correlation_id: str) -> int:
         correlation = _uid(correlation_id, "correlation_id")
@@ -565,9 +672,17 @@ class PostgresArtifactResolver:
         self.engine = engine
         self.artifact_store = artifact_store
 
-    def resolve(self, *, binding_id: str, correlation_id: str) -> dict[str, Any]:
+    def resolve(
+        self,
+        *,
+        binding_id: str,
+        correlation_id: str,
+        kind: str,
+    ) -> dict[str, Any]:
         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")
         statement = text(
             """
             SELECT
@@ -583,6 +698,7 @@ class PostgresArtifactResolver:
               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.artifact_kind = :artifact_kind
               AND a.expires_at > CURRENT_TIMESTAMP
               AND b.object_kind = 'parquet_artifact'
               AND b.access_mode IN ('read', 'read_write')
@@ -596,6 +712,7 @@ class PostgresArtifactResolver:
                 {
                     "binding_id": binding,
                     "correlation_id": correlation,
+                    "artifact_kind": kind,
                 },
             ).mappings().one_or_none()
         if row is None:
@@ -614,12 +731,11 @@ class PostgresArtifactResolver:
             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 {
             **described,
+            "schema_fields": _normalized_schema_fields(row_fields),
             "binding_hash": str(row["binding_hash"]),
         }
 
@@ -667,7 +783,7 @@ class PostgresArtifactResolver:
         artifact: dict[str, Any],
         kind: str,
         binding_hash: str,
-    ) -> None:
+    ) -> dict[str, Any]:
         binding = _uid(binding_id, "artifact binding id")
         correlation = _uid(correlation_id, "artifact correlation id")
         if kind not in {"input", "lookup", "output"}:
@@ -691,13 +807,16 @@ class PostgresArtifactResolver:
             "digest",
             "row_count",
             "schema_hash",
-            "schema_fields",
             "expires_at",
         ):
             if described[key] != artifact.get(key):
                 raise ValueError("artifact metadata does not match storage")
+        fields = _normalized_schema_fields(artifact.get("schema_fields"))
+        if canonical_schema_hash(fields) != described["schema_hash"]:
+            raise ValueError("artifact schema contract does not match storage")
+        inserted = None
         with self.engine.begin() as connection:
-            connection.execute(
+            inserted = connection.execute(
                 text(
                     """
                     INSERT INTO public.rule_run_artifacts (
@@ -712,8 +831,10 @@ class PostgresArtifactResolver:
                         CAST(:expires_at AS timestamptz)
                     )
                     ON CONFLICT (
-                        correlation_id, binding_id, artifact_digest
+                        correlation_id, binding_id, artifact_kind
                     ) DO NOTHING
+                    RETURNING artifact_ref, artifact_digest, row_count,
+                              schema_hash, schema_fields, expires_at
                     """
                 ),
                 {
@@ -725,11 +846,107 @@ class PostgresArtifactResolver:
                     "row_count": described["row_count"],
                     "schema_hash": described["schema_hash"],
                     "schema_fields": json.dumps(
-                        described["schema_fields"],
+                        fields,
                         sort_keys=True,
                         separators=(",", ":"),
                     ),
                     "artifact_kind": kind,
                     "expires_at": described["expires_at"],
                 },
+            ).mappings().one_or_none()
+            if inserted is None:
+                inserted = connection.execute(
+                    text(
+                        """
+                        SELECT artifact_ref, artifact_digest, row_count,
+                               schema_hash, schema_fields, expires_at
+                        FROM public.rule_run_artifacts
+                        WHERE correlation_id = CAST(:correlation_id AS uuid)
+                          AND binding_id = CAST(:binding_id AS uuid)
+                          AND artifact_kind = :artifact_kind
+                        FOR UPDATE
+                        """
+                    ),
+                    {
+                        "correlation_id": correlation,
+                        "binding_id": binding,
+                        "artifact_kind": kind,
+                    },
+                ).mappings().one_or_none()
+        if inserted is None:
+            self.artifact_store.delete(artifact_ref)
+            raise ValueError("immutable artifact catalog handoff was not found")
+        existing = dict(inserted)
+        existing_fields = existing["schema_fields"]
+        if isinstance(existing_fields, str):
+            existing_fields = json.loads(existing_fields)
+        registered = {
+            "artifact_ref": str(existing["artifact_ref"]),
+            "digest": str(existing["artifact_digest"]),
+            "row_count": int(existing["row_count"]),
+            "schema_hash": str(existing["schema_hash"]),
+            "schema_fields": _normalized_schema_fields(existing_fields),
+            "expires_at": (
+                _timestamp(existing["expires_at"])
+                if isinstance(existing["expires_at"], datetime)
+                else str(existing["expires_at"])
+            ),
+        }
+        if registered["digest"] != described["digest"]:
+            self.artifact_store.delete(artifact_ref)
+            raise ValueError(
+                "immutable artifact catalog digest conflicts with retry"
+            )
+        if registered["artifact_ref"] != artifact_ref:
+            self.artifact_store.delete(artifact_ref)
+            stored = self.artifact_store.describe(
+                registered["artifact_ref"]
             )
+            for key in (
+                "digest",
+                "row_count",
+                "schema_hash",
+                "expires_at",
+            ):
+                if stored[key] != registered[key]:
+                    raise ValueError(
+                        "immutable artifact catalog does not match storage"
+                    )
+        return registered
+
+    def cleanup_expired(self, *, limit: int = 100) -> int:
+        if (
+            isinstance(limit, bool)
+            or not isinstance(limit, int)
+            or limit < 1
+            or limit > 1_000
+        ):
+            raise ValueError("artifact cleanup limit is invalid")
+        removed = 0
+        with self.engine.begin() as connection:
+            rows = connection.execute(
+                text(
+                    """
+                    SELECT id::text AS id, artifact_ref
+                    FROM public.rule_run_artifacts
+                    WHERE expires_at <= CURRENT_TIMESTAMP
+                    ORDER BY expires_at, id
+                    LIMIT :limit
+                    FOR UPDATE SKIP LOCKED
+                    """
+                ),
+                {"limit": limit},
+            ).mappings().all()
+            for row in rows:
+                self.artifact_store.delete(str(row["artifact_ref"]))
+                connection.execute(
+                    text(
+                        """
+                        DELETE FROM public.rule_run_artifacts
+                        WHERE id = CAST(:id AS uuid)
+                        """
+                    ),
+                    {"id": str(row["id"])},
+                )
+                removed += 1
+        return removed

app/runner/bootstrap.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import os
 from dataclasses import dataclass, field
 
+import click
 from minio import Minio
 
 from app.core.data_source.runtime import (
@@ -77,6 +78,16 @@ class RunnerSettings:
     max_query_rows: int = 1000
 
 
+def register_artifact_cleanup_cli(app, artifact_resolver):
+    """Register the bounded production maintenance entrypoint."""
+
+    @app.cli.command("cleanup-rule-artifacts")
+    @click.option("--limit", type=click.IntRange(1, 1_000), default=100)
+    def cleanup_rule_artifacts(limit):
+        removed = artifact_resolver.cleanup_expired(limit=limit)
+        click.echo(f"removed {removed} expired rule artifacts")
+
+
 def runner_settings_from_env():
     task_token_secret = _required("RUNNER_TASK_TOKEN_SECRET")
     worker_count = _integer("RUNNER_WORKERS", 2, 1, 8)
@@ -182,11 +193,12 @@ def build_runner_application(settings=None):
         memory_limit_bytes=settings.artifact_memory_limit_bytes,
         max_ttl_seconds=settings.artifact_ttl_seconds,
     )
+    artifact_resolver = PostgresArtifactResolver(
+        runtime.platform_engine, artifact_store
+    )
     polars_rule_adapter = PolarsRulePlanAdapter(
         artifact_store=artifact_store,
-        artifact_resolver=PostgresArtifactResolver(
-            runtime.platform_engine, artifact_store
-        ),
+        artifact_resolver=artifact_resolver,
         masking_policies={
             "customer_mobile_last4": "preserve_last_4",
             "redact": "redact",
@@ -218,6 +230,7 @@ def build_runner_application(settings=None):
         ledger=PostgresTaskLedger(runtime.platform_engine),
         registry=registry,
     )
+    register_artifact_cleanup_cli(application, artifact_resolver)
     application.extensions["dataops_runner_runtime"] = runtime
     application.extensions["dataops_runner_settings"] = settings
     return application

app/runner/polars_worker.py

@@ -0,0 +1,557 @@
+"""OS-isolated execution boundary for allocation-heavy Polars work."""
+
+from __future__ import annotations
+
+import multiprocessing
+import os
+import time
+from contextlib import suppress
+from decimal import Decimal
+from typing import Any
+
+try:
+    import resource
+except ImportError:  # pragma: no cover - fail-closed portability guard
+    resource = None
+
+import polars as pl
+import psutil
+
+from app.core.data_rules.compilers.polars import validate_bound_polars_plan
+
+
+class PolarsWorkerError(RuntimeError):
+    """A bounded worker failed without exposing internal data."""
+
+
+class PolarsWorkerResourceError(PolarsWorkerError):
+    """The worker crossed a hard process resource boundary."""
+
+
+def _set_process_memory_limit(
+    *,
+    baseline_vms: int,
+    memory_limit_bytes: int,
+) -> None:
+    if resource is None or not hasattr(resource, "RLIMIT_AS"):
+        raise OSError("OS address-space limits are unavailable")
+    hard_limit = baseline_vms + memory_limit_bytes
+    resource.setrlimit(resource.RLIMIT_AS, (hard_limit, hard_limit))
+
+
+def _memory_probe_entry(
+    connection,
+    allocate_bytes: int,
+    memory_limit_bytes: int,
+) -> None:
+    process = psutil.Process()
+    memory = process.memory_info()
+    connection.send(
+        {
+            "kind": "ready",
+            "pid": os.getpid(),
+            "baseline_rss": memory.rss,
+        }
+    )
+    try:
+        _set_process_memory_limit(
+            baseline_vms=memory.vms,
+            memory_limit_bytes=memory_limit_bytes,
+        )
+        allocated = bytearray(allocate_bytes)
+        for index in range(0, len(allocated), 4096):
+            allocated[index] = 1
+        connection.send(
+            {
+                "kind": "result",
+                "worker_pid": os.getpid(),
+                "allocated_bytes": len(allocated),
+            }
+        )
+    except (MemoryError, OSError):
+        with suppress(Exception):
+            connection.send({"kind": "resource_error"})
+    except BaseException:
+        with suppress(Exception):
+            connection.send({"kind": "worker_error"})
+    finally:
+        connection.close()
+
+
+class _ExpressionCompiler:
+    def __init__(self, timezone: str):
+        self.timezone = timezone
+
+    def compile(self, ast: dict[str, Any]) -> pl.Expr:
+        kind = ast["kind"]
+        if kind == "identifier":
+            return pl.col(ast["name"])
+        if kind == "literal":
+            value = ast["value"]
+            if ast["type"] == "decimal":
+                value = Decimal(value)
+            return pl.lit(value)
+        if kind == "unary":
+            operand = self.compile(ast["operand"])
+            return ~operand if ast["operator"] == "!" else -operand
+        if kind == "binary":
+            left = self.compile(ast["left"])
+            right = self.compile(ast["right"])
+            return {
+                "||": lambda: left | right,
+                "&&": lambda: left & right,
+                "==": lambda: left == right,
+                "!=": lambda: left != right,
+                "<": lambda: left < right,
+                "<=": lambda: left <= right,
+                ">": lambda: left > right,
+                ">=": lambda: left >= right,
+                "+": lambda: left + right,
+                "-": lambda: left - right,
+                "*": lambda: left * right,
+                "/": lambda: left / right,
+                "%": lambda: left % right,
+            }[ast["operator"]]()
+        function = ast["function"]
+        arguments = [self.compile(item) for item in ast["arguments"]]
+        if function == "matches":
+            return arguments[0].str.contains(
+                ast["arguments"][1]["value"], strict=True
+            )
+        if function == "lower":
+            return arguments[0].str.to_lowercase()
+        if function == "upper":
+            return arguments[0].str.to_uppercase()
+        if function == "trim":
+            return arguments[0].str.strip_chars()
+        if function == "length":
+            return arguments[0].str.len_chars()
+        if function == "coalesce":
+            return pl.coalesce(arguments)
+        if function == "date":
+            return (
+                arguments[0]
+                .cast(pl.String)
+                .str.to_datetime(time_zone=self.timezone, strict=True)
+                .dt.date()
+            )
+        if function == "timestamp":
+            return (
+                arguments[0]
+                .cast(pl.String)
+                .str.to_datetime(time_zone=self.timezone, strict=True)
+                .dt.replace_time_zone(None)
+            )
+        if function == "abs":
+            return arguments[0].abs()
+        raise ValueError("published Polars expression is unsupported")
+
+
+def _cast_type(field: dict[str, Any]) -> pl.DataType:
+    field_type = field["type"]
+    if field_type == "boolean":
+        return pl.Boolean
+    if field_type == "date":
+        return pl.Date
+    if field_type == "decimal":
+        return pl.Decimal(
+            precision=field["precision"],
+            scale=field["scale"],
+        )
+    if field_type == "double":
+        return pl.Float64
+    if field_type == "float":
+        return pl.Float32
+    if field_type == "integer":
+        return pl.Int64
+    if field_type == "string":
+        return pl.String
+    if field_type == "timestamp":
+        return pl.Datetime
+    if field_type == "timestamptz":
+        return pl.Datetime(time_zone=field["timezone"])
+    raise ValueError("published Polars cast type is unsupported")
+
+
+def _count(frame: pl.LazyFrame) -> int:
+    return int(frame.select(pl.len()).collect(engine="streaming").item())
+
+
+def _validate_lazy_schema(
+    frame: pl.LazyFrame,
+    fields: list[dict[str, Any]],
+) -> None:
+    schema = frame.collect_schema()
+    if set(schema.names()) != {field["name"] for field in fields}:
+        raise ValueError("artifact schema fields do not match")
+    for field in fields:
+        if schema[field["name"]] != _cast_type(field):
+            raise ValueError("artifact schema type does not match")
+    required = [
+        field["name"] for field in fields if not field["nullable"]
+    ]
+    if required:
+        counts = frame.select(
+            [pl.col(name).null_count().alias(name) for name in required]
+        ).collect(engine="streaming")
+        if any(counts[name].item() for name in required):
+            raise ValueError("artifact nullable contract does not match")
+
+
+def _execute_polars_job(job: dict[str, Any]) -> dict[str, Any]:
+    plan = validate_bound_polars_plan(job["plan"])
+    frame = pl.scan_parquet(job["input_path"])
+    _validate_lazy_schema(frame, plan["input_fields"])
+    rows_in = _count(frame)
+    expressions = _ExpressionCompiler(plan["timezone"])
+    violations = []
+    metrics = {
+        "rows_rejected": 0,
+        "rows_filtered": 0,
+        "rows_deduplicated": 0,
+        "rows_join_dropped": 0,
+        "rows_aggregated": 0,
+    }
+    for operation in plan["operations"]:
+        op = operation["op"]
+        before = _count(frame)
+        if op == "normalize_text":
+            expression = pl.col(operation["column"])
+            if operation["trim"]:
+                expression = expression.str.strip_chars()
+            if operation["lowercase"]:
+                expression = expression.str.to_lowercase()
+            if operation["uppercase"]:
+                expression = expression.str.to_uppercase()
+            frame = frame.with_columns(
+                expression.alias(operation["column"])
+            )
+        elif op == "regex_replace":
+            frame = frame.with_columns(
+                pl.col(operation["column"])
+                .str.replace_all(
+                    operation["pattern"], operation["replacement"]
+                )
+                .alias(operation["column"])
+            )
+        elif op == "fill_null":
+            frame = frame.with_columns(
+                pl.col(operation["column"])
+                .fill_null(operation["value"])
+                .alias(operation["column"])
+            )
+        elif op == "filter":
+            frame = frame.filter(
+                expressions.compile(operation["expression_ast"]).fill_null(
+                    False
+                )
+            )
+            metrics["rows_filtered"] += before - _count(frame)
+        elif op == "assert":
+            predicate = expressions.compile(
+                operation["expression_ast"]
+            ).fill_null(False)
+            invalid = int(
+                frame.select((~predicate).sum().alias("count"))
+                .collect(engine="streaming")
+                .item()
+                or 0
+            )
+            violations.append(
+                {"step_id": operation["step_id"], "count": invalid}
+            )
+            metrics["rows_rejected"] += invalid
+            frame = frame.filter(predicate)
+        elif op == "derive":
+            frame = frame.with_columns(
+                expressions.compile(operation["expression_ast"]).alias(
+                    operation["target"]
+                )
+            )
+        elif op == "map_values":
+            column = pl.col(operation["column"])
+            frame = frame.with_columns(
+                column.replace_strict(
+                    operation["mapping"], default=column
+                ).alias(operation["column"])
+            )
+        elif op == "cast":
+            target = operation["target_field"]
+            expression = pl.col(operation["column"])
+            source_type = frame.collect_schema()[operation["column"]]
+            if source_type == pl.String and target["type"] == "timestamp":
+                expression = (
+                    expression.str.to_datetime(
+                        time_zone=plan["timezone"],
+                        strict=True,
+                    ).dt.replace_time_zone(None)
+                )
+            elif (
+                source_type == pl.String
+                and target["type"] == "timestamptz"
+            ):
+                expression = expression.str.to_datetime(
+                    time_zone=target["timezone"], strict=True
+                )
+            else:
+                expression = expression.cast(
+                    _cast_type(target), strict=True
+                )
+            frame = frame.with_columns(
+                expression.alias(operation["column"])
+            )
+        elif op == "deduplicate":
+            order = [
+                *operation["order_by"],
+                *sorted(
+                    name
+                    for name in frame.collect_schema().names()
+                    if name not in operation["order_by"]
+                ),
+            ]
+            descending = operation["keep"] == "last"
+            frame = (
+                frame.sort(
+                    order,
+                    descending=[descending] * len(order),
+                    nulls_last=True,
+                )
+                .unique(
+                    subset=operation["keys"],
+                    keep="first",
+                    maintain_order=True,
+                )
+            )
+            metrics["rows_deduplicated"] += before - _count(frame)
+        elif op == "mask":
+            if job["masking_policies"].get(
+                operation["policy_id"]
+            ) != operation["policy_kind"]:
+                raise ValueError("masking policy is not registered")
+            column = pl.col(operation["column"])
+            if operation["policy_kind"] == "redact":
+                masked = pl.when(column.is_null()).then(None).otherwise(
+                    pl.lit("***")
+                )
+            else:
+                masked = pl.when(column.is_null()).then(None).otherwise(
+                    pl.lit("***") + column.str.slice(-4)
+                )
+            frame = frame.with_columns(masked.alias(operation["column"]))
+        elif op == "aggregate":
+            aggregations = []
+            for aggregate in operation["aggregations"]:
+                expression = pl.col(aggregate["column"])
+                function = aggregate["function"]
+                if function == "count":
+                    expression = expression.count()
+                elif function == "sum":
+                    expression = expression.sum()
+                elif function == "min":
+                    expression = expression.min()
+                elif function == "max":
+                    expression = expression.max()
+                elif function == "mean":
+                    expression = expression.mean()
+                aggregations.append(
+                    expression.alias(aggregate["target"])
+                )
+            frame = frame.group_by(
+                operation["group_by"], maintain_order=True
+            ).agg(aggregations)
+            metrics["rows_aggregated"] += before - _count(frame)
+        elif op == "lookup_join":
+            lookup = pl.scan_parquet(
+                job["lookup_paths"][operation["lookup_binding_id"]]
+            )
+            _validate_lazy_schema(lookup, operation["lookup_fields"])
+            duplicates = (
+                lookup.group_by(operation["right_on"])
+                .len()
+                .filter(pl.col("len") > 1)
+                .select(pl.len())
+                .collect(engine="streaming")
+                .item()
+            )
+            if duplicates:
+                raise ValueError("lookup keys are not unique")
+            selected = list(operation["select"].values())
+            lookup = lookup.select(
+                list(
+                    dict.fromkeys(
+                        [*operation["right_on"], *selected]
+                    )
+                )
+            ).rename(
+                {
+                    source: target
+                    for target, source in operation["select"].items()
+                    if source not in operation["right_on"]
+                }
+            )
+            frame = frame.join(
+                lookup,
+                left_on=operation["left_on"],
+                right_on=operation["right_on"],
+                how=operation["how"],
+            )
+            after = _count(frame)
+            if after > before:
+                raise ValueError("lookup join expanded its input")
+            metrics["rows_join_dropped"] += before - after
+
+    frame = frame.select(
+        [field["name"] for field in plan["output_fields"]]
+    )
+    _validate_lazy_schema(frame, plan["output_fields"])
+    rows_out = _count(frame)
+    if rows_out > plan["resource_limits"]["max_rows"]:
+        raise ValueError("output row limit exceeded")
+    frame.sink_parquet(job["output_path"], engine="streaming")
+    return {
+        "worker_pid": os.getpid(),
+        "rows_in": rows_in,
+        "rows_out": rows_out,
+        **metrics,
+        "violation_count": sum(item["count"] for item in violations),
+        "violations": violations,
+    }
+
+
+def _polars_job_entry(
+    connection,
+    job: dict[str, Any],
+    memory_limit_bytes: int,
+) -> None:
+    # Initialize the Rust engine/thread pool before measuring the worker's
+    # immutable runtime baseline. The plan allowance applies only afterwards.
+    pl.DataFrame({"_warmup": [1]}).lazy().collect(engine="streaming")
+    process = psutil.Process()
+    memory = process.memory_info()
+    connection.send(
+        {
+            "kind": "ready",
+            "pid": os.getpid(),
+            "baseline_rss": memory.rss,
+        }
+    )
+    try:
+        _set_process_memory_limit(
+            baseline_vms=memory.vms,
+            memory_limit_bytes=memory_limit_bytes,
+        )
+        result = _execute_polars_job(job)
+        connection.send({"kind": "result", **result})
+    except (MemoryError, OSError):
+        with suppress(Exception):
+            connection.send({"kind": "resource_error"})
+    except BaseException as exc:
+        message = str(exc).lower()
+        kind = (
+            "resource_error"
+            if any(
+                token in message
+                for token in ("memory", "allocate", "allocation")
+            )
+            else "worker_error"
+        )
+        with suppress(Exception):
+            connection.send({"kind": kind})
+    finally:
+        connection.close()
+
+
+def _run_process(
+    *,
+    target,
+    args: tuple[Any, ...],
+    memory_limit_bytes: int,
+    timeout_seconds: float = 15.0,
+) -> dict[str, Any]:
+    context = multiprocessing.get_context("spawn")
+    parent, child = context.Pipe(duplex=False)
+    process = context.Process(
+        target=target,
+        args=(child, *args, memory_limit_bytes),
+        daemon=True,
+    )
+    process.start()
+    child.close()
+    deadline = time.monotonic() + timeout_seconds
+    baseline_rss = None
+    try:
+        while time.monotonic() < deadline:
+            if parent.poll(0.01):
+                try:
+                    message = parent.recv()
+                except EOFError as exc:
+                    raise PolarsWorkerResourceError(
+                        "Polars worker crossed its hard memory limit"
+                    ) from exc
+                kind = message.get("kind")
+                if kind == "ready":
+                    baseline_rss = int(message["baseline_rss"])
+                elif kind == "result":
+                    process.join(timeout=1)
+                    return message
+                elif kind == "resource_error":
+                    raise PolarsWorkerResourceError(
+                        "Polars worker exceeded its hard memory limit"
+                    )
+                else:
+                    raise PolarsWorkerError(
+                        "Polars worker execution failed"
+                    )
+            if baseline_rss is not None and process.is_alive():
+                try:
+                    rss = psutil.Process(process.pid).memory_info().rss
+                except (psutil.Error, ProcessLookupError):
+                    rss = baseline_rss
+                if rss - baseline_rss > memory_limit_bytes:
+                    process.kill()
+                    process.join(timeout=1)
+                    raise PolarsWorkerResourceError(
+                        "Polars worker exceeded its hard memory limit"
+                    )
+            if not process.is_alive():
+                process.join(timeout=1)
+                raise PolarsWorkerResourceError(
+                    "Polars worker crossed its hard memory limit"
+                )
+        process.kill()
+        process.join(timeout=1)
+        raise PolarsWorkerResourceError(
+            "Polars worker exceeded its bounded execution time"
+        )
+    finally:
+        parent.close()
+        if process.is_alive():
+            process.kill()
+        process.join(timeout=1)
+
+
+def run_isolated_memory_probe(
+    *,
+    allocate_bytes: int,
+    memory_limit_bytes: int,
+) -> dict[str, Any]:
+    """Exercise the same process boundary used by plan execution."""
+
+    return _run_process(
+        target=_memory_probe_entry,
+        args=(int(allocate_bytes),),
+        memory_limit_bytes=int(memory_limit_bytes),
+    )
+
+
+def execute_isolated_polars_plan(
+    job: dict[str, Any],
+    *,
+    memory_limit_bytes: int,
+) -> dict[str, Any]:
+    return _run_process(
+        target=_polars_job_entry,
+        args=(job,),
+        memory_limit_bytes=memory_limit_bytes,
+        timeout_seconds=60.0,
+    )

app/runner/rule_polars.py

@@ -1,34 +1,47 @@
-"""Runner adapter that reconstructs allowlisted Polars LazyFrame operations."""
+"""Runner adapter for OS-isolated, allowlisted Polars execution."""
 
 from __future__ import annotations
 
-from contextlib import suppress
-from decimal import Decimal
+import os
+import tempfile
+from contextlib import ExitStack, suppress
 from typing import Any
 
-import polars as pl
-
 from app.core.common.identifiers import ensure_governance_uid
 from app.core.data_rules.compilers.polars import (
     bound_polars_plan_hash,
     validate_bound_polars_plan,
 )
 from app.runner.nodes import NodeExecutionError
+from app.runner.polars_worker import (
+    PolarsWorkerError,
+    PolarsWorkerResourceError,
+    execute_isolated_polars_plan,
+)
 
-_TYPE_MAP = {
-    "boolean": pl.Boolean,
-    "date": pl.Date,
-    "double": pl.Float64,
-    "float": pl.Float32,
-    "integer": pl.Int64,
-    "string": pl.String,
-    "timestamp": pl.Datetime,
-}
 _IDEMPOTENCY = {
     "deduplication_key",
     "partition_replace",
     "upsert",
 }
+_PUBLIC_ARTIFACT_KEYS = (
+    "artifact_ref",
+    "digest",
+    "row_count",
+    "schema_hash",
+    "expires_at",
+)
+_RESULT_KEYS = (
+    "rows_in",
+    "rows_out",
+    "rows_rejected",
+    "rows_filtered",
+    "rows_deduplicated",
+    "rows_join_dropped",
+    "rows_aggregated",
+    "violation_count",
+    "violations",
+)
 
 
 def _uid(value: Any, label: str) -> str:
@@ -38,119 +51,19 @@ def _uid(value: Any, label: str) -> str:
         raise NodeExecutionError(f"{label} is invalid") from exc
 
 
-class _ExpressionCompiler:
-    def compile(self, ast: dict[str, Any]) -> pl.Expr:
-        kind = ast["kind"]
-        if kind == "identifier":
-            return pl.col(ast["name"])
-        if kind == "literal":
-            value = ast["value"]
-            if ast["type"] == "decimal":
-                value = Decimal(value)
-            return pl.lit(value)
-        if kind == "unary":
-            operand = self.compile(ast["operand"])
-            return ~operand if ast["operator"] == "!" else -operand
-        if kind == "binary":
-            left = self.compile(ast["left"])
-            right = self.compile(ast["right"])
-            return {
-                "||": lambda: left | right,
-                "&&": lambda: left & right,
-                "==": lambda: left == right,
-                "!=": lambda: left != right,
-                "<": lambda: left < right,
-                "<=": lambda: left <= right,
-                ">": lambda: left > right,
-                ">=": lambda: left >= right,
-                "+": lambda: left + right,
-                "-": lambda: left - right,
-                "*": lambda: left * right,
-                "/": lambda: left / right,
-                "%": lambda: left % right,
-            }[ast["operator"]]()
-        function = ast["function"]
-        arguments = [self.compile(item) for item in ast["arguments"]]
-        if function == "matches":
-            pattern = ast["arguments"][1]["value"]
-            return arguments[0].str.contains(pattern, strict=True)
-        if function == "lower":
-            return arguments[0].str.to_lowercase()
-        if function == "upper":
-            return arguments[0].str.to_uppercase()
-        if function == "trim":
-            return arguments[0].str.strip_chars()
-        if function == "length":
-            return arguments[0].str.len_chars()
-        if function == "coalesce":
-            return pl.coalesce(arguments)
-        if function == "date":
-            return arguments[0].cast(pl.Date, strict=True)
-        if function == "timestamp":
-            return arguments[0].cast(pl.Datetime, strict=True)
-        if function == "abs":
-            return arguments[0].abs()
-        raise NodeExecutionError("published Polars expression is unsupported")
-
-
-def _cast_type(field: dict[str, Any]) -> pl.DataType:
-    name = field["type"]
-    if name == "decimal":
-        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,
     *,
     binding_id: str,
     binding_hash: str,
     correlation_id: str,
+    kind: str,
 ) -> dict[str, Any]:
     try:
         artifact = resolver.resolve(
             binding_id=binding_id,
             correlation_id=correlation_id,
+            kind=kind,
         )
     except Exception as exc:
         raise NodeExecutionError(
@@ -161,6 +74,7 @@ def _resolve_artifact(
         or artifact.get("binding_hash") != binding_hash
         or not isinstance(artifact.get("artifact_ref"), str)
         or not isinstance(artifact.get("digest"), str)
+        or not isinstance(artifact.get("schema_fields"), list)
     ):
         raise NodeExecutionError(
             "published Polars artifact binding does not match"
@@ -169,7 +83,7 @@ def _resolve_artifact(
 
 
 class PolarsRulePlanAdapter:
-    """Execute one digest-bound Polars plan and write one bounded artifact."""
+    """Stage bounded inputs and execute the plan in a hard-limited process."""
 
     def __init__(
         self,
@@ -203,9 +117,13 @@ class PolarsRulePlanAdapter:
         if config.get("execution_plan_hash") != bound_polars_plan_hash(
             normalized
         ):
-            raise NodeExecutionError("published Polars rule plan hash does not match")
+            raise NodeExecutionError(
+                "published Polars rule plan hash does not match"
+            )
         if config.get("rule_version_id") != normalized["rule_version_id"]:
-            raise NodeExecutionError("published Polars rule id does not match")
+            raise NodeExecutionError(
+                "published Polars rule id does not match"
+            )
         idempotency = node.get("idempotency")
         if (
             node.get("type") != "rule.apply"
@@ -238,296 +156,99 @@ class PolarsRulePlanAdapter:
             binding_id=normalized["input_binding_id"],
             binding_hash=normalized["input_binding_hash"],
             correlation_id=correlation,
+            kind="input",
         )
-        try:
-            frame = self.artifact_store.read(
-                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
         limits = normalized["resource_limits"]
-        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 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"]:
-                    expression = expression.str.strip_chars()
-                if operation["lowercase"]:
-                    expression = expression.str.to_lowercase()
-                if operation["uppercase"]:
-                    expression = expression.str.to_uppercase()
-                frame = frame.with_columns(
-                    expression.alias(operation["column"])
-                )
-            elif op == "regex_replace":
-                frame = frame.with_columns(
-                    pl.col(operation["column"])
-                    .str.replace_all(
-                        operation["pattern"],
-                        operation["replacement"],
-                    )
-                    .alias(operation["column"])
-                )
-            elif op == "fill_null":
-                frame = frame.with_columns(
-                    pl.col(operation["column"])
-                    .fill_null(operation["value"])
-                    .alias(operation["column"])
-                )
-            elif op == "filter":
-                frame = frame.filter(
-                    expressions.compile(operation["expression_ast"]).fill_null(
-                        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"]
-                ).fill_null(False)
-                invalid = int(
-                    frame.select((~predicate).sum().alias("count"))
-                    .collect(engine="streaming")
-                    .item()
-                    or 0
-                )
-                violations.append(
-                    {"step_id": operation["step_id"], "count": invalid}
-                )
-                rows_rejected += invalid
-                frame = frame.filter(predicate)
-            elif op == "derive":
-                frame = frame.with_columns(
-                    expressions.compile(operation["expression_ast"]).alias(
-                        operation["target"]
-                    )
-                )
-            elif op == "map_values":
-                column = pl.col(operation["column"])
-                frame = frame.with_columns(
-                    column.replace_strict(
-                        operation["mapping"],
-                        default=column,
-                    ).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(
-                    column_expression.alias(operation["column"])
-                )
-            elif op == "deduplicate":
-                order = [
-                    *operation["order_by"],
-                    *sorted(
-                        name
-                        for name in frame.collect_schema().names()
-                        if name not in operation["order_by"]
-                    ),
-                ]
-                descending = operation["keep"] == "last"
-                frame = (
-                    frame.sort(
-                        order,
-                        descending=[descending] * len(order),
-                        nulls_last=True,
-                    )
-                    .unique(
-                        subset=operation["keys"],
-                        keep="first",
-                        maintain_order=True,
+        output_path = None
+        try:
+            with ExitStack() as stack:
+                try:
+                    input_path = stack.enter_context(
+                        self.artifact_store.stage(
+                            source["artifact_ref"],
+                            source["digest"],
+                            expected_schema_fields=normalized[
+                                "input_fields"
+                            ],
+                            limits=limits,
+                        )
                     )
-                )
-                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"]
-                ) != operation["policy_kind"]:
+                    lookup_paths = {}
+                    for operation in normalized["operations"]:
+                        if operation["op"] != "lookup_join":
+                            continue
+                        lookup_artifact = _resolve_artifact(
+                            self.artifact_resolver,
+                            binding_id=operation[
+                                "lookup_binding_id"
+                            ],
+                            binding_hash=operation[
+                                "lookup_binding_hash"
+                            ],
+                            correlation_id=correlation,
+                            kind="lookup",
+                        )
+                        lookup_paths[
+                            operation["lookup_binding_id"]
+                        ] = stack.enter_context(
+                            self.artifact_store.stage(
+                                lookup_artifact["artifact_ref"],
+                                lookup_artifact["digest"],
+                                expected_schema_fields=operation[
+                                    "lookup_fields"
+                                ],
+                                limits=limits,
+                            )
+                        )
+                except ValueError as exc:
                     raise NodeExecutionError(
-                        "published masking policy is not registered"
-                    )
-                column = pl.col(operation["column"])
-                if operation["policy_kind"] == "redact":
-                    masked = pl.when(column.is_null()).then(None).otherwise(
-                        pl.lit("***")
-                    )
-                elif operation["policy_kind"] == "preserve_last_4":
-                    masked = pl.when(column.is_null()).then(None).otherwise(
-                        pl.lit("***") + column.str.slice(-4)
+                        "published Polars input artifact is invalid"
+                    ) from exc
+                with tempfile.NamedTemporaryFile(
+                    prefix="dataops-polars-worker-output-",
+                    suffix=".parquet",
+                    delete=False,
+                ) as handle:
+                    output_path = handle.name
+                try:
+                    worker_result = execute_isolated_polars_plan(
+                        {
+                            "plan": normalized,
+                            "input_path": input_path,
+                            "lookup_paths": lookup_paths,
+                            "output_path": output_path,
+                            "masking_policies": self.masking_policies,
+                        },
+                        memory_limit_bytes=limits[
+                            "memory_limit_bytes"
+                        ],
                     )
-                else:
+                except PolarsWorkerResourceError as exc:
                     raise NodeExecutionError(
-                        "published masking policy is unsupported"
-                    )
-                frame = frame.with_columns(masked.alias(operation["column"]))
-            elif op == "aggregate":
-                aggregations = []
-                for aggregate in operation["aggregations"]:
-                    expression = pl.col(aggregate["column"])
-                    function = aggregate["function"]
-                    if function == "count":
-                        expression = expression.count()
-                    elif function == "sum":
-                        expression = expression.sum()
-                    elif function == "min":
-                        expression = expression.min()
-                    elif function == "max":
-                        expression = expression.max()
-                    elif function == "mean":
-                        expression = expression.mean()
-                    aggregations.append(
-                        expression.alias(aggregate["target"])
-                    )
-                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,
-                    binding_id=operation["lookup_binding_id"],
-                    binding_hash=operation["lookup_binding_hash"],
-                    correlation_id=correlation,
-                )
+                        "published Polars worker exceeded its hard memory limit"
+                    ) from exc
+                except PolarsWorkerError as exc:
+                    raise NodeExecutionError(
+                        "published Polars worker execution failed"
+                    ) from exc
                 try:
-                    lookup = self.artifact_store.read(
-                        lookup_artifact["artifact_ref"],
-                        lookup_artifact["digest"],
-                        expected_schema_fields=operation["lookup_fields"],
+                    artifact = self.artifact_store.write_path(
+                        output_path,
+                        correlation,
+                        self.artifact_ttl_seconds,
+                        schema_fields=normalized["output_fields"],
                         limits=limits,
                     )
                 except ValueError as exc:
                     raise NodeExecutionError(
-                        "published Polars lookup artifact is invalid"
+                        "published Polars output artifact write failed"
                     ) from exc
-                lookup = _bounded_materialize(
-                    lookup, limits, f"lookup input {index}"
-                ).lazy()
-                duplicate_count = (
-                    lookup.group_by(operation["right_on"])
-                    .len()
-                    .filter(pl.col("len") > 1)
-                    .select(pl.len())
-                    .collect(engine="streaming")
-                    .item()
-                )
-                if duplicate_count:
-                    raise NodeExecutionError(
-                        "published Polars lookup keys are not unique"
-                    )
-                selected_sources = list(operation["select"].values())
-                lookup = lookup.select(
-                    list(dict.fromkeys(
-                        [*operation["right_on"], *selected_sources]
-                    ))
-                ).rename(
-                    {
-                        source_name: target_name
-                        for target_name, source_name in operation[
-                            "select"
-                        ].items()
-                        if source_name not in operation["right_on"]
-                    }
-                )
-                frame = frame.join(
-                    lookup,
-                    left_on=operation["left_on"],
-                    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)
-        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
+        finally:
+            if output_path is not None:
+                with suppress(FileNotFoundError):
+                    os.unlink(output_path)
         try:
-            self.artifact_resolver.register(
+            registered = self.artifact_resolver.register(
                 binding_id=normalized["output_binding_id"],
                 correlation_id=correlation,
                 artifact=artifact,
@@ -535,24 +256,13 @@ class PolarsRulePlanAdapter:
                 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"])
+            with suppress(Exception):
+                self.artifact_store.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": 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,
+            **{key: registered[key] for key in _PUBLIC_ARTIFACT_KEYS},
+            **{key: worker_result[key] for key in _RESULT_KEYS},
             "commit_outcome": "committed",
         }

docs/security/cyclonedx-1.5-schema/README.md

@@ -0,0 +1,15 @@
+# CycloneDX 1.5 JSON Schema
+
+These fixed validation inputs were retrieved from the official CycloneDX
+schema host on 2026-07-23:
+
+- `https://cyclonedx.org/schema/bom-1.5.schema.json`
+  (`sha256:a00fcba23a44b72179ac1f288be4c3529b59f6e1b3719709a40685f177516b46`)
+- `https://cyclonedx.org/schema/spdx.schema.json`
+  (`sha256:c87aa7bb5eb503d40b52ec6bf00de8045df15da7a13cea48d290cf6d36a8d2ea`)
+- `https://cyclonedx.org/schema/jsf-0.82.schema.json`
+  (`sha256:2faf5eb3651f2ae5f46091a131770d8d847bbd121139d19c85fc7051bfa58c46`)
+
+The primary schema states that the CycloneDX JSON schema is published under
+the Apache License 2.0. The files are vendored so SBOM validation is
+deterministic and does not require network access.

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

@@ -5,27 +5,169 @@
   "metadata": {
     "component": {
       "type": "application",
+      "bom-ref": "dataops-governed-rule-runtime",
       "name": "dataops-governed-rule-runtime"
     }
   },
   "components": [
     {
       "type": "library",
+      "bom-ref": "pkg:pypi/polars@1.42.1",
       "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"
+      "licenses": [
+        {
+          "license": {
+            "id": "MIT"
+          }
+        }
+      ],
+      "externalReferences": [
+        {
+          "type": "vcs",
+          "url": "https://github.com/pola-rs/polars"
+        }
+      ],
+      "properties": [
+        {
+          "name": "dataops:purpose",
+          "value": "Python API for governed Polars batch execution"
+        }
+      ]
     },
     {
       "type": "library",
+      "bom-ref": "pkg:pypi/polars-runtime-32@1.42.1",
+      "name": "polars-runtime-32",
+      "version": "1.42.1",
+      "purl": "pkg:pypi/polars-runtime-32@1.42.1",
+      "licenses": [
+        {
+          "license": {
+            "id": "MIT"
+          }
+        }
+      ],
+      "externalReferences": [
+        {
+          "type": "vcs",
+          "url": "https://github.com/pola-rs/polars"
+        }
+      ],
+      "properties": [
+        {
+          "name": "dataops:purpose",
+          "value": "Native 32-bit Polars runtime wheel"
+        }
+      ]
+    },
+    {
+      "type": "library",
+      "bom-ref": "pkg:pypi/pyarrow@21.0.0",
+      "name": "pyarrow",
+      "version": "21.0.0",
+      "purl": "pkg:pypi/pyarrow@21.0.0",
+      "licenses": [
+        {
+          "license": {
+            "id": "Apache-2.0"
+          }
+        }
+      ],
+      "externalReferences": [
+        {
+          "type": "vcs",
+          "url": "https://github.com/apache/arrow"
+        }
+      ],
+      "properties": [
+        {
+          "name": "dataops:purpose",
+          "value": "Trusted Parquet footer and row-group resource preflight"
+        }
+      ]
+    },
+    {
+      "type": "library",
+      "bom-ref": "pkg:pypi/psutil@5.9.8",
+      "name": "psutil",
+      "version": "5.9.8",
+      "purl": "pkg:pypi/psutil@5.9.8",
+      "licenses": [
+        {
+          "license": {
+            "id": "BSD-3-Clause"
+          }
+        }
+      ],
+      "externalReferences": [
+        {
+          "type": "vcs",
+          "url": "https://github.com/giampaolo/psutil"
+        }
+      ],
+      "properties": [
+        {
+          "name": "dataops:purpose",
+          "value": "Isolated worker RSS enforcement and process monitoring"
+        }
+      ]
+    },
+    {
+      "type": "library",
+      "bom-ref": "pkg:pypi/minio@7.2.10",
       "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"
+      "licenses": [
+        {
+          "license": {
+            "id": "Apache-2.0"
+          }
+        }
+      ],
+      "externalReferences": [
+        {
+          "type": "vcs",
+          "url": "https://github.com/minio/minio-py"
+        }
+      ],
+      "properties": [
+        {
+          "name": "dataops:purpose",
+          "value": "Digest-bound Parquet artifact transport"
+        }
+      ]
+    }
+  ],
+  "dependencies": [
+    {
+      "ref": "dataops-governed-rule-runtime",
+      "dependsOn": [
+        "pkg:pypi/minio@7.2.10",
+        "pkg:pypi/polars@1.42.1",
+        "pkg:pypi/psutil@5.9.8",
+        "pkg:pypi/pyarrow@21.0.0"
+      ]
+    },
+    {
+      "ref": "pkg:pypi/polars@1.42.1",
+      "dependsOn": [
+        "pkg:pypi/polars-runtime-32@1.42.1"
+      ]
+    },
+    {
+      "ref": "pkg:pypi/polars-runtime-32@1.42.1"
+    },
+    {
+      "ref": "pkg:pypi/pyarrow@21.0.0"
+    },
+    {
+      "ref": "pkg:pypi/psutil@5.9.8"
+    },
+    {
+      "ref": "pkg:pypi/minio@7.2.10"
     }
   ]
 }

migrations/versions/20260723_140_rule_run_artifacts.py

@@ -26,7 +26,7 @@ def upgrade() -> None:
                 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)
+            UNIQUE (correlation_id, binding_id, artifact_kind)
         );
         CREATE INDEX idx_rule_run_artifacts_resolve
             ON public.rule_run_artifacts
@@ -36,5 +36,7 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
-    # Runtime handoff evidence is immutable and intentionally retained.
-    pass
+    raise RuntimeError(
+        "rule run artifact catalog is forward-only and cannot downgrade "
+        "without violating immutable runtime handoff evidence"
+    )

requirements.txt

@@ -12,6 +12,7 @@ psycopg2-binary==2.9.9
 PyMySQL==1.1.1
 sqlglot==30.13.0
 polars==1.42.1
+pyarrow==21.0.0
 neo4j==5.26.0
 argon2-cffi==25.1.0
 PyJWT==2.10.1

tests/core/data_rules/test_polars_compiler.py

@@ -592,6 +592,74 @@ def test_polars_plan_carries_timezone_and_complete_decimal_temporal_semantics():
     )["timezone"] == "Asia/Shanghai"
 
 
+def test_polars_cast_operations_carry_complete_typed_targets():
+    from app.core.data_rules.compilers.polars import (
+        PolarsRuleCompiler,
+        validate_bound_polars_plan,
+    )
+
+    input_schema = _schema(
+        "bd:finance:raw",
+        [("amount", "string", False), ("occurred_at", "string", False)],
+    )
+    output_schema = _schema(
+        "bd:finance: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(input_schema, access_mode="read")
+    target = _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_occurred_at",
+                "op": "cast",
+                "column": "occurred_at",
+                "to": "timestamptz",
+                "on_error": "fail",
+            },
+        ],
+    )
+    plan = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=source,
+        output_binding=target,
+        backend=_backend(),
+    )["plan"]
+
+    amount, occurred_at = plan["operations"]
+    assert amount["target_field"] == {
+        "name": "amount",
+        "type": "decimal",
+        "nullable": False,
+        "precision": 12,
+        "scale": 2,
+    }
+    assert occurred_at["target_field"]["timezone"] == "Asia/Shanghai"
+
+    del amount["target_field"]
+    with pytest.raises(ValueError, match="closed shape|target_field"):
+        validate_bound_polars_plan(plan)
+
+
 def test_shared_polars_semantic_validator_rejects_tampered_expression_type():
     from app.core.data_rules.compilers.polars import (
         validate_bound_polars_plan,

tests/integration/test_data_rule_polars_execution.py

@@ -5,6 +5,7 @@ import re
 from pathlib import Path
 
 import polars as pl
+import pytest
 from minio import Minio
 from sqlalchemy import create_engine, text
 
@@ -105,7 +106,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
         bucket=bucket,
         max_artifact_bytes=4 * 1024 * 1024,
         max_rows=1_000,
-        memory_limit_bytes=16 * 1024 * 1024,
+        memory_limit_bytes=256 * 1024 * 1024,
         max_ttl_seconds=3600,
     )
     correlation_id = new_governance_uid()
@@ -119,6 +120,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
     deployment_id = new_governance_uid()
     component_binding_id = new_governance_uid()
     plan_id = new_governance_uid()
+    ledger_jti = None
 
     try:
         with postgres.begin() as connection:
@@ -283,7 +285,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
             backend={
                 "max_rows": 1_000,
                 "max_artifact_bytes": 4 * 1024 * 1024,
-                "memory_limit_bytes": 16 * 1024 * 1024,
+                "memory_limit_bytes": 256 * 1024 * 1024,
                 "masking_policies": {},
                 "lookup_bindings": {
                     lookup_binding["id"]: {
@@ -589,20 +591,110 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
             )
         )
         assert repeated["rows_out"] == 2
+        assert repeated["artifact_ref"] == result["artifact_ref"]
+        assert "schema_fields" not in result
+
+        from app.runner.api import create_runner_app
+        from app.runner.auth import TaskTokenIssuer, TaskTokenVerifier
+        from app.runner.ledger import PostgresTaskLedger
+        from app.runner.nodes import NodeRegistry
+
+        task_secret = "task5-real-http-secret-value-32-bytes"
+        verifier = TaskTokenVerifier(task_secret)
+        task_token = TaskTokenIssuer(task_secret).issue(
+            task_uid=new_governance_uid(),
+            dataflow_uid=dataflow_uid,
+            workflow_version=1,
+            correlation_id=correlation_id,
+            node=node,
+            write_authorized=True,
+        )
+        ledger_jti = verifier.verify(task_token, node=node).jti
+        runner_app = create_runner_app(
+            verifier=verifier,
+            ledger=PostgresTaskLedger(platform),
+            registry=NodeRegistry({"rule.apply": executor}),
+        )
+        with runner_app.test_client() as client:
+            http_result = client.post(
+                "/v1/tasks/execute",
+                json={
+                    "task_token": task_token,
+                    "node": node,
+                    "parameters": {},
+                },
+            )
+            replay = client.post(
+                "/v1/tasks/execute",
+                json={
+                    "task_token": task_token,
+                    "node": node,
+                    "parameters": {},
+                },
+            )
+        assert http_result.status_code == 200
+        assert http_result.get_json()["result"]["artifact_ref"] == result[
+            "artifact_ref"
+        ]
+        assert replay.status_code == 409
+        ledger_record = PostgresTaskLedger(platform).get(ledger_jti)
+        assert ledger_record is not None
+        assert ledger_record.status == "success"
+        assert ledger_record.commit_outcome == "committed"
+        conflict = store.write(
+            pl.DataFrame(
+                {
+                    "customer_id": [99],
+                    "mobile": ["13800138000"],
+                    "name": ["Conflict"],
+                    "segment_code": ["A"],
+                    "segment_name": ["Gold"],
+                    "version_no": [1],
+                }
+            ),
+            correlation_id,
+            900,
+            schema_fields=output_schema["fields"],
+            limits=compiled["plan"]["resource_limits"],
+        )
+        with pytest.raises(ValueError, match="immutable|digest"):
+            resolver.register(
+                binding_id=output_binding["id"],
+                binding_hash=compiled["plan"]["output_binding_hash"],
+                correlation_id=correlation_id,
+                artifact=conflict,
+                kind="output",
+            )
+        assert not any(
+            item.object_name
+            == conflict["artifact_ref"].split("/", 3)[-1]
+            for item in minio.list_objects(
+                bucket, prefix=prefix, recursive=True
+            )
+        )
         with platform.connect() as connection:
-            catalog_count = connection.execute(
+            catalog_rows = connection.execute(
                 text(
                     """
-                    SELECT COUNT(*)
+                    SELECT artifact_kind, artifact_ref
                     FROM public.rule_run_artifacts
                     WHERE correlation_id = CAST(:correlation_id AS uuid)
+                    ORDER BY artifact_kind
                     """
                 ),
                 {"correlation_id": correlation_id},
-            ).scalar_one()
+            ).mappings().all()
         # The repeated deterministic output has the same digest and is
         # idempotently retained as one stable catalog handoff.
-        assert catalog_count == 3
+        assert len(catalog_rows) == 3
+        assert next(
+            row["artifact_ref"]
+            for row in catalog_rows
+            if row["artifact_kind"] == "output"
+        ) == result["artifact_ref"]
+        assert len(
+            list(minio.list_objects(bucket, prefix=prefix, recursive=True))
+        ) == 3
     finally:
         for item in list(
             minio.list_objects(bucket, prefix=prefix, recursive=True)
@@ -616,6 +708,14 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
         with mysql.begin() as connection:
             connection.execute(text(f"DROP TABLE IF EXISTS {segment_table}"))
         with platform.begin() as connection:
+            if ledger_jti is not None:
+                connection.execute(
+                    text(
+                        "DELETE FROM public.runner_task_executions "
+                        "WHERE token_jti = CAST(:jti AS uuid)"
+                    ),
+                    {"jti": ledger_jti},
+                )
             connection.execute(
                 text(
                     "DELETE FROM public.rule_run_artifacts "

tests/runner/test_artifacts.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import io
+import json
 from datetime import UTC, datetime
 from types import SimpleNamespace
 
@@ -112,8 +113,27 @@ def test_artifact_store_generates_key_and_round_trips_digest_bound_lazyframe():
     assert artifact["schema_hash"]
     assert artifact["expires_at"].endswith("Z")
     assert "dataops-test" not in repr(artifact)
-    assert store.describe(artifact["artifact_ref"]) == artifact
-    frame = store.read(artifact["artifact_ref"], artifact["digest"])
+    assert store.describe(artifact["artifact_ref"]) == {
+        key: artifact[key]
+        for key in (
+            "artifact_ref",
+            "digest",
+            "row_count",
+            "schema_hash",
+            "expires_at",
+        )
+    }
+    stored = next(iter(client.objects.values()))
+    assert "x-amz-meta-schema-contract" not in stored["metadata"]
+    assert sum(
+        len(key) + len(value)
+        for key, value in stored["metadata"].items()
+    ) <= 2_048
+    frame = store.read(
+        artifact["artifact_ref"],
+        artifact["digest"],
+        expected_schema_fields=artifact["schema_fields"],
+    )
     assert isinstance(frame, pl.LazyFrame)
     assert frame.collect().to_dicts() == [
         {"customer_id": 1, "name": "Alice"},
@@ -163,7 +183,11 @@ def test_artifact_store_rejects_tampered_digest_schema_ttl_and_content(
     mutation(client.objects[("dataops-rules", key)])
 
     with pytest.raises(ValueError, match=message):
-        store.read(artifact["artifact_ref"], artifact["digest"])
+        store.read(
+            artifact["artifact_ref"],
+            artifact["digest"],
+            expected_schema_fields=artifact["schema_fields"],
+        )
 
 
 def test_artifact_store_rejects_rows_size_ttl_and_unowned_references():
@@ -186,6 +210,9 @@ def test_artifact_store_rejects_rows_size_ttl_and_unowned_references():
         store.read(
             "minio://other-bucket/rules/unsafe/value.parquet",
             "0" * 64,
+            expected_schema_fields=[
+                {"name": "id", "type": "integer", "nullable": True}
+            ],
         )
 
 
@@ -224,6 +251,7 @@ def test_artifact_store_applies_plan_limits_before_download_and_serialization():
         store.read(
             artifact["artifact_ref"],
             artifact["digest"],
+            expected_schema_fields=artifact["schema_fields"],
             limits={
                 "max_rows": 1,
                 "max_artifact_bytes": 1024 * 1024,
@@ -245,6 +273,74 @@ def test_artifact_store_applies_plan_limits_before_download_and_serialization():
         )
 
 
+def test_artifact_stage_streams_to_tempfile_and_preflights_parquet_footer(
+):
+    import inspect
+
+    from app.runner.artifacts import ArtifactStore
+
+    client = FakeMinio()
+    store = _store(client)
+    correlation_id = new_governance_uid()
+    artifact = store.write(
+        pl.DataFrame({"id": [1, 2]}),
+        correlation_id,
+        60,
+    )
+
+    assert "BytesIO" not in inspect.getsource(ArtifactStore.stage)
+
+    with store.stage(
+        artifact["artifact_ref"],
+        artifact["digest"],
+        expected_schema_fields=artifact["schema_fields"],
+    ) as staged:
+        assert pl.scan_parquet(staged).collect().height == 2
+
+
+def test_artifact_stage_rejects_compression_bomb_from_footer_before_scan():
+    from app.runner.artifacts import ArtifactStore
+
+    client = FakeMinio()
+    store = ArtifactStore(
+        client,
+        bucket="dataops-rules",
+        max_artifact_bytes=4 * 1024 * 1024,
+        max_rows=1_000,
+        memory_limit_bytes=64 * 1024 * 1024,
+        max_ttl_seconds=3600,
+    )
+    correlation_id = new_governance_uid()
+    artifact = store.write(
+        pl.DataFrame(
+            {
+                "payload": [
+                    f"{index}-" + ("compressible-value-" * 10_000)
+                    for index in range(100)
+                ]
+            }
+        ),
+        correlation_id,
+        60,
+    )
+    calls_before = len(client.get_calls)
+
+    with pytest.raises(
+        ValueError, match="uncompressed|footer"
+    ), store.stage(
+        artifact["artifact_ref"],
+        artifact["digest"],
+        expected_schema_fields=artifact["schema_fields"],
+        limits={
+            "max_rows": 1_000,
+            "max_artifact_bytes": 4 * 1024 * 1024,
+            "memory_limit_bytes": 1024 * 1024,
+        },
+    ):
+        raise AssertionError("compression bomb must not be exposed")
+    assert len(client.get_calls) == calls_before + 1
+
+
 def test_artifact_schema_contract_covers_nullability_decimal_and_timezone():
     client = FakeMinio()
     store = _store(client)
@@ -346,6 +442,11 @@ class _Rows:
     def one_or_none(self):
         return self.row
 
+    def all(self):
+        if self.row is None:
+            return []
+        return self.row if isinstance(self.row, list) else [self.row]
+
 
 class _Connection:
     def __init__(self, engine):
@@ -401,6 +502,7 @@ def test_postgres_artifact_resolver_uses_catalog_and_rechecks_binding_hash():
     resolved = PostgresArtifactResolver(engine, store).resolve(
         binding_id=binding_id,
         correlation_id=correlation_id,
+        kind="input",
     )
 
     assert resolved["artifact_ref"] == artifact["artifact_ref"]
@@ -414,6 +516,7 @@ def test_postgres_artifact_resolver_uses_catalog_and_rechecks_binding_hash():
         PostgresArtifactResolver(engine, store).resolve(
             binding_id=binding_id,
             correlation_id=new_governance_uid(),
+            kind="input",
         )
 
 
@@ -440,7 +543,14 @@ def test_postgres_artifact_resolver_attests_and_registers_stable_handoff():
         if "INSERT INTO public.rule_run_artifacts" in sql:
             assert parameters["artifact_digest"] == artifact["digest"]
             assert parameters["schema_fields"]
-            return None
+            return {
+                "artifact_ref": parameters["artifact_ref"],
+                "artifact_digest": parameters["artifact_digest"],
+                "row_count": parameters["row_count"],
+                "schema_hash": parameters["schema_hash"],
+                "schema_fields": json.loads(parameters["schema_fields"]),
+                "expires_at": parameters["expires_at"],
+            }
         raise AssertionError(sql)
 
     engine = _Engine(handler)
@@ -466,3 +576,129 @@ def test_postgres_artifact_resolver_attests_and_registers_stable_handoff():
         "INSERT INTO public.rule_run_artifacts" in sql
         for sql, _parameters in engine.calls
     )
+
+
+def test_artifact_catalog_reuses_same_digest_and_rejects_digest_conflict():
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    store = _store(FakeMinio())
+    correlation_id = new_governance_uid()
+    binding_id = new_governance_uid()
+    binding_hash = "c" * 64
+    catalog = {}
+
+    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:
+            key = (
+                parameters["correlation_id"],
+                parameters["binding_id"],
+                parameters["artifact_kind"],
+            )
+            if key in catalog:
+                return None
+            catalog[key] = {
+                "artifact_ref": parameters["artifact_ref"],
+                "artifact_digest": parameters["artifact_digest"],
+                "row_count": parameters["row_count"],
+                "schema_hash": parameters["schema_hash"],
+                "schema_fields": json.loads(parameters["schema_fields"]),
+                "expires_at": parameters["expires_at"],
+            }
+            return catalog[key]
+        if "FROM public.rule_run_artifacts" in sql:
+            key = (
+                parameters["correlation_id"],
+                parameters["binding_id"],
+                parameters["artifact_kind"],
+            )
+            return catalog.get(key)
+        raise AssertionError(sql)
+
+    engine = _Engine(handler)
+    resolver = PostgresArtifactResolver(engine, store)
+    first = store.write(
+        pl.DataFrame({"id": [1]}),
+        correlation_id,
+        60,
+    )
+    first_registered = resolver.register(
+        binding_id=binding_id,
+        correlation_id=correlation_id,
+        artifact=first,
+        kind="output",
+        binding_hash=binding_hash,
+    )
+    retry = store.write(
+        pl.DataFrame({"id": [1]}),
+        correlation_id,
+        60,
+    )
+    retry_registered = resolver.register(
+        binding_id=binding_id,
+        correlation_id=correlation_id,
+        artifact=retry,
+        kind="output",
+        binding_hash=binding_hash,
+    )
+
+    assert retry_registered["artifact_ref"] == first_registered["artifact_ref"]
+    assert retry["artifact_ref"] != first_registered["artifact_ref"]
+    assert (
+        "dataops-rules",
+        retry["artifact_ref"].split("/", 3)[-1],
+    ) in store.client.removed
+
+    conflict = store.write(
+        pl.DataFrame({"id": [2]}),
+        correlation_id,
+        60,
+    )
+    with pytest.raises(ValueError, match="immutable|digest"):
+        resolver.register(
+            binding_id=binding_id,
+            correlation_id=correlation_id,
+            artifact=conflict,
+            kind="output",
+            binding_hash=binding_hash,
+        )
+    assert (
+        "dataops-rules",
+        conflict["artifact_ref"].split("/", 3)[-1],
+    ) in store.client.removed
+    assert len(catalog) == 1
+
+
+def test_catalog_cleanup_deletes_expired_object_and_directory_row():
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    store = _store(FakeMinio())
+    correlation_id = new_governance_uid()
+    artifact = store.write(
+        pl.DataFrame({"id": [1]}),
+        correlation_id,
+        60,
+    )
+    row_id = new_governance_uid()
+    deleted_ids = []
+
+    def handler(sql, parameters):
+        if "DELETE FROM public.rule_run_artifacts" in sql:
+            deleted_ids.append(parameters["id"])
+            return None
+        if "FROM public.rule_run_artifacts" in sql:
+            assert "FOR UPDATE SKIP LOCKED" in sql
+            assert parameters["limit"] == 10
+            return [{"id": row_id, "artifact_ref": artifact["artifact_ref"]}]
+        raise AssertionError(sql)
+
+    resolver = PostgresArtifactResolver(_Engine(handler), store)
+
+    assert resolver.cleanup_expired(limit=10) == 1
+    assert deleted_ids == [row_id]
+    assert not store.client.objects

tests/runner/test_bootstrap.py

@@ -1,6 +1,10 @@
 import pytest
+from flask import Flask
 
-from app.runner.bootstrap import runner_settings_from_env
+from app.runner.bootstrap import (
+    register_artifact_cleanup_cli,
+    runner_settings_from_env,
+)
 
 
 def test_runner_settings_require_signing_secret_and_explicit_dependencies(
@@ -45,3 +49,25 @@ def test_runner_refuses_to_start_without_a_task_token_secret(monkeypatch):
 
     with pytest.raises(ValueError, match="RUNNER_TASK_TOKEN_SECRET"):
         runner_settings_from_env()
+
+
+def test_runner_registers_bounded_artifact_cleanup_cli():
+    class Resolver:
+        def __init__(self):
+            self.limit = None
+
+        def cleanup_expired(self, *, limit):
+            self.limit = limit
+            return 3
+
+    resolver = Resolver()
+    app = Flask("runner-cleanup-test")
+    register_artifact_cleanup_cli(app, resolver)
+
+    result = app.test_cli_runner().invoke(
+        args=["cleanup-rule-artifacts", "--limit", "25"]
+    )
+
+    assert result.exit_code == 0
+    assert result.output.strip() == "removed 3 expired rule artifacts"
+    assert resolver.limit == 25

tests/runner/test_polars_worker.py

@@ -0,0 +1,231 @@
+from __future__ import annotations
+
+import os
+
+import polars as pl
+import pytest
+
+from tests.core.data_rules.test_polars_compiler import (
+    _backend,
+    _binding,
+    _published_rule,
+    _schema,
+)
+
+
+def _compile_worker_plan(
+    *,
+    input_schema,
+    output_schema,
+    steps,
+    lookup_context=None,
+):
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    input_binding = _binding(input_schema, access_mode="read")
+    output_binding = _binding(output_schema, access_mode="write")
+    backend = _backend(
+        max_rows=1_000_000,
+        max_artifact_bytes=256 * 1024 * 1024,
+        memory_limit_bytes=16 * 1024 * 1024,
+        lookup_bindings=lookup_context or {},
+    )
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        steps,
+    )
+    return PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=input_binding,
+        output_binding=output_binding,
+        backend=backend,
+    )["plan"]
+
+
+def _assert_worker_resource_failure(plan, tmp_path, *, lookup_paths=None):
+    from app.runner.polars_worker import (
+        PolarsWorkerResourceError,
+        execute_isolated_polars_plan,
+    )
+
+    with pytest.raises(
+        PolarsWorkerResourceError,
+        match="hard memory limit",
+    ):
+        execute_isolated_polars_plan(
+            {
+                "plan": plan,
+                "input_path": str(tmp_path / "input.parquet"),
+                "lookup_paths": lookup_paths or {},
+                "output_path": str(tmp_path / "output.parquet"),
+                "masking_policies": {},
+            },
+            memory_limit_bytes=plan["resource_limits"][
+                "memory_limit_bytes"
+            ],
+        )
+
+
+def test_isolated_polars_worker_runs_outside_runner_process():
+    from app.runner.polars_worker import run_isolated_memory_probe
+
+    result = run_isolated_memory_probe(
+        allocate_bytes=1_024,
+        memory_limit_bytes=16 * 1024 * 1024,
+    )
+
+    assert result["worker_pid"] != os.getpid()
+    assert result["allocated_bytes"] == 1_024
+
+
+def test_isolated_polars_worker_fails_deterministically_at_hard_memory_limit():
+    from app.runner.polars_worker import (
+        PolarsWorkerResourceError,
+        run_isolated_memory_probe,
+    )
+
+    with pytest.raises(
+        PolarsWorkerResourceError,
+        match="hard memory limit",
+    ):
+        run_isolated_memory_probe(
+            allocate_bytes=128 * 1024 * 1024,
+            memory_limit_bytes=8 * 1024 * 1024,
+        )
+
+
+def test_regex_peak_allocation_fails_inside_isolated_worker(tmp_path):
+    schema = _schema(
+        "bd:regex:raw",
+        [("text", "string", False)],
+    )
+    plan = _compile_worker_plan(
+        input_schema=schema,
+        output_schema={
+            **schema,
+            "id": _schema(
+                "bd:regex:clean",
+                [("text", "string", False)],
+            )["id"],
+            "schema_ref": "bd:regex:clean",
+        },
+        steps=[
+            {
+                "id": "expand",
+                "op": "regex_replace",
+                "column": "text",
+                "pattern": "a",
+                "replacement": "x" * 64,
+            }
+        ],
+    )
+    pl.DataFrame({"text": ["a" * 1_000_000]}).write_parquet(
+        tmp_path / "input.parquet"
+    )
+
+    _assert_worker_resource_failure(plan, tmp_path)
+
+
+def test_group_peak_allocation_fails_inside_isolated_worker(tmp_path):
+    input_schema = _schema(
+        "bd:group:raw",
+        [("group_key", "string", False), ("amount", "integer", False)],
+    )
+    output_schema = _schema(
+        "bd:group:clean",
+        [
+            ("group_key", "string", False),
+            ("total_amount", "integer", False),
+        ],
+    )
+    plan = _compile_worker_plan(
+        input_schema=input_schema,
+        output_schema=output_schema,
+        steps=[
+            {
+                "id": "sum_by_key",
+                "op": "aggregate",
+                "group_by": ["group_key"],
+                "aggregations": {
+                    "total_amount": {
+                        "function": "sum",
+                        "column": "amount",
+                    }
+                },
+            }
+        ],
+    )
+    row_count = 300_000
+    pl.DataFrame(
+        {
+            "group_key": [
+                f"group-{index:040d}" for index in range(row_count)
+            ],
+            "amount": [1] * row_count,
+        }
+    ).write_parquet(tmp_path / "input.parquet")
+
+    _assert_worker_resource_failure(plan, tmp_path)
+
+
+def test_join_peak_allocation_fails_inside_isolated_worker(tmp_path):
+    input_schema = _schema(
+        "bd:join:raw",
+        [("id", "integer", False)],
+    )
+    lookup_schema = _schema(
+        "bd:join:lookup",
+        [("lookup_id", "integer", False), ("label", "string", False)],
+    )
+    output_schema = _schema(
+        "bd:join:clean",
+        [("id", "integer", False), ("label", "string", False)],
+    )
+    lookup_binding = _binding(
+        lookup_schema,
+        access_mode="read",
+        object_ref="join-lookup",
+    )
+    plan = _compile_worker_plan(
+        input_schema=input_schema,
+        output_schema=output_schema,
+        steps=[
+            {
+                "id": "join_label",
+                "op": "lookup_join",
+                "lookup": {
+                    "binding_id": lookup_binding["id"],
+                    "left_on": ["id"],
+                    "right_on": ["lookup_id"],
+                    "select": {"label": "label"},
+                    "how": "left",
+                },
+            }
+        ],
+        lookup_context={
+            lookup_binding["id"]: {
+                "binding": lookup_binding,
+                "schema": lookup_schema,
+            }
+        },
+    )
+    row_count = 250_000
+    pl.DataFrame({"id": range(row_count)}).write_parquet(
+        tmp_path / "input.parquet"
+    )
+    lookup_path = tmp_path / "lookup.parquet"
+    pl.DataFrame(
+        {
+            "lookup_id": range(row_count),
+            "label": [f"label-{index:048d}" for index in range(row_count)],
+        }
+    ).write_parquet(lookup_path)
+
+    _assert_worker_resource_failure(
+        plan,
+        tmp_path,
+        lookup_paths={lookup_binding["id"]: str(lookup_path)},
+    )

tests/runner/test_rule_polars.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import copy
+from datetime import date, datetime
 
 import polars as pl
 import pytest
@@ -24,7 +25,7 @@ def _plan_store(client):
         bucket="dataops-rules",
         max_artifact_bytes=8 * 1024 * 1024,
         max_rows=10_000,
-        memory_limit_bytes=32 * 1024 * 1024,
+        memory_limit_bytes=256 * 1024 * 1024,
         max_ttl_seconds=3600,
     )
 
@@ -37,9 +38,9 @@ class Resolver:
         self.attest_error = None
         self.registrations = []
 
-    def resolve(self, *, binding_id, correlation_id):
+    def resolve(self, *, binding_id, correlation_id, kind):
         self.calls.append((binding_id, correlation_id))
-        self.events.append(("resolve", binding_id))
+        self.events.append(("resolve", binding_id, kind))
         return self.values[binding_id]
 
     def attest_binding(self, *, binding_id, binding_hash, access_mode):
@@ -67,6 +68,7 @@ class Resolver:
                 "binding_hash": binding_hash,
             }
         )
+        return artifact
 
 
 def _compiled_plan(steps):
@@ -88,7 +90,7 @@ def _compiled_plan(steps):
         output_schema=output_schema,
         input_binding=input_binding,
         output_binding=output_binding,
-        backend=_backend(),
+        backend=_backend(memory_limit_bytes=128 * 1024 * 1024),
     )
     return compiled, input_binding
 
@@ -189,6 +191,7 @@ def test_polars_adapter_reconstructs_assert_and_deduplicate_and_writes_artifact(
         {"step_id": "mobile_format", "count": 1}
     ]
     assert result["commit_outcome"] == "committed"
+    assert "schema_fields" not in result
     assert resolver.events[0] == (
         "attest",
         compiled["plan"]["output_binding_id"],
@@ -400,7 +403,7 @@ def test_polars_adapter_uses_exact_decimal_and_timestamptz_output_contracts():
         output_schema=output_schema,
         input_binding=source_binding,
         output_binding=output_binding,
-        backend=_backend(),
+        backend=_backend(memory_limit_bytes=128 * 1024 * 1024),
     )
     store = _plan_store(FakeMinio())
     correlation_id = new_governance_uid()
@@ -448,6 +451,107 @@ def test_polars_adapter_uses_exact_decimal_and_timestamptz_output_contracts():
     )
 
 
+def test_polars_expression_date_and_timestamp_use_plan_timezone():
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+    from app.runner.rule_polars import PolarsRulePlanAdapter
+
+    input_schema = _schema(
+        "bd:event:raw",
+        [
+            ("raw_time", "string", False),
+            ("cast_time", "string", False),
+        ],
+    )
+    output_schema = _schema(
+        "bd:event:clean",
+        [
+            ("raw_time", "string", False),
+            ("cast_time", "timestamp", False),
+            ("local_date", "date", False),
+            ("local_time", "timestamp", False),
+        ],
+    )
+    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_timestamp",
+                "op": "cast",
+                "column": "cast_time",
+                "to": "timestamp",
+                "on_error": "fail",
+            },
+            {
+                "id": "derive_date",
+                "op": "derive",
+                "target": "local_date",
+                "expression": "date(raw_time)",
+            },
+            {
+                "id": "derive_timestamp",
+                "op": "derive",
+                "target": "local_time",
+                "expression": "timestamp(raw_time)",
+            },
+        ],
+    )
+    compiled = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=source_binding,
+        output_binding=output_binding,
+        backend=_backend(memory_limit_bytes=128 * 1024 * 1024),
+    )
+    store = _plan_store(FakeMinio())
+    correlation_id = new_governance_uid()
+    source = store.write(
+        pl.DataFrame(
+            {
+                "raw_time": ["2026-07-22T16:30:00+00:00"],
+                "cast_time": ["2026-07-22T16:30:00+00: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["local_date"].item() == date(2026, 7, 23)
+    assert output["local_time"].item() == datetime(2026, 7, 23, 0, 30)
+    assert output["cast_time"].item() == datetime(2026, 7, 23, 0, 30)
+    assert output.schema["local_time"] == pl.Datetime
+
+
 def test_rule_executor_attests_polars_canonical_hashes_and_forwards_correlation():
     from app.runner.rules import RulePlanExecutor
 

tests/test_data_rule_runtime_sbom.py

@@ -1,10 +1,65 @@
 from __future__ import annotations
 
+import hashlib
 import json
 from pathlib import Path
 
+from jsonschema import Draft7Validator
+from referencing import Registry, Resource
+
 ROOT = Path(__file__).resolve().parents[1]
 SBOM = ROOT / "docs" / "security" / "data-rule-runtime-sbom.json"
+SCHEMAS = ROOT / "docs" / "security" / "cyclonedx-1.5-schema"
+
+
+def _load(name: str):
+    return json.loads((SCHEMAS / name).read_text(encoding="utf-8"))
+
+
+def test_data_rule_runtime_sbom_validates_against_cyclonedx_15_schema():
+    expected_schema_digests = {
+        "bom-1.5.schema.json": (
+            "a00fcba23a44b72179ac1f288be4c3529"
+            "b59f6e1b3719709a40685f177516b46"
+        ),
+        "spdx.schema.json": (
+            "c87aa7bb5eb503d40b52ec6bf00de8045"
+            "df15da7a13cea48d290cf6d36a8d2ea"
+        ),
+        "jsf-0.82.schema.json": (
+            "2faf5eb3651f2ae5f46091a131770d8d8"
+            "47bbd121139d19c85fc7051bfa58c46"
+        ),
+    }
+    for name, expected in expected_schema_digests.items():
+        assert hashlib.sha256((SCHEMAS / name).read_bytes()).hexdigest() == (
+            expected
+        )
+    document = json.loads(SBOM.read_text(encoding="utf-8"))
+    schema = _load("bom-1.5.schema.json")
+    registry = Registry().with_resources(
+        [
+            (
+                "http://cyclonedx.org/schema/spdx.schema.json",
+                Resource.from_contents(_load("spdx.schema.json")),
+            ),
+            (
+                "http://cyclonedx.org/schema/jsf-0.82.schema.json",
+                Resource.from_contents(_load("jsf-0.82.schema.json")),
+            ),
+        ]
+    )
+
+    Draft7Validator.check_schema(schema)
+    errors = sorted(
+        Draft7Validator(
+            schema,
+            registry=registry,
+        ).iter_errors(document),
+        key=lambda error: list(error.absolute_path),
+    )
+
+    assert errors == []
 
 
 def test_data_rule_runtime_sbom_pins_official_oss_dependencies():
@@ -16,24 +71,61 @@ def test_data_rule_runtime_sbom_pins_official_oss_dependencies():
 
     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",
+    assert set(components) == {
+        "minio",
+        "polars",
+        "polars-runtime-32",
+        "psutil",
+        "pyarrow",
     }
+    for name, version, license_id, source in (
+        (
+            "polars",
+            "1.42.1",
+            "MIT",
+            "https://github.com/pola-rs/polars",
+        ),
+        (
+            "polars-runtime-32",
+            "1.42.1",
+            "MIT",
+            "https://github.com/pola-rs/polars",
+        ),
+        (
+            "pyarrow",
+            "21.0.0",
+            "Apache-2.0",
+            "https://github.com/apache/arrow",
+        ),
+        (
+            "psutil",
+            "5.9.8",
+            "BSD-3-Clause",
+            "https://github.com/giampaolo/psutil",
+        ),
+        (
+            "minio",
+            "7.2.10",
+            "Apache-2.0",
+            "https://github.com/minio/minio-py",
+        ),
+    ):
+        component = components[name]
+        assert component["version"] == version
+        assert component["licenses"] == [
+            {"license": {"id": license_id}}
+        ]
+        assert component["externalReferences"] == [
+            {"type": "vcs", "url": source}
+        ]
+        assert component["properties"][0]["name"] == "dataops:purpose"
+        assert not {"license", "officialSource", "purpose"} & set(component)
+
     requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8")
-    assert "polars==1.42.1" in requirements
-    assert "minio==7.2.10" in requirements
+    for requirement in (
+        "polars==1.42.1",
+        "pyarrow==21.0.0",
+        "psutil==5.9.8",
+        "minio==7.2.10",
+    ):
+        assert requirement in requirements

tests/test_data_rule_schema.py

@@ -148,9 +148,18 @@ def test_rule_run_artifact_catalog_is_correlation_scoped_and_forward_preserving(
         "artifact_digest CHAR(64) NOT NULL",
         "schema_fields JSONB NOT NULL",
         "expires_at TIMESTAMPTZ NOT NULL",
-        "UNIQUE (correlation_id, binding_id, artifact_digest)",
+        "UNIQUE (correlation_id, binding_id, artifact_kind)",
     ):
         assert expected in source
     downgrade = source.split("def downgrade()", 1)[1]
     assert "DROP TABLE" not in downgrade.upper()
-    assert "pass" in downgrade
+    assert "raise RuntimeError" in downgrade
+    spec = importlib.util.spec_from_file_location(
+        "rule_run_artifact_catalog_migration",
+        ARTIFACT_CATALOG_MIGRATION,
+    )
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    with pytest.raises(RuntimeError, match="forward-only|cannot downgrade"):
+        module.downgrade()