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

[jegaths]

feat: add deterministic verifiers to replace LLM judge for correctness

Introduce composable, fact-based verifier primitives that grade tasks by
querying live cluster state via kubectl instead of the slow, reward-hackable
LLM judge.

New verifiers (pkg/agents/verifier/):

• resource_exists: asserts a k8s object exists (proves manifests applied)
• resource_field: asserts a field equals expected via kubectl -o jsonpath
  (no new deps; kubectl performs extraction)

Grading pipeline (pkg/evaluator/evaluate.py):

• Run verification_spec against the live cluster before teardown for all
  tasks (previously only chaos tasks consumed it)
• When a task defines verification_spec, use its result as the authoritative
  ChecklistScore and skip the LLM critical-requirements/OutcomeValidity/
  ToolInvocation checks entirely
• Add _strip_named_bundles shim so the existing chaos "named bundle" format
  works outside the chaos flow (documented, with removal path noted)

Tasks wired to the deterministic gate:

• fix-config: resource_field asserts USE_GEMINI_API == false
• deploy-config: resource_exists for deployment/service/hpa + hpa minReplicas

Register new verifiers in spec.py union and add mocked unit tests
(no cluster required). Verified end-to-end on a live GKE cluster:
deterministic pass in ~3.5s with zero LLM judge calls.

[gemini-code-assist]

Code Review

This pull request introduces two new Kubernetes verifiers, ResourceExistsVerifier and ResourceFieldVerifier, and integrates them into the evaluation pipeline to enable deterministic correctness checks instead of relying on LLM-based grading. The review feedback highlights several key areas for improvement: using time.monotonic() instead of time.time() for reliable timeout measurements, correcting invalid type hint syntax, handling FileNotFoundError when kubectl is missing from the system PATH, and updating the _strip_named_bundles helper to handle dictionary inputs to prevent validation errors.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Using time.time() for measuring elapsed time or timeouts is susceptible to system clock adjustments (e.g., NTP synchronization or manual clock changes). This can lead to unexpected behavior or infinite loops in the timeout logic.

Using time.monotonic() is the standard and robust way to measure elapsed time in Python.

        start_time = time.monotonic()
        delay = 1
        max_delay = 10

        while True:
            success, details = self._check_exists()
            if success:
                return VerificationResult(
                    success=True,
                    elapsed_time=time.monotonic() - start_time,
                    reason=details.get("reason"),
                    details=details,
                )
            if time.monotonic() - start_time >= timeout_sec:
                return VerificationResult(
                    success=False,
                    elapsed_time=time.monotonic() - start_time,
                    reason=details.get("reason"),
                    details=details,
                )
            time.sleep(delay)
            delay = min(delay * 2, max_delay)

[gemini-code-assist]

Using time.time() for measuring elapsed time or timeouts is susceptible to system clock adjustments (e.g., NTP synchronization or manual clock changes). This can lead to unexpected behavior or infinite loops in the timeout logic.

Using time.monotonic() is the standard and robust way to measure elapsed time in Python.

        start_time = time.monotonic()
        delay = 1
        max_delay = 10

        while True:
            success, details = self._check_field()
            if success:
                return VerificationResult(
                    success=True,
                    elapsed_time=time.monotonic() - start_time,
                    reason=details.get("reason"),
                    details=details,
                )
            if time.monotonic() - start_time >= timeout_sec:
                return VerificationResult(
                    success=False,
                    elapsed_time=time.monotonic() - start_time,
                    reason=details.get("reason"),
                    details=details,
                )
            time.sleep(delay)
            delay = min(delay * 2, max_delay)

[gemini-code-assist]

The type hint syntax (bool, dict) is invalid in Python. It evaluates to a tuple of type objects at runtime and is not recognized as a valid type hint by static analysis tools like Mypy.

Use tuple[bool, dict] (Python 3.9+) or Tuple[bool, dict] from the typing module instead.

Suggested change

	def _check_field(self) -> (bool, dict):
	def _check_field(self) -> tuple[bool, dict]:

[gemini-code-assist]

If kubectl is not installed or not available in the system's PATH, subprocess.run will raise a FileNotFoundError rather than a subprocess.CalledProcessError. This will cause the verifier to crash with an unhandled exception.

Catching FileNotFoundError explicitly allows the verifier to fail gracefully with a clear error message.

        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()}"
            }
        except FileNotFoundError:
            return False, {
                "reason": "kubectl command not found in PATH. Please ensure kubectl is installed."
            }

[gemini-code-assist]

The _strip_named_bundles helper function only handles cases where spec is a list. If a user provides a single named bundle as a dict (which is a valid YAML/JSON structure), the function returns it unchanged. This will subsequently cause a ValidationError in VerifierAgent.wait_for_condition because the top-level "name" string is not a valid verifier.

Updating the function to handle both list and dict inputs makes the parser much more robust.

def _strip_named_bundles(spec):
    """Converts chaos-style named verification bundles into plain verifier specs.

    A named bundle (the format used by chaos tasks, e.g. optimize-scale) looks
    like:
        [{ "name": "...", "pod_spec": {..verifier..}, "scaling_spec": {..} }]
    The "name" exists so a chaos_spec can reference the bundle by name; it is NOT
    a verifier. We drop the "name" key and keep only the verifier sub-specs.
    Plain specs (already valid SingleVerificationSpec shapes) pass through
    unchanged.

    WHY THIS SHIM EXISTS
    --------------------
    A bundle dict is not a valid VerificationSpec: it carries a top-level "name"
    string, and if that dict is handed straight to VerifierAgent.wait_for_condition
    the dict branch (see pkg/agents/verifier/verifier.py) tries to validate the
    "name" *string* as a verifier and raises a ValidationError (verified against
    the current code). To reuse the exact same task format for non-chaos
    deterministic grading, we normalize the bundle here (drop "name", keep the
    verifier sub-specs) before validation.

    Note: the existing chaos path (ScenarioManager -> wait_for_condition) passes
    the bundle through WITHOUT this normalization, so it is subject to the same
    "name"-key validation issue. The proper long-term fix is at the root (see
    below), which would cover both paths.

    HOW TO REMOVE THIS SHIM
    -----------------------
    Move the fix to the single source of truth instead: in
    VerifierAgent.wait_for_condition's dict branch, skip the reserved "name" key
    while iterating (e.g. `if key == "name": continue`). That fixes both the
    chaos path and this path at the root, after which this function can be
    deleted and callers can pass the spec through directly.
    """
    if isinstance(spec, list):
        converted = []
        for entry in spec:
            if isinstance(entry, dict) and "type" not in entry and "name" in entry:
                converted.extend(
                    v for k, v in entry.items() if k != "name" and isinstance(v, dict)
                )
            else:
                converted.append(entry)
        return converted
    elif isinstance(spec, dict) and "type" not in spec and "name" in spec:
        return {
            k: v for k, v in spec.items() if k != "name" and isinstance(v, dict)
        }
    return spec