fix: harden artifact reconciliation races

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

@@ -136,6 +136,47 @@ Third-remediation GREEN:
 - `git diff f4d798b -- migrations/versions/20260723_140_rule_run_artifacts.py`:
   empty.
 
+## Fourth-remediation RED/GREEN evidence
+
+The fourth review closed concurrency and storage-classification gaps in
+reconciliation. Fail-first tests reproduced all four defects:
+
+- a one-second prepared TTL became zero after a microsecond of validation delay
+  because `int()` rounded the remaining interval down;
+- a pending snapshot raced with another finalizer, observed `UPDATE` rowcount
+  zero, and the old broad failure update downgraded the new ready row;
+- timeout, network, and authentication failures from MinIO were treated as
+  proof of an invalid object for both pending and ready catalog rows;
+- a same-size mutated Parquet payload with unchanged trusted-looking stat
+  metadata was finalized because reconciliation never streamed its content.
+
+The GREEN implementation and coverage now prove:
+
+- upload validation uses ceiling semantics for the absolute expiry interval, so
+  `ttl_seconds=1` remains valid while it has positive lifetime;
+- every failure transition is an expected-state CAS; a stale pending snapshot
+  cannot update a ready row;
+- a zero-row pending finalize rereads by immutable row id, accepts an exact
+  matching ready row, marks only a still-pending binding mismatch, and leaves
+  other current states unchanged;
+- confirmed `NotFound` is distinct from deterministic metadata/content/schema
+  invalidity and from temporary timeout/network/AccessDenied failures;
+- temporary failures during either stat or streaming download preserve both
+  pending and ready catalog state and never delete the object;
+- pending recovery streams and hashes the object and validates the Parquet
+  footer and exact schema before ready; same-size content mutation fails;
+- a freshly observed ready row with deterministically invalid content uses a
+  ready-only CAS to transition to failed.
+
+Fourth-remediation verification:
+
+- focused artifact/handoff tests: `36 passed`;
+- focused schema/artifact/handoff/bootstrap/worker/adapter plus real old140
+  migration and real cross-source integration: `63 passed in 5.21s`;
+- final full suite: `578 passed, 26 skipped, 59 subtests passed`;
+- Ruff, `git diff --check`, Compose config validation, immutable-140 diff, and
+  unchanged-150 diff: passed.
+
 ## Durable artifact handoff, attestation, and reconciliation
 
 Migration `20260723_140` is historical and unchanged. It retains its original
@@ -179,6 +220,16 @@ 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.
 
+Reconciliation does not trust object stat metadata as content attestation.
+Before a pending row becomes ready it streams the exact object to a temporary
+file and verifies digest, Parquet footer row/uncompressed bounds, and the exact
+catalog schema. Failure transitions include the snapshot's expected state in
+their `WHERE` clause. If a pending finalization loses its race, the resolver
+rereads the row and treats only an exact matching ready handoff as success.
+Confirmed absence and deterministic invalidity follow catalog repair policy;
+temporary MinIO authentication, server, transport, and timeout errors cause no
+catalog or object mutation in that reconciliation pass.
+
 `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

app/runner/artifacts.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import math
 import os
 import re
 import tempfile
@@ -27,6 +28,13 @@ PARQUET_CONTENT_TYPE = "application/x-parquet"
 _DIGEST = re.compile(r"^[0-9a-f]{64}$")
 
 
+def _confirmed_object_missing(exc: BaseException) -> bool:
+    return isinstance(exc, KeyError) or (
+        isinstance(exc, S3Error)
+        and exc.code in {"NoSuchKey", "NoSuchObject", "NotFound"}
+    )
+
+
 class ArtifactCommitUnknown(RuntimeError):
     """A catalog transaction may have committed but cannot be confirmed."""
 
@@ -521,7 +529,7 @@ class ArtifactStore:
             raise ValueError("prepared artifact metadata has a closed shape")
         key = self._parse_ref(artifact["artifact_ref"])
         correlation = key.split("/", 2)[1]
-        ttl_seconds = int(
+        ttl_seconds = math.ceil(
             (
                 _parse_timestamp(artifact["expires_at"])
                 - _now_utc(self.clock)
@@ -616,10 +624,8 @@ class ArtifactStore:
 
         try:
             return self.describe(ref)
-        except KeyError:
-            return None
-        except S3Error as exc:
-            if exc.code in {"NoSuchKey", "NoSuchObject", "NotFound"}:
+        except Exception as exc:
+            if _confirmed_object_missing(exc):
                 return None
             raise
 
@@ -1291,10 +1297,13 @@ class PostgresArtifactResolver:
         self,
         *,
         row_id: str,
+        expected_status: str,
         failure_code: str,
-    ) -> None:
+    ) -> bool:
+        if expected_status not in {"pending", "ready"}:
+            raise ValueError("artifact expected handoff status is invalid")
         with self.engine.begin() as connection:
-            connection.execute(
+            updated = connection.execute(
                 text(
                     """
                     UPDATE public.rule_run_artifacts
@@ -1303,11 +1312,92 @@ class PostgresArtifactResolver:
                         updated_at = CURRENT_TIMESTAMP,
                         failure_code = :failure_code
                     WHERE id = CAST(:id AS uuid)
-                      AND handoff_status IN ('pending','ready')
+                      AND handoff_status = :expected_status
                     """
                 ),
-                {"id": row_id, "failure_code": failure_code},
+                {
+                    "id": row_id,
+                    "expected_status": expected_status,
+                    "failure_code": failure_code,
+                },
             )
+        return int(updated.rowcount or 0) == 1
+
+    def _lookup_handoff_by_id(
+        self,
+        row_id: 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 id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": row_id},
+            ).mappings().one_or_none()
+        return dict(row) if row is not None else None
+
+    @staticmethod
+    def _matching_ready(
+        snapshot: Mapping[str, Any],
+        current: Mapping[str, Any] | None,
+    ) -> bool:
+        if current is None or str(current.get("handoff_status")) != "ready":
+            return False
+        return all(
+            str(current.get(key)) == str(snapshot.get(key))
+            for key in (
+                "id",
+                "correlation_id",
+                "binding_id",
+                "artifact_ref",
+                "artifact_digest",
+                "row_count",
+                "schema_hash",
+                "artifact_kind",
+                "binding_hash",
+            )
+        )
+
+    def _verify_reconcile_object(
+        self,
+        row: Mapping[str, Any],
+    ) -> dict[str, Any] | None:
+        artifact_ref = str(row["artifact_ref"])
+        self.artifact_store._parse_ref(artifact_ref)
+        stored = self.artifact_store.describe_optional(artifact_ref)
+        if stored is None:
+            return None
+        if any(
+            (
+                stored["digest"] != str(row["artifact_digest"]),
+                stored["row_count"] != int(row["row_count"]),
+                stored["schema_hash"] != str(row["schema_hash"]),
+            )
+        ):
+            raise ValueError("catalog artifact metadata does not match storage")
+        fields = row["schema_fields"]
+        if isinstance(fields, str):
+            fields = json.loads(fields)
+        try:
+            with self.artifact_store.stage(
+                artifact_ref,
+                str(row["artifact_digest"]),
+                expected_schema_fields=fields,
+            ):
+                pass
+        except Exception as exc:
+            if _confirmed_object_missing(exc):
+                return None
+            raise
+        return stored
 
     def reconcile(
         self,
@@ -1363,37 +1453,40 @@ class PostgresArtifactResolver:
             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:
+                stored = self._verify_reconcile_object(row)
+            except ValueError:
                 if status == "pending":
-                    with suppress(Exception):
-                        self.artifact_store.delete(row["artifact_ref"])
-                    self._mark_failed(
+                    failed = self._mark_failed(
                         row_id=row_id,
+                        expected_status="pending",
                         failure_code="pending_object_invalid",
                     )
+                    if failed:
+                        with suppress(Exception):
+                            self.artifact_store.delete(row["artifact_ref"])
                 else:
-                    self._mark_failed(
+                    failed = self._mark_failed(
                         row_id=row_id,
+                        expected_status="ready",
                         failure_code="ready_object_invalid",
                     )
-                    result["ready_failed"] += 1
+                    if failed:
+                        result["ready_failed"] += 1
+                continue
+            except Exception:
+                # Authentication, timeout, transport, and server failures are
+                # not evidence that a cataloged object is invalid.
                 continue
             if status == "ready":
                 if stored is None:
-                    self._mark_failed(
+                    failed = self._mark_failed(
                         row_id=row_id,
+                        expected_status="ready",
                         failure_code="ready_object_missing",
                     )
-                    result["ready_failed"] += 1
+                    if failed:
+                        result["ready_failed"] += 1
                 continue
             if stored is None:
                 with self.engine.begin() as connection:
@@ -1410,20 +1503,6 @@ class PostgresArtifactResolver:
                 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(
@@ -1454,10 +1533,18 @@ class PostgresArtifactResolver:
             if int(finalized.rowcount or 0) == 1:
                 result["pending_finalized"] += 1
             else:
-                self._mark_failed(
-                    row_id=row_id,
-                    failure_code="pending_binding_changed",
-                )
+                current = self._lookup_handoff_by_id(row_id)
+                if self._matching_ready(row, current):
+                    result["pending_finalized"] += 1
+                elif (
+                    current is not None
+                    and str(current.get("handoff_status")) == "pending"
+                ):
+                    self._mark_failed(
+                        row_id=row_id,
+                        expected_status="pending",
+                        failure_code="pending_binding_changed",
+                    )
 
         remaining = limit - len(rows)
         if remaining <= 0:

tests/runner/test_artifact_handoff.py

@@ -7,6 +7,7 @@ from types import SimpleNamespace
 
 import polars as pl
 import pytest
+from minio.error import S3Error
 
 from app.core.common.identifiers import new_governance_uid
 from tests.runner.test_artifacts import FakeMinio, _store
@@ -295,6 +296,7 @@ class ReconcileEngine:
     def __init__(self, rows):
         self.rows = [dict(row) for row in rows]
         self.events = []
+        self.finalize_race_to_ready = False
 
     def begin(self):
         return _ReconcileConnection(self)
@@ -323,6 +325,19 @@ class _ReconcileConnection:
             return _Result(
                 copy.deepcopy(self.engine.rows[: parameters["limit"]])
             )
+        if (
+            "SELECT id::text AS id" in sql
+            and "WHERE id = CAST(:id AS uuid)" in sql
+        ):
+            row = next(
+                (
+                    item
+                    for item in self.engine.rows
+                    if item["id"] == parameters["id"]
+                ),
+                None,
+            )
+            return _Result(copy.deepcopy(row))
         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]
@@ -339,17 +354,25 @@ class _ReconcileConnection:
             "UPDATE public.rule_run_artifacts" in sql
             and "handoff_status = 'ready'" in sql
         ):
-            if row is not None:
+            if self.engine.finalize_race_to_ready and row is not None:
                 row["handoff_status"] = "ready"
-            return _Result(copy.deepcopy(row), rowcount=int(row is not None))
+                return _Result(None, rowcount=0)
+            if row is not None and row["handoff_status"] == "pending":
+                row["handoff_status"] = "ready"
+                return _Result(copy.deepcopy(row), rowcount=1)
+            return _Result(None, rowcount=0)
         if (
             "UPDATE public.rule_run_artifacts" in sql
             and "handoff_status = 'failed'" in sql
         ):
-            if row is not None:
+            if (
+                row is not None
+                and row["handoff_status"] == parameters["expected_status"]
+            ):
                 row["handoff_status"] = "failed"
                 row["failure_code"] = parameters["failure_code"]
-            return _Result(rowcount=int(row is not None))
+                return _Result(rowcount=1)
+            return _Result(rowcount=0)
         if "DELETE FROM public.rule_run_artifacts" in sql:
             before = len(self.engine.rows)
             self.engine.rows = [
@@ -473,3 +496,225 @@ def test_reconcile_repairs_both_catalog_and_object_store_safely():
         and row["failure_code"] == "pending_object_invalid"
         for row in engine.rows
     )
+
+
+def test_reconcile_accepts_concurrent_matching_ready_without_downgrade():
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    now = datetime(2026, 7, 23, 12, 0, tzinfo=UTC)
+    client = ReconcileMinio(now)
+    store = _store(client, clock=lambda: now)
+    artifact = store.write(
+        pl.DataFrame({"id": [1]}),
+        new_governance_uid(),
+        600,
+    )
+    engine = ReconcileEngine([_catalog_row(artifact, status="pending")])
+    engine.finalize_race_to_ready = True
+
+    result = PostgresArtifactResolver(engine, store).reconcile(
+        limit=1,
+        grace_seconds=300,
+    )
+
+    assert result["pending_finalized"] == 1
+    assert engine.rows[0]["handoff_status"] == "ready"
+    assert engine.rows[0].get("failure_code") is None
+
+
+@pytest.mark.parametrize(
+    ("status", "storage_error"),
+    [
+        ("pending", TimeoutError("MinIO timed out")),
+        ("ready", TimeoutError("MinIO timed out")),
+        ("pending", ConnectionError("MinIO network unavailable")),
+        ("ready", ConnectionError("MinIO network unavailable")),
+        (
+            "pending",
+            S3Error(
+                "AccessDenied",
+                "authentication unavailable",
+                None,
+                None,
+                None,
+                None,
+            ),
+        ),
+        (
+            "ready",
+            S3Error(
+                "AccessDenied",
+                "authentication unavailable",
+                None,
+                None,
+                None,
+                None,
+            ),
+        ),
+    ],
+)
+def test_reconcile_preserves_catalog_on_transient_storage_error(
+    status, storage_error
+):
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    now = datetime(2026, 7, 23, 12, 0, tzinfo=UTC)
+    client = ReconcileMinio(now)
+    store = _store(client, clock=lambda: now)
+    artifact = store.write(
+        pl.DataFrame({"id": [1]}),
+        new_governance_uid(),
+        600,
+    )
+    key = artifact["artifact_ref"].split("/", 3)[-1]
+    original_stat = client.stat_object
+
+    def fail_stat(bucket, object_key):
+        if object_key == key:
+            raise storage_error
+        return original_stat(bucket, object_key)
+
+    client.stat_object = fail_stat
+    engine = ReconcileEngine([_catalog_row(artifact, status=status)])
+
+    result = PostgresArtifactResolver(engine, store).reconcile(
+        limit=1,
+        grace_seconds=300,
+    )
+
+    assert result == {
+        "pending_finalized": 0,
+        "pending_deleted": 0,
+        "ready_failed": 0,
+        "orphans_deleted": 0,
+    }
+    assert engine.rows[0]["handoff_status"] == status
+    assert ("dataops-rules", key) in client.objects
+
+
+@pytest.mark.parametrize("status", ["pending", "ready"])
+def test_reconcile_preserves_catalog_on_transient_download_error(status):
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    now = datetime(2026, 7, 23, 12, 0, tzinfo=UTC)
+    client = ReconcileMinio(now)
+    store = _store(client, clock=lambda: now)
+    artifact = store.write(
+        pl.DataFrame({"id": [1]}),
+        new_governance_uid(),
+        600,
+    )
+    key = artifact["artifact_ref"].split("/", 3)[-1]
+
+    def fail_download(_bucket, object_key):
+        assert object_key == key
+        raise TimeoutError("MinIO download timed out")
+
+    client.get_object = fail_download
+    engine = ReconcileEngine([_catalog_row(artifact, status=status)])
+
+    result = PostgresArtifactResolver(engine, store).reconcile(
+        limit=1,
+        grace_seconds=300,
+    )
+
+    assert result["pending_finalized"] == 0
+    assert result["ready_failed"] == 0
+    assert engine.rows[0]["handoff_status"] == status
+    assert ("dataops-rules", key) in client.objects
+
+
+@pytest.mark.parametrize("status", ["pending", "ready"])
+def test_reconcile_handles_confirmed_not_found_during_stream_validation(status):
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    now = datetime(2026, 7, 23, 12, 0, tzinfo=UTC)
+    client = ReconcileMinio(now)
+    store = _store(client, clock=lambda: now)
+    artifact = store.write(
+        pl.DataFrame({"id": [1]}),
+        new_governance_uid(),
+        600,
+    )
+
+    def disappear_on_download(bucket, object_key):
+        client.objects.pop((bucket, object_key), None)
+        raise KeyError(object_key)
+
+    client.get_object = disappear_on_download
+    engine = ReconcileEngine([_catalog_row(artifact, status=status)])
+
+    result = PostgresArtifactResolver(engine, store).reconcile(
+        limit=1,
+        grace_seconds=300,
+    )
+
+    if status == "pending":
+        assert result["pending_deleted"] == 1
+        assert engine.rows == []
+    else:
+        assert result["ready_failed"] == 1
+        assert engine.rows[0]["handoff_status"] == "failed"
+        assert engine.rows[0]["failure_code"] == "ready_object_missing"
+
+
+def test_reconcile_streams_pending_content_before_finalize():
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    now = datetime(2026, 7, 23, 12, 0, tzinfo=UTC)
+    client = ReconcileMinio(now)
+    store = _store(client, clock=lambda: now)
+    artifact = store.write(
+        pl.DataFrame({"id": [1, 2, 3]}),
+        new_governance_uid(),
+        600,
+    )
+    key = artifact["artifact_ref"].split("/", 3)[-1]
+    stored = client.objects[("dataops-rules", key)]
+    payload = stored["payload"]
+    midpoint = len(payload) // 2
+    stored["payload"] = (
+        payload[:midpoint]
+        + bytes([payload[midpoint] ^ 1])
+        + payload[midpoint + 1 :]
+    )
+    client.get_calls.clear()
+    engine = ReconcileEngine([_catalog_row(artifact, status="pending")])
+
+    result = PostgresArtifactResolver(engine, store).reconcile(
+        limit=1,
+        grace_seconds=300,
+    )
+
+    assert result["pending_finalized"] == 0
+    assert engine.rows[0]["handoff_status"] == "failed"
+    assert engine.rows[0]["failure_code"] == "pending_object_invalid"
+    assert ("dataops-rules", key) not in client.objects
+    assert client.get_calls == [("dataops-rules", key)]
+
+
+def test_reconcile_marks_fresh_ready_snapshot_failed_for_invalid_content():
+    from app.runner.artifacts import PostgresArtifactResolver
+
+    now = datetime(2026, 7, 23, 12, 0, tzinfo=UTC)
+    client = ReconcileMinio(now)
+    store = _store(client, clock=lambda: now)
+    artifact = store.write(
+        pl.DataFrame({"id": [1, 2]}),
+        new_governance_uid(),
+        600,
+    )
+    key = artifact["artifact_ref"].split("/", 3)[-1]
+    stored = client.objects[("dataops-rules", key)]
+    payload = stored["payload"]
+    stored["payload"] = bytes([payload[0] ^ 1]) + payload[1:]
+    engine = ReconcileEngine([_catalog_row(artifact, status="ready")])
+
+    result = PostgresArtifactResolver(engine, store).reconcile(
+        limit=1,
+        grace_seconds=300,
+    )
+
+    assert result["ready_failed"] == 1
+    assert engine.rows[0]["handoff_status"] == "failed"
+    assert engine.rows[0]["failure_code"] == "ready_object_invalid"

tests/runner/test_artifacts.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import io
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 
 import polars as pl
@@ -215,6 +215,27 @@ def test_artifact_store_rejects_rows_size_ttl_and_unowned_references():
         )
 
 
+def test_upload_path_preserves_one_second_ttl_after_validation_delay(tmp_path):
+    now = datetime(2026, 7, 23, 10, 0, tzinfo=UTC)
+    clock_value = [now]
+    store = _store(FakeMinio(), clock=lambda: clock_value[0])
+    path = tmp_path / "one-second.parquet"
+    pl.DataFrame({"id": [1]}).write_parquet(path)
+    artifact = store.prepare_path(
+        str(path),
+        new_governance_uid(),
+        1,
+        schema_fields=[
+            {"name": "id", "type": "integer", "nullable": False}
+        ],
+    )
+    clock_value[0] = now + timedelta(microseconds=1)
+
+    uploaded = store.upload_path(str(path), artifact)
+
+    assert uploaded["artifact_ref"] == artifact["artifact_ref"]
+
+
 def test_artifact_store_does_not_return_ref_for_corrupted_server_content():
     class CorruptingMinio(FakeMinio):
         def put_object(self, bucket, key, *args, **kwargs):