feat(evals): agent evals harness for before-and-after comparisons

.gitignore

@@ -41,3 +41,6 @@ src/deadline/client/ui/_translation_keys.py
 # Install builder license
 license.xml
 /THIRD_PARTY_LICENSES
+
+# Agent evals run artifacts
+/evals/output/

evals/README.md

@@ -0,0 +1,96 @@
+# Agent evals
+
+Measure how well an AI agent achieves goals with Deadline Cloud tooling and
+docs — and prove, with a before-and-after comparison, that a change to the CLI,
+the docs, or any reference material actually helps.
+
+## How it works
+
+1. An **eval file** (JSON) gives an agent a goal, the tools it may use, optional
+   reference material, and a **rubric** describing what a passing answer must do.
+2. The **runner** launches an isolated headless agent (`claude -p`) per run in a
+   throwaway sandbox and captures telemetry (tool calls, turns, cost).
+3. An **LLM judge** grades the agent's final answer against the rubric.
+4. In **A/B mode**, every case runs twice — with this repo at the current ref
+   (baseline) and at `--revised-ref` (candidate) — and the summary reports the
+   paired delta plus the diff, which is the PR-ready proposed change.
+
+The material under test can be almost anything the agent relies on:
+
+| Data source | How |
+| --- | --- |
+| `deadline` CLI | `pip install -e .` this repo; A/B two git refs of `src/` |
+| AWS CLI usage | goal + rubric only — no subject needed |
+| AWS documentation / blog / web page | fetch it to markdown, pass as `materials`, seed a corpus to revise |
+| This repo's docs | A/B with `--pathspec ':(glob)docs/**/*.md'` |
+
+## Setup
+
+```bash
+pip install -e .          # the agent's `deadline` is this checkout (A/B needs this)
+which claude              # Claude Code must be on PATH and authenticated
+```
+
+That's it — no extra dependencies; the evals use only the Python standard library.
+
+## Run
+
+```bash
+cd evals
+
+# score how an agent does today (baseline only)
+python -m agent_evals.runner run examples/deadline_cli.json --k 3
+
+# A/B a change: current branch vs a revision of the CLI source
+python -m agent_evals.runner run examples/deadline_cli.json --k 3 --revised-ref my-improvement
+
+# A/B a docs change instead of code
+python -m agent_evals.runner run my_docs_eval.json --revised-ref docs-fix --pathspec ':(glob)docs/**/*.md'
+```
+
+Artifacts land under `evals/output/<eval>/<timestamp>/`: per-run telemetry and
+transcripts, `summary.json`, and — when a revision measurably improved an eval —
+`proposal.patch`. Exit code 4 means the revision regressed.
+
+## Write an eval
+
+```json
+[
+  {
+    "id": "my_case",
+    "prompt": "The goal, stated imperatively and self-contained.",
+    "tools": ["Bash", "Read"],
+    "rubric": "What a passing answer must do, in plain language.",
+    "materials": {"guide.md": "optional reference text the agent can read"},
+    "max_turns": 20,
+    "k": 3
+  }
+]
+```
+
+The rubric is the only per-eval authoring step that matters: it should state the
+*material's own* success criterion (for a docs page, what the page promises the
+reader can do), including what a correct answer looks like when the evidence is
+incomplete — a good judge passes an agent that refuses to invent missing details.
+
+## Close the loop automatically
+
+`reviser.revise()` hands a struggling run's transcript to an agent that edits the
+subject (code or docs), commits to a scratch ref, and returns it — feed that ref
+back to `--revised-ref` to A/B-prove the improvement:
+
+```python
+from agent_evals import reviser, subject
+
+subj = subject.repo_subject()                       # or corpus_subject(markdown)
+ref = reviser.revise(subj, run_dir, goal="...", base_ref="mainline")
+# python -m agent_evals.runner run my_eval.json --revised-ref <ref>
+```
+
+## Notes
+
+- The tested agent always runs isolated — the orchestrating session must never do
+  the goal itself, or the telemetry measures the wrong thing.
+- Runs that talk to real AWS use whatever credentials/config the environment has;
+  point them at a non-production sandbox account.
+- The judge is a single vote per run; for gate-quality decisions increase `k`.

evals/agent_evals/__init__.py

@@ -0,0 +1,4 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+"""Agent evals: measure how well an AI agent achieves goals with Deadline Cloud
+tooling and docs, and A/B-prove improvements to whatever the agent relied on."""

evals/agent_evals/harness.py

@@ -0,0 +1,130 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+"""Run Claude Code headless against a goal and capture telemetry.
+
+The tested agent runs as an ISOLATED subprocess (`claude -p` with stream-json
+output) in a sandbox directory. The JSON event stream carries both per-tool-call
+events and a final result event with token/cost telemetry, so no extra
+instrumentation is needed.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+
+@dataclass
+class RunResult:
+    """One headless agent run: outcome + telemetry, plus the raw event log."""
+
+    success: bool  # did the CLI complete without error
+    subtype: Optional[str]  # result subtype, e.g. "success" / "error_max_turns"
+    tool_calls: list = field(default_factory=list)
+    num_turns: int = 0
+    total_cost_usd: float = 0.0
+    duration_ms: int = 0
+    final_text: str = ""  # the agent's final response text
+    workdir: Optional[Path] = None
+    raw_events: list = field(default_factory=list)
+
+    @property
+    def tool_call_count(self) -> int:
+        return len(self.tool_calls)
+
+    def telemetry_dict(self) -> dict:
+        """Serializable telemetry (excludes the bulky raw event log)."""
+        return {
+            "success": self.success,
+            "subtype": self.subtype,
+            "tool_calls": self.tool_calls,
+            "tool_call_count": self.tool_call_count,
+            "num_turns": self.num_turns,
+            "total_cost_usd": self.total_cost_usd,
+            "duration_ms": self.duration_ms,
+            "final_text": self.final_text,
+        }
+
+
+def run_agent(
+    prompt: str,
+    workdir: Path,
+    allowed_tools: list,
+    *,
+    max_turns: int = 20,
+    model: Optional[str] = None,
+    claude_bin: str = "claude",
+) -> RunResult:
+    """Run Claude Code headless in `workdir`, restricted to `allowed_tools`."""
+    cmd = [
+        claude_bin,
+        "-p",
+        prompt,
+        "--output-format",
+        "stream-json",
+        "--verbose",  # required for stream-json event detail
+        "--permission-mode",
+        "bypassPermissions",
+        "--max-turns",
+        str(max_turns),
+    ]
+    if allowed_tools:
+        cmd += ["--allowedTools", *allowed_tools]
+    if model:
+        cmd += ["--model", model]
+
+    # stdin=DEVNULL: with -p the CLI still waits on stdin and can exit non-zero on a
+    # closed pipe; DEVNULL makes the call cleanly non-interactive.
+    proc = subprocess.run(
+        cmd, cwd=str(workdir), capture_output=True, text=True, stdin=subprocess.DEVNULL
+    )
+
+    events = []
+    for line in proc.stdout.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            events.append(json.loads(line))
+        except json.JSONDecodeError:
+            continue  # non-JSON lines (rare) are ignored
+
+    return _parse_events(events, proc.returncode, workdir)
+
+
+def _parse_events(events: list, returncode: int, workdir: Path) -> RunResult:
+    tool_calls = []
+    final = None
+
+    for ev in events:
+        if ev.get("type") == "assistant":
+            for blk in ev.get("message", {}).get("content", []):
+                if blk.get("type") == "tool_use":
+                    tool_calls.append(blk["name"])
+        elif ev.get("type") == "result":
+            final = ev
+
+    if final is None:
+        # CLI died before emitting a result event.
+        return RunResult(
+            success=False,
+            subtype="no_result_event",
+            tool_calls=tool_calls,
+            workdir=workdir,
+            raw_events=events,
+        )
+
+    return RunResult(
+        success=(not final.get("is_error", False)) and returncode == 0,
+        subtype=final.get("subtype"),
+        tool_calls=tool_calls,
+        num_turns=final.get("num_turns", 0),
+        total_cost_usd=final.get("total_cost_usd", 0.0),
+        duration_ms=final.get("duration_ms", 0),
+        final_text=final.get("result", "") if isinstance(final.get("result"), str) else "",
+        workdir=workdir,
+        raw_events=events,
+    )

evals/agent_evals/judge.py

@@ -0,0 +1,141 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+"""Grade an agent's answer against a natural-language rubric (LLM-as-judge).
+
+The judge is a SCORER, not an agent: a headless, tool-less, single-turn model call.
+It sees only the rubric, the task prompt, and the agent's final answer, so its
+verdict is reproducible from the run artifact alone. The rubric being a plain
+string is what makes new eval domains code-free: describing what "success" means
+is the only per-eval authoring step.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from dataclasses import dataclass
+from typing import Optional
+
+
+class JudgeError(RuntimeError):
+    """Raised when the judge call fails to produce a usable verdict."""
+
+
+@dataclass
+class Verdict:
+    passed: bool
+    reasoning: str
+
+
+_JUDGE_PROMPT = """\
+You are grading an AI agent's answer to a task. Decide ONLY whether the answer
+satisfies the rubric below. Judge the substance of the answer, not its wording or
+format. A correct conclusion stated in unexpected phrasing still passes; a
+confident answer that is wrong or unsupported fails. If the rubric asks the agent
+to reach a conclusion the available evidence does not support, then correctly
+declining to invent one SATISFIES the rubric.
+
+You have NO tools. Do not attempt to read any file or run any command -- files the
+answer mentions are not available to you. Grade purely from the text below.
+
+=== RUBRIC (what a passing answer must do) ===
+{rubric}
+
+=== THE TASK THE AGENT WAS GIVEN ===
+{prompt}
+
+=== THE AGENT'S FINAL ANSWER ===
+{answer}
+
+Respond with ONLY a JSON object on a single line, no prose, no code fence:
+{{"passed": true or false, "reasoning": "one or two sentences citing the rubric"}}
+"""
+
+
+def judge_answer(
+    rubric: str,
+    prompt: str,
+    answer: str,
+    *,
+    model: Optional[str] = None,
+    claude_bin: str = "claude",
+) -> Verdict:
+    """Grade `answer` against `rubric` with a headless, tool-less model call."""
+    if not (answer or "").strip():
+        return Verdict(passed=False, reasoning="agent produced no final answer")
+
+    full_prompt = _JUDGE_PROMPT.format(rubric=rubric, prompt=prompt or "(none)", answer=answer)
+    cmd = [
+        claude_bin,
+        "-p",
+        full_prompt,
+        "--output-format",
+        "json",
+        "--permission-mode",
+        "bypassPermissions",
+        # The judge must answer from the given text alone, so all tools are denied.
+        # A denied tool attempt still consumes a turn, so leave headroom for the
+        # model to recover and answer instead of dying on error_max_turns.
+        "--max-turns",
+        "5",
+        "--disallowedTools",
+        "Bash",
+        "Read",
+        "Write",
+        "Edit",
+        "Grep",
+        "Glob",
+        "WebFetch",
+        "WebSearch",
+        "Agent",
+        "TodoWrite",
+        "NotebookEdit",
+    ]
+    if model:
+        cmd += ["--model", model]
+
+    try:
+        # stdin=DEVNULL: with -p the CLI still waits on stdin and can exit non-zero
+        # on a closed pipe.
+        proc = subprocess.run(cmd, capture_output=True, text=True, stdin=subprocess.DEVNULL)
+    except OSError as e:
+        raise JudgeError(f"could not launch {claude_bin}: {e}") from e
+
+    # --output-format json wraps the reply in an envelope whose `result` field is
+    # the text we asked for.
+    reply = proc.stdout
+    try:
+        envelope = json.loads(proc.stdout)
+        if isinstance(envelope, dict) and isinstance(envelope.get("result"), str):
+            reply = envelope["result"]
+    except json.JSONDecodeError:
+        pass
+
+    # Parse the verdict from stdout FIRST: the CLI sometimes exits non-zero after
+    # emitting a valid result (e.g. a model-availability warning). Only surface the
+    # exit code when stdout carried nothing gradable.
+    try:
+        return _extract_verdict(reply)
+    except JudgeError:
+        if proc.returncode != 0:
+            raise JudgeError(
+                f"judge exited {proc.returncode} with no usable verdict: "
+                f"{(proc.stderr or reply).strip()[:200]}"
+            ) from None
+        raise
+
+
+def _extract_verdict(text: str) -> Verdict:
+    """Pull the JSON verdict out of the judge's reply, tolerating code fences or
+    stray prose around it (scan first '{' to last '}')."""
+    start = text.find("{")
+    end = text.rfind("}")
+    if start == -1 or end == -1 or end < start:
+        raise JudgeError(f"judge response had no JSON object: {text[:200]!r}")
+    try:
+        obj = json.loads(text[start : end + 1])
+    except json.JSONDecodeError as e:
+        raise JudgeError(f"judge JSON did not parse: {e}") from e
+    if "passed" not in obj:
+        raise JudgeError(f"judge JSON missing 'passed': {obj!r}")
+    return Verdict(passed=bool(obj["passed"]), reasoning=str(obj.get("reasoning", "")))