feat: add deterministic verifiers (resource_exists, resource_field) to replace LLM judge for fix-config and deploy-config

pkg/agents/verifier/resource_exists.py

@@ -0,0 +1,61 @@
+import subprocess
+import time
+from typing import Literal, Optional
+from pkg.agents.verifier.base import BaseVerifier, VerificationResult
+
+
+class ResourceExistsVerifier(BaseVerifier):
+    """Verifies that a named Kubernetes resource exists in the cluster.
+
+    Generic primitive over `kubectl get <kind> <name>`. Because it reads live
+    cluster state, a successful check also proves the resource was actually
+    applied (not merely printed by the agent).
+    """
+
+    type: Literal["resource_exists"] = "resource_exists"
+    kind: str
+    name: str
+    namespace: Optional[str] = None
+
+    def verify(self, timeout_sec: int) -> VerificationResult:
+        start_time = time.time()
+        delay = 1
+        max_delay = 10
+
+        while True:
+            success, details = self._check_exists()
+            if success:
+                return VerificationResult(
+                    success=True,
+                    elapsed_time=time.time() - start_time,
+                    reason=details.get("reason"),
+                    details=details,
+                )
+            if time.time() - start_time >= timeout_sec:
+                return VerificationResult(
+                    success=False,
+                    elapsed_time=time.time() - start_time,
+                    reason=details.get("reason"),
+                    details=details,
+                )
+            time.sleep(delay)
+            delay = min(delay * 2, max_delay)
+
+    def _check_exists(self) -> tuple[bool, dict]:
+        cmd = ["kubectl", "get", self.kind, self.name]
+        if self.namespace:
+            cmd.extend(["-n", self.namespace])
+        try:
+            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
+            return True, {
+                "reason": f"{self.kind}/{self.name} exists",
+                "output": result.stdout.strip(),
+            }
+        except subprocess.CalledProcessError as e:
+            return False, {
+                "reason": f"{self.kind}/{self.name} not found: {e.stderr.strip()}"
+            }
+        except FileNotFoundError:
+            return False, {
+                "reason": "kubectl command not found in PATH. Please ensure kubectl is installed."
+            }

pkg/agents/verifier/resource_field.py

@@ -0,0 +1,86 @@
+import subprocess
+import time
+from typing import Literal, Optional
+from pkg.agents.verifier.base import BaseVerifier, VerificationResult
+
+
+class ResourceFieldVerifier(BaseVerifier):
+    """Verifies that a specific field on a Kubernetes resource equals an
+    expected value.
+
+    Generic primitive over `kubectl get <kind> <name> -o jsonpath=<json_path>`.
+    kubectl performs the extraction, so no extra Python dependency is required.
+    Reading the live field value also proves the resource was actually applied
+    with the correct configuration.
+
+    Example (container env var):
+        json_path = "{.spec.template.spec.containers[?(@.name=='frontend')]"
+                    ".env[?(@.name=='USE_GEMINI_API')].value}"
+        expected  = "false"
+    """
+
+    type: Literal["resource_field"] = "resource_field"
+    kind: str
+    name: str
+    json_path: str
+    expected: str
+    namespace: Optional[str] = None
+
+    def verify(self, timeout_sec: int) -> VerificationResult:
+        start_time = time.time()
+        delay = 1
+        max_delay = 10
+
+        while True:
+            success, details = self._check_field()
+            if success:
+                return VerificationResult(
+                    success=True,
+                    elapsed_time=time.time() - start_time,
+                    reason=details.get("reason"),
+                    details=details,
+                )
+            if time.time() - start_time >= timeout_sec:
+                return VerificationResult(
+                    success=False,
+                    elapsed_time=time.time() - start_time,
+                    reason=details.get("reason"),
+                    details=details,
+                )
+            time.sleep(delay)
+            delay = min(delay * 2, max_delay)
+
+    def _check_field(self) -> (bool, dict):
+        cmd = [
+            "kubectl",
+            "get",
+            self.kind,
+            self.name,
+            "-o",
+            f"jsonpath={self.json_path}",
+        ]
+        if self.namespace:
+            cmd.extend(["-n", self.namespace])
+        try:
+            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
+        except subprocess.CalledProcessError as e:
+            return False, {
+                "reason": f"Failed to read {self.kind}/{self.name}: {e.stderr.strip()}"
+            }
+
+        actual = result.stdout.strip()
+        success = actual == self.expected
+        reason = (
+            f"{self.kind}/{self.name} field matched expected value '{self.expected}'"
+            if success
+            else (
+                f"{self.kind}/{self.name} field mismatch: expected "
+                f"'{self.expected}', got '{actual}'"
+            )
+        )
+        return success, {
+            "reason": reason,
+            "json_path": self.json_path,
+            "expected": self.expected,
+            "actual": actual,
+        }

pkg/agents/verifier/spec.py

@@ -2,9 +2,16 @@
 from pydantic import RootModel
 from pkg.agents.verifier.pod_healthy import PodHealthyVerifier
 from pkg.agents.verifier.scaling_complete import ScalingCompleteVerifier
+from pkg.agents.verifier.resource_exists import ResourceExistsVerifier
+from pkg.agents.verifier.resource_field import ResourceFieldVerifier
 
 # SingleVerificationSpec is a discriminated union of all supported checker types
-SingleVerificationSpec = Union[PodHealthyVerifier, ScalingCompleteVerifier]
+SingleVerificationSpec = Union[
+    PodHealthyVerifier,
+    ScalingCompleteVerifier,
+    ResourceExistsVerifier,
+    ResourceFieldVerifier,
+]
 
 # Top-level VerificationSpec which can parse a dict, a list, or a single checker spec.
 class VerificationSpec(RootModel[Union[Dict[str, SingleVerificationSpec], List[SingleVerificationSpec], SingleVerificationSpec]]):

pkg/agents/verifier/test_verifier.py

@@ -6,6 +6,8 @@
 from pkg.agents.verifier.verifier import VerifierAgent
 from pkg.agents.verifier.pod_healthy import PodHealthyVerifier
 from pkg.agents.verifier.scaling_complete import ScalingCompleteVerifier
+from pkg.agents.verifier.resource_exists import ResourceExistsVerifier
+from pkg.agents.verifier.resource_field import ResourceFieldVerifier
 
 class TestVerifierAgent(unittest.TestCase):
 
@@ -110,6 +112,153 @@ def test_scaling_complete_verifier_verify_polling_success(self, mock_sleep, mock
         self.assertEqual(result.reason, "Scaling complete: done")
         self.assertEqual(mock_sleep.call_count, 1)
 
+    # --- ResourceExistsVerifier ---
+
+    @patch("subprocess.run")
+    def test_resource_exists_verifier_success(self, mock_run):
+        mock_run.return_value = MagicMock(
+            stdout="deployment.apps/my-dep", returncode=0
+        )
+
+        r_verifier = ResourceExistsVerifier(
+            kind="deployment", name="my-dep", namespace="default"
+        )
+        result = r_verifier.verify(timeout_sec=5)
+
+        self.assertTrue(result.success)
+        self.assertIn("exists", result.reason)
+
+    @patch("time.sleep")
+    @patch("subprocess.run")
+    def test_resource_exists_verifier_not_found(self, mock_run, mock_sleep):
+        mock_run.side_effect = subprocess.CalledProcessError(
+            1, "kubectl get", stderr='Error from server (NotFound)'
+        )
+
+        r_verifier = ResourceExistsVerifier(
+            kind="deployment", name="does-not-exist", namespace="default"
+        )
+        result = r_verifier.verify(timeout_sec=1)
+
+        self.assertFalse(result.success)
+        self.assertIn("not found", result.reason)
+
+    @patch("subprocess.run")
+    def test_resource_exists_verifier_omits_namespace_when_none(self, mock_run):
+        mock_run.return_value = MagicMock(stdout="namespace/foo", returncode=0)
+
+        r_verifier = ResourceExistsVerifier(kind="namespace", name="foo")
+        result = r_verifier.verify(timeout_sec=5)
+
+        self.assertTrue(result.success)
+        called_cmd = mock_run.call_args[0][0]
+        self.assertNotIn("-n", called_cmd)
+
+    # --- ResourceFieldVerifier ---
+
+    @patch("subprocess.run")
+    def test_resource_field_verifier_match(self, mock_run):
+        mock_run.return_value = MagicMock(stdout="false", returncode=0)
+
+        f_verifier = ResourceFieldVerifier(
+            kind="deployment",
+            name="hypercomputer-d1-frontend",
+            namespace="default",
+            json_path="{.spec.template.spec.containers[?(@.name=='frontend')].env[?(@.name=='USE_GEMINI_API')].value}",
+            expected="false",
+        )
+        result = f_verifier.verify(timeout_sec=5)
+
+        self.assertTrue(result.success)
+        self.assertIn("matched expected value", result.reason)
+        self.assertEqual(result.details["actual"], "false")
+
+    @patch("time.sleep")
+    @patch("subprocess.run")
+    def test_resource_field_verifier_mismatch(self, mock_run, mock_sleep):
+        mock_run.return_value = MagicMock(stdout="true", returncode=0)
+
+        f_verifier = ResourceFieldVerifier(
+            kind="deployment",
+            name="hypercomputer-d1-frontend",
+            namespace="default",
+            json_path="{.spec.template.spec.containers[?(@.name=='frontend')].env[?(@.name=='USE_GEMINI_API')].value}",
+            expected="false",
+        )
+        result = f_verifier.verify(timeout_sec=1)
+
+        self.assertFalse(result.success)
+        self.assertIn("mismatch", result.reason)
+        self.assertEqual(result.details["expected"], "false")
+        self.assertEqual(result.details["actual"], "true")
+
+    @patch("time.sleep")
+    @patch("subprocess.run")
+    def test_resource_field_verifier_get_fails(self, mock_run, mock_sleep):
+        mock_run.side_effect = subprocess.CalledProcessError(
+            1, "kubectl get", stderr="Error from server (NotFound)"
+        )
+
+        f_verifier = ResourceFieldVerifier(
+            kind="deployment",
+            name="missing",
+            namespace="default",
+            json_path="{.spec.replicas}",
+            expected="1",
+        )
+        result = f_verifier.verify(timeout_sec=1)
+
+        self.assertFalse(result.success)
+        self.assertIn("Failed to read", result.reason)
+
+    @patch("subprocess.run")
+    def test_resource_field_verifier_uses_jsonpath_output(self, mock_run):
+        mock_run.return_value = MagicMock(stdout="1", returncode=0)
+
+        f_verifier = ResourceFieldVerifier(
+            kind="hpa",
+            name="my-hpa",
+            namespace="default",
+            json_path="{.spec.minReplicas}",
+            expected="1",
+        )
+        result = f_verifier.verify(timeout_sec=5)
+
+        self.assertTrue(result.success)
+        called_cmd = mock_run.call_args[0][0]
+        self.assertIn("jsonpath={.spec.minReplicas}", called_cmd)
+
+    # --- Compound spec that mixes the new verifiers ---
+
+    @patch("subprocess.run")
+    def test_wait_for_condition_compound_new_verifiers(self, mock_run):
+        def run_side_effect(cmd, *args, **kwargs):
+            if "jsonpath" in " ".join(cmd):
+                return MagicMock(stdout="false", returncode=0)
+            return MagicMock(stdout="service/my-svc", returncode=0)
+
+        mock_run.side_effect = run_side_effect
+
+        spec = {
+            "cfg": {
+                "type": "resource_field",
+                "kind": "deployment",
+                "name": "my-dep",
+                "json_path": "{.spec.replicas}",
+                "expected": "false",
+            },
+            "svc": {
+                "type": "resource_exists",
+                "kind": "service",
+                "name": "my-svc",
+            },
+        }
+        result = self.verifier.wait_for_condition(spec, timeout_sec=30)
+
+        self.assertTrue(result.success)
+        self.assertIn("cfg succeeded", result.reason)
+        self.assertIn("svc succeeded", result.reason)
+
     @patch("subprocess.run")
     def test_wait_for_condition_compound_success(self, mock_run):
         def run_side_effect(cmd, *args, **kwargs):