fix: make Polars artifact handoff durable

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

@@ -2,7 +2,9 @@
 
 Date: 2026-07-23
 Branch: `codex/data-rule-execution-m3a-m5`
-Implementation commits: `69d83ed`, `f4d798b`, `8bffdb6`
+Implementation commits before the third remediation: `69d83ed`, `f4d798b`,
+`8bffdb6`, `c152e74`. The third-remediation commit is the commit containing
+this report revision.
 
 `8bffdb6` is the final hash of the second remediation commit. It replaces the
 intermediate `a17bcd0` after restoring the historical M3A acceptance document
@@ -33,9 +35,11 @@ The final implementation includes:
   group/aggregate, counts, and output write in an isolated spawn child;
 - parent RSS monitoring, child OS `RLIMIT_AS`, bounded execution time, and
   fail-closed worker errors;
-- immutable catalog handoff keyed by
-  `(correlation_id, binding_id, artifact_kind)`;
-- production cleanup that removes the expired object and its catalog row;
+- durable `pending` → object upload → `ready` catalog handoff keyed by
+  `(correlation_id, binding_id, artifact_kind)`, with the exact attested
+  `binding_hash`;
+- bounded bidirectional reconciliation for pending, ready, missing, invalid,
+  and orphaned objects, plus the existing expiry cleanup;
 - minimal public artifact results without full schema fields;
 - standards-valid CycloneDX 1.5 runtime SBOM with deterministic offline schema
   validation.
@@ -93,39 +97,100 @@ Second-remediation GREEN:
 - `git diff --check`: passed.
 - `docker compose -f deploy/docker/docker-compose.yml config -q`: passed.
 
-## Immutable artifact handoff, TTL, and cleanup
-
-Migration `20260723_140` now enforces:
+The third remediation supersedes the second-remediation migration and
+handoff implementation described below wherever they conflict.
+
+## Third-remediation RED/GREEN evidence
+
+The third review was also implemented in fail-first slices:
+
+- schema source tests failed until committed migration 140 was restored
+  byte-for-byte from `f4d798b` and the state change moved to new migration
+  `20260723_150`;
+- reservation-order tests failed until the resolver inserted a durable
+  `pending` row before the first MinIO PUT;
+- crash-injection tests failed until reserve, upload, and finalize had distinct
+  failure semantics and commit-acknowledgement loss was rechecked;
+- stale-binding tests failed until reservation and finalization locked and
+  attested the canonical binding in the same database transaction;
+- reconciliation tests failed until missing pending rows, valid pending
+  objects, missing ready objects, invalid objects, old orphans, fresh objects,
+  and unsafe keys had separate bounded policies;
+- worker-start injection initially leaked raw startup errors and pipe/process
+  handles; it now returns a safe error and closes every created endpoint in a
+  unified `finally`.
+
+Third-remediation GREEN:
+
+- focused schema/artifact/handoff/bootstrap/worker/adapter plus both real
+  integrations: `49 passed in 5.62s`;
+- isolated temporary PostgreSQL migration test: real old 140 schema, old-format
+  row insertion, upgrade to head, migrated `ready` row and `binding_hash`,
+  same-digest retry, different-digest CAS conflict: passed;
+- real source PostgreSQL + source MySQL + platform PostgreSQL + MinIO +
+  isolated Polars + Flask HTTP + durable ledger: passed;
+- final full suite: `564 passed, 26 skipped, 59 subtests passed in 9.01s`;
+- Ruff across every changed Python file: passed;
+- `git diff --check`: passed;
+- `docker compose -f deploy/docker/docker-compose.yml config -q`: passed;
+- `git diff f4d798b -- migrations/versions/20260723_140_rule_run_artifacts.py`:
+  empty.
+
+## Durable artifact handoff, attestation, and reconciliation
+
+Migration `20260723_140` is historical and unchanged. It retains its original
+unique constraint:
 
 ```text
-UNIQUE (correlation_id, binding_id, artifact_kind)
+UNIQUE (correlation_id, binding_id, artifact_digest)
 ```
 
-Its downgrade raises `RuntimeError` because removing the catalog would violate
-immutable runtime handoff evidence.
-
-`PostgresArtifactResolver.register` implements the retry invariant:
-
-- first writer inserts and returns the new catalog artifact;
-- same triple + same digest reuses the existing catalog ref and deletes the
-  newly generated duplicate object;
-- same triple + different digest fails closed and deletes the new object;
-- existing catalog metadata is rechecked against the retained store object.
-
-Resolution requires the exact artifact kind, current binding hash, matching
-correlation prefix, valid TTL, digest, row count, and schema hash. The
-integration asserts one input, one lookup, and one output row/object after
-repeated deterministic execution.
+New forward-only migration `20260723_150`:
+
+- adds `binding_hash`, `handoff_status`, `ready_at`, `failed_at`,
+  `failure_code`, and `updated_at`;
+- migrates old rows to `ready` with the current canonical binding hash;
+- refuses unattested or conflicting old rows;
+- discovers and drops only the exact old-140 digest uniqueness constraint;
+- adds named uniqueness on
+  `(correlation_id, binding_id, artifact_kind)` and a reconciliation index.
+
+`PostgresArtifactResolver.publish_path` implements the durable protocol:
+
+1. validate the worker path and reserve a server-generated exact object key;
+2. in one transaction, lock the canonical binding with `FOR SHARE`, attest its
+   hash/object/access contract, and insert the exact `pending` catalog row;
+3. upload only to that reserved key and revalidate the stored object;
+4. in a new transaction, re-lock and re-attest the binding, then atomically
+   finalize only the matching pending row to `ready`.
+
+Resolution sees only `ready` rows whose persisted binding hash still equals the
+canonical binding hash. Same-triple/same-digest retries reuse the ready object
+without uploading; different digests fail before upload. Upload failure removes
+the exact object and pending row. A lost finalize acknowledgement is rechecked:
+confirmed ready is success; unconfirmed state returns
+`commit_outcome="unknown"` and deliberately does not delete the possibly
+cataloged object.
+
+`reconcile-rule-artifacts --limit --grace-seconds` processes only aged,
+bounded candidates. It deletes missing pending rows, finalizes valid pending
+objects only while the binding still matches, marks ready rows failed when
+their object is missing or invalid, deletes invalid pending objects safely, and
+deletes only old safe-key orphans after the grace period. Fresh objects,
+referenced objects, and malformed/unowned keys are retained.
 
 `cleanup-rule-artifacts --limit` uses a bounded
 `FOR UPDATE SKIP LOCKED` selection, removes each expired store object, and
 deletes its exact catalog row in the same database transaction scope. Unit
 tests cover the CLI and object+row cleanup behavior.
 
-The local database had an earlier applied copy of migration 140. Before real
-integration, its artifact table was verified empty and its old unique
-constraint was aligned to the final triple invariant. Migration source tests
-verify the clean-install definition.
+The local database had an earlier applied empty copy of migration 140 with a
+manually altered constraint. It was explicitly not accepted as migration
+evidence. After verifying zero rows, only that empty table was rebuilt by
+rewinding its Alembic marker to 130 and running the formal
+130 → original-140 → 150 chain. Independently, the automated acceptance test
+creates and drops a fresh temporary database and proves old140 → head without
+manual ALTER.
 
 ## Resource enforcement and worker isolation
 
@@ -163,6 +228,11 @@ The parent independently watches incremental RSS with psutil, kills the child
 on limit breach, enforces a hard timeout, and returns only bounded metrics.
 Worker PID is deliberately not exposed in the public result.
 
+Pipe creation, process construction, `process.start()`, message handling, and
+resource enforcement now share one guarded lifecycle. Startup errors are
+sanitized, both pipe endpoints are closed when created, and a started process
+is killed/joined/closed from the unified `finally`.
+
 Tests prove:
 
 - worker PID differs from the Runner PID;
@@ -215,7 +285,7 @@ Evidence:
 - 1 deduplicated row;
 - exact enriched values verified after reread;
 - direct repeat reuses the stable output ref;
-- conflicting output digest fails and its new object is absent;
+- conflicting output digest fails before a new object is uploaded;
 - HTTP result returns the same stable output ref;
 - replaying the same token returns HTTP 409;
 - durable ledger records `success` / `committed`;
@@ -258,6 +328,7 @@ Production:
 - `app/runner/rule_polars.py`
 - `app/runner/bootstrap.py`
 - `migrations/versions/20260723_140_rule_run_artifacts.py`
+- `migrations/versions/20260723_150_rule_artifact_handoff_state.py`
 - `requirements.txt`
 - `docs/security/data-rule-runtime-sbom.json`
 - `docs/security/cyclonedx-1.5-schema/`
@@ -266,10 +337,12 @@ Tests:
 
 - `tests/core/data_rules/test_polars_compiler.py`
 - `tests/runner/test_artifacts.py`
+- `tests/runner/test_artifact_handoff.py`
 - `tests/runner/test_polars_worker.py`
 - `tests/runner/test_rule_polars.py`
 - `tests/runner/test_bootstrap.py`
 - `tests/integration/test_data_rule_polars_execution.py`
+- `tests/integration/test_rule_artifact_migration_upgrade.py`
 - `tests/test_data_rule_runtime_sbom.py`
 - `tests/test_data_rule_schema.py`
 

app/runner/artifacts.py

@@ -14,6 +14,7 @@ from typing import Any
 
 import polars as pl
 import pyarrow.parquet as pq
+from minio.error import S3Error
 from sqlalchemy import text
 
 from app.core.common.identifiers import (
@@ -26,6 +27,14 @@ PARQUET_CONTENT_TYPE = "application/x-parquet"
 _DIGEST = re.compile(r"^[0-9a-f]{64}$")
 
 
+class ArtifactCommitUnknown(RuntimeError):
+    """A catalog transaction may have committed but cannot be confirmed."""
+
+
+class ArtifactHandoffPending(RuntimeError):
+    """Another publisher owns the durable pending handoff."""
+
+
 def _parquet_footer_bounds(path: str) -> tuple[int, int]:
     try:
         metadata = pq.ParquetFile(path).metadata
@@ -436,7 +445,7 @@ class ArtifactStore:
             "expires_at": expires_at,
         }
 
-    def write_path(
+    def prepare_path(
         self,
         path: str,
         correlation_id: str,
@@ -445,7 +454,7 @@ class ArtifactStore:
         schema_fields: list[dict[str, Any]],
         limits: dict[str, int] | None = None,
     ) -> dict[str, Any]:
-        """Upload a worker-produced Parquet file without collecting it."""
+        """Validate a local Parquet file and reserve its server-owned key."""
 
         effective = self._limits(limits)
         correlation = _uid(correlation_id, "correlation_id")
@@ -480,6 +489,63 @@ class ArtifactStore:
         )
         artifact_id = new_governance_uid()
         key = f"rules/{correlation}/{artifact_id}.parquet"
+        return {
+            "artifact_ref": f"minio://{self.bucket}/{key}",
+            "digest": digest_hex,
+            "row_count": rows,
+            "schema_hash": schema_digest,
+            "schema_fields": fields,
+            "expires_at": expires_at,
+        }
+
+    def upload_path(
+        self,
+        path: str,
+        artifact: dict[str, Any],
+        *,
+        limits: dict[str, int] | None = None,
+    ) -> dict[str, Any]:
+        """Upload to an already reserved exact key and verify the object."""
+
+        if not isinstance(artifact, dict):
+            raise ValueError("prepared artifact metadata is invalid")
+        required = {
+            "artifact_ref",
+            "digest",
+            "row_count",
+            "schema_hash",
+            "schema_fields",
+            "expires_at",
+        }
+        if set(artifact) != required:
+            raise ValueError("prepared artifact metadata has a closed shape")
+        key = self._parse_ref(artifact["artifact_ref"])
+        correlation = key.split("/", 2)[1]
+        ttl_seconds = int(
+            (
+                _parse_timestamp(artifact["expires_at"])
+                - _now_utc(self.clock)
+            ).total_seconds()
+        )
+        if ttl_seconds < 1 or ttl_seconds > self.max_ttl_seconds:
+            raise ValueError("prepared artifact TTL is invalid")
+        expected = self.prepare_path(
+            path,
+            correlation,
+            ttl_seconds,
+            schema_fields=artifact["schema_fields"],
+            limits=limits,
+        )
+        for name in (
+            "digest",
+            "row_count",
+            "schema_hash",
+            "schema_fields",
+        ):
+            if expected[name] != artifact[name]:
+                raise ValueError("prepared artifact no longer matches its path")
+        effective = self._limits(limits)
+        size = os.path.getsize(path)
         uploaded = False
         try:
             with open(path, "rb") as handle:
@@ -490,19 +556,18 @@ class ArtifactStore:
                     size,
                     content_type=PARQUET_CONTENT_TYPE,
                     metadata={
-                        "sha256": digest_hex,
-                        "row-count": str(rows),
-                        "schema-sha256": schema_digest,
-                        "expires-at": expires_at,
+                        "sha256": artifact["digest"],
+                        "row-count": str(artifact["row_count"]),
+                        "schema-sha256": artifact["schema_hash"],
+                        "expires-at": artifact["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,
+                artifact["artifact_ref"],
+                artifact["digest"],
+                expected_schema_fields=artifact["schema_fields"],
                 limits=effective,
             ):
                 pass
@@ -511,14 +576,27 @@ class ArtifactStore:
                 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,
-        }
+        return dict(artifact)
+
+    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]:
+        """Prepare and upload a non-cataloged compatibility artifact."""
+
+        artifact = self.prepare_path(
+            path,
+            correlation_id,
+            ttl_seconds,
+            schema_fields=schema_fields,
+            limits=limits,
+        )
+        return self.upload_path(path, artifact, limits=limits)
 
     def describe(self, ref: str) -> dict[str, Any]:
         """Return validated object metadata without exposing MinIO credentials."""
@@ -533,6 +611,18 @@ class ArtifactStore:
             "expires_at": metadata["expires-at"],
         }
 
+    def describe_optional(self, ref: str) -> dict[str, Any] | None:
+        """Return None only for a confirmed missing object."""
+
+        try:
+            return self.describe(ref)
+        except KeyError:
+            return None
+        except S3Error as exc:
+            if exc.code in {"NoSuchKey", "NoSuchObject", "NotFound"}:
+                return None
+            raise
+
     def read(
         self,
         ref: str,
@@ -692,14 +782,17 @@ class PostgresArtifactResolver:
                 a.schema_hash,
                 a.schema_fields,
                 a.expires_at,
-                b.binding_hash
+                a.binding_hash AS catalog_binding_hash,
+                b.binding_hash AS current_binding_hash
             FROM public.rule_run_artifacts a
             JOIN public.dataflow_dataset_bindings b
               ON b.id = a.binding_id
             WHERE a.binding_id = CAST(:binding_id AS uuid)
               AND a.correlation_id = CAST(:correlation_id AS uuid)
               AND a.artifact_kind = :artifact_kind
+              AND a.handoff_status = 'ready'
               AND a.expires_at > CURRENT_TIMESTAMP
+              AND a.binding_hash = b.binding_hash
               AND b.object_kind = 'parquet_artifact'
               AND b.access_mode IN ('read', 'read_write')
             ORDER BY a.created_at DESC
@@ -736,7 +829,7 @@ class PostgresArtifactResolver:
         return {
             **described,
             "schema_fields": _normalized_schema_fields(row_fields),
-            "binding_hash": str(row["binding_hash"]),
+            "binding_hash": str(row["catalog_binding_hash"]),
         }
 
     def attest_binding(
@@ -775,7 +868,88 @@ class PostgresArtifactResolver:
             raise ValueError("canonical artifact binding no longer matches")
         return {"binding_hash": str(row["binding_hash"])}
 
-    def register(
+    @staticmethod
+    def _catalog_artifact(row: Mapping[str, Any]) -> dict[str, Any]:
+        fields = row["schema_fields"]
+        if isinstance(fields, str):
+            fields = json.loads(fields)
+        expires_at = row["expires_at"]
+        return {
+            "artifact_ref": str(row["artifact_ref"]),
+            "digest": str(row["artifact_digest"]),
+            "row_count": int(row["row_count"]),
+            "schema_hash": str(row["schema_hash"]),
+            "schema_fields": _normalized_schema_fields(fields),
+            "expires_at": (
+                _timestamp(expires_at)
+                if isinstance(expires_at, datetime)
+                else str(expires_at)
+            ),
+        }
+
+    @staticmethod
+    def _attest_binding_locked(
+        connection,
+        *,
+        binding_id: str,
+        binding_hash: str,
+        kind: str,
+    ) -> None:
+        row = connection.execute(
+            text(
+                """
+                SELECT binding_hash, access_mode, object_kind
+                FROM public.dataflow_dataset_bindings
+                WHERE id = CAST(:binding_id AS uuid)
+                FOR SHARE
+                """
+            ),
+            {"binding_id": binding_id},
+        ).mappings().one_or_none()
+        allowed = (
+            {"write", "read_write"}
+            if kind == "output"
+            else {"read", "read_write"}
+        )
+        if (
+            row is None
+            or row["object_kind"] != "parquet_artifact"
+            or row["access_mode"] not in allowed
+            or str(row["binding_hash"]) != binding_hash
+        ):
+            raise ValueError("canonical artifact binding no longer matches")
+
+    def _lookup_handoff(
+        self,
+        *,
+        correlation_id: str,
+        binding_id: str,
+        kind: str,
+    ) -> dict[str, Any] | None:
+        with self.engine.connect() as connection:
+            row = connection.execute(
+                text(
+                    """
+                    SELECT id::text AS id, correlation_id::text,
+                           binding_id::text, artifact_ref, artifact_digest,
+                           row_count, schema_hash, schema_fields,
+                           artifact_kind, binding_hash, expires_at,
+                           handoff_status
+                    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
+                    """
+                ),
+                {
+                    "correlation_id": correlation_id,
+                    "binding_id": binding_id,
+                    "artifact_kind": kind,
+                },
+            ).mappings().one_or_none()
+        return dict(row) if row is not None else None
+
+    def reserve(
         self,
         *,
         binding_id: str,
@@ -784,15 +958,14 @@ class PostgresArtifactResolver:
         kind: str,
         binding_hash: str,
     ) -> dict[str, Any]:
+        """Atomically attest the binding and reserve one pending handoff."""
+
         binding = _uid(binding_id, "artifact binding id")
         correlation = _uid(correlation_id, "artifact correlation id")
         if kind not in {"input", "lookup", "output"}:
             raise ValueError("artifact kind is invalid")
-        self.attest_binding(
-            binding_id=binding,
-            binding_hash=binding_hash,
-            access_mode="write" if kind == "output" else "read",
-        )
+        if _DIGEST.fullmatch(str(binding_hash or "")) is None:
+            raise ValueError("artifact binding hash is invalid")
         if not isinstance(artifact, dict):
             raise ValueError("artifact metadata is invalid")
         artifact_ref = artifact.get("artifact_ref")
@@ -801,118 +974,544 @@ class PostgresArtifactResolver:
             raise ValueError(
                 "artifact does not match the execution correlation"
             )
-        described = self.artifact_store.describe(artifact_ref)
-        for key in (
-            "artifact_ref",
-            "digest",
-            "row_count",
-            "schema_hash",
-            "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
+        if canonical_schema_hash(fields) != artifact.get("schema_hash"):
+            raise ValueError("artifact schema contract does not match")
+        if _DIGEST.fullmatch(str(artifact.get("digest") or "")) is None:
+            raise ValueError("artifact digest is invalid")
+        reservation_id = new_governance_uid()
+        parameters = {
+            "id": reservation_id,
+            "correlation_id": correlation,
+            "binding_id": binding,
+            "artifact_ref": artifact_ref,
+            "artifact_digest": artifact["digest"],
+            "row_count": int(artifact["row_count"]),
+            "schema_hash": artifact["schema_hash"],
+            "schema_fields": json.dumps(
+                fields,
+                sort_keys=True,
+                separators=(",", ":"),
+            ),
+            "artifact_kind": kind,
+            "binding_hash": binding_hash,
+            "expires_at": artifact["expires_at"],
+        }
+        selected = None
+        inserted = False
+        try:
+            with self.engine.begin() as connection:
+                self._attest_binding_locked(
+                    connection,
+                    binding_id=binding,
+                    binding_hash=binding_hash,
+                    kind=kind,
+                )
+                selected = connection.execute(
+                    text(
+                        """
+                        INSERT INTO public.rule_run_artifacts (
+                            id, correlation_id, binding_id, artifact_ref,
+                            artifact_digest, row_count, schema_hash,
+                            schema_fields, artifact_kind, binding_hash,
+                            handoff_status, expires_at
+                        ) VALUES (
+                            CAST(:id AS uuid),
+                            CAST(:correlation_id AS uuid),
+                            CAST(:binding_id AS uuid), :artifact_ref,
+                            :artifact_digest, :row_count, :schema_hash,
+                            CAST(:schema_fields AS jsonb), :artifact_kind,
+                            :binding_hash, 'pending',
+                            CAST(:expires_at AS timestamptz)
+                        )
+                        ON CONFLICT (
+                            correlation_id, binding_id, artifact_kind
+                        ) DO NOTHING
+                        RETURNING id::text AS id, correlation_id::text,
+                                  binding_id::text, artifact_ref,
+                                  artifact_digest, row_count, schema_hash,
+                                  schema_fields, artifact_kind, binding_hash,
+                                  expires_at, handoff_status
+                        """
+                    ),
+                    parameters,
+                ).mappings().one_or_none()
+                inserted = selected is not None
+                if selected is None:
+                    selected = connection.execute(
+                        text(
+                            """
+                            SELECT id::text AS id, correlation_id::text,
+                                   binding_id::text, artifact_ref,
+                                   artifact_digest, row_count, schema_hash,
+                                   schema_fields, artifact_kind, binding_hash,
+                                   expires_at, handoff_status
+                            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
+                            """
+                        ),
+                        parameters,
+                    ).mappings().one_or_none()
+        except ValueError:
+            raise
+        except Exception as exc:
+            try:
+                selected = self._lookup_handoff(
+                    correlation_id=correlation,
+                    binding_id=binding,
+                    kind=kind,
+                )
+            except Exception as recheck_exc:
+                raise ArtifactCommitUnknown(
+                    "artifact reservation commit outcome is unknown"
+                ) from recheck_exc
+            if (
+                selected is None
+                or str(selected["artifact_ref"]) != artifact_ref
+                or str(selected["artifact_digest"]) != artifact["digest"]
+                or str(selected["binding_hash"]) != binding_hash
+            ):
+                raise ArtifactCommitUnknown(
+                    "artifact reservation commit outcome is unknown"
+                ) from exc
+            inserted = True
+        if selected is None:
+            raise ArtifactCommitUnknown(
+                "artifact reservation commit outcome is unknown"
+            )
+        row = dict(selected)
+        if str(row["binding_hash"]) != binding_hash:
+            raise ValueError("artifact reservation binding hash conflicts")
+        if str(row["artifact_digest"]) != artifact["digest"]:
+            raise ValueError(
+                "immutable artifact catalog digest conflicts with retry"
+            )
+        status = str(row["handoff_status"])
+        if not inserted:
+            if status == "ready":
+                return {
+                    **self._catalog_artifact(row),
+                    "correlation_id": correlation,
+                    "reservation_id": str(row["id"]),
+                    "handoff_status": "ready",
+                    "upload_required": False,
+                }
+            if status == "pending":
+                raise ArtifactHandoffPending(
+                    "artifact handoff is already pending"
+                )
+            raise ValueError("artifact handoff has failed")
+        return {
+            **self._catalog_artifact(row),
+            "correlation_id": correlation,
+            "reservation_id": str(row["id"]),
+            "handoff_status": status,
+            "upload_required": status == "pending",
+        }
+
+    def _abort_pending(self, reservation_id: str) -> None:
         with self.engine.begin() as connection:
-            inserted = connection.execute(
+            connection.execute(
                 text(
                     """
-                    INSERT INTO public.rule_run_artifacts (
-                        id, correlation_id, binding_id, artifact_ref,
-                        artifact_digest, row_count, schema_hash, schema_fields,
-                        artifact_kind, expires_at
-                    ) VALUES (
-                        CAST(:id AS uuid), CAST(:correlation_id AS uuid),
-                        CAST(:binding_id AS uuid), :artifact_ref,
-                        :artifact_digest, :row_count, :schema_hash,
-                        CAST(:schema_fields AS jsonb), :artifact_kind,
-                        CAST(:expires_at AS timestamptz)
-                    )
-                    ON CONFLICT (
-                        correlation_id, binding_id, artifact_kind
-                    ) DO NOTHING
-                    RETURNING artifact_ref, artifact_digest, row_count,
-                              schema_hash, schema_fields, expires_at
+                    DELETE FROM public.rule_run_artifacts
+                    WHERE id = CAST(:id AS uuid)
+                      AND handoff_status = 'pending'
                     """
                 ),
-                {
-                    "id": new_governance_uid(),
-                    "correlation_id": correlation,
-                    "binding_id": binding,
-                    "artifact_ref": described["artifact_ref"],
-                    "artifact_digest": described["digest"],
-                    "row_count": described["row_count"],
-                    "schema_hash": described["schema_hash"],
-                    "schema_fields": json.dumps(
-                        fields,
-                        sort_keys=True,
-                        separators=(",", ":"),
-                    ),
-                    "artifact_kind": kind,
-                    "expires_at": described["expires_at"],
-                },
-            ).mappings().one_or_none()
-            if inserted is None:
-                inserted = connection.execute(
+                {"id": reservation_id},
+            )
+
+    def finalize(
+        self,
+        *,
+        reservation: dict[str, Any],
+        binding_id: str,
+        binding_hash: str,
+        kind: str,
+    ) -> dict[str, Any]:
+        reservation_id = _uid(
+            reservation.get("reservation_id"), "artifact reservation id"
+        )
+        binding = _uid(binding_id, "artifact binding id")
+        selected = None
+        try:
+            with self.engine.begin() as connection:
+                self._attest_binding_locked(
+                    connection,
+                    binding_id=binding,
+                    binding_hash=binding_hash,
+                    kind=kind,
+                )
+                selected = 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)
+                        UPDATE public.rule_run_artifacts
+                        SET handoff_status = 'ready',
+                            ready_at = CURRENT_TIMESTAMP,
+                            updated_at = CURRENT_TIMESTAMP,
+                            failure_code = NULL,
+                            failed_at = NULL
+                        WHERE id = CAST(:id AS uuid)
                           AND binding_id = CAST(:binding_id AS uuid)
-                          AND artifact_kind = :artifact_kind
-                        FOR UPDATE
+                          AND binding_hash = :binding_hash
+                          AND artifact_digest = :artifact_digest
+                          AND handoff_status = 'pending'
+                        RETURNING id::text AS id, correlation_id::text,
+                                  binding_id::text, artifact_ref,
+                                  artifact_digest, row_count, schema_hash,
+                                  schema_fields, artifact_kind, binding_hash,
+                                  expires_at, handoff_status
                         """
                     ),
                     {
-                        "correlation_id": correlation,
+                        "id": reservation_id,
                         "binding_id": binding,
-                        "artifact_kind": kind,
+                        "binding_hash": binding_hash,
+                        "artifact_digest": reservation["digest"],
                     },
                 ).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)
+                if selected is None:
+                    raise ValueError(
+                        "pending artifact handoff no longer matches"
+                    )
+        except ValueError:
+            raise
+        except Exception as exc:
+            try:
+                selected = self._lookup_handoff(
+                    correlation_id=_uid(
+                        reservation["correlation_id"],
+                        "artifact correlation id",
+                    ),
+                    binding_id=binding,
+                    kind=kind,
+                )
+            except Exception as recheck_exc:
+                raise ArtifactCommitUnknown(
+                    "artifact finalize commit outcome is unknown"
+                ) from recheck_exc
+            if (
+                selected is None
+                or str(selected["id"]) != reservation_id
+                or str(selected["artifact_digest"])
+                != reservation["digest"]
+                or str(selected["binding_hash"]) != binding_hash
+                or str(selected["handoff_status"]) != "ready"
+            ):
+                raise ArtifactCommitUnknown(
+                    "artifact finalize commit outcome is unknown"
+                ) from exc
+        return self._catalog_artifact(selected)
+
+    def publish_path(
+        self,
+        path: str,
+        *,
+        binding_id: str,
+        binding_hash: str,
+        correlation_id: str,
+        kind: str,
+        ttl_seconds: int,
+        schema_fields: list[dict[str, Any]],
+        limits: dict[str, int] | None = None,
+    ) -> dict[str, Any]:
+        """Reserve, upload, and finalize one durable artifact handoff."""
+
+        prepared = self.artifact_store.prepare_path(
+            path,
+            correlation_id,
+            ttl_seconds,
+            schema_fields=schema_fields,
+            limits=limits,
+        )
+        reservation = self.reserve(
+            binding_id=binding_id,
+            correlation_id=correlation_id,
+            artifact=prepared,
+            kind=kind,
+            binding_hash=binding_hash,
+        )
+        if not reservation["upload_required"]:
             stored = self.artifact_store.describe(
-                registered["artifact_ref"]
+                reservation["artifact_ref"]
             )
+            if stored["digest"] != reservation["digest"]:
+                raise ValueError(
+                    "ready artifact catalog does not match storage"
+                )
+            return {
+                key: reservation[key]
+                for key in (
+                    "artifact_ref",
+                    "digest",
+                    "row_count",
+                    "schema_hash",
+                    "schema_fields",
+                    "expires_at",
+                )
+            }
+        reserved_artifact = {
+            key: reservation[key]
             for key in (
+                "artifact_ref",
                 "digest",
                 "row_count",
                 "schema_hash",
+                "schema_fields",
                 "expires_at",
-            ):
-                if stored[key] != registered[key]:
-                    raise ValueError(
-                        "immutable artifact catalog does not match storage"
+            )
+        }
+        try:
+            self.artifact_store.upload_path(
+                path,
+                reserved_artifact,
+                limits=limits,
+            )
+        except Exception:
+            with suppress(Exception):
+                self.artifact_store.delete(
+                    reserved_artifact["artifact_ref"]
+                )
+            with suppress(Exception):
+                self._abort_pending(reservation["reservation_id"])
+            raise
+        return self.finalize(
+            reservation=reservation,
+            binding_id=binding_id,
+            binding_hash=binding_hash,
+            kind=kind,
+        )
+
+    def _mark_failed(
+        self,
+        *,
+        row_id: str,
+        failure_code: str,
+    ) -> None:
+        with self.engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_run_artifacts
+                    SET handoff_status = 'failed',
+                        failed_at = CURRENT_TIMESTAMP,
+                        updated_at = CURRENT_TIMESTAMP,
+                        failure_code = :failure_code
+                    WHERE id = CAST(:id AS uuid)
+                      AND handoff_status IN ('pending','ready')
+                    """
+                ),
+                {"id": row_id, "failure_code": failure_code},
+            )
+
+    def reconcile(
+        self,
+        *,
+        limit: int = 100,
+        grace_seconds: int = 300,
+    ) -> dict[str, int]:
+        """Repair bounded catalog/store drift after the grace period."""
+
+        if (
+            isinstance(limit, bool)
+            or not isinstance(limit, int)
+            or limit < 1
+            or limit > 1_000
+        ):
+            raise ValueError("artifact reconciliation limit is invalid")
+        if (
+            isinstance(grace_seconds, bool)
+            or not isinstance(grace_seconds, int)
+            or grace_seconds < 30
+            or grace_seconds > 86_400
+        ):
+            raise ValueError("artifact reconciliation grace is invalid")
+        result = {
+            "pending_finalized": 0,
+            "pending_deleted": 0,
+            "ready_failed": 0,
+            "orphans_deleted": 0,
+        }
+        with self.engine.connect() as connection:
+            rows = connection.execute(
+                text(
+                    """
+                    SELECT id::text AS id, correlation_id::text,
+                           binding_id::text, artifact_ref, artifact_digest,
+                           row_count, schema_hash, schema_fields,
+                           artifact_kind, binding_hash, expires_at,
+                           handoff_status
+                    FROM public.rule_run_artifacts
+                    WHERE handoff_status IN ('pending','ready')
+                      AND updated_at <= CURRENT_TIMESTAMP
+                          - make_interval(secs => :grace_seconds)
+                    ORDER BY updated_at, id
+                    LIMIT :limit
+                    """
+                ),
+                {
+                    "grace_seconds": grace_seconds,
+                    "limit": limit,
+                },
+            ).mappings().all()
+        for raw_row in rows:
+            row = dict(raw_row)
+            row_id = str(row["id"])
+            status = str(row["handoff_status"])
+            invalid_object = False
+            try:
+                self.artifact_store._parse_ref(row["artifact_ref"])
+                stored = self.artifact_store.describe_optional(
+                    row["artifact_ref"]
+                )
+            except Exception:
+                invalid_object = True
+                stored = None
+            if invalid_object:
+                if status == "pending":
+                    with suppress(Exception):
+                        self.artifact_store.delete(row["artifact_ref"])
+                    self._mark_failed(
+                        row_id=row_id,
+                        failure_code="pending_object_invalid",
+                    )
+                else:
+                    self._mark_failed(
+                        row_id=row_id,
+                        failure_code="ready_object_invalid",
                     )
-        return registered
+                    result["ready_failed"] += 1
+                continue
+            if status == "ready":
+                if stored is None:
+                    self._mark_failed(
+                        row_id=row_id,
+                        failure_code="ready_object_missing",
+                    )
+                    result["ready_failed"] += 1
+                continue
+            if stored is None:
+                with self.engine.begin() as connection:
+                    deleted = connection.execute(
+                        text(
+                            """
+                            DELETE FROM public.rule_run_artifacts
+                            WHERE id = CAST(:id AS uuid)
+                              AND handoff_status = 'pending'
+                            """
+                        ),
+                        {"id": row_id},
+                    )
+                if int(deleted.rowcount or 0) == 1:
+                    result["pending_deleted"] += 1
+                continue
+            if any(
+                (
+                    stored["digest"] != str(row["artifact_digest"]),
+                    stored["row_count"] != int(row["row_count"]),
+                    stored["schema_hash"] != str(row["schema_hash"]),
+                )
+            ):
+                with suppress(Exception):
+                    self.artifact_store.delete(row["artifact_ref"])
+                self._mark_failed(
+                    row_id=row_id,
+                    failure_code="pending_object_invalid",
+                )
+                continue
+            with self.engine.begin() as connection:
+                finalized = connection.execute(
+                    text(
+                        """
+                        UPDATE public.rule_run_artifacts a
+                        SET handoff_status = 'ready',
+                            ready_at = CURRENT_TIMESTAMP,
+                            updated_at = CURRENT_TIMESTAMP,
+                            failure_code = NULL,
+                            failed_at = NULL
+                        WHERE a.id = CAST(:id AS uuid)
+                          AND a.handoff_status = 'pending'
+                          AND EXISTS (
+                              SELECT 1
+                              FROM public.dataflow_dataset_bindings b
+                              WHERE b.id = a.binding_id
+                                AND b.binding_hash = a.binding_hash
+                                AND b.object_kind = 'parquet_artifact'
+                                AND b.access_mode IN (
+                                    'read','write','read_write'
+                                )
+                          )
+                        RETURNING a.id
+                        """
+                    ),
+                    {"id": row_id},
+                )
+            if int(finalized.rowcount or 0) == 1:
+                result["pending_finalized"] += 1
+            else:
+                self._mark_failed(
+                    row_id=row_id,
+                    failure_code="pending_binding_changed",
+                )
+
+        remaining = limit - len(rows)
+        if remaining <= 0:
+            return result
+        now = _now_utc(self.artifact_store.clock)
+        candidates = []
+        scanned = 0
+        for item in self.artifact_store.client.list_objects(
+            self.artifact_store.bucket,
+            prefix="rules/",
+            recursive=True,
+        ):
+            scanned += 1
+            if scanned > limit * 10 or len(candidates) >= remaining:
+                break
+            key = str(getattr(item, "object_name", ""))
+            ref = f"minio://{self.artifact_store.bucket}/{key}"
+            try:
+                self.artifact_store._parse_ref(ref)
+            except ValueError:
+                continue
+            modified = getattr(item, "last_modified", None)
+            if not isinstance(modified, datetime):
+                continue
+            if modified.tzinfo is None:
+                modified = modified.replace(tzinfo=UTC)
+            if modified.astimezone(UTC) > now - timedelta(
+                seconds=grace_seconds
+            ):
+                continue
+            candidates.append(ref)
+        if not candidates:
+            return result
+        with self.engine.connect() as connection:
+            referenced = {
+                str(row["artifact_ref"])
+                for row in connection.execute(
+                    text(
+                        """
+                        SELECT artifact_ref
+                        FROM public.rule_run_artifacts
+                        WHERE artifact_ref =
+                              ANY(CAST(:artifact_refs AS text[]))
+                        """
+                    ),
+                    {"artifact_refs": candidates},
+                ).mappings().all()
+            }
+        for ref in candidates:
+            if ref in referenced:
+                continue
+            self.artifact_store.delete(ref)
+            result["orphans_deleted"] += 1
+        return result
 
     def cleanup_expired(self, *, limit: int = 100) -> int:
         if (

app/runner/bootstrap.py

@@ -81,6 +81,30 @@ class RunnerSettings:
 def register_artifact_cleanup_cli(app, artifact_resolver):
     """Register the bounded production maintenance entrypoint."""
 
+    @app.cli.command("reconcile-rule-artifacts")
+    @click.option("--limit", type=click.IntRange(1, 1_000), default=100)
+    @click.option(
+        "--grace-seconds",
+        type=click.IntRange(30, 86_400),
+        default=300,
+    )
+    def reconcile_rule_artifacts(limit, grace_seconds):
+        result = artifact_resolver.reconcile(
+            limit=limit,
+            grace_seconds=grace_seconds,
+        )
+        click.echo(
+            " ".join(
+                f"{key}={result[key]}"
+                for key in (
+                    "pending_finalized",
+                    "pending_deleted",
+                    "ready_failed",
+                    "orphans_deleted",
+                )
+            )
+        )
+
     @app.cli.command("cleanup-rule-artifacts")
     @click.option("--limit", type=click.IntRange(1, 1_000), default=100)
     def cleanup_rule_artifacts(limit):

app/runner/polars_worker.py

@@ -468,18 +468,23 @@ def _run_process(
     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
+    parent = None
+    child = None
+    process = None
+    started = False
     try:
+        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()
+        started = True
+        child.close()
+        deadline = time.monotonic() + timeout_seconds
+        baseline_rss = None
         while time.monotonic() < deadline:
             if parent.poll(0.01):
                 try:
@@ -523,11 +528,30 @@ def _run_process(
         raise PolarsWorkerResourceError(
             "Polars worker exceeded its bounded execution time"
         )
+    except (PolarsWorkerError, PolarsWorkerResourceError):
+        raise
+    except Exception as exc:
+        raise PolarsWorkerError(
+            "Polars worker failed to start safely"
+        ) from exc
     finally:
-        parent.close()
-        if process.is_alive():
-            process.kill()
-        process.join(timeout=1)
+        if child is not None:
+            with suppress(Exception):
+                child.close()
+        if parent is not None:
+            with suppress(Exception):
+                parent.close()
+        if process is not None:
+            if started:
+                with suppress(Exception):
+                    if process.is_alive():
+                        process.kill()
+                with suppress(Exception):
+                    process.join(timeout=1)
+            close = getattr(process, "close", None)
+            if callable(close):
+                with suppress(Exception):
+                    close()
 
 
 def run_isolated_memory_probe(

app/runner/rule_polars.py

@@ -12,6 +12,7 @@ from app.core.data_rules.compilers.polars import (
     bound_polars_plan_hash,
     validate_bound_polars_plan,
 )
+from app.runner.artifacts import ArtifactCommitUnknown
 from app.runner.nodes import NodeExecutionError
 from app.runner.polars_worker import (
     PolarsWorkerError,
@@ -232,35 +233,29 @@ class PolarsRulePlanAdapter:
                         "published Polars worker execution failed"
                     ) from exc
                 try:
-                    artifact = self.artifact_store.write_path(
+                    registered = self.artifact_resolver.publish_path(
                         output_path,
-                        correlation,
-                        self.artifact_ttl_seconds,
+                        binding_id=normalized["output_binding_id"],
+                        binding_hash=normalized["output_binding_hash"],
+                        correlation_id=correlation,
+                        kind="output",
+                        ttl_seconds=self.artifact_ttl_seconds,
                         schema_fields=normalized["output_fields"],
                         limits=limits,
                     )
-                except ValueError as exc:
+                except ArtifactCommitUnknown as exc:
+                    raise NodeExecutionError(
+                        "published Polars artifact commit outcome is unknown",
+                        commit_outcome="unknown",
+                    ) from exc
+                except Exception as exc:
                     raise NodeExecutionError(
-                        "published Polars output artifact write failed"
+                        "published Polars output artifact publication failed"
                     ) from exc
         finally:
             if output_path is not None:
                 with suppress(FileNotFoundError):
                     os.unlink(output_path)
-        try:
-            registered = self.artifact_resolver.register(
-                binding_id=normalized["output_binding_id"],
-                correlation_id=correlation,
-                artifact=artifact,
-                kind="output",
-                binding_hash=normalized["output_binding_hash"],
-            )
-        except Exception as exc:
-            with suppress(Exception):
-                self.artifact_store.delete(artifact["artifact_ref"])
-            raise NodeExecutionError(
-                "published Polars output artifact registration failed"
-            ) from exc
         return {
             **{key: registered[key] for key in _PUBLIC_ARTIFACT_KEYS},
             **{key: worker_result[key] for key in _RESULT_KEYS},

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

migrations/versions/20260723_150_rule_artifact_handoff_state.py

@@ -0,0 +1,97 @@
+"""Upgrade artifact handoff from old-140 rows to a durable state machine."""
+
+from alembic import op
+
+revision = "20260723_150"
+down_revision = "20260723_140"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE public.rule_run_artifacts
+            ADD COLUMN binding_hash CHAR(64),
+            ADD COLUMN handoff_status VARCHAR(20) NOT NULL
+                DEFAULT 'pending'
+                CHECK (handoff_status IN ('pending','ready','failed')),
+            ADD COLUMN ready_at TIMESTAMPTZ,
+            ADD COLUMN failed_at TIMESTAMPTZ,
+            ADD COLUMN failure_code VARCHAR(100),
+            ADD COLUMN updated_at TIMESTAMPTZ NOT NULL
+                DEFAULT CURRENT_TIMESTAMP;
+
+        UPDATE public.rule_run_artifacts a
+        SET binding_hash = (
+                SELECT binding_hash
+                FROM public.dataflow_dataset_bindings b
+                WHERE b.id = a.binding_id
+            ),
+            handoff_status = 'ready',
+            ready_at = a.created_at,
+            updated_at = a.created_at;
+
+        DO $$
+        BEGIN
+            IF EXISTS (
+                SELECT 1
+                FROM public.rule_run_artifacts
+                WHERE binding_hash IS NULL
+            ) THEN
+                RAISE EXCEPTION
+                    'artifact handoff migration found unattested bindings';
+            END IF;
+            IF EXISTS (
+                SELECT 1
+                FROM public.rule_run_artifacts
+                GROUP BY correlation_id, binding_id, artifact_kind
+                HAVING COUNT(*) > 1
+            ) THEN
+                RAISE EXCEPTION
+                    'artifact handoff migration found conflicting old rows';
+            END IF;
+        END
+        $$;
+
+        ALTER TABLE public.rule_run_artifacts
+            ALTER COLUMN binding_hash SET NOT NULL;
+
+        DO $$
+        DECLARE
+            old_constraint TEXT;
+        BEGIN
+            SELECT c.conname
+            INTO old_constraint
+            FROM pg_constraint c
+            WHERE c.conrelid = 'public.rule_run_artifacts'::regclass
+              AND c.contype = 'u'
+              AND pg_get_constraintdef(c.oid) =
+                  'UNIQUE (correlation_id, binding_id, artifact_digest)';
+            IF old_constraint IS NULL THEN
+                RAISE EXCEPTION
+                    'old-140 artifact uniqueness constraint was not found';
+            END IF;
+            EXECUTE format(
+                'ALTER TABLE public.rule_run_artifacts DROP CONSTRAINT %I',
+                old_constraint
+            );
+        END
+        $$;
+
+        ALTER TABLE public.rule_run_artifacts
+            ADD CONSTRAINT rule_run_artifacts_handoff_key
+            UNIQUE (correlation_id, binding_id, artifact_kind);
+
+        CREATE INDEX idx_rule_run_artifacts_reconcile
+            ON public.rule_run_artifacts
+            (handoff_status, updated_at, id);
+        """
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError(
+        "artifact handoff state is forward-only and cannot downgrade "
+        "without violating durable publication evidence"
+    )

tests/integration/test_data_rule_polars_execution.py

@@ -55,7 +55,7 @@ def _binding(schema, *, source_uid, access_mode, object_ref):
     }
 
 
-def test_real_postgres_mysql_minio_polars_cross_source_execution():
+def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
     from app.core.data_rules.compilers.polars import PolarsRuleCompiler
     from app.runner.artifacts import ArtifactStore, PostgresArtifactResolver
     from app.runner.rule_polars import PolarsRulePlanAdapter
@@ -483,34 +483,30 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
                     "schema_hashes": json.dumps(schema_hashes),
                 },
             )
-        customer_artifact = store.write(
-            pl.DataFrame(customer_rows),
-            correlation_id,
-            900,
-            schema_fields=input_schema["fields"],
-            limits=compiled["plan"]["resource_limits"],
-        )
-        segment_artifact = store.write(
-            pl.DataFrame(segment_rows),
-            correlation_id,
-            900,
-            schema_fields=lookup_schema["fields"],
-            limits=compiled["plan"]["resource_limits"],
-        )
+        customer_path = tmp_path / "customers.parquet"
+        segment_path = tmp_path / "segments.parquet"
+        pl.DataFrame(customer_rows).write_parquet(customer_path)
+        pl.DataFrame(segment_rows).write_parquet(segment_path)
         resolver = PostgresArtifactResolver(platform, store)
-        resolver.register(
+        customer_artifact = resolver.publish_path(
+            str(customer_path),
             binding_id=input_binding["id"],
             binding_hash=compiled["plan"]["input_binding_hash"],
             correlation_id=correlation_id,
-            artifact=customer_artifact,
             kind="input",
+            ttl_seconds=900,
+            schema_fields=input_schema["fields"],
+            limits=compiled["plan"]["resource_limits"],
         )
-        resolver.register(
+        segment_artifact = resolver.publish_path(
+            str(segment_path),
             binding_id=lookup_binding["id"],
             binding_hash=lookup_operation["lookup_binding_hash"],
             correlation_id=correlation_id,
-            artifact=segment_artifact,
             kind="lookup",
+            ttl_seconds=900,
+            schema_fields=lookup_schema["fields"],
+            limits=compiled["plan"]["resource_limits"],
         )
         node = {
             "id": "task5_real_polars",
@@ -641,42 +637,42 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
         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"],
+        conflict_path = tmp_path / "conflict.parquet"
+        pl.DataFrame(
+            {
+                "customer_id": [99],
+                "mobile": ["13800138000"],
+                "name": ["Conflict"],
+                "segment_code": ["A"],
+                "segment_name": ["Gold"],
+                "version_no": [1],
+            }
+        ).write_parquet(conflict_path)
+        object_count_before_conflict = len(
+            list(minio.list_objects(bucket, prefix=prefix, recursive=True))
         )
         with pytest.raises(ValueError, match="immutable|digest"):
-            resolver.register(
+            resolver.publish_path(
+                str(conflict_path),
                 binding_id=output_binding["id"],
                 binding_hash=compiled["plan"]["output_binding_hash"],
                 correlation_id=correlation_id,
-                artifact=conflict,
                 kind="output",
+                ttl_seconds=900,
+                schema_fields=output_schema["fields"],
+                limits=compiled["plan"]["resource_limits"],
             )
-        assert not any(
-            item.object_name
-            == conflict["artifact_ref"].split("/", 3)[-1]
-            for item in minio.list_objects(
-                bucket, prefix=prefix, recursive=True
-            )
-        )
+        assert len(
+            list(minio.list_objects(bucket, prefix=prefix, recursive=True))
+        ) == object_count_before_conflict
+        assert customer_artifact["digest"]
+        assert segment_artifact["digest"]
         with platform.connect() as connection:
             catalog_rows = connection.execute(
                 text(
                     """
-                    SELECT artifact_kind, artifact_ref
+                    SELECT artifact_kind, artifact_ref, handoff_status,
+                           binding_hash
                     FROM public.rule_run_artifacts
                     WHERE correlation_id = CAST(:correlation_id AS uuid)
                     ORDER BY artifact_kind
@@ -687,6 +683,8 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution():
         # The repeated deterministic output has the same digest and is
         # idempotently retained as one stable catalog handoff.
         assert len(catalog_rows) == 3
+        assert all(row["handoff_status"] == "ready" for row in catalog_rows)
+        assert all(len(row["binding_hash"]) == 64 for row in catalog_rows)
         assert next(
             row["artifact_ref"]
             for row in catalog_rows

tests/integration/test_rule_artifact_migration_upgrade.py

@@ -0,0 +1,330 @@
+from __future__ import annotations
+
+import logging
+import os
+import re
+from pathlib import Path
+
+import polars as pl
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, text
+from sqlalchemy.exc import IntegrityError
+
+from app.core.common.identifiers import new_governance_uid
+from tests.runner.test_artifacts import FakeMinio, _store
+
+ROOT = Path(__file__).resolve().parents[2]
+COMPOSE = ROOT / "deploy" / "docker" / "docker-compose.yml"
+
+
+def _compose_value(pattern: str) -> str:
+    match = re.search(
+        pattern,
+        COMPOSE.read_text(encoding="utf-8"),
+        flags=re.DOTALL,
+    )
+    assert match is not None
+    return match.group(1)
+
+
+def _upgrade(database_url: str, revision: str) -> None:
+    previous = os.environ.get("DATABASE_URL")
+    root_logger = logging.getLogger()
+    root_handlers = list(root_logger.handlers)
+    root_level = root_logger.level
+    logger_disabled = {
+        name: logger.disabled
+        for name, logger in logging.Logger.manager.loggerDict.items()
+        if isinstance(logger, logging.Logger)
+    }
+    os.environ["DATABASE_URL"] = database_url
+    try:
+        command.upgrade(Config(str(ROOT / "alembic.ini")), revision)
+    finally:
+        root_logger.handlers[:] = root_handlers
+        root_logger.setLevel(root_level)
+        for name, disabled in logger_disabled.items():
+            logging.getLogger(name).disabled = disabled
+        if previous is None:
+            os.environ.pop("DATABASE_URL", None)
+        else:
+            os.environ["DATABASE_URL"] = previous
+
+
+def test_old_140_upgrades_to_durable_handoff_and_enforces_cas(tmp_path):
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    platform_user = _compose_value(
+        r"\n  postgres:.*?POSTGRES_USER:\s*([^\s]+)"
+    )
+    platform_password = _compose_value(
+        r"\n  postgres:.*?POSTGRES_PASSWORD:\s*([^\s]+)"
+    )
+    platform_port = _compose_value(r'"(15432):5432"')
+    admin_url = (
+        f"postgresql+psycopg2://{platform_user}:{platform_password}"
+        f"@127.0.0.1:{platform_port}/postgres"
+    )
+    database_name = f"task5_migration_{new_governance_uid().replace('-', '')}"
+    database_url = (
+        f"postgresql+psycopg2://{platform_user}:{platform_password}"
+        f"@127.0.0.1:{platform_port}/{database_name}"
+    )
+    admin = create_engine(admin_url, isolation_level="AUTOCOMMIT")
+    engine = None
+    try:
+        with admin.connect() as connection:
+            connection.execute(text(f'CREATE DATABASE "{database_name}"'))
+        _upgrade(database_url, "20260723_140")
+        engine = create_engine(database_url, pool_pre_ping=True)
+        dataflow_version_id = new_governance_uid()
+        deployment_id = new_governance_uid()
+        schema_id = new_governance_uid()
+        binding_id = new_governance_uid()
+        binding_hash = "b" * 64
+        correlation_id = new_governance_uid()
+        old_artifact_id = new_governance_uid()
+        old_ref = (
+            f"minio://dataops-rules/rules/{correlation_id}/"
+            f"{new_governance_uid()}.parquet"
+        )
+        with engine.begin() as connection:
+            columns_before = {
+                row[0]
+                for row in connection.execute(
+                    text(
+                        """
+                        SELECT column_name
+                        FROM information_schema.columns
+                        WHERE table_schema = 'public'
+                          AND table_name = 'rule_run_artifacts'
+                        """
+                    )
+                )
+            }
+            assert "handoff_status" not in columns_before
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.dataflow_versions (
+                        id, dataflow_uid, version_no, name, dataflow_spec,
+                        input_schema_hashes, output_schema_hash, status
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:uid AS uuid), 1, 'migration',
+                        '{}'::jsonb, '[]'::jsonb, :schema_hash, 'released'
+                    )
+                    """
+                ),
+                {
+                    "id": dataflow_version_id,
+                    "uid": new_governance_uid(),
+                    "schema_hash": "a" * 64,
+                },
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.dataflow_deployments (
+                        id, dataflow_version_id, environment,
+                        deployment_config, status
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:version_id AS uuid), 'test',
+                        '{}'::jsonb, 'active'
+                    )
+                    """
+                ),
+                {"id": deployment_id, "version_id": dataflow_version_id},
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.data_schema_snapshots (
+                        id, schema_ref, schema_hash, fields, source_revision
+                    ) VALUES (
+                        CAST(:id AS uuid), 'migration:id', :schema_hash,
+                        CAST(:fields AS jsonb), 'old-140'
+                    )
+                    """
+                ),
+                {
+                    "id": schema_id,
+                    "schema_hash": "a" * 64,
+                    "fields": (
+                        '[{"name":"id","type":"integer",'
+                        '"nullable":false}]'
+                    ),
+                },
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.dataflow_dataset_bindings (
+                        id, dataflow_deployment_id, logical_ref,
+                        object_kind, object_ref, schema_snapshot_id, dialect,
+                        access_mode, write_mode, binding_hash
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:deployment_id AS uuid),
+                        'output', 'parquet_artifact', 'migration-output',
+                        CAST(:schema_id AS uuid), 'parquet', 'write',
+                        'append', :binding_hash
+                    )
+                    """
+                ),
+                {
+                    "id": binding_id,
+                    "deployment_id": deployment_id,
+                    "schema_id": schema_id,
+                    "binding_hash": binding_hash,
+                },
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_run_artifacts (
+                        id, correlation_id, binding_id, artifact_ref,
+                        artifact_digest, row_count, schema_hash, schema_fields,
+                        artifact_kind, expires_at
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:correlation_id AS uuid),
+                        CAST(:binding_id AS uuid), :artifact_ref,
+                        :digest, 1, :schema_hash, CAST(:fields AS jsonb),
+                        'output', CURRENT_TIMESTAMP + INTERVAL '1 hour'
+                    )
+                    """
+                ),
+                {
+                    "id": old_artifact_id,
+                    "correlation_id": correlation_id,
+                    "binding_id": binding_id,
+                    "artifact_ref": old_ref,
+                    "digest": "d" * 64,
+                    "schema_hash": "a" * 64,
+                    "fields": (
+                        '[{"name":"id","type":"integer",'
+                        '"nullable":false}]'
+                    ),
+                },
+            )
+
+        engine.dispose()
+        engine = None
+        _upgrade(database_url, "head")
+        engine = create_engine(database_url, pool_pre_ping=True)
+        with engine.begin() as connection:
+            migrated = connection.execute(
+                text(
+                    """
+                    SELECT binding_hash, handoff_status, ready_at
+                    FROM public.rule_run_artifacts
+                    WHERE id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": old_artifact_id},
+            ).mappings().one()
+            assert migrated["binding_hash"] == binding_hash
+            assert migrated["handoff_status"] == "ready"
+            assert migrated["ready_at"] is not None
+            connection.execute(
+                text(
+                    """
+                    DELETE FROM public.rule_run_artifacts
+                    WHERE id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": old_artifact_id},
+            )
+
+        store = _store(FakeMinio())
+        resolver = PostgresArtifactResolver(engine, store)
+        schema_fields = [
+            {"name": "id", "type": "integer", "nullable": False}
+        ]
+        same_path = tmp_path / "same.parquet"
+        conflict_path = tmp_path / "conflict.parquet"
+        pl.DataFrame({"id": [1]}).write_parquet(same_path)
+        pl.DataFrame({"id": [2]}).write_parquet(conflict_path)
+        first = resolver.publish_path(
+            str(same_path),
+            binding_id=binding_id,
+            binding_hash=binding_hash,
+            correlation_id=correlation_id,
+            kind="output",
+            ttl_seconds=300,
+            schema_fields=schema_fields,
+        )
+        repeated = resolver.publish_path(
+            str(same_path),
+            binding_id=binding_id,
+            binding_hash=binding_hash,
+            correlation_id=correlation_id,
+            kind="output",
+            ttl_seconds=300,
+            schema_fields=schema_fields,
+        )
+        assert repeated["artifact_ref"] == first["artifact_ref"]
+        assert len(store.client.objects) == 1
+        with pytest.raises(ValueError, match="immutable|digest"):
+            resolver.publish_path(
+                str(conflict_path),
+                binding_id=binding_id,
+                binding_hash=binding_hash,
+                correlation_id=correlation_id,
+                kind="output",
+                ttl_seconds=300,
+                schema_fields=schema_fields,
+            )
+        assert len(store.client.objects) == 1
+
+        with pytest.raises(IntegrityError), engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_run_artifacts (
+                        id, correlation_id, binding_id, artifact_ref,
+                        artifact_digest, row_count, schema_hash,
+                        schema_fields, artifact_kind, binding_hash,
+                        handoff_status, expires_at
+                    ) VALUES (
+                        CAST(:id AS uuid),
+                        CAST(:correlation_id AS uuid),
+                        CAST(:binding_id AS uuid), :artifact_ref,
+                        :artifact_digest, 1, :schema_hash,
+                        CAST(:fields AS jsonb), 'output', :binding_hash,
+                        'pending',
+                        CURRENT_TIMESTAMP + INTERVAL '5 minutes'
+                    )
+                    """
+                ),
+                {
+                    "id": new_governance_uid(),
+                    "correlation_id": correlation_id,
+                    "binding_id": binding_id,
+                    "artifact_ref": old_ref,
+                    "artifact_digest": "e" * 64,
+                    "schema_hash": "a" * 64,
+                    "fields": (
+                        '[{"name":"id","type":"integer",'
+                        '"nullable":false}]'
+                    ),
+                    "binding_hash": binding_hash,
+                },
+            )
+    finally:
+        if engine is not None:
+            engine.dispose()
+        with admin.connect() as connection:
+            connection.execute(
+                text(
+                    """
+                    SELECT pg_terminate_backend(pid)
+                    FROM pg_stat_activity
+                    WHERE datname = :database_name
+                      AND pid <> pg_backend_pid()
+                    """
+                ),
+                {"database_name": database_name},
+            )
+            connection.execute(text(f'DROP DATABASE IF EXISTS "{database_name}"'))
+        admin.dispose()

tests/runner/test_artifact_handoff.py

@@ -0,0 +1,475 @@
+from __future__ import annotations
+
+import copy
+import json
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import polars as pl
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+from tests.runner.test_artifacts import FakeMinio, _store
+
+
+class _Result:
+    def __init__(self, row=None, *, rowcount=0):
+        self.row = row
+        self.rowcount = rowcount
+
+    def mappings(self):
+        return self
+
+    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 _HandoffConnection:
+    def __init__(self, engine, *, transactional):
+        self.engine = engine
+        self.transactional = transactional
+        self.backup = None
+        self.finalized = False
+
+    def __enter__(self):
+        self.backup = copy.deepcopy(self.engine.row)
+        return self
+
+    def __exit__(self, exc_type, *_args):
+        if exc_type is not None:
+            self.engine.row = self.backup
+            return False
+        if self.finalized and self.engine.finalize_mode:
+            if self.engine.finalize_mode == "rollback_unknown":
+                self.engine.row = self.backup
+                self.engine.fail_recheck = True
+            raise RuntimeError("database commit acknowledgement lost")
+        return False
+
+    def execute(self, statement, parameters):
+        sql = str(statement)
+        self.engine.events.append(sql)
+        if (
+            "FROM public.dataflow_dataset_bindings" in sql
+            and "JOIN" not in sql
+        ):
+            if self.engine.fail_reserve:
+                raise RuntimeError("database unavailable")
+            return _Result(copy.deepcopy(self.engine.binding))
+        if "INSERT INTO public.rule_run_artifacts" in sql:
+            if self.engine.row is not None:
+                return _Result(None)
+            self.engine.row = {
+                "id": parameters["id"],
+                "correlation_id": parameters["correlation_id"],
+                "binding_id": parameters["binding_id"],
+                "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"]),
+                "artifact_kind": parameters["artifact_kind"],
+                "binding_hash": parameters["binding_hash"],
+                "expires_at": parameters["expires_at"],
+                "handoff_status": "pending",
+            }
+            return _Result(copy.deepcopy(self.engine.row), rowcount=1)
+        if (
+            "UPDATE public.rule_run_artifacts" in sql
+            and "handoff_status = 'ready'" in sql
+        ):
+            if self.engine.row is not None:
+                self.engine.row["handoff_status"] = "ready"
+                self.finalized = True
+                return _Result(copy.deepcopy(self.engine.row), rowcount=1)
+            return _Result(None)
+        if (
+            "DELETE FROM public.rule_run_artifacts" in sql
+            and "handoff_status = 'pending'" in sql
+        ):
+            deleted = self.engine.row is not None
+            self.engine.row = None
+            return _Result(rowcount=int(deleted))
+        if "FROM public.rule_run_artifacts" in sql:
+            if self.engine.fail_recheck:
+                raise RuntimeError("database recheck unavailable")
+            row = copy.deepcopy(self.engine.row)
+            if "JOIN public.dataflow_dataset_bindings" in sql:
+                if (
+                    row is None
+                    or row["handoff_status"] != "ready"
+                    or row["binding_hash"] != self.engine.binding[
+                        "binding_hash"
+                    ]
+                ):
+                    row = None
+                elif row is not None:
+                    row["current_binding_hash"] = self.engine.binding[
+                        "binding_hash"
+                    ]
+            return _Result(row)
+        raise AssertionError(sql)
+
+
+class HandoffEngine:
+    def __init__(self, binding):
+        self.binding = dict(binding)
+        self.row = None
+        self.events = []
+        self.fail_reserve = False
+        self.fail_recheck = False
+        self.finalize_mode = None
+
+    def begin(self):
+        return _HandoffConnection(self, transactional=True)
+
+    def connect(self):
+        return _HandoffConnection(self, transactional=False)
+
+
+class EventMinio(FakeMinio):
+    def __init__(self, events):
+        super().__init__()
+        self.events = events
+        self.fail_upload = False
+
+    def put_object(self, *args, **kwargs):
+        self.events.append("MINIO PUT")
+        if self.fail_upload:
+            raise RuntimeError("object upload failed")
+        return super().put_object(*args, **kwargs)
+
+
+def _binding(binding_hash):
+    return {
+        "binding_hash": binding_hash,
+        "access_mode": "read_write",
+        "object_kind": "parquet_artifact",
+    }
+
+
+def _publish_fixture(tmp_path, *, engine=None, client=None):
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    binding_id = new_governance_uid()
+    binding_hash = "b" * 64
+    engine = engine or HandoffEngine(_binding(binding_hash))
+    client = client or EventMinio(engine.events)
+    store = _store(client)
+    path = tmp_path / "handoff.parquet"
+    pl.DataFrame({"id": [1]}).write_parquet(path)
+    resolver = PostgresArtifactResolver(engine, store)
+    return {
+        "binding_id": binding_id,
+        "binding_hash": binding_hash,
+        "client": client,
+        "engine": engine,
+        "path": path,
+        "resolver": resolver,
+        "store": store,
+    }
+
+
+def _publish(fixture):
+    return fixture["resolver"].publish_path(
+        str(fixture["path"]),
+        binding_id=fixture["binding_id"],
+        binding_hash=fixture["binding_hash"],
+        correlation_id=new_governance_uid(),
+        kind="output",
+        ttl_seconds=60,
+        schema_fields=[
+            {"name": "id", "type": "integer", "nullable": False}
+        ],
+    )
+
+
+def test_publish_reserves_pending_before_upload_and_finalizes_ready(tmp_path):
+    fixture = _publish_fixture(tmp_path)
+
+    result = _publish(fixture)
+
+    events = fixture["engine"].events
+    assert next(
+        index
+        for index, event in enumerate(events)
+        if "INSERT INTO public.rule_run_artifacts" in event
+    ) < events.index("MINIO PUT")
+    assert events.index("MINIO PUT") < next(
+        index
+        for index, event in enumerate(events)
+        if "handoff_status = 'ready'" in event
+    )
+    assert fixture["engine"].row["handoff_status"] == "ready"
+    assert fixture["engine"].row["binding_hash"] == fixture["binding_hash"]
+    assert result["artifact_ref"] == fixture["engine"].row["artifact_ref"]
+
+
+def test_reserve_locks_and_rejects_stale_binding_before_upload(tmp_path):
+    fixture = _publish_fixture(tmp_path)
+    fixture["engine"].binding["binding_hash"] = "c" * 64
+
+    with pytest.raises(ValueError, match="binding"):
+        _publish(fixture)
+
+    assert any(
+        "FOR SHARE" in event
+        for event in fixture["engine"].events
+        if "dataflow_dataset_bindings" in event
+    )
+    assert "MINIO PUT" not in fixture["engine"].events
+    assert fixture["engine"].row is None
+
+
+def test_reserve_crash_never_uploads_an_object(tmp_path):
+    from app.runner.artifacts import ArtifactCommitUnknown
+
+    fixture = _publish_fixture(tmp_path)
+    fixture["engine"].fail_reserve = True
+
+    with pytest.raises(ArtifactCommitUnknown):
+        _publish(fixture)
+
+    assert "MINIO PUT" not in fixture["engine"].events
+    assert fixture["client"].objects == {}
+
+
+def test_upload_failure_removes_pending_reservation(tmp_path):
+    fixture = _publish_fixture(tmp_path)
+    fixture["client"].fail_upload = True
+
+    with pytest.raises(RuntimeError, match="upload"):
+        _publish(fixture)
+
+    assert fixture["engine"].row is None
+    assert fixture["client"].objects == {}
+
+
+def test_finalize_commit_after_client_error_is_rechecked_as_success(tmp_path):
+    fixture = _publish_fixture(tmp_path)
+    fixture["engine"].finalize_mode = "committed_then_error"
+
+    result = _publish(fixture)
+
+    assert fixture["engine"].row["handoff_status"] == "ready"
+    assert result["artifact_ref"] == fixture["engine"].row["artifact_ref"]
+    assert fixture["client"].objects
+
+
+def test_unconfirmed_finalize_returns_unknown_without_deleting_object(tmp_path):
+    from app.runner.artifacts import ArtifactCommitUnknown
+
+    fixture = _publish_fixture(tmp_path)
+    fixture["engine"].finalize_mode = "rollback_unknown"
+
+    with pytest.raises(ArtifactCommitUnknown):
+        _publish(fixture)
+
+    assert fixture["client"].objects
+
+
+class ReconcileMinio(FakeMinio):
+    def __init__(self, now):
+        super().__init__()
+        self.now = now
+        self.ages = {}
+
+    def list_objects(self, bucket, *, prefix, recursive):
+        assert recursive is True
+        return [
+            SimpleNamespace(
+                object_name=key,
+                last_modified=self.ages.get(key, self.now),
+            )
+            for object_bucket, key in sorted(self.objects)
+            if object_bucket == bucket and key.startswith(prefix)
+        ]
+
+
+class ReconcileEngine:
+    def __init__(self, rows):
+        self.rows = [dict(row) for row in rows]
+        self.events = []
+
+    def begin(self):
+        return _ReconcileConnection(self)
+
+    def connect(self):
+        return _ReconcileConnection(self)
+
+
+class _ReconcileConnection:
+    def __init__(self, engine):
+        self.engine = engine
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+    def execute(self, statement, parameters):
+        sql = str(statement)
+        self.engine.events.append(sql)
+        if (
+            "SELECT id::text AS id" in sql
+            and "handoff_status IN" in sql
+        ):
+            return _Result(
+                copy.deepcopy(self.engine.rows[: parameters["limit"]])
+            )
+        if "SELECT artifact_ref" in sql and "handoff_status" not in sql:
+            return _Result(
+                [{"artifact_ref": row["artifact_ref"]} for row in self.engine.rows]
+            )
+        row = next(
+            (
+                item
+                for item in self.engine.rows
+                if item["id"] == parameters.get("id")
+            ),
+            None,
+        )
+        if (
+            "UPDATE public.rule_run_artifacts" in sql
+            and "handoff_status = 'ready'" in sql
+        ):
+            if row is not None:
+                row["handoff_status"] = "ready"
+            return _Result(copy.deepcopy(row), rowcount=int(row is not None))
+        if (
+            "UPDATE public.rule_run_artifacts" in sql
+            and "handoff_status = 'failed'" in sql
+        ):
+            if row is not None:
+                row["handoff_status"] = "failed"
+                row["failure_code"] = parameters["failure_code"]
+            return _Result(rowcount=int(row is not None))
+        if "DELETE FROM public.rule_run_artifacts" in sql:
+            before = len(self.engine.rows)
+            self.engine.rows = [
+                item
+                for item in self.engine.rows
+                if item["id"] != parameters["id"]
+            ]
+            return _Result(rowcount=before - len(self.engine.rows))
+        raise AssertionError(sql)
+
+
+def _catalog_row(artifact, *, status, binding_hash="d" * 64):
+    return {
+        "id": new_governance_uid(),
+        "artifact_ref": artifact["artifact_ref"],
+        "artifact_digest": artifact["digest"],
+        "row_count": artifact["row_count"],
+        "schema_hash": artifact["schema_hash"],
+        "schema_fields": artifact["schema_fields"],
+        "expires_at": artifact["expires_at"],
+        "handoff_status": status,
+        "binding_hash": binding_hash,
+        "binding_id": new_governance_uid(),
+        "artifact_kind": "output",
+        "correlation_id": artifact["artifact_ref"].split("/")[4],
+    }
+
+
+def test_reconcile_repairs_both_catalog_and_object_store_safely():
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    now = datetime(2026, 7, 23, 12, 0, tzinfo=UTC)
+    client = ReconcileMinio(now)
+    store = _store(client, clock=lambda: now)
+    valid_pending = store.write(
+        pl.DataFrame({"id": [1]}),
+        new_governance_uid(),
+        600,
+    )
+    missing_pending = store.write(
+        pl.DataFrame({"id": [2]}),
+        new_governance_uid(),
+        600,
+    )
+    store.delete(missing_pending["artifact_ref"])
+    missing_ready = store.write(
+        pl.DataFrame({"id": [3]}),
+        new_governance_uid(),
+        600,
+    )
+    store.delete(missing_ready["artifact_ref"])
+    invalid_pending = store.write(
+        pl.DataFrame({"id": [6]}),
+        new_governance_uid(),
+        600,
+    )
+    invalid_key = invalid_pending["artifact_ref"].split("/", 3)[-1]
+    client.objects[("dataops-rules", invalid_key)][
+        "content_type"
+    ] = "text/plain"
+    old_orphan = store.write(
+        pl.DataFrame({"id": [4]}),
+        new_governance_uid(),
+        600,
+    )
+    fresh_orphan = store.write(
+        pl.DataFrame({"id": [5]}),
+        new_governance_uid(),
+        600,
+    )
+    old_key = old_orphan["artifact_ref"].split("/", 3)[-1]
+    fresh_key = fresh_orphan["artifact_ref"].split("/", 3)[-1]
+    client.ages[old_key] = now - timedelta(minutes=20)
+    client.ages[fresh_key] = now - timedelta(seconds=30)
+    client.objects[
+        ("dataops-rules", "rules/not-a-safe-catalog-key.parquet")
+    ] = copy.deepcopy(next(iter(client.objects.values())))
+    client.ages["rules/not-a-safe-catalog-key.parquet"] = now - timedelta(
+        hours=1
+    )
+    engine = ReconcileEngine(
+        [
+            _catalog_row(valid_pending, status="pending"),
+            _catalog_row(missing_pending, status="pending"),
+            _catalog_row(missing_ready, status="ready"),
+            _catalog_row(invalid_pending, status="pending"),
+        ]
+    )
+
+    result = PostgresArtifactResolver(engine, store).reconcile(
+        limit=10,
+        grace_seconds=300,
+    )
+
+    assert result == {
+        "pending_finalized": 1,
+        "pending_deleted": 1,
+        "ready_failed": 1,
+        "orphans_deleted": 1,
+    }
+    assert ("dataops-rules", old_key) not in client.objects
+    assert ("dataops-rules", fresh_key) in client.objects
+    assert ("dataops-rules", invalid_key) not in client.objects
+    assert (
+        "dataops-rules",
+        "rules/not-a-safe-catalog-key.parquet",
+    ) in client.objects
+    assert any(
+        row["handoff_status"] == "ready"
+        and row["artifact_ref"] == valid_pending["artifact_ref"]
+        for row in engine.rows
+    )
+    assert any(
+        row["handoff_status"] == "failed"
+        and row["artifact_ref"] == missing_ready["artifact_ref"]
+        for row in engine.rows
+    )
+    assert any(
+        row["handoff_status"] == "failed"
+        and row["artifact_ref"] == invalid_pending["artifact_ref"]
+        and row["failure_code"] == "pending_object_invalid"
+        for row in engine.rows
+    )

tests/runner/test_artifacts.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import io
-import json
 from datetime import UTC, datetime
 from types import SimpleNamespace
 
@@ -341,6 +340,34 @@ def test_artifact_stage_rejects_compression_bomb_from_footer_before_scan():
     assert len(client.get_calls) == calls_before + 1
 
 
+def test_artifact_prepare_reserves_key_before_exact_path_upload(tmp_path):
+    client = FakeMinio()
+    store = _store(client)
+    correlation_id = new_governance_uid()
+    path = tmp_path / "prepared.parquet"
+    pl.DataFrame({"id": [1, 2]}).write_parquet(path)
+    fields = [{"name": "id", "type": "integer", "nullable": False}]
+
+    prepared = store.prepare_path(
+        str(path),
+        correlation_id,
+        60,
+        schema_fields=fields,
+    )
+
+    assert client.objects == {}
+    assert prepared["artifact_ref"].startswith(
+        f"minio://dataops-rules/rules/{correlation_id}/"
+    )
+
+    uploaded = store.upload_path(str(path), prepared)
+
+    assert uploaded == prepared
+    assert store.describe(prepared["artifact_ref"])["digest"] == prepared[
+        "digest"
+    ]
+
+
 def test_artifact_schema_contract_covers_nullability_decimal_and_timezone():
     client = FakeMinio()
     store = _store(client)
@@ -494,7 +521,8 @@ def test_postgres_artifact_resolver_uses_catalog_and_rechecks_binding_hash():
             return {
                 **artifact,
                 "artifact_digest": artifact["digest"],
-                "binding_hash": binding_hash,
+                "catalog_binding_hash": binding_hash,
+                "current_binding_hash": binding_hash,
             }
         raise AssertionError(sql)
 
@@ -511,6 +539,8 @@ def test_postgres_artifact_resolver_uses_catalog_and_rechecks_binding_hash():
     assert "a.correlation_id = CAST(:correlation_id AS uuid)" in (
         engine.calls[0][0]
     )
+    assert "a.handoff_status = 'ready'" in engine.calls[0][0]
+    assert "a.binding_hash = b.binding_hash" in engine.calls[0][0]
 
     with pytest.raises(ValueError, match="correlation"):
         PostgresArtifactResolver(engine, store).resolve(
@@ -520,160 +550,6 @@ def test_postgres_artifact_resolver_uses_catalog_and_rechecks_binding_hash():
         )
 
 
-def test_postgres_artifact_resolver_attests_and_registers_stable_handoff():
-    from app.runner.artifacts import PostgresArtifactResolver
-
-    store = _store(FakeMinio())
-    correlation_id = new_governance_uid()
-    binding_id = new_governance_uid()
-    binding_hash = "b" * 64
-    artifact = store.write(
-        pl.DataFrame({"id": [1]}),
-        correlation_id,
-        60,
-    )
-
-    def handler(sql, parameters):
-        if "FROM public.dataflow_dataset_bindings" in sql:
-            return {
-                "binding_hash": binding_hash,
-                "access_mode": "read_write",
-                "object_kind": "parquet_artifact",
-            }
-        if "INSERT INTO public.rule_run_artifacts" in sql:
-            assert parameters["artifact_digest"] == artifact["digest"]
-            assert parameters["schema_fields"]
-            return {
-                "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)
-    resolver = PostgresArtifactResolver(engine, store)
-    resolver.attest_binding(
-        binding_id=binding_id,
-        binding_hash=binding_hash,
-        access_mode="write",
-    )
-    resolver.register(
-        binding_id=binding_id,
-        correlation_id=correlation_id,
-        artifact=artifact,
-        kind="output",
-        binding_hash=binding_hash,
-    )
-
-    assert sum(
-        "FROM public.dataflow_dataset_bindings" in sql
-        for sql, _parameters in engine.calls
-    ) == 2
-    assert any(
-        "INSERT INTO public.rule_run_artifacts" in sql
-        for sql, _parameters in engine.calls
-    )
-
-
-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
 

tests/runner/test_bootstrap.py

@@ -51,23 +51,38 @@ def test_runner_refuses_to_start_without_a_task_token_secret(monkeypatch):
         runner_settings_from_env()
 
 
-def test_runner_registers_bounded_artifact_cleanup_cli():
+def test_runner_registers_bounded_artifact_reconciliation_cli():
     class Resolver:
         def __init__(self):
             self.limit = None
+            self.grace_seconds = None
 
-        def cleanup_expired(self, *, limit):
+        def reconcile(self, *, limit, grace_seconds):
             self.limit = limit
-            return 3
+            self.grace_seconds = grace_seconds
+            return {
+                "pending_finalized": 1,
+                "pending_deleted": 1,
+                "ready_failed": 1,
+                "orphans_deleted": 1,
+            }
 
     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"]
+        args=[
+            "reconcile-rule-artifacts",
+            "--limit",
+            "25",
+            "--grace-seconds",
+            "600",
+        ]
     )
 
     assert result.exit_code == 0
-    assert result.output.strip() == "removed 3 expired rule artifacts"
+    assert "pending_finalized=1" in result.output
+    assert "orphans_deleted=1" in result.output
     assert resolver.limit == 25
+    assert resolver.grace_seconds == 600

tests/runner/test_polars_worker.py

@@ -53,7 +53,7 @@ def _assert_worker_resource_failure(plan, tmp_path, *, lookup_paths=None):
 
     with pytest.raises(
         PolarsWorkerResourceError,
-        match="hard memory limit",
+        match="hard memory limit|bounded execution time",
     ):
         execute_isolated_polars_plan(
             {
@@ -97,6 +97,70 @@ def test_isolated_polars_worker_fails_deterministically_at_hard_memory_limit():
         )
 
 
+def test_worker_start_failure_closes_pipe_endpoints_and_maps_safe_error(
+    monkeypatch,
+):
+    from app.runner import polars_worker
+
+    class Endpoint:
+        def __init__(self):
+            self.closed = False
+
+        def close(self):
+            self.closed = True
+
+    class Process:
+        pid = None
+
+        def __init__(self):
+            self.joined = False
+            self.closed = False
+
+        def start(self):
+            raise OSError("sensitive spawn detail")
+
+        def is_alive(self):
+            return False
+
+        def join(self, timeout=None):
+            self.joined = True
+
+        def close(self):
+            self.closed = True
+
+    parent = Endpoint()
+    child = Endpoint()
+    process = Process()
+
+    class Context:
+        def Pipe(self, duplex):
+            assert duplex is False
+            return parent, child
+
+        def Process(self, **_kwargs):
+            return process
+
+    monkeypatch.setattr(
+        polars_worker.multiprocessing,
+        "get_context",
+        lambda _method: Context(),
+    )
+
+    with pytest.raises(
+        polars_worker.PolarsWorkerError,
+        match="failed to start safely",
+    ) as error:
+        polars_worker.run_isolated_memory_probe(
+            allocate_bytes=1,
+            memory_limit_bytes=1024,
+        )
+
+    assert "sensitive" not in str(error.value)
+    assert parent.closed is True
+    assert child.closed is True
+    assert process.closed is True
+
+
 def test_regex_peak_allocation_fails_inside_isolated_worker(tmp_path):
     schema = _schema(
         "bd:regex:raw",

tests/runner/test_rule_polars.py

@@ -31,11 +31,13 @@ def _plan_store(client):
 
 
 class Resolver:
-    def __init__(self, values):
+    def __init__(self, values, artifact_store):
         self.values = values
+        self.artifact_store = artifact_store
         self.calls = []
         self.events = []
         self.attest_error = None
+        self.publish_error = None
         self.registrations = []
 
     def resolve(self, *, binding_id, correlation_id, kind):
@@ -49,16 +51,28 @@ class Resolver:
             raise self.attest_error
         return {"binding_hash": binding_hash}
 
-    def register(
+    def publish_path(
         self,
+        path,
         *,
         binding_id,
         correlation_id,
-        artifact,
         kind,
         binding_hash,
+        ttl_seconds,
+        schema_fields,
+        limits,
     ):
-        self.events.append(("register", binding_id, kind))
+        self.events.append(("publish", binding_id, kind))
+        if self.publish_error is not None:
+            raise self.publish_error
+        artifact = self.artifact_store.write_path(
+            path,
+            correlation_id,
+            ttl_seconds,
+            schema_fields=schema_fields,
+            limits=limits,
+        )
         self.registrations.append(
             {
                 "binding_id": binding_id,
@@ -160,7 +174,8 @@ def test_polars_adapter_reconstructs_assert_and_deduplicate_and_writes_artifact(
                 **source,
                 "binding_hash": compiled["plan"]["input_binding_hash"],
             }
-        }
+        },
+        store,
     )
     adapter = PolarsRulePlanAdapter(
         artifact_store=store,
@@ -213,6 +228,57 @@ def test_polars_adapter_reconstructs_assert_and_deduplicate_and_writes_artifact(
     ]
 
 
+def test_polars_adapter_reports_unknown_catalog_commit_outcome():
+    from app.runner.artifacts import ArtifactCommitUnknown
+    from app.runner.rule_polars import PolarsRulePlanAdapter
+
+    compiled, input_binding = _compiled_plan(
+        [
+            {
+                "id": "trim_name",
+                "op": "normalize_text",
+                "column": "name",
+                "trim": True,
+            }
+        ]
+    )
+    store = _plan_store(FakeMinio())
+    correlation_id = new_governance_uid()
+    source = store.write(
+        pl.DataFrame(
+            {"customer_id": [1], "name": [" A "], "mobile": ["1"]}
+        ),
+        correlation_id,
+        600,
+        schema_fields=compiled["plan"]["input_fields"],
+        limits=compiled["plan"]["resource_limits"],
+    )
+    resolver = Resolver(
+        {
+            input_binding["id"]: {
+                **source,
+                "binding_hash": compiled["plan"]["input_binding_hash"],
+            }
+        },
+        store,
+    )
+    resolver.publish_error = ArtifactCommitUnknown("lost acknowledgement")
+
+    with pytest.raises(NodeExecutionError, match="commit outcome") as error:
+        PolarsRulePlanAdapter(
+            artifact_store=store,
+            artifact_resolver=resolver,
+        ).execute(
+            plan=compiled["plan"],
+            node=_node(compiled),
+            parameters={},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+
+    assert error.value.commit_outcome == "unknown"
+
+
 def test_polars_adapter_fails_closed_for_plan_hash_binding_and_authorization():
     from app.runner.rule_polars import PolarsRulePlanAdapter
 
@@ -243,7 +309,8 @@ def test_polars_adapter_fails_closed_for_plan_hash_binding_and_authorization():
                 **source,
                 "binding_hash": "0" * 64,
             }
-        }
+        },
+        store,
     )
     adapter = PolarsRulePlanAdapter(
         artifact_store=store,
@@ -323,7 +390,8 @@ def test_polars_adapter_attests_current_output_binding_before_reading_input():
                 **source,
                 "binding_hash": compiled["plan"]["input_binding_hash"],
             }
-        }
+        },
+        store,
     )
     resolver.attest_error = ValueError("binding changed")
     reads_before_execute = list(store.client.get_calls)
@@ -425,7 +493,8 @@ def test_polars_adapter_uses_exact_decimal_and_timestamptz_output_contracts():
                 **source,
                 "binding_hash": compiled["plan"]["input_binding_hash"],
             }
-        }
+        },
+        store,
     )
 
     result = PolarsRulePlanAdapter(
@@ -526,7 +595,8 @@ def test_polars_expression_date_and_timestamp_use_plan_timezone():
                 **source,
                 "binding_hash": compiled["plan"]["input_binding_hash"],
             }
-        }
+        },
+        store,
     )
 
     result = PolarsRulePlanAdapter(

tests/test_data_rule_schema.py

@@ -30,6 +30,12 @@ ARTIFACT_CATALOG_MIGRATION = (
     / "versions"
     / "20260723_140_rule_run_artifacts.py"
 )
+ARTIFACT_HANDOFF_MIGRATION = (
+    ROOT
+    / "migrations"
+    / "versions"
+    / "20260723_150_rule_artifact_handoff_state.py"
+)
 
 EXPECTED_TABLES = {
     "data_rules",
@@ -148,15 +154,38 @@ 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)",
+    ):
+        assert expected in source
+    downgrade = source.split("def downgrade()", 1)[1]
+    assert "DROP TABLE" not in downgrade.upper()
+    assert "pass" in downgrade
+
+
+def test_artifact_handoff_state_migration_upgrades_old_140_forward_only():
+    source = ARTIFACT_HANDOFF_MIGRATION.read_text(encoding="utf-8")
+
+    assert 'revision = "20260723_150"' in source
+    assert 'down_revision = "20260723_140"' in source
+    for expected in (
+        "binding_hash CHAR(64)",
+        "handoff_status VARCHAR(20)",
+        "ready_at TIMESTAMPTZ",
+        "failed_at TIMESTAMPTZ",
+        "updated_at TIMESTAMPTZ",
         "UNIQUE (correlation_id, binding_id, artifact_kind)",
+        "'pending','ready','failed'",
+        "SELECT binding_hash",
     ):
         assert expected in source
+    assert "artifact_digest" in source
+    assert "RAISE EXCEPTION" in source
     downgrade = source.split("def downgrade()", 1)[1]
     assert "DROP TABLE" not in downgrade.upper()
     assert "raise RuntimeError" in downgrade
     spec = importlib.util.spec_from_file_location(
-        "rule_run_artifact_catalog_migration",
-        ARTIFACT_CATALOG_MIGRATION,
+        "rule_artifact_handoff_state_migration",
+        ARTIFACT_HANDOFF_MIGRATION,
     )
     assert spec is not None and spec.loader is not None
     module = importlib.util.module_from_spec(spec)