fix(security): split API from executor identity

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -2458,14 +2458,6 @@ def test_runtime_execution_loop_uses_terminal_apply_without_hiding_inflight_appl
     ledger = runtime_control_module._autonomous_execution_loop_ledger(
         project_id="awoooi",
         operation_latest_rows=[
-            {
-                "op_id": "pending-apply",
-                "operation_type": "ansible_apply_executed",
-                "status": "pending",
-                "automation_run_id": "pending-run",
-                "incident_id": "INC-PENDING",
-                "created_at": "2026-07-10T15:00:00Z",
-            },
             {
                 "op_id": "terminal-apply",
                 "operation_type": "ansible_apply_executed",
@@ -2474,6 +2466,14 @@ def test_runtime_execution_loop_uses_terminal_apply_without_hiding_inflight_appl
                 "incident_id": "INC-TERMINAL",
                 "created_at": "2026-07-10T14:59:00Z",
             },
+            {
+                "op_id": "pending-apply",
+                "operation_type": "ansible_apply_executed",
+                "status": "pending",
+                "automation_run_id": "pending-run",
+                "incident_id": "INC-PENDING",
+                "created_at": "2026-07-10T15:00:00Z",
+            },
         ],
         verifier_latest_rows=[],
         km_latest_rows=[],

apps/api/tests/test_awooop_truth_chain_service.py

@@ -1970,6 +1970,16 @@ def test_ansible_subprocess_is_terminated_when_worker_task_is_cancelled() -> Non
     assert "await process.communicate()" in source
 
 
+def test_controlled_apply_shutdown_marks_pending_row_before_cancelling() -> None:
+    source = inspect.getsource(run_controlled_apply_for_claim)
+
+    assert "except asyncio.CancelledError" in source
+    assert "ansible_controlled_apply_interrupted_by_worker_shutdown" in source
+    assert '"status": status' in source
+    assert '"op_id": apply_op_id' in source
+    assert "ansible_controlled_apply_interrupted" in source
+
+
 def test_ansible_post_apply_verifier_helpers_are_deterministic() -> None:
     assert _post_apply_km_path_type("03ca6836-1b76-4da2-8e3e-6d3b6df9254a") == (
         "ansible_apply_receipt:03ca6836"

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 import inspect
 from pathlib import Path
 
+import yaml
+
 from src.workers import signal_worker
 
 
@@ -38,6 +40,62 @@ def test_worker_manifest_mounts_existing_ssh_mcp_transport() -> None:
     assert "optional: true" in manifest
 
 
+def test_api_manifest_has_read_only_identity_and_no_ssh_executor_mounts() -> None:
+    repo_root = Path(__file__).resolve().parents[3]
+    deployment = next(
+        document
+        for document in yaml.safe_load_all(
+            (repo_root / "k8s/awoooi-prod/06-deployment-api.yaml").read_text()
+        )
+        if document and document.get("kind") == "Deployment"
+    )
+    pod_spec = deployment["spec"]["template"]["spec"]
+    container = pod_spec["containers"][0]
+    env = {item["name"]: item.get("value") for item in container["env"]}
+    mount_names = {item["name"] for item in container.get("volumeMounts", [])}
+    volume_names = {item["name"] for item in pod_spec.get("volumes", [])}
+
+    assert pod_spec["serviceAccountName"] == "awoooi-api"
+    assert env["SSH_MCP_ENABLED"] == "false"
+    assert env["ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER"] == "false"
+    assert env["ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER"] == "false"
+    assert {"repair-ssh-key", "repair-known-hosts", "ssh-mcp-key"}.isdisjoint(
+        mount_names | volume_names
+    )
+
+
+def test_api_reader_rbac_has_no_workload_write_verbs() -> None:
+    repo_root = Path(__file__).resolve().parents[3]
+    documents = [
+        document
+        for document in yaml.safe_load_all(
+            (repo_root / "k8s/awoooi-prod/07-rbac.yaml").read_text()
+        )
+        if document
+    ]
+    by_name = {
+        (document["kind"], document["metadata"]["name"]): document
+        for document in documents
+    }
+    reader = by_name[("ClusterRole", "awoooi-api-reader-role")]
+    binding = by_name[("ClusterRoleBinding", "awoooi-api-reader-binding")]
+    api_account = by_name[("ServiceAccount", "awoooi-api")]
+
+    assert api_account["metadata"]["namespace"] == "awoooi-prod"
+    assert binding["subjects"] == [
+        {
+            "kind": "ServiceAccount",
+            "name": "awoooi-api",
+            "namespace": "awoooi-prod",
+        }
+    ]
+    assert binding["roleRef"]["name"] == "awoooi-api-reader-role"
+    assert all(
+        set(rule["verbs"]) <= {"get", "list", "watch"}
+        for rule in reader["rules"]
+    )
+
+
 def test_api_log_does_not_claim_skipped_executor_was_scheduled() -> None:
     repo_root = Path(__file__).resolve().parents[3]
     main_source = (repo_root / "apps/api/src/main.py").read_text()