fix(alerts): guard approval actions and wire playbook learning

apps/api/tests/test_alert_approval_guard.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import pytest
+
+from src.services.alert_approval_guard import guard_alert_approval_action
+from src.services.resource_resolver import ResolveResult, set_resource_resolver
+from src.utils.k8s_naming import ResourceType
+
+
+class StubResolver:
+    def __init__(self, *, success: bool, candidates: list[str] | None = None) -> None:
+        self.success = success
+        self.candidates = candidates or []
+        self.calls: list[dict[str, str]] = []
+
+    async def resolve(
+        self,
+        raw_resource: str,
+        namespace: str = "awoooi-prod",
+        resource_kind: str = "deployment",
+    ) -> ResolveResult:
+        self.calls.append(
+            {
+                "raw_resource": raw_resource,
+                "namespace": namespace,
+                "resource_kind": resource_kind,
+            }
+        )
+        return ResolveResult(
+            success=self.success,
+            resource_name=raw_resource if self.success else None,
+            namespace=namespace,
+            resource_type=ResourceType(resource_kind)
+            if resource_kind in ResourceType._value2member_map_
+            else ResourceType.UNKNOWN,
+            confidence=1.0 if self.success else 0.0,
+            candidates=self.candidates,
+            original_input=raw_resource,
+        )
+
+
+@pytest.mark.asyncio
+async def test_blocks_llm_kubectl_default_namespace_for_prod_alert() -> None:
+    result = await guard_alert_approval_action(
+        action="kubectl logs deployment/sentry-self-hosted-snuba-metrics-consumer-1 -n default",
+        alert_namespace="awoooi-prod",
+        alertname="SentryRpsZero",
+        alert_category="infrastructure",
+    )
+
+    assert result.blocked is True
+    assert result.action.startswith("NO_ACTION - INVALID_TARGET")
+    assert result.reason == "namespace_not_allowed:default"
+
+
+@pytest.mark.asyncio
+async def test_blocks_hallucinated_mutating_k8s_resource() -> None:
+    resolver = StubResolver(success=False, candidates=["awoooi-api"])
+    set_resource_resolver(resolver)
+    try:
+        result = await guard_alert_approval_action(
+            action="kubectl scale deployment flywheelexecutionratemissing --replicas=5 -n awoooi-prod",
+            alert_namespace="awoooi-prod",
+            alertname="FlywheelExecutionRateMissing",
+            alert_category="infrastructure",
+        )
+    finally:
+        set_resource_resolver(None)
+
+    assert result.blocked is True
+    assert result.reason == "k8s_resource_not_found:deployment/flywheelexecutionratemissing"
+    assert result.metadata["candidates"] == ["awoooi-api"]
+    assert resolver.calls == [
+        {
+            "raw_resource": "flywheelexecutionratemissing",
+            "namespace": "awoooi-prod",
+            "resource_kind": "deployment",
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_allows_sane_readonly_cluster_inventory() -> None:
+    result = await guard_alert_approval_action(
+        action="kubectl get pods -n awoooi-prod",
+        alert_namespace="awoooi-prod",
+        alertname="ColdStartCheck",
+        alert_category="infrastructure",
+    )
+
+    assert result.blocked is False
+    assert result.action == "kubectl get pods -n awoooi-prod"