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()