fix(security): isolate allowlisted execution broker

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -982,16 +982,30 @@ def test_runtime_receipt_auxiliary_sql_keeps_source_family_counts_schema_safe():
     )
 
     assert "semantic_operation_type" in _RUNTIME_OPERATION_COUNTS_SQL
-    assert "km_linked" in _RUNTIME_OPERATION_COUNTS_SQL
+    assert "coalesce(input ->> 'semantic_operation_type', operation_type)" in (
+        _RUNTIME_OPERATION_COUNTS_SQL
+    )
     assert "log_controlled_writeback_dispatched" in _RUNTIME_OPERATION_COUNTS_SQL
     assert "semantic_operation_type" in _RUNTIME_OPERATION_LATEST_SQL
-    assert "km_linked" in _RUNTIME_OPERATION_LATEST_SQL
+    assert "coalesce(input ->> 'semantic_operation_type', operation_type)" in (
+        _RUNTIME_OPERATION_LATEST_SQL
+    )
     assert "automation_run_id" in _RUNTIME_OPERATION_LATEST_SQL
     assert "automation_run_id" in runtime_control_module._RUNTIME_OPERATION_LATEST_DIRECT_SQL
     assert "automation_run_id" in runtime_control_module._RUNTIME_AUTO_REPAIR_LATEST_SQL
     assert "automation_run_id" in runtime_control_module._RUNTIME_VERIFIER_LATEST_SQL
     assert "automation_run_id" in runtime_control_module._RUNTIME_KM_LATEST_SQL
     assert "automation_run_id" in _RUNTIME_TELEGRAM_LATEST_SQL
+    for operation_type in (
+        "ansible_executor_capability_issued",
+        "ansible_executor_capability_revoked",
+        "ansible_executor_capability_expired",
+    ):
+        assert operation_type in _RUNTIME_OPERATION_COUNTS_SQL
+        assert operation_type in _RUNTIME_OPERATION_LATEST_SQL
+        assert operation_type in runtime_control_module._RUNTIME_OPERATION_COUNTS_DIRECT_SQL
+        assert operation_type in runtime_control_module._RUNTIME_OPERATION_LATEST_DIRECT_SQL
+        assert operation_type in runtime_control_module._RUNTIME_EXECUTOR_LOG_COUNTS_SQL
     assert "GROUP BY coalesce(status, 'unknown')" in _RUNTIME_TIMELINE_COUNTS_SQL
     assert "FROM timeline_events" in _RUNTIME_TIMELINE_COUNTS_SQL
     assert "count(*) AS total" in _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL
@@ -2581,6 +2595,66 @@ def test_runtime_operation_latest_queries_prioritize_latest_apply_chain() -> Non
         assert "operation_row.op_id::text = latest_apply_chain.candidate_op_id" in sql
 
 
+def test_execution_capability_lifecycle_requires_same_run_terminal_receipt() -> None:
+    rows = [
+        {
+            "op_id": "apply-1",
+            "operation_type": "ansible_apply_executed",
+            "status": "success",
+            "automation_run_id": "run-1",
+            "capability_op_id": "capability-1",
+        },
+        {
+            "op_id": "capability-1",
+            "operation_type": "ansible_executor_capability_issued",
+            "status": "success",
+            "automation_run_id": "run-1",
+            "capability_issued_at": "2026-07-11T01:00:00+00:00",
+            "capability_expires_at": "2026-07-11T01:10:00+00:00",
+        },
+        {
+            "op_id": "revoke-1",
+            "parent_op_id": "capability-1",
+            "operation_type": "ansible_executor_capability_revoked",
+            "status": "success",
+            "automation_run_id": "run-1",
+            "capability_terminal_status": "controlled_apply_verified",
+        },
+    ]
+    summary = {
+        operation_type: {"total": 1, "recent": 1}
+        for operation_type in (
+            "ansible_executor_capability_issued",
+            "ansible_executor_capability_revoked",
+            "ansible_executor_capability_expired",
+        )
+    }
+
+    lifecycle = runtime_control_module._latest_execution_capability_lifecycle(
+        operation_latest_rows=rows,
+        operation_summary=summary,
+    )
+
+    assert lifecycle["closed"] is True
+    assert lifecycle["same_run_correlation"] is True
+    assert lifecycle["capability_op_id"] == "capability-1"
+    assert lifecycle["apply_capability_op_id"] == "capability-1"
+    assert lifecycle["terminal_operation_type"] == (
+        "ansible_executor_capability_revoked"
+    )
+    assert lifecycle["terminal_status"] == "controlled_apply_verified"
+    assert lifecycle["missing"] == []
+
+    rows[-1]["automation_run_id"] = "run-2"
+    mismatch = runtime_control_module._latest_execution_capability_lifecycle(
+        operation_latest_rows=rows,
+        operation_summary=summary,
+    )
+    assert mismatch["closed"] is False
+    assert mismatch["same_run_correlation"] is False
+    assert "capability_terminal_receipt" in mismatch["missing"]
+
+
 def test_runtime_execution_loop_uses_terminal_apply_without_hiding_inflight_apply():
     ledger = runtime_control_module._autonomous_execution_loop_ledger(
         project_id="awoooi",

apps/api/tests/test_awoooi_priority_work_order_readback_api.py

@@ -109,6 +109,35 @@ def _autonomous_runtime_control_closed() -> dict:
     }
 
 
+def _autonomous_runtime_control_trust_boundary_closed() -> dict:
+    payload = _autonomous_runtime_control_closed()
+    run_id = payload["strict_runtime_completion"]["automation_run_id"]
+    payload["runtime_receipt_readback"] = {
+        "execution_capability_lifecycle": {
+            "schema_version": "ansible_execution_capability_lifecycle_v1",
+            "automation_run_id": run_id,
+            "apply_op_id": "apply-p0-011",
+            "capability_op_id": "capability-p0-011",
+            "apply_capability_op_id": "capability-p0-011",
+            "terminal_operation_type": "ansible_executor_capability_revoked",
+            "same_run_correlation": True,
+            "closed": True,
+            "missing": [],
+        }
+    }
+    payload["executor_trust_boundary"] = {
+        "schema_version": "awoooi_executor_trust_boundary_readback_v1",
+        "status": "verified_ready",
+        "production_boundary_verified": True,
+        "source_sha_matches_deployment": True,
+        "verified_source_sha": "source-p0-011",
+        "receipt_ref": (
+            "configmap:awoooi-prod/awoooi-executor-boundary-verification"
+        ),
+    }
+    return payload
+
+
 async def _autonomous_runtime_control_closed_async() -> dict:
     return _autonomous_runtime_control_closed()
 
@@ -1923,6 +1952,29 @@ def test_ai_automation_live_closure_readbacks_close_node_receipts():
     ] == 6
 
 
+def test_ai_automation_trust_boundary_runtime_receipts_advance_order() -> None:
+    payload = load_latest_awoooi_priority_work_order_readback()
+
+    apply_ai_automation_live_closure_readbacks(
+        payload,
+        autonomous_runtime_control=(
+            _autonomous_runtime_control_trust_boundary_closed()
+        ),
+    )
+
+    program = payload["ai_automation_program_ledger"]
+    items = {item["id"]: item for item in program["work_items"]}
+    assert items["AIA-P0-001"]["status"] == "done"
+    assert items["AIA-P0-011"]["status"] == "done"
+    assert items["AIA-P0-011"]["runtime_progress"]["runtime_closed"] is True
+    assert items["AIA-P0-011"]["completion_evidence"][
+        "capability_op_id"
+    ] == "capability-p0-011"
+    assert program["summary"]["next_work_item_id"] == "AIA-P0-002"
+    assert program["summary"]["done_count"] == 2
+    assert program["summary"]["completion_percent"] == 10
+
+
 def test_awoooi_priority_work_order_readback_endpoint_returns_snapshot(
     monkeypatch: pytest.MonkeyPatch,
 ):

apps/api/tests/test_awooop_truth_chain_service.py

@@ -15,12 +15,16 @@ from src.services.awooop_ansible_audit_service import (
     record_ansible_decision_audit,
 )
 from src.services.awooop_ansible_check_mode_service import (
+    _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE,
     AnsibleCheckModeClaim,
     AnsibleRunResult,
     _automation_operation_log_incident_id,
     _build_auto_repair_execution_receipt,
     _claim_from_apply_operation_row,
     _claim_from_stale_check_mode_row,
+    _execution_capability_timeout_seconds,
+    _expire_stale_ansible_execution_capabilities,
+    _issue_ansible_execution_capability,
     _load_pre_decision_context_runtime_stage_receipts,
     _post_apply_km_path_type,
     _post_apply_verification_result,
@@ -29,6 +33,8 @@ from src.services.awooop_ansible_check_mode_service import (
     _record_retry_runtime_stage_receipt,
     _record_runtime_stage_receipts,
     _record_timeline_projection_receipt,
+    _resolve_execution_capability_operation_type,
+    _revoke_ansible_execution_capability,
     _run_ansible_command,
     _send_controlled_apply_telegram_receipt,
     backfill_missing_auto_repair_execution_receipts_once,
@@ -1728,6 +1734,116 @@ def test_stale_check_mode_reclaim_uses_lease_lock_and_current_policy() -> None:
     assert "effective_timeout_seconds + 120" in run_source
 
 
+def test_ansible_execution_broker_issues_expires_and_revokes_bounded_capabilities() -> None:
+    issue_source = inspect.getsource(_issue_ansible_execution_capability)
+    expire_source = inspect.getsource(_expire_stale_ansible_execution_capabilities)
+    revoke_source = inspect.getsource(_revoke_ansible_execution_capability)
+    resolver_source = inspect.getsource(_resolve_execution_capability_operation_type)
+    run_source = inspect.getsource(run_pending_check_modes_once)
+
+    assert "execution_capability_catalog_not_allowlisted" in issue_source
+    assert "execution_capability_inventory_outside_allowlist" in issue_source
+    assert "ansible_executor_capability_issued" in issue_source
+    assert "execution_capability_active_lease_exists" in issue_source
+    assert "WHERE NOT EXISTS" in issue_source
+    assert "raw_credential_exposed_to_requester" in issue_source
+    assert "expires_at" in issue_source
+    assert "ansible_executor_capability_expired" in expire_source
+    assert "FOR UPDATE SKIP LOCKED" in expire_source
+    assert "ansible_executor_capability_revoked" in revoke_source
+    assert "further_execution_allowed" in revoke_source
+    assert "automation_operation_log_type_valid" in resolver_source
+    assert "semantic_operation_type" in resolver_source
+    assert _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE == "remediation_executed"
+    assert "_issue_ansible_execution_capability" in run_source
+    assert "_revoke_ansible_execution_capability" in run_source
+    assert "_expire_stale_ansible_execution_capabilities" in run_source
+
+
+def test_ansible_execution_capability_enforces_scope_mode_and_expiry() -> None:
+    now = datetime(2026, 7, 11, 1, 0, tzinfo=UTC)
+    run_id = "00000000-0000-0000-0000-000000000201"
+    check_op_id = "00000000-0000-0000-0000-000000000202"
+    scope = {
+        "catalog_id": "ansible:110-devops",
+        "playbook_path": "infra/ansible/playbooks/110-devops.yml",
+        "apply_playbook_path": "infra/ansible/playbooks/110-devops.yml",
+        "inventory_hosts": ["host_110"],
+        "risk_level": "medium",
+        "allowed_execution_modes": ["check_mode", "controlled_apply"],
+    }
+
+    def _claim(*, capability_scope: dict, expires_at: datetime) -> AnsibleCheckModeClaim:
+        return AnsibleCheckModeClaim(
+            op_id=check_op_id,
+            source_candidate_op_id=run_id,
+            incident_id="INC-CAPABILITY",
+            catalog_id="ansible:110-devops",
+            playbook_path="infra/ansible/playbooks/110-devops.yml",
+            apply_playbook_path="infra/ansible/playbooks/110-devops.yml",
+            inventory_hosts=("host_110",),
+            risk_level="medium",
+            input_payload={
+                "automation_run_id": run_id,
+                "controlled_apply_allowed": True,
+                "execution_capability": {
+                    "capability_op_id": (
+                        "00000000-0000-0000-0000-000000000203"
+                    ),
+                    "automation_run_id": run_id,
+                    "check_mode_op_id": check_op_id,
+                    "issued_at": (now - timedelta(seconds=1)).isoformat(),
+                    "expires_at": expires_at.isoformat(),
+                    "scope": capability_scope,
+                },
+            },
+        )
+
+    claim = _claim(
+        capability_scope=scope,
+        expires_at=now + timedelta(seconds=100),
+    )
+    assert _execution_capability_timeout_seconds(
+        claim,
+        execution_mode="controlled_apply",
+        requested_timeout_seconds=180,
+        now=now,
+    ) == 95
+
+    with pytest.raises(ValueError, match="execution_capability_expired"):
+        _execution_capability_timeout_seconds(
+            _claim(capability_scope=scope, expires_at=now),
+            execution_mode="check_mode",
+            requested_timeout_seconds=180,
+            now=now,
+        )
+
+    with pytest.raises(ValueError, match="execution_capability_mode_not_allowed"):
+        _execution_capability_timeout_seconds(
+            _claim(
+                capability_scope={
+                    **scope,
+                    "allowed_execution_modes": ["check_mode"],
+                },
+                expires_at=now + timedelta(seconds=100),
+            ),
+            execution_mode="controlled_apply",
+            requested_timeout_seconds=180,
+            now=now,
+        )
+
+    with pytest.raises(ValueError, match="execution_capability_scope_mismatch"):
+        _execution_capability_timeout_seconds(
+            _claim(
+                capability_scope={**scope, "inventory_hosts": ["host_188"]},
+                expires_at=now + timedelta(seconds=100),
+            ),
+            execution_mode="check_mode",
+            requested_timeout_seconds=180,
+            now=now,
+        )
+
+
 def test_failed_check_mode_catalog_drift_replays_once() -> None:
     source = inspect.getsource(claim_catalog_drift_failed_check_modes)
     run_source = inspect.getsource(run_pending_check_modes_once)

apps/api/tests/test_config_url_validation.py

@@ -13,7 +13,6 @@ from __future__ import annotations
 import pytest
 from pydantic import ValidationError
 
-
 # =============================================================================
 # Helpers
 # =============================================================================
@@ -172,3 +171,4 @@ def test_database_pool_budget_defaults_to_production_safe_values():
     assert s.DATABASE_POOL_SIZE == 1
     assert s.DATABASE_MAX_OVERFLOW == 0
     assert s.DATABASE_POOL_TIMEOUT_SECONDS == 5.0
+    assert s.DATABASE_NULL_POOL is False

apps/api/tests/test_executor_trust_boundary_readback.py

@@ -0,0 +1,55 @@
+from src.services.executor_trust_boundary_readback import (
+    RECEIPT_SCHEMA_VERSION,
+    build_executor_trust_boundary_readback,
+)
+
+
+def _verified_receipt(source_sha: str = "abc123") -> dict[str, str]:
+    return {
+        "schema_version": RECEIPT_SCHEMA_VERSION,
+        "verified_source_sha": source_sha,
+        "verified_at": "2026-07-11T01:00:00Z",
+        "workflow_run_id": "4870",
+        "verifier": "kubectl_auth_can_i_and_live_mount_readback",
+        "api_service_account": "awoooi-api",
+        "worker_service_account": "awoooi-worker",
+        "broker_service_account": "awoooi-executor-broker",
+        "api_mutation_denied": "true",
+        "api_ssh_mount_absent": "true",
+        "worker_mutation_denied": "true",
+        "worker_ssh_mount_absent": "true",
+        "broker_kubernetes_token_absent": "true",
+        "broker_ssh_mount_present": "true",
+        "legacy_executor_binding_absent": "true",
+    }
+
+
+def test_executor_trust_boundary_requires_live_controls_and_matching_source() -> None:
+    readback = build_executor_trust_boundary_readback(
+        _verified_receipt(),
+        deployed_source_sha="abc123",
+    )
+
+    assert readback["status"] == "verified_ready"
+    assert readback["production_boundary_verified"] is True
+    assert readback["source_sha_matches_deployment"] is True
+    assert readback["active_blockers"] == []
+    assert readback["secret_values_read"] is False
+
+
+def test_executor_trust_boundary_rejects_stale_or_incomplete_receipt() -> None:
+    stale = build_executor_trust_boundary_readback(
+        _verified_receipt("old-source"),
+        deployed_source_sha="new-source",
+    )
+    assert stale["production_boundary_verified"] is False
+    assert "verified_source_sha_mismatch" in stale["active_blockers"]
+
+    incomplete_receipt = _verified_receipt()
+    incomplete_receipt["worker_ssh_mount_absent"] = "false"
+    incomplete = build_executor_trust_boundary_readback(
+        incomplete_receipt,
+        deployed_source_sha="abc123",
+    )
+    assert incomplete["production_boundary_verified"] is False
+    assert "worker_ssh_mount_absent_not_verified" in incomplete["active_blockers"]

apps/api/tests/test_runtime_bootstrap_guards.py

@@ -80,6 +80,7 @@ def test_get_engine_uses_database_pool_budget(monkeypatch):
     monkeypatch.setattr(db_base.settings, "DATABASE_POOL_SIZE", 1)
     monkeypatch.setattr(db_base.settings, "DATABASE_MAX_OVERFLOW", 0)
     monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 5.0)
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", False)
     monkeypatch.setattr(db_base, "create_async_engine", fake_create_async_engine)
 
     assert db_base.get_engine() is fake_engine
@@ -89,6 +90,36 @@ def test_get_engine_uses_database_pool_budget(monkeypatch):
     assert captured["pool_timeout"] == 5.0
 
 
+def test_get_engine_uses_null_pool_for_bounded_background_processes(monkeypatch):
+    from sqlalchemy.pool import NullPool
+
+    from src.db import base as db_base
+
+    captured: dict[str, object] = {}
+    fake_engine = object()
+
+    def fake_create_async_engine(database_url: str, **kwargs: object) -> object:
+        captured["database_url"] = database_url
+        captured.update(kwargs)
+        return fake_engine
+
+    monkeypatch.setattr(db_base, "_engine", None)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(
+        db_base.settings,
+        "DATABASE_URL",
+        "postgresql+asyncpg://u:p@localhost/db",
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
+    monkeypatch.setattr(db_base, "create_async_engine", fake_create_async_engine)
+
+    assert db_base.get_engine() is fake_engine
+    assert captured["poolclass"] is NullPool
+    assert "pool_size" not in captured
+    assert "max_overflow" not in captured
+    assert "pool_timeout" not in captured
+
+
 @pytest.mark.asyncio
 async def test_init_db_serializes_bootstrap_ddl_with_advisory_lock(monkeypatch):
     from src.db import base as db_base

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -5,39 +5,68 @@ from pathlib import Path
 
 import yaml
 
-from src.workers import signal_worker
+from src.workers import ansible_executor_broker, signal_worker
 
 
-def test_signal_worker_owns_ansible_executor_loop() -> None:
+def test_signal_worker_only_produces_ansible_candidates() -> None:
     source = inspect.getsource(signal_worker._main)
 
     assert "run_awooop_ansible_candidate_backfill_loop" in source
     assert "signal_worker_ansible_candidate_backfill_loop_started" in source
     assert "ansible_candidate_backfill_task.cancel()" in source
     assert "await ansible_candidate_backfill_task" in source
+    assert "run_awooop_ansible_check_mode_loop" not in source
+    assert "signal_worker_ansible_check_mode_loop_started" not in source
+
+
+def test_dedicated_broker_owns_ansible_executor_loop() -> None:
+    source = inspect.getsource(ansible_executor_broker._main)
+
     assert "run_awooop_ansible_check_mode_loop" in source
-    assert "asyncio.create_task" in source
-    assert "signal_worker_ansible_check_mode_loop_started" in source
-    assert "ansible_check_mode_task.cancel()" in source
-    assert "await ansible_check_mode_task" in source
+    assert "ansible_execution_broker_transport_not_mounted" in source
+    assert "asyncio.FIRST_COMPLETED" in source
+    assert "ansible_execution_broker_loop_exited_unexpectedly" in source
+    assert "execution_task.cancel()" in source
+    assert "await task" in source
 
 
-def test_worker_manifest_mounts_existing_ssh_mcp_transport() -> None:
+def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None:
     repo_root = Path(__file__).resolve().parents[3]
     manifest = (repo_root / "k8s/awoooi-prod/08-deployment-worker.yaml").read_text()
 
     assert 'command: ["python", "-m", "src.workers.signal_worker"]' in manifest
-    assert "serviceAccountName: awoooi-executor" in manifest
-    assert "fsGroup: 1000" in manifest
+    assert "serviceAccountName: awoooi-worker" in manifest
     assert "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER" in manifest
     assert "AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS" in manifest
     assert "ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER" in manifest
-    assert "ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY" in manifest
-    assert "AWOOOP_ANSIBLE_CHECK_MODE_STARTUP_SLEEP_SECONDS" in manifest
+    assert 'value: "false"' in manifest
+    assert "mountPath: /run/secrets/ssh_mcp_key" not in manifest
+    assert "mountPath: /etc/ssh-mcp/known_hosts" not in manifest
+    assert "secretName: ssh-mcp-key" not in manifest
+
+
+def test_execution_broker_is_only_workload_with_ssh_transport() -> None:
+    repo_root = Path(__file__).resolve().parents[3]
+    manifest = (
+        repo_root
+        / "k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml"
+    ).read_text()
+
+    assert "serviceAccountName: awoooi-executor-broker" in manifest
+    assert "automountServiceAccountToken: false" in manifest
+    assert "src.workers.ansible_executor_broker" in manifest
+    assert "DATABASE_NULL_POOL" in manifest
     assert "mountPath: /run/secrets/ssh_mcp_key" in manifest
     assert "mountPath: /etc/ssh-mcp/known_hosts" in manifest
     assert "secretName: ssh-mcp-key" in manifest
-    assert "optional: true" in manifest
+    deployment = yaml.safe_load(manifest)
+    ssh_volume = next(
+        volume
+        for volume in deployment["spec"]["template"]["spec"]["volumes"]
+        if volume["name"] == "ssh-mcp-key"
+    )
+    assert ssh_volume["secret"]["secretName"] == "ssh-mcp-key"
+    assert ssh_volume["secret"].get("optional") is not True
 
 
 def test_api_manifest_has_read_only_identity_and_no_ssh_executor_mounts() -> None:
@@ -79,7 +108,10 @@ def test_api_reader_rbac_has_no_workload_write_verbs() -> None:
     }
     reader = by_name[("ClusterRole", "awoooi-api-reader-role")]
     binding = by_name[("ClusterRoleBinding", "awoooi-api-reader-binding")]
+    worker_binding = by_name[("ClusterRoleBinding", "awoooi-worker-reader-binding")]
     api_account = by_name[("ServiceAccount", "awoooi-api")]
+    worker_account = by_name[("ServiceAccount", "awoooi-worker")]
+    broker_account = by_name[("ServiceAccount", "awoooi-executor-broker")]
 
     assert api_account["metadata"]["namespace"] == "awoooi-prod"
     assert binding["subjects"] == [
@@ -90,10 +122,15 @@ def test_api_reader_rbac_has_no_workload_write_verbs() -> None:
         }
     ]
     assert binding["roleRef"]["name"] == "awoooi-api-reader-role"
+    assert worker_account["metadata"]["namespace"] == "awoooi-prod"
+    assert broker_account["metadata"]["namespace"] == "awoooi-prod"
+    assert worker_binding["roleRef"]["name"] == "awoooi-api-reader-role"
     assert all(
         set(rule["verbs"]) <= {"get", "list", "watch"}
         for rule in reader["rules"]
     )
+    assert ("ClusterRole", "awoooi-executor-role") not in by_name
+    assert ("ClusterRoleBinding", "awoooi-executor-binding") not in by_name
 
 
 def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None:
@@ -110,10 +147,16 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None:
     assert "argocd.argoproj.io/sync-options" not in deployment["metadata"].get(
         "annotations", {}
     )
-    assert "AIA-P0-011 API/executor production boundary verified" in workflow
+    assert "awoooi-ansible-executor-broker" in workflow
+    assert "AIA-P0-011 API/worker/broker production boundary verified" in workflow
     assert "patch deployments.apps --all-namespaces" in workflow
     assert "delete pods --all-namespaces" in workflow
     assert "repair-ssh-key|repair-known-hosts|ssh-mcp-key" in workflow
+    assert "legacy cluster-wide executor binding still exists" in workflow
+    assert "awoooi-executor-boundary-verification" in workflow
+    assert "awoooi_executor_boundary_verification_v1" in workflow
+    assert "executor_boundary_public_readback=verified_ready" in workflow
+    assert "verified_source_sha" in workflow
     assert "github.com/kubernetes-sigs/kustomize" not in workflow
     assert "git checkout --force -B main FETCH_HEAD" in workflow