DataOps-platform/tests/integration/test_wp13_plugin_platform_postgres.py

"""Real PostgreSQL migration and ACL probes for the WP13 fixture baseline."""

# ruff: noqa: B017

from __future__ import annotations

import hashlib
import json
import os
import subprocess
import uuid
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from threading import Barrier

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.engine import make_url

ROOT = Path(__file__).resolve().parents[2]


def _alembic(url: str, command: str, revision: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [str(ROOT / ".venv/bin/alembic"), "-c", str(ROOT / "alembic.ini"), command, revision],
        cwd=ROOT, env={**os.environ, "MIGRATION_DATABASE_URL": url}, text=True, capture_output=True,
    )


@pytest.mark.integration
def test_wp13_restricted_migrator_creates_closed_facts_and_empty_round_trip(monkeypatch):
    base_url = os.getenv("TEST_DATABASE_URL")
    if not base_url:
        pytest.skip("TEST_DATABASE_URL is required")
    base = make_url(base_url)
    database = f"wp13_{uuid.uuid4().hex}"
    migrator = f"wp13_migrator_{uuid.uuid4().hex[:10]}"
    runtime_login = f"wp13_runtime_{uuid.uuid4().hex[:10]}"
    control_login = f"wp13_control_{uuid.uuid4().hex[:10]}"
    password = f"m-{uuid.uuid4().hex}"
    runtime_password = f"r-{uuid.uuid4().hex}"
    control_password = f"c-{uuid.uuid4().hex}"
    admin = create_engine(base.set(database="postgres").render_as_string(hide_password=False), isolation_level="AUTOCOMMIT")
    quoted_database = admin.dialect.identifier_preparer.quote(database)
    quoted_migrator = admin.dialect.identifier_preparer.quote(migrator)
    created_owner = False
    created_control = False
    try:
        with admin.connect() as connection:
            connection.execute(text(f"CREATE DATABASE {quoted_database}"))
            connection.execute(text(f"CREATE ROLE {admin.dialect.identifier_preparer.quote(runtime_login)} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": runtime_password})
            connection.execute(text(f"CREATE ROLE {admin.dialect.identifier_preparer.quote(control_login)} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION INHERIT PASSWORD :password"), {"password": control_password})
        url = base.set(database=database).render_as_string(hide_password=False)
        engine = create_engine(url)
        with engine.begin() as connection:
            connection.execute(text("CREATE EXTENSION IF NOT EXISTS pgcrypto"))
            created_owner = not connection.execute(text("SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='dataops_plugin_platform_owner')")).scalar_one()
            created_control = not connection.execute(text("SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='dataops_plugin_platform_control')")).scalar_one()
            if created_owner:
                connection.execute(text("CREATE ROLE dataops_plugin_platform_owner NOLOGIN NOSUPERUSER NOCREATEROLE"))
            if created_control:
                connection.execute(text("CREATE ROLE dataops_plugin_platform_control NOLOGIN NOSUPERUSER NOCREATEROLE"))
            connection.execute(text("GRANT USAGE,CREATE ON SCHEMA public TO dataops_plugin_platform_owner"))
        assert _alembic(url, "stamp", "20260818_552").returncode == 0
        with engine.begin() as connection:
            connection.execute(text(f"CREATE ROLE {quoted_migrator} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": password})
            connection.execute(text(f"GRANT dataops_plugin_platform_owner TO {quoted_migrator}"))
            connection.execute(text(f"GRANT USAGE,CREATE ON SCHEMA public TO {quoted_migrator}"))
            connection.execute(text(f"GRANT SELECT,UPDATE ON public.alembic_version TO {quoted_migrator}"))
        restricted_url = base.set(database=database, username=migrator, password=password).render_as_string(hide_password=False)
        upgraded = _alembic(restricted_url, "upgrade", "20260818_559")
        assert upgraded.returncode == 0, upgraded.stdout + upgraded.stderr
        with engine.begin() as connection:
            connection.execute(text(f"GRANT dataops_app_runtime TO {engine.dialect.identifier_preparer.quote(runtime_login)}"))
            connection.execute(text(f"GRANT dataops_plugin_platform_control TO {engine.dialect.identifier_preparer.quote(control_login)}"))
        with engine.connect() as connection:
            assert connection.execute(text("SELECT count(*) FROM public.plugin_registry_versions")).scalar_one() == 0
            assert not connection.execute(text("SELECT has_table_privilege('dataops_app_runtime','public.plugin_runs','SELECT')")).scalar_one()
            assert connection.execute(text("SELECT to_regprocedure('public.plugin_platform_runtime_execute(jsonb)') IS NULL")).scalar_one()
        with engine.begin() as connection:
            connection.execute(text("""INSERT INTO public.plugin_registry_versions(plugin_uid,version,plugin_type,manifest_digest,artifact_digest,trust_store_key_id,signature_digest,sbom_digest,license_digest,vulnerability_digest,provenance_digest,fixture_id,permissions,capabilities,resource)
                VALUES('fixture-fence-db','1.0.0','quality',:manifest_digest,:artifact_digest,'local-fixture-key-v1',:signature_digest,:sbom_digest,:license_digest,:vulnerability_digest,:provenance_digest,'ENGINEERING_EVIDENCE_ONLY',CAST(:permissions AS jsonb),CAST(:capabilities AS jsonb),CAST(:resource AS jsonb))"""), {"manifest_digest": "a" * 64, "artifact_digest": "b" * 64, "signature_digest": "c" * 64, "sbom_digest": "d" * 64, "license_digest": "e" * 64, "vulnerability_digest": "f" * 64, "provenance_digest": "0" * 64, "permissions": json.dumps({"file": False, "network": False, "secret": False, "child_process": False}), "capabilities": json.dumps(["evaluate"]), "resource": json.dumps({"timeout_ms": 100, "max_output_bytes": 4096, "max_concurrency": 1, "max_retries": 1})})
        refused_fence = _alembic(restricted_url, "downgrade", "20260818_556")
        assert refused_fence.returncode != 0
        refused_legacy_fence = _alembic(restricted_url, "downgrade", "20260818_555")
        assert refused_legacy_fence.returncode != 0
        with engine.connect() as connection:
            assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == "20260818_559"
            assert connection.execute(text("SELECT count(*) FROM public.plugin_registry_versions WHERE plugin_uid='fixture-fence-db'")).scalar_one() == 1
        with engine.begin() as connection:
            connection.execute(text("DELETE FROM public.plugin_registry_versions WHERE plugin_uid='fixture-fence-db'"))
        assert _alembic(restricted_url, "downgrade", "20260818_556").returncode == 0
        assert _alembic(restricted_url, "upgrade", "20260818_559").returncode == 0
        from app import create_app
        from app.core.plugins.governance import (
            fixture_signature_digest,
        )

        runtime_url = base.set(database=database, username=runtime_login, password=runtime_password).render_as_string(hide_password=False)
        control_url = base.set(database=database, username=control_login, password=control_password).render_as_string(hide_password=False)
        monkeypatch.setenv("DB_ROLE_INIT_DATABASE_URL", url)
        monkeypatch.setenv("MIGRATION_DATABASE_URL", restricted_url)
        monkeypatch.setenv("DATABASE_URL", runtime_url)
        monkeypatch.setenv("DATAOPS_RUNTIME_USER", runtime_login)
        monkeypatch.setenv("DATAOPS_RUNTIME_PASSWORD", runtime_password)
        monkeypatch.setenv("DATAOPS_MIGRATOR_USER", migrator)
        monkeypatch.setenv("BI_AI_CATALOG_CONTROL_DATABASE_URL", control_url)
        monkeypatch.setenv("DATAOPS_BI_AI_CATALOG_CONTROL_USER", control_login)
        monkeypatch.setenv("DATAOPS_BI_AI_CATALOG_CONTROL_PASSWORD", control_password)
        from app.core.edge_gateway.runtime_roles import provision_runtime_login
        provision_runtime_login()
        control_attack = create_engine(control_url)
        try:
            with control_attack.connect() as connection:
                assert connection.execute(text("SELECT has_function_privilege(current_user,'public.plugin_platform_control_v2(jsonb)','EXECUTE')")).scalar_one() is False
                assert connection.execute(text("SELECT has_function_privilege(current_user,'public.plugin_platform_control_v3(jsonb)','EXECUTE')")).scalar_one() is False
                assert connection.execute(text("SELECT has_function_privilege(current_user,'public.plugin_platform_control_v4(jsonb)','EXECUTE')")).scalar_one() is False
                assert connection.execute(text("SELECT has_function_privilege(current_user,'public.plugin_platform_control_v5(jsonb)','EXECUTE')")).scalar_one() is True
                with pytest.raises(Exception):
                    connection.execute(text("SELECT public.plugin_platform_control_v2('{}'::jsonb)"))
        finally:
            control_attack.dispose()
        runtime_attack = create_engine(runtime_url)
        try:
            with runtime_attack.connect() as connection:
                with pytest.raises(Exception):
                    connection.execute(text("SET ROLE dataops_plugin_platform_control"))
                connection.rollback()
            with runtime_attack.connect() as connection:
                with pytest.raises(Exception):
                    connection.execute(text("SELECT public.plugin_platform_runtime_execute('{}'::jsonb)"))
                with pytest.raises(Exception):
                    connection.execute(text("SELECT public.plugin_platform_runtime_v2('{}'::jsonb)"))
        finally:
            runtime_attack.dispose()
        monkeypatch.setenv("PLUGIN_PLATFORM_CONTROL_DATABASE_URL", control_url)
        monkeypatch.setenv("TRUSTED_PLUGIN_TENANT", "tenant-a")
        monkeypatch.setenv("TRUSTED_PLUGIN_DOMAIN", "default")
        api_identities = {
            "submitter": {"id": "01900000-0000-7000-8000-000000000012", "roles": ["admin"]},
            "reviewer": {"id": "01900000-0000-7000-8000-000000000013", "roles": ["admin"]},
            "operator": {"id": "01900000-0000-7000-8000-000000000014", "roles": ["admin"]},
        }
        monkeypatch.setattr("app.core.system.auth.load_identity_from_token", lambda token, secret: api_identities.get(token))
        api_manifest = {"schema_version": 1, "plugin_uid": "fixture-api-db", "name": "fixture-api-db", "version": "1.0.0", "api_version": "1", "type": "quality", "capabilities": ["evaluate"], "permissions": {"network": False, "file": False, "secret": False, "child_process": False}, "resource": {"timeout_ms": 100, "max_output_bytes": 4096, "max_concurrency": 1, "max_retries": 1}, "compatibility": {"platform_api": "1"}, "distribution": {"kind": "builtin_fixture", "fixture_id": "ENGINEERING_EVIDENCE_ONLY", "artifact_digest": "c" * 64}}
        api_record = {"artifact_digest": "c" * 64, "signature": {"trust_store_key_id": "local-fixture-key-v1", "signature_digest": fixture_signature_digest("c" * 64)}, "sbom_digest": "d" * 64, "license_digest": "e" * 64, "vulnerability_digest": "f" * 64, "provenance_digest": "0" * 64}
        app = create_app()
        app.config.update(TESTING=True)
        client = app.test_client()
        api_created = client.post("/api/system/plugins/registry", json={"manifest": api_manifest, "registry_record": api_record}, headers={"Authorization": "Bearer submitter"})
        assert api_created.status_code == 201 and api_created.headers["Cache-Control"] == "no-store"
        pre_review_api_approval = client.post("/api/system/plugins/fixture-api-db/1.0.0/approvals", json={"approval_action": "approve"}, headers={"Authorization": "Bearer operator"})
        assert pre_review_api_approval.status_code == 400 and pre_review_api_approval.headers["Cache-Control"] == "no-store"
        api_reviewed = client.post("/api/system/plugins/fixture-api-db/1.0.0/review", json={}, headers={"Authorization": "Bearer reviewer"})
        assert api_reviewed.status_code == 200 and api_reviewed.headers["Cache-Control"] == "no-store"
        self_approval = client.post("/api/system/plugins/fixture-api-db/1.0.0/approvals", json={"approval_action": "approve"}, headers={"Authorization": "Bearer reviewer"})
        assert self_approval.status_code == 400 and self_approval.headers["Cache-Control"] == "no-store"
        submitter_approval = client.post("/api/system/plugins/fixture-api-db/1.0.0/approvals", json={"approval_action": "approve"}, headers={"Authorization": "Bearer submitter"})
        assert submitter_approval.status_code == 400 and submitter_approval.headers["Cache-Control"] == "no-store"
        api_approval = create_app().test_client().post("/api/system/plugins/fixture-api-db/1.0.0/approvals", json={"approval_action": "approve"}, headers={"Authorization": "Bearer operator"})
        assert api_approval.status_code == 201 and api_approval.headers["Cache-Control"] == "no-store"
        forged_reviewer = client.post("/api/system/plugins/fixture-api-db/1.0.0/approvals", json={"approval_action": "approve", "reviewer_ref": "forged"}, headers={"Authorization": "Bearer operator"})
        assert forged_reviewer.status_code == 400 and forged_reviewer.headers["Cache-Control"] == "no-store"
        duplicate = create_app().test_client().post("/api/system/plugins/registry", json={"manifest": api_manifest, "registry_record": api_record}, headers={"Authorization": "Bearer submitter"})
        assert duplicate.status_code == 400 and duplicate.headers["Cache-Control"] == "no-store"
        with engine.connect() as connection:
            assert connection.execute(text("SELECT state FROM public.plugin_registry_versions WHERE plugin_uid='fixture-api-db'")).scalar_one() == "reviewed"
            assert connection.execute(text("SELECT count(*) FROM public.plugin_registry_versions WHERE plugin_uid='fixture-api-db'")).scalar_one() == 1
        def control(payload: dict) -> dict:
            with engine.begin() as connection:
                connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_plugin_platform_control"))
                return dict(connection.execute(text("SELECT public.plugin_platform_control_v5(CAST(:payload AS jsonb))"), {"payload": json.dumps(payload)}).scalar_one())

        artifact_digest = "b" * 64
        registered = control({"action": "register", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "author-a", "plugin_type": "quality", "manifest_digest": "a" * 64, "artifact_digest": artifact_digest, "signature_digest": hashlib.sha256(f"local-fixture-key-v1|{artifact_digest}".encode("ascii")).hexdigest(), "sbom_digest": "d" * 64, "license_digest": "e" * 64, "vulnerability_digest": "f" * 64, "provenance_digest": "0" * 64, "capabilities": ["evaluate"], "resource": {"timeout_ms": 100, "max_output_bytes": 4096, "max_concurrency": 1, "max_retries": 1}})
        assert registered["state"] == "draft"
        with pytest.raises(Exception):
            control({"action": "issue_approval", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "operator-a", "approval_action": "approve", "tenant_ref": "tenant-a", "domain_ref": "default", "expires_in_seconds": 300})
        with pytest.raises(Exception):
            control({"action": "review", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "author-a"})
        reviewed = control({"action": "review", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "reviewer-a"})
        with engine.connect() as connection:
            review_fact = connection.execute(text("SELECT state,review_actor,review_generation,reviewed_at,review_manifest_digest,manifest_digest FROM public.plugin_registry_versions WHERE plugin_uid='fixture-quality-db' AND version='1.0.0'")).mappings().one()
        assert review_fact["state"] == "reviewed" and review_fact["review_actor"] == "reviewer-a"
        assert review_fact["review_generation"] == 1 and review_fact["reviewed_at"] is not None
        assert review_fact["review_manifest_digest"] == review_fact["manifest_digest"]
        with pytest.raises(Exception):
            control({"action": "issue_approval", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "author-a", "approval_action": "approve", "tenant_ref": "tenant-a", "domain_ref": "default", "expires_in_seconds": 300})
        def approve(action: str) -> str:
            return control({"action": "issue_approval", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "operator-a", "approval_action": action, "tenant_ref": "tenant-a", "domain_ref": "default", "expires_in_seconds": 300})["approval_uid"]
        approval_ids = {action: approve(action) for action in ("approve", "canary", "activate", "recover")}
        with pytest.raises(Exception):
            control({"action": "issue_approval", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "reviewer-a", "approval_action": "pause", "tenant_ref": "tenant-a", "domain_ref": "default", "expires_in_seconds": 300})
        def transition(target: str, fence: int) -> dict:
            action = {"approved": "approve", "canary": "canary", "active": "activate", "paused": "pause", "rolled_back": "rollback", "revoked": "revoke", "recovery": "recover"}[target]
            return control({"action": "transition", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "operator-a", "tenant_ref": "tenant-a", "domain_ref": "default", "approval_uid": approval_ids[action], "target_state": target, "expected_fence": fence, "incident_uid": ""})
        approved = transition("approved", reviewed["fence"])
        with pytest.raises(Exception):
            control({"action": "transition", "plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "operator-a", "tenant_ref": "tenant-b", "domain_ref": "default", "approval_uid": approval_ids["canary"], "target_state": "canary", "expected_fence": approved["fence"], "incident_uid": ""})
        transition_barrier = Barrier(2)
        def canary_worker() -> tuple[str, dict | str]:
            try:
                transition_barrier.wait(timeout=3)
                return "ok", transition("canary", approved["fence"])
            except Exception as exc:
                return "denied", str(exc)
        with ThreadPoolExecutor(max_workers=2) as pool:
            transition_outcomes = list(pool.map(lambda _: canary_worker(), range(2)))
        assert sorted(item[0] for item in transition_outcomes) == ["denied", "ok"]
        canary = next(item[1] for item in transition_outcomes if item[0] == "ok")
        assert isinstance(canary, dict)
        active = transition("active", canary["fence"])
        assert active["state"] == "active"
        with pytest.raises(Exception):
            transition("active", active["fence"])
        with engine.connect() as connection:
            connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_plugin_platform_control"))
            with pytest.raises(Exception):
                connection.execute(text("SELECT public.plugin_platform_issue_claim_v3(CAST(:payload AS jsonb))"), {"payload": json.dumps({"plugin_uid": "fixture-quality-db", "version": "1.0.0", "tenant_ref": "tenant-a", "domain_ref": "default", "principal_ref": "operator-a", "operation_name": "escalate", "input_digest": "1" * 64, "idempotency_key": "operation-escalation"})})
            connection.rollback()
        with engine.begin() as connection:
            connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_plugin_platform_control"))
            claim = dict(connection.execute(text("SELECT public.plugin_platform_issue_claim_v3(CAST(:payload AS jsonb))"), {"payload": json.dumps({"plugin_uid": "fixture-quality-db", "version": "1.0.0", "tenant_ref": "tenant-a", "domain_ref": "default", "principal_ref": "operator-a", "operation_name": "evaluate", "input_digest": "2" * 64, "idempotency_key": "run-001"})}).scalar_one())
        with engine.begin() as connection:
            connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_app_runtime"))
            queued = dict(connection.execute(text("SELECT public.plugin_platform_runtime_v3(CAST(:payload AS jsonb))"), {"payload": json.dumps({"action": "enqueue", "claim_uid": claim["claim_uid"], "idempotency_key": "run-001", "input_digest": "2" * 64, "operation_name": "evaluate"})}).scalar_one())
        barrier = Barrier(2)
        def claim_worker(worker: str):
            try:
                barrier.wait(timeout=3)
                with engine.begin() as connection:
                    connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_app_runtime"))
                    return "ok", dict(connection.execute(text("SELECT public.plugin_platform_runtime_v3(CAST(:payload AS jsonb))"), {"payload": json.dumps({"action": "claim", "run_uid": queued["run_uid"], "worker": worker})}).scalar_one())
            except Exception as exc:
                return "denied", str(exc)
        with ThreadPoolExecutor(max_workers=2) as pool:
            outcomes = list(pool.map(claim_worker, ("worker-a", "worker-b")))
        assert sorted(item[0] for item in outcomes) == ["denied", "ok"]
        winning = next(item[1] for item in outcomes if item[0] == "ok")
        with engine.begin() as connection:
            connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_app_runtime"))
            settled = dict(connection.execute(text("SELECT public.plugin_platform_runtime_v3(CAST(:payload AS jsonb))"), {"payload": json.dumps({"action": "settle", "run_uid": queued["run_uid"], "worker": winning["lease_owner"], "fence": winning["fence"], "success": True, "output_digest": "3" * 64, "failure_digest": "4" * 64})}).scalar_one())
        assert settled["state"] == "succeeded"

        def issue(input_digest: str, idempotency_key: str) -> dict:
            with engine.begin() as connection:
                connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_plugin_platform_control"))
                return dict(connection.execute(text("SELECT public.plugin_platform_issue_claim_v3(CAST(:payload AS jsonb))"), {"payload": json.dumps({"plugin_uid": "fixture-quality-db", "version": "1.0.0", "tenant_ref": "tenant-a", "domain_ref": "default", "principal_ref": "operator-a", "operation_name": "evaluate", "input_digest": input_digest, "idempotency_key": idempotency_key})}).scalar_one())

        def runtime(payload: dict) -> dict:
            with engine.begin() as connection:
                connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_app_runtime"))
                return dict(connection.execute(text("SELECT public.plugin_platform_runtime_v3(CAST(:payload AS jsonb))"), {"payload": json.dumps(payload)}).scalar_one())

        replay_claim = issue("2" * 64, "run-001")
        replay = runtime({"action": "enqueue", "claim_uid": replay_claim["claim_uid"], "idempotency_key": "run-001", "input_digest": "2" * 64, "operation_name": "evaluate"})
        assert replay["run_uid"] == queued["run_uid"] and replay["replay"] is True
        changed_claim = issue("5" * 64, "run-001")
        with pytest.raises(Exception):
            runtime({"action": "enqueue", "claim_uid": changed_claim["claim_uid"], "idempotency_key": "run-001", "input_digest": "5" * 64, "operation_name": "evaluate"})

        failed_claim = issue("8" * 64, "run-fail")
        failed = runtime({"action": "enqueue", "claim_uid": failed_claim["claim_uid"], "idempotency_key": "run-fail", "input_digest": "8" * 64, "operation_name": "evaluate"})
        late = runtime({"action": "claim", "run_uid": failed["run_uid"], "worker": "worker-late"})
        with engine.begin() as connection:
            connection.execute(text("UPDATE public.plugin_runs SET lease_expires_at=clock_timestamp()-interval '1 second' WHERE run_uid=:run_uid"), {"run_uid": failed["run_uid"]})
        replacement = runtime({"action": "claim", "run_uid": failed["run_uid"], "worker": "worker-replacement"})
        with pytest.raises(Exception):
            runtime({"action": "settle", "run_uid": failed["run_uid"], "worker": late["lease_owner"], "fence": late["fence"], "success": False, "output_digest": "0" * 64, "failure_digest": "9" * 64})
        first_failed = runtime({"action": "settle", "run_uid": failed["run_uid"], "worker": replacement["lease_owner"], "fence": replacement["fence"], "success": False, "output_digest": "0" * 64, "failure_digest": "9" * 64})
        assert first_failed["state"] == "pending"
        for retry in range(2):
            with engine.begin() as connection:
                connection.execute(text("UPDATE public.plugin_runs SET next_attempt_at=clock_timestamp() WHERE run_uid=:run_uid"), {"run_uid": failed["run_uid"]})
            lease = runtime({"action": "claim", "run_uid": failed["run_uid"], "worker": f"retry-{retry}"})
            dead = runtime({"action": "settle", "run_uid": failed["run_uid"], "worker": lease["lease_owner"], "fence": lease["fence"], "success": False, "output_digest": "0" * 64, "failure_digest": "9" * 64})
        assert dead["state"] == "dead_letter"
        with pytest.raises(Exception):
            runtime({"action": "recover", "run_uid": failed["run_uid"], "worker": "runtime-attacker"})
        recovery_incident = str(uuid.uuid4())
        recovery_approval = approval_ids["recover"]
        with engine.connect() as connection:
            connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_plugin_platform_control"))
            with pytest.raises(Exception):
                connection.execute(text("SELECT public.plugin_platform_issue_recovery_claim(CAST(:payload AS jsonb))"), {"payload": json.dumps({"plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "operator-a", "tenant_ref": "tenant-b", "domain_ref": "default", "run_uid": failed["run_uid"], "approval_uid": recovery_approval, "incident_uid": recovery_incident, "expected_fence": dead["fence"]})})
            connection.rollback()
        with engine.connect() as connection:
            connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_plugin_platform_control"))
            with pytest.raises(Exception):
                connection.execute(text("SELECT public.plugin_platform_issue_recovery_claim(CAST(:payload AS jsonb))"), {"payload": json.dumps({"plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "operator-a", "tenant_ref": "tenant-a", "domain_ref": "default", "run_uid": failed["run_uid"], "approval_uid": recovery_approval, "incident_uid": recovery_incident, "expected_fence": dead["fence"] + 1})})
            connection.rollback()
        with engine.begin() as connection:
            connection.execute(text("SET LOCAL SESSION AUTHORIZATION dataops_plugin_platform_control"))
            recovery_claim = dict(connection.execute(text("SELECT public.plugin_platform_issue_recovery_claim(CAST(:payload AS jsonb))"), {"payload": json.dumps({"plugin_uid": "fixture-quality-db", "version": "1.0.0", "actor_ref": "operator-a", "tenant_ref": "tenant-a", "domain_ref": "default", "run_uid": failed["run_uid"], "approval_uid": recovery_approval, "incident_uid": recovery_incident, "expected_fence": dead["fence"]})}).scalar_one())
        recovered = runtime({"action": "recover", "claim_uid": recovery_claim["claim_uid"], "worker": "recovery-worker"})
        assert recovered["state"] == "pending"
        with pytest.raises(Exception):
            runtime({"action": "recover", "claim_uid": recovery_claim["claim_uid"], "worker": "recovery-replay"})
        with engine.connect() as connection:
            assert connection.execute(text("SELECT count(*) FROM public.plugin_dead_letters WHERE run_uid=:run_uid"), {"run_uid": failed["run_uid"]}).scalar_one() == 0
            assert connection.execute(text("SELECT count(*) FROM public.plugin_audit_outbox WHERE event_type IN ('active','invoked','dead_letter')")).scalar_one() >= 3
        assert _alembic(restricted_url, "downgrade", "20260818_554").returncode != 0
        with engine.begin() as connection:
            for table in ("plugin_recovery_claims", "plugin_dead_letters", "plugin_runs", "plugin_request_claims", "plugin_runtime_breakers", "plugin_approvals", "plugin_audit_outbox", "plugin_registry_versions"):
                connection.execute(text(f"DELETE FROM public.{table}"))
        assert _alembic(restricted_url, "downgrade", "20260818_554").returncode == 0
        assert _alembic(restricted_url, "upgrade", "20260818_555").returncode == 0
        assert _alembic(restricted_url, "upgrade", "20260818_556").returncode == 0
        assert _alembic(restricted_url, "upgrade", "20260818_557").returncode == 0
        assert _alembic(restricted_url, "upgrade", "20260818_559").returncode == 0
    finally:
        engine.dispose() if "engine" in locals() else None
        with admin.connect() as connection:
            connection.execute(text("SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=:name"), {"name": database})
            connection.execute(text(f"DROP DATABASE IF EXISTS {quoted_database}"))
            connection.execute(text(f"DROP ROLE IF EXISTS {quoted_migrator}"))
            connection.execute(text(f"DROP ROLE IF EXISTS {admin.dialect.identifier_preparer.quote(runtime_login)}"))
            connection.execute(text(f"DROP ROLE IF EXISTS {admin.dialect.identifier_preparer.quote(control_login)}"))
            if created_control:
                connection.execute(text("DROP ROLE IF EXISTS dataops_plugin_platform_control"))
            if created_owner:
                connection.execute(text("DROP ROLE IF EXISTS dataops_plugin_platform_owner"))
        admin.dispose()