fix: expire legacy cleanup claims on upgrade

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

@@ -204,6 +204,42 @@ Third-review verification:
 - Alembic inside the rebuilt backend container:
   `20260723_190 (head)`.
 
+## Fourth-review revision-180 claim compatibility
+
+The fourth review closes the upgrade boundary for cleanup claims created by
+revision-180 workers:
+
+- revision 180 already allowed a cleanup worker to persist a non-null
+  `cleanup_claim`, but it had no claim-expiry column;
+- migration 190 now backfills every such historical sample and SQL-receipt
+  claim with `cleanup_claim_expires_at = CURRENT_TIMESTAMP` immediately after
+  adding the column;
+- rows without a claim remain unclaimed with a null expiry;
+- the deployment contract requires revision-180 cleanup workers to be stopped
+  during the schema upgrade. This makes immediate expiry safe and prevents an
+  old worker from racing a revision-190 takeover;
+- after upgrade, the normal owner-and-expiry compare-and-set contract can
+  replace the old claim, so a crash before migration cannot permanently lock
+  either evidence table.
+
+The real migration acceptance creates an isolated PostgreSQL database, upgrades
+it to revision 180, inserts non-null legacy claims into both
+`rule_violation_samples` and `rule_sql_staging_receipts`, and then upgrades to
+190. It verifies both expiry values are non-null and already reclaimable, then
+successfully replaces both old owners through the production-compatible CAS
+predicate.
+
+Fourth-review verification:
+
+- focused schema, real migration, Runner, evidence, ledger, and token suite:
+  `42 passed`;
+- isolated real PostgreSQL `180 -> 190` legacy-claim upgrade:
+  passed for both evidence tables;
+- full suite: `606 passed, 28 skipped, 59 subtests passed`;
+- changed-file Ruff and `git diff --check`: passed;
+- migration 180 remains byte-for-byte unchanged from the second-review
+  baseline.
+
 ## Real PostgreSQL and MinIO acceptance
 
 The production-path integration uses:

migrations/versions/20260723_190_rule_cleanup_leases.py

@@ -9,11 +9,19 @@ depends_on = None
 
 
 def upgrade() -> None:
+    # Revision-180 cleanup workers must be stopped during this schema upgrade.
+    # Their owner-only claims have no lease timestamp, so making those claims
+    # immediately expired is the only deterministic, fail-safe takeover path.
     op.execute(
         """
         ALTER TABLE public.rule_violation_samples
             ADD COLUMN cleanup_claim_expires_at TIMESTAMPTZ;
 
+        UPDATE public.rule_violation_samples
+        SET cleanup_claim_expires_at = CURRENT_TIMESTAMP
+        WHERE cleanup_claim IS NOT NULL
+          AND cleanup_claim_expires_at IS NULL;
+
         CREATE INDEX idx_rule_violation_sample_cleanup_lease
             ON public.rule_violation_samples(
                 cleanup_claim_expires_at, expires_at, id
@@ -22,6 +30,11 @@ def upgrade() -> None:
         ALTER TABLE public.rule_sql_staging_receipts
             ADD COLUMN cleanup_claim_expires_at TIMESTAMPTZ;
 
+        UPDATE public.rule_sql_staging_receipts
+        SET cleanup_claim_expires_at = CURRENT_TIMESTAMP
+        WHERE cleanup_claim IS NOT NULL
+          AND cleanup_claim_expires_at IS NULL;
+
         CREATE INDEX idx_rule_sql_receipt_cleanup_lease
             ON public.rule_sql_staging_receipts(
                 cleanup_claim_expires_at, expires_at, id

tests/integration/test_rule_artifact_migration_upgrade.py

@@ -328,3 +328,219 @@ def test_old_140_upgrades_to_durable_handoff_and_enforces_cas(tmp_path):
             )
             connection.execute(text(f'DROP DATABASE IF EXISTS "{database_name}"'))
         admin.dispose()
+
+
+def test_old_180_cleanup_claims_become_expired_and_cas_takeover_ready():
+    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"task6_claim_{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_180")
+        engine = create_engine(database_url, pool_pre_ping=True)
+        rule_run_id = new_governance_uid()
+        sample_id = new_governance_uid()
+        receipt_id = new_governance_uid()
+        old_sample_claim = new_governance_uid()
+        old_receipt_claim = new_governance_uid()
+        with engine.begin() as connection:
+            expiry_columns = connection.execute(
+                text(
+                    """
+                    SELECT table_name
+                    FROM information_schema.columns
+                    WHERE table_schema = 'public'
+                      AND table_name IN (
+                          'rule_violation_samples',
+                          'rule_sql_staging_receipts'
+                      )
+                      AND column_name = 'cleanup_claim_expires_at'
+                    """
+                )
+            ).all()
+            assert expiry_columns == []
+            connection.execute(text("SET session_replication_role = replica"))
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_runs (
+                        id, deployment_id, component_binding_id,
+                        rule_version_id, plan_hash, status, correlation_id
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:deployment_id AS uuid),
+                        CAST(:component_id AS uuid),
+                        CAST(:rule_version_id AS uuid),
+                        :plan_hash, 'failed', CAST(:correlation_id AS uuid)
+                    )
+                    """
+                ),
+                {
+                    "id": rule_run_id,
+                    "deployment_id": new_governance_uid(),
+                    "component_id": new_governance_uid(),
+                    "rule_version_id": new_governance_uid(),
+                    "plan_hash": "a" * 64,
+                    "correlation_id": new_governance_uid(),
+                },
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_violation_samples (
+                        id, rule_run_id, artifact_ref, sample_count,
+                        redaction_policy, expires_at, cleanup_claim
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:rule_run_id AS uuid),
+                        :artifact_ref, 1, 'all-fields',
+                        CURRENT_TIMESTAMP - INTERVAL '1 hour',
+                        CAST(:cleanup_claim AS uuid)
+                    )
+                    """
+                ),
+                {
+                    "id": sample_id,
+                    "rule_run_id": rule_run_id,
+                    "artifact_ref": (
+                        f"minio://legacy/{new_governance_uid()}.parquet"
+                    ),
+                    "cleanup_claim": old_sample_claim,
+                },
+            )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_sql_staging_receipts (
+                        id, producer_rule_run_id, deployment_id,
+                        correlation_id, output_binding_id,
+                        output_binding_hash, relation_ref, relation_digest,
+                        commit_outcome, status, expires_at, cleanup_claim
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:run_id AS uuid),
+                        CAST(:deployment_id AS uuid),
+                        CAST(:correlation_id AS uuid),
+                        CAST(:binding_id AS uuid), :binding_hash,
+                        :relation_ref, :relation_digest,
+                        'committed', 'expired',
+                        CURRENT_TIMESTAMP - INTERVAL '1 hour',
+                        CAST(:cleanup_claim AS uuid)
+                    )
+                    """
+                ),
+                {
+                    "id": receipt_id,
+                    "run_id": rule_run_id,
+                    "deployment_id": new_governance_uid(),
+                    "correlation_id": new_governance_uid(),
+                    "binding_id": new_governance_uid(),
+                    "binding_hash": "b" * 64,
+                    "relation_ref": "legacy.receipt",
+                    "relation_digest": "c" * 64,
+                    "cleanup_claim": old_receipt_claim,
+                },
+            )
+            connection.execute(text("SET session_replication_role = origin"))
+
+        engine.dispose()
+        engine = None
+        _upgrade(database_url, "20260723_190")
+        engine = create_engine(database_url, pool_pre_ping=True)
+        new_sample_claim = new_governance_uid()
+        new_receipt_claim = new_governance_uid()
+        with engine.begin() as connection:
+            migrated = connection.execute(
+                text(
+                    """
+                    SELECT
+                        (
+                            SELECT cleanup_claim_expires_at
+                            FROM public.rule_violation_samples
+                            WHERE id = CAST(:sample_id AS uuid)
+                        ) AS sample_claim_expires_at,
+                        (
+                            SELECT cleanup_claim_expires_at
+                            FROM public.rule_sql_staging_receipts
+                            WHERE id = CAST(:receipt_id AS uuid)
+                        ) AS receipt_claim_expires_at,
+                        CURRENT_TIMESTAMP AS observed_at
+                    """
+                ),
+                {"sample_id": sample_id, "receipt_id": receipt_id},
+            ).mappings().one()
+            assert migrated["sample_claim_expires_at"] is not None
+            assert migrated["receipt_claim_expires_at"] is not None
+            assert migrated["sample_claim_expires_at"] <= migrated["observed_at"]
+            assert migrated["receipt_claim_expires_at"] <= migrated["observed_at"]
+
+            sample_takeover = connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_violation_samples
+                    SET cleanup_claim = CAST(:new_claim AS uuid),
+                        cleanup_claim_expires_at =
+                            CURRENT_TIMESTAMP + INTERVAL '5 minutes'
+                    WHERE id = CAST(:id AS uuid)
+                      AND cleanup_claim = CAST(:old_claim AS uuid)
+                      AND cleanup_claim_expires_at <= CURRENT_TIMESTAMP
+                    RETURNING cleanup_claim
+                    """
+                ),
+                {
+                    "id": sample_id,
+                    "old_claim": old_sample_claim,
+                    "new_claim": new_sample_claim,
+                },
+            ).scalar_one()
+            receipt_takeover = connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_sql_staging_receipts
+                    SET cleanup_claim = CAST(:new_claim AS uuid),
+                        cleanup_claim_expires_at =
+                            CURRENT_TIMESTAMP + INTERVAL '5 minutes'
+                    WHERE id = CAST(:id AS uuid)
+                      AND cleanup_claim = CAST(:old_claim AS uuid)
+                      AND cleanup_claim_expires_at <= CURRENT_TIMESTAMP
+                    RETURNING cleanup_claim
+                    """
+                ),
+                {
+                    "id": receipt_id,
+                    "old_claim": old_receipt_claim,
+                    "new_claim": new_receipt_claim,
+                },
+            ).scalar_one()
+            assert str(sample_takeover) == new_sample_claim
+            assert str(receipt_takeover) == new_receipt_claim
+    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/test_data_rule_schema.py

@@ -278,4 +278,9 @@ def test_rule_cleanup_lease_migration_extends_180_forward_only():
     assert source.count("cleanup_claim_expires_at TIMESTAMPTZ") == 2
     assert "rule_violation_samples" in source
     assert "rule_sql_staging_receipts" in source
+    assert source.count(
+        "SET cleanup_claim_expires_at = CURRENT_TIMESTAMP"
+    ) == 2
+    assert source.count("WHERE cleanup_claim IS NOT NULL") == 2
+    assert source.count("AND cleanup_claim_expires_at IS NULL") == 2
     assert "forward-only" in source