DataOps-platform/tests/integration/test_wp09_approval_consumption_postgres.py

from __future__ import annotations

import hashlib
import json
import os
import uuid
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from typing import Any

import pytest
from sqlalchemy import create_engine, text

pytestmark = pytest.mark.integration


def _database_url() -> str:
    value = os.getenv("TEST_DATABASE_URL")
    if not value:
        pytest.skip("TEST_DATABASE_URL is required")
    return value


def _seed_http_runtime_fixture(engine) -> dict[str, Any]:
    """Commit only the precise persisted scope consumed by the HTTP runtime."""
    suffix = uuid.uuid4().hex[:12]
    ids = {
        name: str(uuid.uuid4())
        for name in ("workflow", "version", "agent", "grant", "route")
    }
    fixture = {
        **ids,
        "tenant": f"wp09_http_{suffix}",
        "domain": str(uuid.uuid4()),
        "code": f"wp09-http-{suffix}",
        "control_incident": None,
        "task_uids": [],
    }
    with engine.begin() as connection:
        users = [
            str(row[0])
            for row in connection.execute(text(
                "SELECT id::text FROM public.users WHERE status='active' ORDER BY id LIMIT 2"
            ))
        ]
        if len(users) < 2:
            pytest.skip("requires two existing test users for independent review")
        fixture["owner"], fixture["reviewer"] = users
        connection.execute(text("""
            INSERT INTO public.governance_workflows
              (uid,code,name,status,current_version,created_by)
            VALUES (CAST(:workflow AS uuid),:code,'wp09 http','published',1,
                    CAST(:owner AS uuid))
        """), fixture)
        connection.execute(text("""
            INSERT INTO public.governance_workflow_versions
              (uid,workflow_uid,version,status,definition,created_by,published_by,published_at)
            VALUES (CAST(:version AS uuid),CAST(:workflow AS uuid),1,'published','{}',
                    CAST(:owner AS uuid),CAST(:owner AS uuid),clock_timestamp())
        """), fixture)
        connection.execute(text("""
            UPDATE public.governance_workflows
               SET active_version_uid=CAST(:version AS uuid)
             WHERE uid=CAST(:workflow AS uuid)
        """), fixture)
        connection.execute(text("""
            INSERT INTO public.governed_agents
              (uid,code,name,purpose,owner_uid,machine_subject,business_domain_uids,
               environments,autonomy_level,prompt_policy,status,created_by,updated_by)
            VALUES (CAST(:agent AS uuid),:code,'wp09 http','test',CAST(:owner AS uuid),
                    :machine_subject,CAST(:domains AS jsonb),'["test"]',
                    'approval_execution','{}','active',CAST(:owner AS uuid),
                    CAST(:owner AS uuid))
        """), {
            **fixture,
            "machine_subject": f"wp09-http-{suffix}",
            "domains": json.dumps([fixture["domain"]]),
        })
        connection.execute(text("""
            INSERT INTO public.agent_tool_grants
              (uid,agent_uid,interface_type,tool_name,action,business_domain_uid,
               environment,risk_level,requires_approval,status,created_by)
            VALUES (CAST(:grant AS uuid),CAST(:agent AS uuid),'mcp','knowledge.search',
                    'execute',CAST(:domain AS uuid),'test','high',true,'active',
                    CAST(:owner AS uuid))
        """), fixture)
        connection.execute(text("""
            INSERT INTO public.model_gateway_routes
              (route_id,tenant_id,principal_id,business_domain_uid,environment,provider,
               model,prompt_version,generation,canary_status)
            VALUES (:route,:tenant,CAST(:owner AS uuid),CAST(:domain AS uuid),'test',
                    'controlled','model','prompt','gen','approved')
        """), fixture)
        connection.execute(text("""
            INSERT INTO public.agent_runtime_budgets
              (tenant_id,token_remaining,cost_remaining_micros,tool_remaining,
               time_remaining_ms,concurrency_remaining)
            VALUES (:tenant,10,10,3,100,3);
            INSERT INTO public.agent_runtime_model_budgets(tenant_id,model,remaining)
            VALUES (:tenant,'model',3)
        """), fixture)
    return fixture


def _seed_http_approval(engine, fixture: dict[str, Any], input_text: str) -> str:
    task_uid = str(uuid.uuid4())
    input_hash = hashlib.sha256(input_text.strip().encode()).hexdigest()
    context = json.dumps({
        "agent_uid": fixture["agent"],
        "principal_id": fixture["owner"],
        "business_domain_uid": fixture["domain"],
        "environment": "test",
        "tool_name": "knowledge.search",
        "action": "execute",
        "request_digest": input_hash,
        "automatic_execution_allowed": False,
    })
    with engine.begin() as connection:
        connection.execute(text("""
            INSERT INTO public.governance_tasks
              (uid,task_code,workflow_uid,workflow_version,task_type,subject_type,
               subject_uid,source_type,source_uid,title,description,business_domain_uid,
               priority,status,assignee_uid,due_at,route_snapshot,context,created_by,updated_by)
            VALUES (CAST(:task AS uuid),:task_code,CAST(:workflow AS uuid),1,'high_risk',
                    'agent',:agent,'wp09',:task,'wp09 http','wp09 http',:domain,
                    'critical','approved',CAST(:owner AS uuid),
                    clock_timestamp()+interval '5 minutes','{}',CAST(:context AS jsonb),
                    CAST(:owner AS uuid),CAST(:owner AS uuid))
        """), {**fixture, "task": task_uid, "task_code": f"WP09-HTTP-{uuid.uuid4().hex[:12]}", "context": context})
        connection.execute(text("""
            INSERT INTO public.governance_task_reviews(uid,task_uid,reviewer_uid,decision,reason)
            VALUES (CAST(:uid AS uuid),CAST(:task AS uuid),CAST(:reviewer AS uuid),
                    'approve','wp09 http')
        """), {**fixture, "task": task_uid, "uid": str(uuid.uuid4())})
    fixture["task_uids"].append(task_uid)
    return task_uid


def _runtime_payload(grant_uid: str, approval_task_uid: str, *, key: str, input_text: str) -> dict[str, Any]:
    return {
        "grant_uid": grant_uid,
        "idempotency_key": key,
        "input_text": input_text,
        "evidence_refs": [],
        "estimated_tokens": 1,
        "estimated_cost_micros": "1",
        "requested_time_ms": 1,
        "approval_task_uid": approval_task_uid,
    }


def _cleanup_http_runtime_fixture(engine, fixture: dict[str, Any]) -> None:
    with engine.begin() as connection:
        connection.execute(text("""
            DELETE FROM public.agent_runtime_control_claims WHERE tenant_id=:tenant;
            DELETE FROM public.agent_runtime_control_events WHERE tenant_id=:tenant;
            DELETE FROM public.agent_runtime_settlement_audits WHERE tenant_id=:tenant;
            DELETE FROM public.agent_invocation_audits WHERE tenant_id=:tenant;
            DELETE FROM public.agent_runtime_reservations WHERE tenant_id=:tenant;
            DELETE FROM public.agent_runtime_model_budgets WHERE tenant_id=:tenant;
            DELETE FROM public.agent_runtime_budgets WHERE tenant_id=:tenant;
            DELETE FROM public.agent_generation_canaries WHERE tenant_id=:tenant;
            DELETE FROM public.agent_runtime_states WHERE tenant_id=:tenant;
            DELETE FROM public.data_incidents WHERE uid=CAST(:control_incident AS uuid);
            DELETE FROM public.governance_task_reviews
             WHERE task_uid = ANY(CAST(:task_uids AS uuid[]));
            DELETE FROM public.governance_tasks WHERE uid = ANY(CAST(:task_uids AS uuid[]));
            DELETE FROM public.agent_governance_events WHERE agent_uid=CAST(:agent AS uuid);
            DELETE FROM public.agent_runtime_control_claims WHERE agent_uid=CAST(:agent AS uuid);
            DELETE FROM public.agent_credentials WHERE agent_uid=CAST(:agent AS uuid);
            DELETE FROM public.model_gateway_routes WHERE route_id=:route;
            DELETE FROM public.agent_tool_grants WHERE uid=CAST(:grant AS uuid);
            DELETE FROM public.governed_agents WHERE uid=CAST(:agent AS uuid);
            UPDATE public.governance_workflows SET active_version_uid=NULL
             WHERE uid=CAST(:workflow AS uuid);
            DELETE FROM public.governance_workflow_versions WHERE uid=CAST(:version AS uuid);
            DELETE FROM public.governance_workflows WHERE uid=CAST(:workflow AS uuid)
        """), fixture)
    # Deliberately use a new connection: no assertion may be satisfied only by
    # the cleanup transaction's uncommitted view.
    with engine.connect() as observer:
        remaining = observer.execute(text("""
            SELECT
              (SELECT count(*) FROM public.agent_runtime_control_claims WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.agent_runtime_control_events WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.agent_runtime_settlement_audits WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.agent_invocation_audits WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.agent_runtime_reservations WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.agent_runtime_model_budgets WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.agent_runtime_budgets WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.agent_generation_canaries WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.agent_runtime_states WHERE tenant_id=:tenant) +
              (SELECT count(*) FROM public.data_incidents WHERE uid=CAST(:control_incident AS uuid)) +
              (SELECT count(*) FROM public.governance_task_reviews WHERE task_uid=ANY(CAST(:task_uids AS uuid[]))) +
              (SELECT count(*) FROM public.governance_tasks WHERE uid=ANY(CAST(:task_uids AS uuid[]))) +
              (SELECT count(*) FROM public.agent_credentials WHERE agent_uid=CAST(:agent AS uuid)) +
              (SELECT count(*) FROM public.agent_governance_events WHERE agent_uid=CAST(:agent AS uuid)) +
              (SELECT count(*) FROM public.model_gateway_routes WHERE route_id=:route) +
              (SELECT count(*) FROM public.agent_tool_grants WHERE uid=CAST(:grant AS uuid)) +
              (SELECT count(*) FROM public.governed_agents WHERE uid=CAST(:agent AS uuid)) +
              (SELECT count(*) FROM public.governance_workflow_versions WHERE uid=CAST(:version AS uuid)) +
              (SELECT count(*) FROM public.governance_workflows WHERE uid=CAST(:workflow AS uuid))
        """), fixture).scalar_one()
    assert remaining == 0


def test_wp09_runtime_http_commits_once_across_new_app_restart_and_rolls_back(monkeypatch):
    """Exercise the real Flask route, credential service and PostgreSQL boundary."""
    database_url = _database_url()
    engine = create_engine(database_url, pool_pre_ping=True)
    fixture = _seed_http_runtime_fixture(engine)
    first_task = _seed_http_approval(engine, fixture, "approved-runtime-input")
    try:
        monkeypatch.setenv("DATABASE_URL", database_url)
        monkeypatch.setattr(
            "app.core.system.permissions.authenticate_request",
            lambda: {"id": fixture["owner"], "sub": fixture["owner"], "roles": ["editor"]},
        )
        from app import create_app, db
        from app.api.knowledge_base.agent_governance_routes import _service

        first_app = create_app()
        first_app.config.update(TESTING=True)
        with first_app.app_context():
            credential = _service().issue_credential(
                fixture["agent"], {"ttl_seconds": 60}, actor_uid=fixture["owner"]
            )
        headers = {
            "Authorization": "Bearer integration-test",
            "X-Agent-Credential": credential["token"],
        }
        payload = _runtime_payload(
            fixture["grant"], first_task, key="restart-once", input_text="approved-runtime-input"
        )
        response = first_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json=payload,
            headers=headers,
        )
        assert response.status_code == 201
        assert response.headers["Cache-Control"] == "no-store"
        assert response.get_json()["data"]["replay"] is False
        with engine.connect() as observer:
            assert observer.execute(text("""
                SELECT token_remaining,cost_remaining_micros,tool_remaining,
                       time_remaining_ms,concurrency_remaining
                  FROM public.agent_runtime_budgets WHERE tenant_id=:tenant
            """), fixture).one() == (9, 9, 2, 99, 2)
            assert observer.execute(text("""
                SELECT count(*) FROM public.agent_invocation_audits WHERE tenant_id=:tenant
            """), fixture).scalar_one() == 1
            assert observer.execute(text("""
                SELECT status FROM public.governance_tasks WHERE uid=CAST(:task AS uuid)
            """), {"task": first_task}).scalar_one() == "closed"

        with first_app.app_context():
            db.session.remove()
        del first_app

        restarted_app = create_app()
        restarted_app.config.update(TESTING=True)
        replay = restarted_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json=payload,
            headers=headers,
        )
        assert replay.status_code == 201
        assert replay.headers["Cache-Control"] == "no-store"
        assert replay.get_json()["data"]["replay"] is True
        with engine.connect() as observer:
            assert observer.execute(text("""
                SELECT token_remaining,cost_remaining_micros,tool_remaining,
                       time_remaining_ms,concurrency_remaining
                  FROM public.agent_runtime_budgets WHERE tenant_id=:tenant
            """), fixture).one() == (9, 9, 2, 99, 2)
            assert observer.execute(text("""
                SELECT count(*) FROM public.agent_invocation_audits WHERE tenant_id=:tenant
            """), fixture).scalar_one() == 1
            assert observer.execute(text("""
                SELECT status FROM public.governance_tasks WHERE uid=CAST(:task AS uuid)
            """), {"task": first_task}).scalar_one() == "closed"

        changed = restarted_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json={**payload, "input_text": "changed-runtime-input"},
            headers=headers,
        )
        assert changed.status_code == 409
        assert changed.headers["Cache-Control"] == "no-store"

        rollback_task = _seed_http_approval(engine, fixture, "rollback-runtime-input")
        rollback_payload = _runtime_payload(
            fixture["grant"], rollback_task, key="rollback-once", input_text="rollback-runtime-input"
        )
        original_commit = db.session.commit

        def fail_commit():
            raise RuntimeError("synthetic commit failure")

        monkeypatch.setattr(db.session, "commit", fail_commit)
        rejected = restarted_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json=rollback_payload,
            headers=headers,
        )
        assert rejected.status_code == 409
        assert rejected.headers["Cache-Control"] == "no-store"
        with engine.connect() as observer:
            assert observer.execute(text("""
                SELECT count(*) FROM public.agent_invocation_audits WHERE tenant_id=:tenant
            """), fixture).scalar_one() == 1
            assert observer.execute(text("""
                SELECT status FROM public.governance_tasks WHERE uid=CAST(:task AS uuid)
            """), {"task": rollback_task}).scalar_one() == "approved"
            assert observer.execute(text("""
                SELECT token_remaining,cost_remaining_micros,tool_remaining,
                       time_remaining_ms,concurrency_remaining
                  FROM public.agent_runtime_budgets WHERE tenant_id=:tenant
            """), fixture).one() == (9, 9, 2, 99, 2)

        monkeypatch.setattr(db.session, "commit", original_commit)
        reused = restarted_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json=rollback_payload,
            headers=headers,
        )
        assert reused.status_code == 201
        assert reused.get_json()["data"]["replay"] is False
    finally:
        _cleanup_http_runtime_fixture(engine, fixture)
        engine.dispose()


def test_wp09_runtime_settlement_is_fenced_replay_safe_and_cross_session(monkeypatch):
    """498 settles the durable reservation exactly once on the real HTTP path."""
    database_url = _database_url()
    engine = create_engine(database_url, pool_pre_ping=True)
    fixture = _seed_http_runtime_fixture(engine)
    task_uid = _seed_http_approval(engine, fixture, "settlement-input")
    try:
        monkeypatch.setenv("DATABASE_URL", database_url)
        monkeypatch.setattr(
            "app.core.system.permissions.authenticate_request",
            lambda: {"id": fixture["owner"], "sub": fixture["owner"], "roles": ["editor"]},
        )
        from app import create_app, db
        from app.api.knowledge_base.agent_governance_routes import _service

        first_app = create_app()
        first_app.config.update(TESTING=True)
        with first_app.app_context():
            credential = _service().issue_credential(
                fixture["agent"], {"ttl_seconds": 60}, actor_uid=fixture["owner"]
            )
        headers = {"Authorization": "Bearer integration-test", "X-Agent-Credential": credential["token"]}
        authorize = first_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json=_runtime_payload(fixture["grant"], task_uid, key="settle-once", input_text="settlement-input"),
            headers=headers,
        )
        assert authorize.status_code == 201
        fence = authorize.get_json()["data"]["lease_fence"]
        settle_url = f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations/settle-once/settle"
        settle_payload = {
            "lease_fence": fence, "outcome": "success", "actual_tokens": 0,
            "actual_cost_micros": 0, "actual_tools": 0, "actual_time_ms": 0,
        }
        settled = first_app.test_client().post(settle_url, json=settle_payload, headers=headers)
        assert settled.status_code == 201
        assert settled.headers["Cache-Control"] == "no-store"
        assert settled.get_json()["data"]["replay"] is False
        with engine.connect() as observer:
            assert observer.execute(text("""
                SELECT token_remaining,cost_remaining_micros,tool_remaining,time_remaining_ms,concurrency_remaining
                  FROM public.agent_runtime_budgets WHERE tenant_id=:tenant
            """), fixture).one() == (10, 10, 3, 100, 3)
            assert observer.execute(text("SELECT count(*) FROM public.agent_runtime_settlement_audits WHERE tenant_id=:tenant"), fixture).scalar_one() == 1

        with first_app.app_context():
            db.session.remove()
        del first_app
        restarted_app = create_app()
        restarted_app.config.update(TESTING=True)
        replay = restarted_app.test_client().post(settle_url, json=settle_payload, headers=headers)
        assert replay.status_code == 201
        assert replay.get_json()["data"]["replay"] is True
        changed = restarted_app.test_client().post(
            settle_url, json={**settle_payload, "actual_tokens": 1}, headers=headers
        )
        assert changed.status_code == 409
        assert changed.headers["Cache-Control"] == "no-store"
        with engine.connect() as observer:
            assert observer.execute(text("SELECT count(*) FROM public.agent_runtime_settlement_audits WHERE tenant_id=:tenant"), fixture).scalar_one() == 1

        expired_task = _seed_http_approval(engine, fixture, "expired-settlement-input")
        expired_key = "settle-expired"
        expired_auth = restarted_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json=_runtime_payload(fixture["grant"], expired_task, key=expired_key, input_text="expired-settlement-input"),
            headers=headers,
        )
        assert expired_auth.status_code == 201
        with engine.begin() as writer:
            writer.execute(text("""
                UPDATE public.agent_runtime_reservations
                   SET lease_expires_at=clock_timestamp()-interval '1 second'
                 WHERE tenant_id=:tenant AND idempotency_key=:key
            """), {**fixture, "key": expired_key})
        expired_url = f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations/{expired_key}/settle"
        late = restarted_app.test_client().post(
            expired_url, json={**settle_payload, "lease_fence": expired_auth.get_json()["data"]["lease_fence"]}, headers=headers
        )
        assert late.status_code == 409
        assert late.headers["Cache-Control"] == "no-store"
        with engine.connect() as observer:
            assert observer.execute(text("""
                SELECT status,settlement FROM public.agent_runtime_reservations
                 WHERE tenant_id=:tenant AND idempotency_key=:key
            """), {**fixture, "key": expired_key}).one() == ("failed", "lease_expired")
            assert observer.execute(text("""
                SELECT token_remaining,cost_remaining_micros,tool_remaining,time_remaining_ms,concurrency_remaining
                FROM public.agent_runtime_budgets WHERE tenant_id=:tenant
            """), fixture).one() == (10, 10, 3, 100, 3)

        rollback_task = _seed_http_approval(engine, fixture, "settlement-rollback-input")
        rollback_key = "settle-rollback"
        rollback_auth = restarted_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json=_runtime_payload(fixture["grant"], rollback_task, key=rollback_key, input_text="settlement-rollback-input"),
            headers=headers,
        )
        assert rollback_auth.status_code == 201
        rollback_url = f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations/{rollback_key}/settle"
        rollback_payload = {**settle_payload, "lease_fence": rollback_auth.get_json()["data"]["lease_fence"]}
        original_commit = db.session.commit
        monkeypatch.setattr(db.session, "commit", lambda: (_ for _ in ()).throw(RuntimeError("synthetic commit")))
        failed_commit = restarted_app.test_client().post(rollback_url, json=rollback_payload, headers=headers)
        assert failed_commit.status_code == 409
        with engine.connect() as observer:
            assert observer.execute(text("""
                SELECT status FROM public.agent_runtime_reservations WHERE tenant_id=:tenant AND idempotency_key=:key
            """), {**fixture, "key": rollback_key}).scalar_one() == "reserved"
            assert observer.execute(text("""
                SELECT count(*) FROM public.agent_runtime_settlement_audits WHERE tenant_id=:tenant AND idempotency_key=:key
            """), {**fixture, "key": rollback_key}).scalar_one() == 0
        monkeypatch.setattr(db.session, "commit", original_commit)
        overflow = restarted_app.test_client().post(
            rollback_url, json={**rollback_payload, "actual_tokens": 2}, headers=headers
        )
        assert overflow.status_code == 409
        settled_after_rollback = restarted_app.test_client().post(rollback_url, json=rollback_payload, headers=headers)
        assert settled_after_rollback.status_code == 201

        concurrent_task = _seed_http_approval(engine, fixture, "settlement-concurrent-input")
        concurrent_key = "settle-concurrent"
        concurrent_auth = restarted_app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
            json=_runtime_payload(fixture["grant"], concurrent_task, key=concurrent_key, input_text="settlement-concurrent-input"),
            headers=headers,
        )
        assert concurrent_auth.status_code == 201
        concurrent_payload = json.dumps({
            "agent_uid": fixture["agent"], "principal_id": fixture["owner"],
            "worker_id": f"agent-runtime-http:{fixture['owner']}", "idempotency_key": concurrent_key,
            "lease_fence": concurrent_auth.get_json()["data"]["lease_fence"], "outcome": "cancelled",
            "actual_tokens": 0, "actual_cost_micros": 0, "actual_tools": 0, "actual_time_ms": 0,
        })
        barrier = Barrier(2)

        def settle_from_independent_connection():
            with engine.begin() as connection:
                connection.execute(text("SET LOCAL ROLE dataops_agent_runtime"))
                barrier.wait(timeout=10)
                return connection.execute(text("""
                    SELECT public.agent_runtime_settle(CAST(:payload AS jsonb))
                """), {"payload": concurrent_payload}).scalar_one()

        with ThreadPoolExecutor(max_workers=2) as pool:
            first, second = list(pool.map(lambda _unused: settle_from_independent_connection(), range(2)))
        assert sorted((first["replay"], second["replay"])) == [False, True]
    finally:
        _cleanup_http_runtime_fixture(engine, fixture)
        engine.dispose()


def test_wp09_runtime_rejects_approval_agent_and_principal_tampering(monkeypatch):
    database_url = _database_url()
    engine = create_engine(database_url, pool_pre_ping=True)
    fixture = _seed_http_runtime_fixture(engine)
    agent_tampered = _seed_http_approval(engine, fixture, "agent-tamper-input")
    principal_tampered = _seed_http_approval(engine, fixture, "principal-tamper-input")
    try:
        with engine.begin() as connection:
            connection.execute(text("""
                UPDATE public.governance_tasks SET context=jsonb_set(context,'{agent_uid}',to_jsonb(CAST(:wrong AS text)))
                 WHERE uid=CAST(:task AS uuid)
            """), {"task": agent_tampered, "wrong": str(uuid.uuid4())})
            connection.execute(text("""
                UPDATE public.governance_tasks SET context=jsonb_set(context,'{principal_id}',to_jsonb(CAST(:wrong AS text)))
                 WHERE uid=CAST(:task AS uuid)
            """), {"task": principal_tampered, "wrong": str(uuid.uuid4())})
        monkeypatch.setenv("DATABASE_URL", database_url)
        monkeypatch.setattr(
            "app.core.system.permissions.authenticate_request",
            lambda: {"id": fixture["owner"], "sub": fixture["owner"], "roles": ["editor"]},
        )
        from app import create_app
        from app.api.knowledge_base.agent_governance_routes import _service

        app = create_app()
        app.config.update(TESTING=True)
        with app.app_context():
            credential = _service().issue_credential(
                fixture["agent"], {"ttl_seconds": 60}, actor_uid=fixture["owner"]
            )
        headers = {"Authorization": "Bearer integration-test", "X-Agent-Credential": credential["token"]}
        for task, value, key in (
            (agent_tampered, "agent-tamper-input", "tampered-agent"),
            (principal_tampered, "principal-tamper-input", "tampered-principal"),
        ):
            response = app.test_client().post(
                f"/api/knowledge/agents/{fixture['agent']}/runtime/invocations",
                json=_runtime_payload(fixture["grant"], task, key=key, input_text=value), headers=headers,
            )
            assert response.status_code == 409
            assert response.headers["Cache-Control"] == "no-store"
        with engine.connect() as observer:
            assert observer.execute(text("SELECT count(*) FROM public.agent_invocation_audits WHERE tenant_id=:tenant"), fixture).scalar_one() == 0
            assert observer.execute(text("SELECT count(*) FROM public.governance_tasks WHERE uid=ANY(CAST(:tasks AS uuid[])) AND status='approved'"), {"tasks": [agent_tampered, principal_tampered]}).scalar_one() == 2
    finally:
        _cleanup_http_runtime_fixture(engine, fixture)
        engine.dispose()


def test_wp09_control_approvals_are_operation_bound_one_shot_and_replay_safe(monkeypatch):
    """499 binds recovery/promotion/rollback to separate locked work-center tasks."""
    engine = create_engine(_database_url(), pool_pre_ping=True)
    fixture = _seed_http_runtime_fixture(engine)
    incident = str(uuid.uuid4())
    fixture["control_incident"] = incident
    fixed = {"dataset_version": "fixture-v1", "dataset_digest": "a" * 64,
             "metrics_digest": "b" * 64, "threshold_policy": "default-v1"}

    def control_task(operation, digest, fence, from_state, to_state, *, incident_uid="", generation="g1"):
        task = str(uuid.uuid4())
        context = {
            "tenant_id": fixture["tenant"], "agent_uid": fixture["agent"],
            "principal_id": fixture["owner"], "business_domain_uid": fixture["domain"],
            "environment": "test", "control_operation": operation,
            "request_digest": digest, "expected_fence": str(fence), "incident_uid": incident_uid,
            "issued_at": "2026-08-13T00:00:00+00:00",
            "from_state": from_state, "to_state": to_state, "generation": generation,
            "from_generation": generation, "to_generation": generation, **fixed,
        }
        with engine.begin() as connection:
            connection.execute(text("""
                INSERT INTO public.governance_tasks(uid,task_code,workflow_uid,workflow_version,task_type,
                  subject_type,subject_uid,source_type,source_uid,title,description,business_domain_uid,
                  priority,status,assignee_uid,due_at,route_snapshot,context,created_by,updated_by)
                VALUES(CAST(:task AS uuid),:code,CAST(:workflow AS uuid),1,'high_risk','agent',:agent,
                  'wp09-control',:task,'wp09 control','wp09 control',:domain,'critical','approved',
                  CAST(:owner AS uuid),clock_timestamp()+interval '5 minutes','{}',CAST(:context AS jsonb),
                  CAST(:owner AS uuid),CAST(:owner AS uuid))
            """), {**fixture, "task": task, "code": f"WP09-CONTROL-{uuid.uuid4().hex[:10]}", "context": json.dumps(context)})
            connection.execute(text("""
                INSERT INTO public.governance_task_reviews(uid,task_uid,reviewer_uid,decision,reason)
                VALUES(CAST(:uid AS uuid),CAST(:task AS uuid),CAST(:reviewer AS uuid),'approve','control')
            """), {**fixture, "task": task, "uid": str(uuid.uuid4())})
        fixture["task_uids"].append(task)
        return task, context

    monkeypatch.setenv("DATABASE_URL", _database_url())
    current_actor = {"id": fixture["owner"]}
    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {"id": current_actor["id"], "sub": current_actor["id"], "roles": ["editor"]},
    )
    from app import create_app, db
    from app.api.knowledge_base.agent_governance_routes import (
        _runtime_service,
        _service,
    )

    app = create_app()
    app.config.update(TESTING=True)
    with app.app_context():
        credential = _service().issue_credential(
            fixture["agent"], {"ttl_seconds": 60}, actor_uid=fixture["owner"]
        )
    headers = {"Authorization": "Bearer integration-test", "X-Agent-Credential": credential["token"]}

    def invoke(payload):
        response = app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/control",
            json={key: value for key, value in payload.items() if key not in {"agent_uid", "principal_id"}},
            headers=headers,
        )
        if response.status_code != 201:
            raise RuntimeError(response.get_json())
        return response.get_json()["data"]

    try:
        with engine.begin() as connection:
            connection.execute(text("""
                INSERT INTO public.data_incidents(uid,code,dedup_key,title,severity,status,owner_uid,
                  escalation_level,first_detected_at,last_observed_at,created_by)
                VALUES(CAST(:incident AS uuid),:code,:digest,'wp09','critical','open',CAST(:owner AS uuid),1,
                  clock_timestamp(),clock_timestamp(),CAST(:owner AS uuid));
                INSERT INTO public.agent_runtime_states(tenant_id,state,incident_ref,lease_fence)
                VALUES(:tenant,'paused',:incident,0);
                INSERT INTO public.agent_generation_canaries(tenant_id,generation,status,lease_fence)
                VALUES(:tenant,'g1','pending',0)
            """), {**fixture, "incident": incident, "code": f"WP09-CONTROL-{uuid.uuid4().hex[:10]}", "digest": hashlib.sha256(incident.encode()).hexdigest()})
            recover_task, recover_context = control_task("recover", "c" * 64, 0, "paused", "active", incident_uid=incident)
            recover = {**{key: value for key, value in recover_context.items() if key != "control_operation"}, "operation": "recover", "approval_task_uid": recover_task, "idempotency_key": "recover-1", "expected_fence": 0}
        # A persisted, credential-bound claim expired by the DB clock cannot be used late.
        with app.app_context():
            stored_credential = _service().repository.get_credential(credential["jti"])
            expired_claim = _runtime_service().repository.issue_control_claim({
                "claim_uid": str(uuid.uuid4()), "credential_uid": stored_credential["uid"],
                "credential_token_digest": hashlib.sha256(credential["token"].encode()).hexdigest(),
                "nonce": str(uuid.uuid4()), "control_payload": recover,
            })
            db.session.commit()
        with engine.begin() as connection:
            connection.execute(text("""
                UPDATE public.agent_runtime_control_claims
                   SET expires_at=clock_timestamp()-interval '1 second'
                 WHERE claim_uid=CAST(:claim AS uuid)
            """), {"claim": expired_claim})
        with app.app_context(), pytest.raises(RuntimeError):
            _runtime_service().repository.control_claimed(recover, expired_claim)
        with app.app_context():
            db.session.rollback()
        assert invoke(recover)["replay"] is False
        # Discard the first Flask application/session before replaying its signed credential.
        app = create_app()
        app.config.update(TESTING=True)
        assert invoke(recover)["replay"] is True
        with engine.connect() as observer:
            assert observer.execute(text("SELECT state FROM public.agent_runtime_states WHERE tenant_id=:tenant"), fixture).scalar_one() == "active"
            assert observer.execute(text("SELECT status FROM public.governance_tasks WHERE uid=CAST(:task AS uuid)"), {"task": recover_task}).scalar_one() == "closed"
        reused = {**recover, "operation": "canary_promote", "idempotency_key": "cross-operation"}
        with pytest.raises(Exception):  # noqa: B017 - PostgreSQL emits a driver-specific DBAPI error.
            invoke(reused)

        promote_task, promote_context = control_task("canary_promote", "d" * 64, 0, "pending", "approved")
        promote = {**{key: value for key, value in promote_context.items() if key != "control_operation"}, "operation": "canary_promote", "approval_task_uid": promote_task, "idempotency_key": "promote-1", "expected_fence": 0}
        for attack in (
            {"tenant_id": "other-tenant"},
            {"incident_uid": str(uuid.uuid4())}, {"dataset_digest": "e" * 64},
            {"metrics_digest": "e" * 64}, {"generation": "other-generation"},
        ):
            with pytest.raises(Exception):  # noqa: B017 - PostgreSQL emits a driver-specific DBAPI error.
                invoke({**promote, **attack, "idempotency_key": f"attack-{uuid.uuid4().hex[:8]}"})
        current_actor["id"] = fixture["reviewer"]
        with pytest.raises(Exception):  # noqa: B017 - Flask returns a stable no-store rejection.
            invoke(promote)
        current_actor["id"] = fixture["owner"]
        with engine.connect() as observer:
            assert observer.execute(text("SELECT status FROM public.governance_tasks WHERE uid=CAST(:task AS uuid)"), {"task": promote_task}).scalar_one() == "approved"
            assert observer.execute(text("SELECT status FROM public.agent_generation_canaries WHERE tenant_id=:tenant AND generation='g1'"), fixture).scalar_one() == "pending"
        assert invoke(promote)["lease_fence"] == 1
        wrong_credential = app.test_client().post(
            f"/api/knowledge/agents/{fixture['agent']}/runtime/control",
            json={key: value for key, value in promote.items() if key not in {"agent_uid", "principal_id"}},
            headers={**headers, "X-Agent-Credential": headers["X-Agent-Credential"] + "x"},
        )
        assert wrong_credential.status_code == 403
        assert wrong_credential.headers["Cache-Control"] == "no-store"
        changed_metrics = {**promote, "metrics_digest": "e" * 64}
        with pytest.raises(Exception):  # noqa: B017 - PostgreSQL emits a driver-specific DBAPI error.
            invoke(changed_metrics)
        rollback_task, rollback_context = control_task("canary_rollback", "f" * 64, 1, "approved", "rolled_back")
        rollback = {**{key: value for key, value in rollback_context.items() if key != "control_operation"}, "operation": "canary_rollback", "approval_task_uid": rollback_task, "idempotency_key": "rollback-1", "expected_fence": 1}
        assert invoke(rollback)["lease_fence"] == 2
        concurrent_task, concurrent_context = control_task("canary_promote", "1" * 64, 0, "pending", "approved", generation="g2")
        concurrent = {**{key: value for key, value in concurrent_context.items() if key != "control_operation"}, "operation": "canary_promote", "approval_task_uid": concurrent_task, "idempotency_key": "promote-concurrent", "expected_fence": 0}
        with engine.begin() as connection:
            connection.execute(text("""
                INSERT INTO public.agent_generation_canaries(tenant_id,generation,status,lease_fence)
                VALUES(:tenant,'g2','pending',0)
            """), fixture)
        barrier = Barrier(2)

        def concurrent_invoke():
            local_app = create_app()
            local_app.config.update(TESTING=True)
            barrier.wait(timeout=10)
            response = local_app.test_client().post(
                f"/api/knowledge/agents/{fixture['agent']}/runtime/control",
                json={key: value for key, value in concurrent.items() if key not in {"agent_uid", "principal_id"}},
                headers=headers,
            )
            assert response.status_code == 201, response.get_json()
            return response.get_json()["data"]

        with ThreadPoolExecutor(max_workers=2) as pool:
            first, second = list(pool.map(lambda _unused: concurrent_invoke(), range(2)))
        assert sorted((first["replay"], second["replay"])) == [False, True]
        with engine.connect() as observer:
            assert observer.execute(text("SELECT status,lease_fence FROM public.agent_generation_canaries WHERE tenant_id=:tenant AND generation='g1'"), fixture).one() == ("rolled_back", 2)
            assert observer.execute(text("SELECT count(*) FROM public.agent_runtime_control_events WHERE tenant_id=:tenant"), fixture).scalar_one() == 4
    finally:
        _cleanup_http_runtime_fixture(engine, fixture)
        engine.dispose()


def test_wp09_approved_high_risk_task_is_consumed_once_and_exact_replay_is_free():
    url = os.getenv("TEST_DATABASE_URL")
    if not url:
        pytest.skip("TEST_DATABASE_URL is required")
    engine = create_engine(url)
    suffix = uuid.uuid4().hex[:12]
    tenant = f"wp09_{suffix}"
    ids = {name: str(uuid.uuid4()) for name in ("workflow", "version", "agent", "grant", "route", "task")}
    try:
        c = engine.connect()
        transaction = c.begin()
        try:
            users = [str(row[0]) for row in c.execute(text("SELECT id::text FROM public.users WHERE status='active' ORDER BY id LIMIT 2"))]
            if len(users) < 2:
                pytest.skip("requires two existing test users for independent review")
            owner, reviewer = users
            domain = str(uuid.uuid4())
            digest = hashlib.sha256(b"approved-runtime-input").hexdigest()
            c.execute(text("INSERT INTO public.governance_workflows(uid,code,name,status,current_version,created_by) VALUES(CAST(:id AS uuid),:code,'wp09','published',1,CAST(:owner AS uuid))"), {"id": ids["workflow"], "code": f"wp09-{suffix}", "owner": owner})
            c.execute(text("INSERT INTO public.governance_workflow_versions(uid,workflow_uid,version,status,definition,created_by,published_by,published_at) VALUES(CAST(:id AS uuid),CAST(:workflow AS uuid),1,'published','{}',CAST(:owner AS uuid),CAST(:owner AS uuid),clock_timestamp())"), {"id": ids["version"], "workflow": ids["workflow"], "owner": owner})
            c.execute(text("UPDATE public.governance_workflows SET active_version_uid=CAST(:version AS uuid) WHERE uid=CAST(:workflow AS uuid)"), {"workflow": ids["workflow"], "version": ids["version"]})
            c.execute(text("INSERT INTO public.governed_agents(uid,code,name,purpose,owner_uid,machine_subject,business_domain_uids,environments,autonomy_level,prompt_policy,status,created_by,updated_by) VALUES(CAST(:id AS uuid),:code,'wp09','test',CAST(:owner AS uuid),:subject,CAST(:domains AS jsonb),'[\"test\"]','approval_execution','{}','active',CAST(:owner AS uuid),CAST(:owner AS uuid))"), {"id": ids["agent"], "code": f"wp09-{suffix}", "owner": owner, "subject": f"wp09-{suffix}", "domains": json.dumps([domain])})
            c.execute(text("INSERT INTO public.agent_tool_grants(uid,agent_uid,interface_type,tool_name,action,business_domain_uid,environment,risk_level,requires_approval,status,created_by) VALUES(CAST(:id AS uuid),CAST(:agent AS uuid),'mcp','knowledge.search','execute',CAST(:domain AS uuid),'test','high',true,'active',CAST(:owner AS uuid))"), {"id": ids["grant"], "agent": ids["agent"], "domain": domain, "owner": owner})
            c.execute(text("INSERT INTO public.model_gateway_routes(route_id,tenant_id,principal_id,business_domain_uid,environment,provider,model,prompt_version,generation,canary_status) VALUES(:id,:tenant,CAST(:owner AS uuid),CAST(:domain AS uuid),'test','controlled','model','prompt','gen','approved')"), {"id": ids["route"], "tenant": tenant, "owner": owner, "domain": domain})
            c.execute(text("INSERT INTO public.agent_runtime_budgets(tenant_id,token_remaining,cost_remaining_micros,tool_remaining,time_remaining_ms,concurrency_remaining) VALUES(:tenant,10,10,2,100,2); INSERT INTO public.agent_runtime_model_budgets(tenant_id,model,remaining) VALUES(:tenant,'model',2)"), {"tenant": tenant})
            context = json.dumps({"agent_uid": ids["agent"], "principal_id": owner, "business_domain_uid": domain, "environment": "test", "tool_name": "knowledge.search", "action": "execute", "request_digest": digest, "automatic_execution_allowed": False})
            c.execute(text("INSERT INTO public.governance_tasks(uid,task_code,workflow_uid,workflow_version,task_type,subject_type,subject_uid,source_type,source_uid,title,description,business_domain_uid,priority,status,assignee_uid,due_at,route_snapshot,context,created_by,updated_by) VALUES(CAST(:id AS uuid),:code,CAST(:workflow AS uuid),1,'high_risk','agent',:agent,'wp09',:agent,'wp09','wp09',:domain,'critical','approved',CAST(:owner AS uuid),clock_timestamp()+interval '5 minutes','{}',CAST(:context AS jsonb),CAST(:owner AS uuid),CAST(:owner AS uuid))"), {"id": ids["task"], "code": f"WP09-{suffix}", "workflow": ids["workflow"], "agent": ids["agent"], "domain": domain, "owner": owner, "context": context})
            c.execute(text("INSERT INTO public.governance_task_reviews(uid,task_uid,reviewer_uid,decision,reason) VALUES(CAST(:id AS uuid),CAST(:task AS uuid),CAST(:reviewer AS uuid),'approve','wp09')"), {"id": str(uuid.uuid4()), "task": ids["task"], "reviewer": reviewer})
            incident = str(uuid.uuid4())
            c.execute(text("INSERT INTO public.data_incidents(uid,code,dedup_key,title,severity,status,owner_uid,escalation_level,first_detected_at,last_observed_at,created_by) VALUES(CAST(:id AS uuid),:code,:dedup,'wp09','critical','open',CAST(:owner AS uuid),1,clock_timestamp(),clock_timestamp(),CAST(:owner AS uuid)); INSERT INTO public.agent_runtime_states(tenant_id,state,incident_ref,lease_fence) VALUES(:tenant,'paused',CAST(:incident AS text),0); INSERT INTO public.agent_generation_canaries(tenant_id,generation,status,lease_fence) VALUES(:tenant,'gen','pending',0)"), {"id": incident, "code": f"INC-{suffix}", "dedup": hashlib.sha256((tenant + '-incident').encode()).hexdigest(), "owner": owner, "tenant": tenant, "incident": incident})
            with pytest.raises(Exception), c.begin_nested():  # noqa: B017 - Legacy direct gateway emits driver-specific error.
                c.execute(text("SELECT public.agent_runtime_transition(:tenant,'active',NULL,CAST(:task AS uuid),0)"), {"tenant": tenant, "task": ids["task"]})
            with pytest.raises(Exception), c.begin_nested():  # noqa: B017 - Legacy direct gateway emits driver-specific error.
                c.execute(text("SELECT public.agent_runtime_canary_transition(:tenant,'gen','approved',CAST(:task AS uuid),0,CAST(:metrics AS jsonb),'engineering-fixture-v1')"), {"tenant": tenant, "task": ids["task"], "metrics": '{"recall":1.0}'})
            payload = {"agent_uid":ids["agent"],"grant_uid":ids["grant"],"tenant_id":tenant,"principal_id":owner,"business_domain_uid":domain,"environment":"test","provider":"controlled","model":"model","prompt_version":"prompt","generation":"gen","interface_type":"mcp","tool_name":"knowledge.search","action":"execute","risk_level":"high","idempotency_key":"once","request_digest":hashlib.sha256(b"approved-runtime-input|[]").hexdigest(),"input_hash":digest,"evidence_digests":[],"estimated_tokens":"1","estimated_cost_micros":"1","requested_time_ms":"1","approval_task_uid":ids["task"]}
            transaction.commit()
            barrier = Barrier(2)
            def invoke():
                with engine.begin() as worker:
                    barrier.wait(timeout=10)
                    return worker.execute(text("SELECT public.agent_runtime_authorize(CAST(:payload AS jsonb))"), {"payload": json.dumps(payload)}).scalar_one()
            with ThreadPoolExecutor(max_workers=2) as pool:
                first, replay = list(pool.map(lambda _unused: invoke(), range(2)))
            assert sum(item["replay"] is False for item in (first, replay)) == 1
            assert c.execute(text("SELECT status FROM public.governance_tasks WHERE uid=CAST(:id AS uuid)"), {"id": ids["task"]}).scalar_one() == "closed"
            changed = {**payload, "input_hash": "b" * 64, "request_digest": "b" * 64}
            with pytest.raises(Exception):  # noqa: B017 - PostgreSQL emits a driver-specific DBAPI error.
                c.execute(text("SELECT public.agent_runtime_authorize(CAST(:payload AS jsonb))"), {"payload": json.dumps(changed)})
            c.commit()
        finally:
            if transaction.is_active:
                transaction.rollback()
            c.close()
    finally:
        with engine.begin() as c:
            c.execute(text("DELETE FROM public.agent_invocation_audits WHERE tenant_id=:tenant; DELETE FROM public.agent_runtime_reservations WHERE tenant_id=:tenant; DELETE FROM public.agent_generation_canaries WHERE tenant_id=:tenant; DELETE FROM public.agent_runtime_states WHERE tenant_id=:tenant; DELETE FROM public.agent_runtime_budgets WHERE tenant_id=:tenant; DELETE FROM public.governance_task_reviews WHERE task_uid=CAST(:task AS uuid); DELETE FROM public.governance_tasks WHERE uid=CAST(:task AS uuid); DELETE FROM public.data_incidents WHERE code=:incident_code; DELETE FROM public.model_gateway_routes WHERE tenant_id=:tenant; DELETE FROM public.agent_tool_grants WHERE uid=CAST(:grant AS uuid); DELETE FROM public.governed_agents WHERE uid=CAST(:agent AS uuid); UPDATE public.governance_workflows SET active_version_uid=NULL WHERE uid=CAST(:workflow AS uuid); DELETE FROM public.governance_workflow_versions WHERE uid=CAST(:version AS uuid); DELETE FROM public.governance_workflows WHERE uid=CAST(:workflow AS uuid)"), {**ids, "tenant": tenant, "incident_code": f"INC-{suffix}"})
        engine.dispose()


def test_wp09_persistent_pause_degrade_and_late_fence_are_database_enforced():
    url = os.getenv("TEST_DATABASE_URL")
    if not url:
        pytest.skip("TEST_DATABASE_URL is required")
    engine = create_engine(url)
    tenant, incident = f"wp09_{uuid.uuid4().hex[:12]}", str(uuid.uuid4())
    try:
        with engine.begin() as c:
            owner = str(c.execute(text("SELECT id::text FROM public.users WHERE status='active' LIMIT 1")).scalar_one())
            c.execute(text("INSERT INTO public.data_incidents(uid,code,dedup_key,title,severity,status,owner_uid,escalation_level,first_detected_at,last_observed_at,created_by) VALUES(CAST(:id AS uuid),:code,:dedup,'wp09','critical','open',CAST(:owner AS uuid),1,clock_timestamp(),clock_timestamp(),CAST(:owner AS uuid))"), {"id": incident, "code": f"WP09-{tenant[-12:]}", "dedup": hashlib.sha256(tenant.encode()).hexdigest(), "owner": owner})
            c.execute(text("INSERT INTO public.agent_runtime_states(tenant_id,state,lease_fence) VALUES(:tenant,'active',0); INSERT INTO public.agent_generation_canaries(tenant_id,generation,status,lease_fence) VALUES(:tenant,'g1','pending',0)"), {"tenant": tenant})
            with pytest.raises(Exception), c.begin_nested():  # noqa: B017 - Legacy direct gateway emits driver-specific error.
                c.execute(text("SELECT public.agent_runtime_transition(:tenant,'degraded',CAST(:incident AS uuid),NULL,0)"), {"tenant": tenant, "incident": incident})
            with pytest.raises(Exception), c.begin_nested():  # noqa: B017 - Legacy direct gateway emits driver-specific error.
                c.execute(text("SELECT public.agent_runtime_transition(:tenant,'paused',CAST(:incident AS uuid),NULL,1)"), {"tenant": tenant, "incident": incident})
            with pytest.raises(Exception):  # noqa: B017 - PostgreSQL emits a driver-specific DBAPI error.
                c.execute(text("SELECT public.agent_runtime_transition(:tenant,'paused',CAST(:incident AS uuid),NULL,1)"), {"tenant": tenant, "incident": incident})
            with pytest.raises(Exception):  # noqa: B017 - PostgreSQL emits a driver-specific DBAPI error.
                c.execute(text("SELECT public.agent_runtime_transition(:tenant,'active',NULL,NULL,2)"), {"tenant": tenant})
            with pytest.raises(Exception):  # noqa: B017 - PostgreSQL emits a driver-specific DBAPI error.
                c.execute(text("SELECT public.agent_runtime_canary_transition(:tenant,'g1','approved',NULL,0,'{}','fixture-v1')"), {"tenant": tenant})
    finally:
        with engine.begin() as c:
            c.execute(text("DELETE FROM public.agent_generation_canaries WHERE tenant_id=:tenant; DELETE FROM public.agent_runtime_states WHERE tenant_id=:tenant; DELETE FROM public.data_incidents WHERE uid=CAST(:incident AS uuid)"), {"tenant": tenant, "incident": incident})
        engine.dispose()