Ehole/verification

devops_bench/cli.py

@@ -24,7 +24,7 @@
 import argparse
 import sys
 from dataclasses import replace
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from devops_bench.core import ConfigError
 
@@ -141,6 +141,65 @@ def args_to_config(args: argparse.Namespace) -> BenchmarkConfig:
     )
 
 
+def _fmt_reason(reason: Any, limit: int = 160) -> str:
+    """Flatten a possibly-multiline reason to a single truncated line."""
+    if not reason:
+        return ""
+    flat = " ".join(str(reason).split())
+    return flat if len(flat) <= limit else flat[: limit - 1] + "…"
+
+
+def print_run_summary(records: list[dict[str, Any]]) -> None:
+    """Print a per-task deterministic + judge scoreboard to stdout.
+
+    Consolidates the deterministic verification rollup (which otherwise only
+    reaches ``results.json``) with the LLM-judge scores, so a single run's
+    outcome is legible without opening the JSON.
+    """
+    if not records:
+        return
+    print("\n================ run summary ================")
+    for r in records:
+        print(f"\n{r.get('name') or '<unknown>'}  [{r.get('status') or ''}]")
+
+        rollup = r.get("verification_rollup")
+        if rollup is not None:
+            c = rollup.get("c")
+            rec_v = rollup.get("rec_v")
+            c_s = "n/a" if c is None else f"{c:.2f}"
+            rec_s = "n/a" if rec_v is None else f"{rec_v:.2f}"
+            print(f"  deterministic:  c={c_s}  rec_v={rec_s}  cat_v={rollup.get('cat_v')}")
+
+        entries = r.get("verification_entries_report") or []
+        objectives = [e for e in entries if e.get("role") == "objective"]
+        safeguards = [e for e in entries if e.get("role") == "safeguard"]
+        if objectives:
+            print("  objectives (feed c):")
+            for e in objectives:
+                mark = "PASS" if e.get("success") else "FAIL"
+                tail = "" if e.get("success") else f"  {_fmt_reason(e.get('reason'))}"
+                print(f"    [{mark}] {e.get('name')} (w{e.get('weight')}){tail}")
+        if safeguards:
+            print("  safeguards:")
+            for e in safeguards:
+                held = e.get("success")
+                mark = "HELD  " if held else "BREACH"
+                tail = "" if held else f"  {_fmt_reason(e.get('reason'))}"
+                print(f"    [{mark}] {e.get('name')} ({e.get('severity')}){tail}")
+
+        scores = r.get("scores") or {}
+        if scores:
+            print("  judge (GEval, non-gating):")
+            for metric, val in scores.items():
+                if isinstance(val, dict):
+                    score = val.get("score")
+                    score_s = "n/a" if score is None else f"{score:.2f}"
+                    mark = "PASS" if val.get("success") else "FAIL"
+                    print(f"    [{mark}] {metric} = {score_s}  {_fmt_reason(val.get('reason'))}")
+                else:
+                    print(f"    {metric} = {val}")
+
+
 def main(argv: list[str] | None = None) -> int:
     """Console entry point: parse args, run the benchmark, return an exit code.
 
@@ -163,6 +222,7 @@ def main(argv: list[str] | None = None) -> int:
         print(f"error: {exc}", file=sys.stderr)
         return 2
 
+    print_run_summary(result.results)
     failed = sum(1 for r in result.results if r.get("status") == "failed")
     print(f"ran {len(result.results)} task(s), {failed} failed; results: {result.results_path}")
     return 1 if failed else 0

devops_bench/evalharness/default.py

@@ -53,7 +53,13 @@
     pick_free_port,
 )
 from devops_bench.tasks import Task
-from devops_bench.verification import VerificationSpec
+from devops_bench.verification import (
+    EvaluatedEntry,
+    VerificationResult,
+    VerificationSpec,
+    VerifierAgent,
+    rollup,
+)
 
 __all__ = ["DefaultEvalHarness"]
 
@@ -510,6 +516,84 @@ def _build_verification_mapping(
                 errors.append({"name": str(name), "reason": str(exc)})
         return mapping, errors
 
+    def _run_verification_entries(
+        self,
+        task: Task,
+        cluster_name: str,
+        target_dep: str,
+        ns: str,
+    ) -> tuple[list[dict[str, Any]], dict[str, float | int | None] | None]:
+        """Evaluate the task's objective/safeguard entries after the agent runs.
+
+        Unconditional (no chaos dependency): each entry's ``check`` is
+        placeholder-substituted, parsed through the verifier registry, and
+        evaluated under the entry's mode via a fresh :class:`VerifierAgent`. The
+        per-entry results are rolled up into ``c`` / ``rec_v`` / ``cat_v``.
+
+        A per-entry failure (bad spec, evaluation error) is recorded as a failed
+        entry rather than raising, so a verification problem never turns an
+        otherwise-successful agent run into a failed record.
+
+        Args:
+            task: The task under evaluation.
+            cluster_name: Active cluster name for placeholder substitution.
+            target_dep: Resolved target deployment name.
+            ns: Resolved namespace.
+
+        Returns:
+            A pair ``(report, rollup)``: the per-entry report list and the rollup
+            mapping (or ``None`` when no entries are declared or the rollup is
+            undefined for lack of an objective).
+        """
+        entries = task.verification_entries or []
+        if not entries:
+            return [], None
+
+        agent = VerifierAgent()
+        evaluated: list[EvaluatedEntry] = []
+        report: list[dict[str, Any]] = []
+        for entry in entries:
+            resolved = self._resolve_spec_placeholders(entry.check, cluster_name, target_dep, ns)
+            try:
+                node = VerificationSpec(resolved).root
+                ev = agent.run_entry(entry, node, timeout_sec=VERIFICATION_TIMEOUT_SEC)
+            except Exception as exc:  # noqa: BLE001 - a bad entry must not fail the run
+                _log.warning("verification entry %r failed to evaluate: %s", entry.name, exc)
+                ev = EvaluatedEntry(
+                    name=entry.name,
+                    role=entry.role,
+                    severity=entry.severity,
+                    weight=entry.weight,
+                    result=VerificationResult(
+                        success=False, elapsed_time=0.0, reason=f"entry error: {exc}"
+                    ),
+                )
+            evaluated.append(ev)
+            report.append(
+                {
+                    "name": ev.name,
+                    "role": ev.role,
+                    "severity": ev.severity,
+                    "weight": ev.weight,
+                    "mode": entry.mode,
+                    "success": ev.result.success,
+                    "reason": ev.result.reason,
+                    "result": ev.result.model_dump(),
+                }
+            )
+
+        try:
+            scores = rollup(evaluated)
+            rollup_dict: dict[str, float | int | None] | None = {
+                "c": scores.c,
+                "rec_v": scores.rec_v,
+                "cat_v": scores.cat_v,
+            }
+        except ValueError as exc:
+            _log.warning("verification rollup skipped: %s", exc)
+            rollup_dict = None
+        return report, rollup_dict
+
     # -- scenario (background chaos) --------------------------------------
 
     def start_scenario(
@@ -743,6 +827,10 @@ def _run_one(self, task: Task, run_dir: Path) -> dict[str, Any]:
 
             chaos_report, perf_report = self._drain_scenario(scenario_manager, scenario_thread)
 
+            verification_entries_report, verification_rollup = self._run_verification_entries(
+                task, active_cluster_name, target_dep, ns
+            )
+
             result = self._build_success_record(
                 task=task,
                 prompt=prompt,
@@ -751,6 +839,8 @@ def _run_one(self, task: Task, run_dir: Path) -> dict[str, Any]:
                 chaos_report=chaos_report,
                 perf_report=perf_report,
                 verification_parse_errors=verification_parse_errors,
+                verification_entries_report=verification_entries_report,
+                verification_rollup=verification_rollup,
             )
             _log.info("agent response for %s:\n%s", task.name, result["output"])
         except Exception as exc:  # noqa: BLE001 - surface every task failure
@@ -800,6 +890,8 @@ def _run_one(self, task: Task, run_dir: Path) -> dict[str, Any]:
             "documentation",
             "capabilities_granted",
             "verification_parse_errors",
+            "verification_entries_report",
+            "verification_rollup",
             "generation_only",
             "validated",
         }
@@ -815,6 +907,8 @@ def _build_success_record(
         chaos_report: dict[str, Any],
         perf_report: dict[str, Any],
         verification_parse_errors: list[dict[str, str]] | None = None,
+        verification_entries_report: list[dict[str, Any]] | None = None,
+        verification_rollup: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Shape a typed :class:`AgentResult` + reports into the on-disk schema.
 
@@ -863,6 +957,8 @@ def _build_success_record(
                 "chaos_report": chaos_report,
                 "perf_report": perf_report,
                 "verification_parse_errors": list(verification_parse_errors or []),
+                "verification_entries_report": list(verification_entries_report or []),
+                "verification_rollup": verification_rollup,
             }
         )
         return record
@@ -953,6 +1049,8 @@ def _empty_record(self, task: Task) -> dict[str, Any]:
                 "skills": list(self._granted_skill_paths),
             },
             "verification_parse_errors": [],
+            "verification_entries_report": [],
+            "verification_rollup": None,
             # Generation-only (``deployer: noop``) tasks have no cluster, so the
             # OutcomeValidity judge must not penalize them for "not applying".
             "generation_only": (task.infrastructure or {}).get("deployer") == "noop",

devops_bench/k8s/__init__.py

@@ -20,6 +20,7 @@
     get_resource,
     port_forward,
     rollout_status,
+    run_pod,
     wait,
 )
 
@@ -29,5 +30,6 @@
     "poll_until",
     "port_forward",
     "rollout_status",
+    "run_pod",
     "wait",
 ]

devops_bench/k8s/kubectl.py

@@ -32,6 +32,7 @@
     "get_resource",
     "port_forward",
     "rollout_status",
+    "run_pod",
     "wait",
 ]
 
@@ -223,6 +224,58 @@ def rollout_status(
     return _run_kubectl(argv, kubeconfig)
 
 
+def run_pod(
+    name: str,
+    image: str,
+    command: list[str],
+    *,
+    namespace: str | None = None,
+    kubeconfig: KubeconfigSource = None,
+    timeout: float | None = None,
+    env: dict[str, str] | None = None,
+) -> str:
+    """Run a one-shot ephemeral pod and return its captured stdout.
+
+    Launches the pod with ``--rm -i --restart=Never`` so the pod is auto-deleted
+    after completion and ``kubectl`` attaches stdin, which is required to capture
+    the container's output.
+
+    Args:
+        name: Pod name.
+        image: Container image to run.
+        command: Command and arguments passed after ``--`` to the container.
+        namespace: Optional namespace (``-n``).
+        kubeconfig: Kubeconfig path or context-like object.
+        timeout: Optional timeout in seconds forwarded to ``core.subprocess.run``.
+        env: Optional env vars injected into the container via ``--env=K=V``.
+
+    Returns:
+        The pod's captured stdout.
+
+    Raises:
+        SubprocessError: If kubectl exits non-zero or times out.
+    """
+    env_args = [f"--env={k}={v}" for k, v in (env or {}).items()]
+    argv = [
+        "kubectl",
+        "run",
+        name,
+        "--rm",
+        "-i",
+        "--restart=Never",
+        f"--image={image}",
+        *env_args,
+        *_namespace_args(namespace),
+        "--",
+        *command,
+    ]
+    extra_kwargs: dict[str, Any] = {}
+    if timeout is not None:
+        extra_kwargs["timeout"] = timeout
+    completed = _run_kubectl(argv, kubeconfig, **extra_kwargs)
+    return completed.stdout
+
+
 @contextlib.contextmanager
 def port_forward(
     target: str,

devops_bench/tasks/__init__.py

@@ -15,12 +15,13 @@
 """Task contracts: the typed schema and loaders for benchmark tasks."""
 
 from devops_bench.tasks.loader import FileSystemTaskLoader, TaskLoader
-from devops_bench.tasks.schema import Constraint, DocumentationEntry, Task
+from devops_bench.tasks.schema import Constraint, DocumentationEntry, Task, VerificationEntry
 
 __all__ = [
     "Task",
     "Constraint",
     "DocumentationEntry",
+    "VerificationEntry",
     "TaskLoader",
     "FileSystemTaskLoader",
 ]