feat: generate auditable agent trial reports

This commit is contained in:
lda
2026-06-23 02:14:03 +07:00 Verified
parent 20293a3781
commit 3bb9eacb25
11 changed files with 1918 additions and 7 deletions
+4
View File
@@ -221,6 +221,10 @@ stable.
manual-audited reports. Two data-driven challenges exist: browser-click and
report-workflow. The central `run_trials.py` runner accepts any challenge
manifest.
- Completed: report projections generate bounded Markdown and JSON reports for
every V2 trial and regenerate both after audit without mutating raw evidence.
Implementation:
[`report projections`](historical/superpowers/plans/2026-06-23-agent-challenge-report-projections.md).
## Historical References
@@ -0,0 +1,452 @@
# Agent Challenge Report Projections Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Generate equivalent bounded Markdown and JSON reports for every V2 trial and regenerate both after audit without mutating raw evidence.
**Architecture:** Add strict Pydantic report DTOs, project one `TrialReport` to Markdown and JSON, wire projection generation after raw result persistence, and make V2 audit rebuild both. Preserve referenced V1 behavior.
**Tech Stack:** Python 3.14, Pydantic v2, PyYAML, pathlib, argparse, pytest, Ruff, basedpyright.
---
## File Structure
- Create `examples/agent_challenges/report_models.py`.
- Modify `examples/agent_challenges/reports.py`, `runner.py`, `run_trials.py`, `audit.py`, and `base-prompt.md`.
- Create `tests/examples/test_agent_challenge_reports.py`.
- Modify `tests/examples/test_agent_challenge_harness_v2.py`.
- Update `docs/current_roadmap.md` and archive this plan.
### Task 1: Define The Normalized Report
**Files:**
- Create: `examples/agent_challenges/report_models.py`
- Modify: `examples/agent_challenges/reports.py`
- Create: `tests/examples/test_agent_challenge_reports.py`
- [ ] **Step 1: Write a failing bounded-report test**
Build a `_raw_result(tmp_path)` fixture matching the current V2 result shape,
including explicit paths, identity, challenge report, parsed text, metrics,
policy, tool calls, and deliberately large stdout/output previews.
```python
def test_trial_report_is_bounded_machine_projection(tmp_path: Path) -> None:
payload = build_trial_report(_raw_result(tmp_path), audit=None).model_dump(
mode="json"
)
assert payload["schema_version"] == 1
assert payload["identity"]["challenge_id"] == "fixture"
assert payload["outcome"]["task_outcome"] == "success"
assert payload["commands_and_tools"][0]["detail"].endswith(
"workflow.plan.json"
)
serialized = json.dumps(payload)
assert "large raw stream" not in serialized
assert "full tool output" not in serialized
assert payload["manual_audit"]["status"] == "pending"
```
- [ ] **Step 2: Run and verify import failure**
```powershell
uv run pytest tests/examples/test_agent_challenge_reports.py::test_trial_report_is_bounded_machine_projection -q
```
- [ ] **Step 3: Implement strict DTOs**
Create `TrialIdentity`, `TrialOutcome`, `CommandToolBrief`, `TokenSummary`,
`AutomaticEvidence`, `ManualAuditSummary`, and `TrialReport` using
`ConfigDict(extra="forbid")`.
```python
class StrictReportModel(BaseModel):
model_config = ConfigDict(extra="forbid")
class TrialIdentity(StrictReportModel):
challenge_id: str
model: str
variant: str
instruction_profile: str
trial_index: int
repository_commit: str | None = None
repository_dirty: bool | None = None
prompt_hashes: dict[str, str] = Field(default_factory=dict)
raw_result_path: str
workspace_path: str
class TrialOutcome(StrictReportModel):
task_outcome: str
evaluation_validity: str
duration_seconds: float
returncode: int | None
assertion_failures: list[str] = Field(default_factory=list)
parse_errors: dict[str, dict[str, str]] = Field(default_factory=dict)
class CommandToolBrief(StrictReportModel):
ordinal: int
tool: str
status: str
title: str
detail: str | None = None
failed: bool
output_chars: int
output_sha256: str
class TokenSummary(StrictReportModel):
total: int = 0
input: int = 0
output: int = 0
reasoning: int = 0
cache_read: int = 0
cache_write: int = 0
class AutomaticEvidence(StrictReportModel):
step_count: int = 0
tool_call_count: int = 0
failed_tool_call_count: int = 0
tool_counts: dict[str, int] = Field(default_factory=dict)
tokens: TokenSummary = Field(default_factory=TokenSummary)
cost: float = 0.0
unknown_event_count: int = 0
reads_by_category: dict[str, list[str]] = Field(default_factory=dict)
escalated_to_product_code: bool = False
disallowed_reads: list[str] = Field(default_factory=list)
opaque_shell_commands: list[str] = Field(default_factory=list)
class ManualAuditSummary(StrictReportModel):
status: Literal["pending", "complete"] = "pending"
official_outcome: str | None = None
auditor: str | None = None
audited_at: str | None = None
corrections: list[str] = Field(default_factory=list)
notes: str = ""
read_flags: dict[str, bool] = Field(default_factory=dict)
evidence: dict[str, Any] = Field(default_factory=dict)
class TrialReport(StrictReportModel):
schema_version: Literal[1] = 1
identity: TrialIdentity
outcome: TrialOutcome
agent_self_report: dict[str, Any] | None = None
final_agent_answer: str | None = None
commands_and_tools: list[CommandToolBrief] = Field(default_factory=list)
automatic_evidence: AutomaticEvidence
policy_findings: list[str] = Field(default_factory=list)
self_report_discrepancies: list[str] = Field(default_factory=list)
manual_audit: ManualAuditSummary = Field(default_factory=ManualAuditSummary)
follow_up_notes: list[str] = Field(default_factory=list)
```
Identity holds challenge/model/variant/profile/index, repository provenance,
prompt hashes, raw path, and workspace path. Outcome holds task outcome,
validity, duration, return code, assertions, and parser errors. Evidence holds
bounded metrics, read categories, disallowed reads, and opaque commands.
- [ ] **Step 4: Implement the builder**
```python
def build_trial_report(
result: dict[str, object],
*,
audit: dict[str, object] | None,
) -> TrialReport:
"""Project immutable raw evidence and optional audit into one report."""
```
Use focused helpers. Require explicit paths; bound final text to 8,000 chars and
command detail to 1,000; retain only tool ordinal/name/status/title/detail,
failure, output size/hash; exclude stdout/stderr/previews/metadata/full input.
Flag observable self-report conflicts. Treat example reads plus
`existing_solution=false` as a manual follow-up, not automatic guilt.
- [ ] **Step 5: Verify and commit**
```powershell
uv run pytest tests/examples/test_agent_challenge_reports.py -q
git add examples/agent_challenges/report_models.py examples/agent_challenges/reports.py tests/examples/test_agent_challenge_reports.py
git commit -m "feat: add normalized agent trial report model"
```
### Task 2: Render And Write Both Projections
**Files:**
- Modify: `examples/agent_challenges/reports.py`
- Modify: `tests/examples/test_agent_challenge_reports.py`
- [ ] **Step 1: Add failing projection tests**
Assert writes to `workspace/final-report.md` and `results/trial.report.json`,
stable heading order, bounded commands, no raw outputs, pending audit, valid
JSON, and no temporary-file residue.
- [ ] **Step 2: Run and verify missing APIs**
```powershell
uv run pytest tests/examples/test_agent_challenge_reports.py -k projection -q
```
- [ ] **Step 3: Implement Markdown renderer**
Use exactly these headings:
```markdown
# Trial Report
## Outcome
## Agent Self-Report
## Commands And Tool Calls
## Automatic Evidence
## Policy Findings
## Self-Report Discrepancies
## Manual Audit
## Follow-Up Notes
```
Render empty sections explicitly and commands as ordered bounded entries.
- [ ] **Step 4: Implement atomic writes**
```python
@dataclass(frozen=True, slots=True)
class TrialReportPaths:
markdown: Path
machine: Path
def _atomic_write_text(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(f".{path.name}.tmp")
temporary.write_text(text, encoding="utf-8")
temporary.replace(path)
def write_trial_report_projections(
report: TrialReport,
*,
markdown_path: Path,
machine_path: Path,
) -> TrialReportPaths:
machine = json.dumps(
report.model_dump(mode="json"), indent=2, sort_keys=True
) + "\n"
markdown = render_trial_report_markdown(report).rstrip() + "\n"
_atomic_write_text(machine_path, machine)
_atomic_write_text(markdown_path, markdown)
return TrialReportPaths(markdown=markdown_path, machine=machine_path)
```
- [ ] **Step 5: Verify and commit**
```powershell
uv run pytest tests/examples/test_agent_challenge_reports.py -q
git add examples/agent_challenges/reports.py tests/examples/test_agent_challenge_reports.py
git commit -m "feat: write human and machine trial reports"
```
### Task 3: Generate Reports From The Runner
**Files:**
- Modify: `examples/agent_challenges/runner.py`
- Modify: `examples/agent_challenges/run_trials.py`
- Modify: `tests/examples/test_agent_challenge_harness_v2.py`
- [ ] **Step 1: Extend runner integration tests**
For success and timeout, assert explicit `workspace_path`, `result_path`, and
`report_paths`; raw, Markdown, and machine files exist; machine challenge id is
correct; Markdown contains the final answer.
- [ ] **Step 2: Run and verify failure**
```powershell
uv run pytest tests/examples/test_agent_challenge_harness_v2.py -k "runner_to_report or timeout" -q
```
- [ ] **Step 3: Add explicit paths before the single raw write**
```python
"challenge_id": challenge.manifest.id,
"workspace_path": str(workspace.root.resolve()),
"result_path": str(result_path.resolve()),
"report_paths": {
"markdown": str((workspace.root / "final-report.md").resolve()),
"machine": str(result_path.with_suffix(".report.json").resolve()),
},
```
Include `reads_by_category` in policy. Write raw result once before projection
generation; never rewrite it.
- [ ] **Step 4: Generate pending-audit projections**
Call `build_trial_report(result, audit=None)` and
`write_trial_report_projections`. On failure preserve raw evidence and return a
concise `report_generation_error`.
- [ ] **Step 5: Print report paths**
Add `result_path` and `report_paths` to each central-runner summary.
- [ ] **Step 6: Verify and commit**
```powershell
uv run pytest tests/examples/test_agent_challenge_harness_v2.py -k "runner or timeout" -q
git add examples/agent_challenges/runner.py examples/agent_challenges/run_trials.py tests/examples/test_agent_challenge_harness_v2.py
git commit -m "feat: generate reports after agent trials"
```
### Task 4: Regenerate Reports After Manual Audit
**Files:**
- Modify: `examples/agent_challenges/audit.py`
- Modify: `examples/agent_challenges/save_manual_audit.py`
- Modify: `tests/examples/test_agent_challenge_reports.py`
- [ ] **Step 1: Add failing regeneration test**
```python
paths = save_v2_manual_audit(
result_path,
official_outcome="pass",
auditor="reviewer",
audited_at="2026-06-23T00:00:00Z",
read_overrides={"existing_solution": True},
corrections=["Agent inspected a ready-made workflow plan."],
notes="Technical run passed; self-report corrected.",
)
```
Assert YAML, JSON, and Markdown contain the official grade/corrections.
- [ ] **Step 2: Add invalid-audit preservation test**
Use outcome `maybe`, expect `ValueError`, and assert previous projections remain
byte-for-byte unchanged.
- [ ] **Step 3: Run and verify missing API**
```powershell
uv run pytest tests/examples/test_agent_challenge_reports.py -k "manual_audit or invalid" -q
```
- [ ] **Step 4: Implement V2 audit writer**
Add:
```python
@dataclass(frozen=True, slots=True)
class V2AuditPaths:
audit: Path
markdown: Path
machine: Path
def save_v2_manual_audit(
result_path: Path,
*,
official_outcome: str,
auditor: str = "human",
audited_at: str | None = None,
read_overrides: dict[str, bool] | None = None,
evidence_overrides: dict[str, object] | None = None,
corrections: list[str] | None = None,
notes: str = "",
) -> V2AuditPaths:
"""Write authoritative audit data and regenerate both report projections."""
```
Allow only `pass`, `fail`, `invalid`; require V2; use explicit raw paths;
validate before writes; atomically write audit; rebuild both projections; never
rewrite raw result.
- [ ] **Step 5: Route existing CLI by harness version**
For V2, interpret `--manual-classification` as official outcome, reject
`--from-report`, and print JSON containing all three paths. Preserve referenced
V1 behavior.
- [ ] **Step 6: Verify and commit**
```powershell
uv run pytest tests/examples/test_agent_challenge_reports.py -q
uv run python examples/agent_challenges/save_manual_audit.py --help
git add examples/agent_challenges/audit.py examples/agent_challenges/save_manual_audit.py tests/examples/test_agent_challenge_reports.py
git commit -m "feat: regenerate trial reports after audit"
```
### Task 5: Clarify Shared Self-Reporting Rules
**Files:**
- Modify: `examples/agent_challenges/base-prompt.md`
- Modify: `tests/examples/test_agent_challenge_harness_v2.py`
- [ ] **Step 1: Add failing assertions**
Assert the prompt mentions `tests/`, `examples/`, `read.product_code: true`,
`read.existing_solution: true`, and `read.adjacent_attempts: true`.
- [ ] **Step 2: Add approved paragraph**
```markdown
Files under `tests/` and `examples/` may contain complete or partial solutions.
If you inspect them, report `read.product_code: true`; also report
`read.existing_solution: true` when they provide a ready-made solution, or
`read.adjacent_attempts: true` when they contain prior trial outputs.
```
- [ ] **Step 3: Verify and commit**
```powershell
uv run pytest tests/examples/test_agent_challenge_harness_v2.py -k base_prompt -q
git add examples/agent_challenges/base-prompt.md tests/examples/test_agent_challenge_harness_v2.py
git commit -m "docs: clarify challenge self-report rules"
```
### Task 6: Final Verification And Documentation
**Files:**
- Modify: `docs/current_roadmap.md`
- Move this plan to `docs/historical/superpowers/plans/`.
- [ ] **Step 1: Run verification**
```powershell
uv run pytest tests/examples/test_agent_challenge_reports.py tests/examples/test_agent_challenge_harness_v2.py tests/examples/test_agent_challenge_skill_bundle.py tests/examples/test_opencode_browser_click_challenge.py tests/examples/test_report_workflow_challenge.py -q
uv run ruff check examples/agent_challenges tests/examples/test_agent_challenge_reports.py tests/examples/test_agent_challenge_harness_v2.py
uv run ruff format --check examples/agent_challenges tests/examples/test_agent_challenge_reports.py tests/examples/test_agent_challenge_harness_v2.py
uv run basedpyright --level error examples/agent_challenges tests/examples/test_agent_challenge_reports.py tests/examples/test_agent_challenge_harness_v2.py
git diff --check
```
- [ ] **Step 2: Smoke CLI help**
```powershell
uv run python examples/agent_challenges/run_trials.py --help
uv run python examples/agent_challenges/save_manual_audit.py --help
```
- [ ] **Step 3: Update roadmap and archive**
Record completion, then move this plan to
`docs/historical/superpowers/plans/2026-06-23-agent-challenge-report-projections.md`.
- [ ] **Step 4: Review boundaries**
Confirm raw result is written once, machine report excludes raw outputs,
Markdown headings are stable, explicit paths are used, audit regenerates both,
V1 remains intact, and no runbook/branching challenge leaked into this slice.
- [ ] **Step 5: Commit completion docs**
```powershell
git add docs/current_roadmap.md docs/superpowers/plans/2026-06-23-agent-challenge-report-projections.md docs/historical/superpowers/plans/2026-06-23-agent-challenge-report-projections.md
git commit -m "docs: record agent challenge report projections"
```
+112 -1
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import argparse
import json
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
@@ -9,7 +10,11 @@ from typing import Any
import yaml
from examples.agent_challenges.classification import extract_challenge_report
from examples.agent_challenges.reports import report_from_result
from examples.agent_challenges.report_models import build_trial_report
from examples.agent_challenges.reports import (
report_from_result,
write_trial_report_projections,
)
def audit_from_result(
@@ -256,3 +261,109 @@ def manual_audit_from_v2_result(
output_path = Path(output_name)
return output_path, audit
_VALID_OFFICIAL_OUTCOMES = frozenset({"pass", "fail", "invalid"})
@dataclass(frozen=True, slots=True)
class V2AuditPaths:
audit: Path
markdown: Path
machine: Path
def save_v2_manual_audit(
result_path: Path,
*,
official_outcome: str,
auditor: str = "human",
audited_at: str | None = None,
read_overrides: dict[str, bool] | None = None,
evidence_overrides: dict[str, object] | None = None,
corrections: list[str] | None = None,
notes: str = "",
) -> V2AuditPaths:
"""Write authoritative audit data and regenerate both report projections.
Args:
result_path: Path to the V2 raw result JSON file.
official_outcome: One of 'pass', 'fail', 'invalid'.
auditor: Name or identifier of the human auditor.
audited_at: ISO-8601 timestamp string.
read_overrides: Override read flags from the challenge report.
evidence_overrides: Override evidence values from the challenge report.
corrections: List of correction strings.
notes: Free-form auditor notes.
Returns:
V2AuditPaths with paths to audit YAML, markdown report, and machine report.
"""
if official_outcome not in _VALID_OFFICIAL_OUTCOMES:
raise ValueError(
f"official_outcome must be one of {sorted(_VALID_OFFICIAL_OUTCOMES)}, "
f"got {official_outcome!r}"
)
result = json.loads(result_path.read_text(encoding="utf-8"))
if not isinstance(result, dict):
raise ValueError("result file must contain a JSON object")
harness_version = result.get("harness_version")
if harness_version != "v2":
raise ValueError(
f"save_v2_manual_audit requires harness_version='v2', "
f"got {harness_version!r}"
)
workspace_path_str = result.get("workspace_path")
if not isinstance(workspace_path_str, str) or not workspace_path_str:
raise ValueError("result is missing workspace_path")
workspace_path = Path(workspace_path_str)
result_path_str = result.get("result_path")
if not isinstance(result_path_str, str) or not result_path_str:
raise ValueError("result is missing result_path")
audit_payload: dict[str, object] = {
"manual_audit": {
"official_outcome": official_outcome,
"auditor": auditor,
"audited_at": audited_at or _utc_now(),
"task_outcome": result.get("task_outcome"),
"evaluation_validity": result.get("evaluation_validity"),
"corrections": corrections or [],
"notes": notes,
"read_flags": dict(read_overrides or {}),
"evidence": dict(evidence_overrides or {}),
}
}
yaml_text = yaml.safe_dump(audit_payload, sort_keys=False, allow_unicode=True)
trial_report = build_trial_report(
result,
audit=audit_payload,
raw_result_path=result_path_str,
workspace_path=workspace_path_str,
)
markdown_path = workspace_path / "final-report.md"
machine_path = result_path.with_suffix(".report.json")
write_trial_report_projections(
trial_report,
markdown_path=markdown_path,
machine_path=machine_path,
)
audit_path = workspace_path / "manual-audit.yaml"
audit_path.parent.mkdir(parents=True, exist_ok=True)
temporary = audit_path.with_name(f".{audit_path.name}.tmp")
temporary.write_text(yaml_text, encoding="utf-8")
temporary.replace(audit_path)
return V2AuditPaths(
audit=audit_path,
markdown=markdown_path,
machine=machine_path,
)
+5
View File
@@ -14,3 +14,8 @@ Use this command prefix:
Your writable trial workspace is `{{workspace_path}}`. Write attempt files only
inside it. End with the challenge's requested YAML self-report. The self-report
will be checked against observed tool calls and manually audited.
Files under `tests/` and `examples/` may contain complete or partial solutions.
If you inspect them, report `read.product_code: true`; also report
`read.existing_solution: true` when they provide a ready-made solution, or
`read.adjacent_attempts: true` when they contain prior trial outputs.
+501
View File
@@ -0,0 +1,501 @@
from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
class StrictReportModel(BaseModel):
model_config = ConfigDict(extra="forbid")
class TrialIdentity(StrictReportModel):
challenge_id: str
model: str
variant: str
instruction_profile: str
trial_index: int
repository_commit: str | None = None
repository_dirty: bool | None = None
prompt_hashes: dict[str, str] = Field(default_factory=dict)
raw_result_path: str
workspace_path: str
class TrialOutcome(StrictReportModel):
task_outcome: str
evaluation_validity: str
duration_seconds: float
returncode: int | None = None
assertion_failures: list[str] = Field(default_factory=list)
parse_errors: dict[str, dict[str, str]] = Field(default_factory=dict)
class CommandToolBrief(StrictReportModel):
ordinal: int
tool: str
status: str
title: str
detail: str | None = None
failed: bool
output_chars: int
output_sha256: str
class TokenSummary(StrictReportModel):
total: int = 0
input: int = 0
output: int = 0
reasoning: int = 0
cache_read: int = 0
cache_write: int = 0
class AutomaticEvidence(StrictReportModel):
step_count: int = 0
tool_call_count: int = 0
failed_tool_call_count: int = 0
tool_counts: dict[str, int] = Field(default_factory=dict)
tokens: TokenSummary = Field(default_factory=TokenSummary)
cost: float = 0.0
unknown_event_count: int = 0
reads_by_category: dict[str, list[str]] = Field(default_factory=dict)
escalated_to_product_code: bool = False
disallowed_reads: list[str] = Field(default_factory=list)
opaque_shell_commands: list[str] = Field(default_factory=list)
class ManualAuditSummary(StrictReportModel):
status: Literal["pending", "complete"] = "pending"
official_outcome: str | None = None
auditor: str | None = None
audited_at: str | None = None
corrections: list[str] = Field(default_factory=list)
notes: str = ""
read_flags: dict[str, bool] = Field(default_factory=dict)
evidence: dict[str, Any] = Field(default_factory=dict)
class TrialReport(StrictReportModel):
schema_version: Literal[1] = 1
identity: TrialIdentity
outcome: TrialOutcome
agent_self_report: dict[str, Any] | None = None
final_agent_answer: str | None = None
commands_and_tools: list[CommandToolBrief] = Field(default_factory=list)
automatic_evidence: AutomaticEvidence
policy_findings: list[str] = Field(default_factory=list)
self_report_discrepancies: list[str] = Field(default_factory=list)
manual_audit: ManualAuditSummary = Field(default_factory=ManualAuditSummary)
follow_up_notes: list[str] = Field(default_factory=list)
_MAX_FINAL_TEXT_CHARS = 8_000
_MAX_COMMAND_DETAIL_CHARS = 1_000
def _build_identity(
result: dict[str, object], raw_result_path: str, workspace_path: str
) -> TrialIdentity:
return TrialIdentity(
challenge_id=_str(result.get("challenge_id")),
model=_str(result.get("model")),
variant=_str(result.get("variant")),
instruction_profile=_str(result.get("instruction_profile")),
trial_index=_int(result.get("trial_index", result.get("index"))),
repository_commit=_str_none(result.get("repository_commit")),
repository_dirty=_bool_none(result.get("repository_dirty")),
prompt_hashes=_dict_str_str(result.get("prompt_hashes")),
raw_result_path=raw_result_path,
workspace_path=workspace_path,
)
def _build_outcome(result: dict[str, object]) -> TrialOutcome:
return TrialOutcome(
task_outcome=_str(result.get("task_outcome")),
evaluation_validity=_str(result.get("evaluation_validity")),
duration_seconds=_float(result.get("duration_seconds")),
returncode=_int_none(result.get("returncode")),
assertion_failures=_list_str(result.get("assertion_failures")),
parse_errors=_parse_errors(result),
)
def _build_tool_briefs(result: dict[str, object]) -> list[CommandToolBrief]:
metrics = result.get("metrics")
if not isinstance(metrics, dict):
return []
tool_calls = metrics.get("tool_calls")
if not isinstance(tool_calls, list):
return []
briefs: list[CommandToolBrief] = []
for tc in tool_calls:
if not isinstance(tc, dict):
continue
raw_input = tc.get("input")
tc_input: dict[str, object] = {}
if isinstance(raw_input, dict):
tc_input = raw_input
detail_str = ""
path_val = (
tc_input.get("path")
or tc_input.get("filePath")
or tc_input.get("file")
or ""
)
if isinstance(path_val, str) and path_val:
detail_str = path_val[:_MAX_COMMAND_DETAIL_CHARS]
elif "command" in tc_input:
cmd = tc_input["command"]
if isinstance(cmd, str):
detail_str = cmd[:_MAX_COMMAND_DETAIL_CHARS]
briefs.append(
CommandToolBrief(
ordinal=_int(tc.get("ordinal")),
tool=_str(tc.get("tool")),
status=_str(tc.get("status")),
title=_str(tc.get("title")),
detail=detail_str or None,
failed=bool(tc.get("failed", False)),
output_chars=_int(tc.get("output_chars")),
output_sha256=_str(tc.get("output_sha256")),
)
)
return briefs
def _build_automatic_evidence(result: dict[str, object]) -> AutomaticEvidence:
metrics = result.get("metrics")
tokens = TokenSummary()
cost = 0.0
tool_counts: dict[str, int] = {}
step_count = 0
tool_call_count = 0
failed_tool_call_count = 0
unknown_event_count = 0
if isinstance(metrics, dict):
step_count = _int(metrics.get("step_count"))
tool_call_count = _int(metrics.get("tool_call_count"))
failed_tool_call_count = _int(metrics.get("failed_tool_call_count"))
tool_counts = _dict_str_int(metrics.get("tool_counts"))
unknown_event_count = _int(metrics.get("unknown_event_count"))
tokens_raw = metrics.get("tokens")
if isinstance(tokens_raw, dict):
tokens = TokenSummary(
total=_int(tokens_raw.get("total")),
input=_int(tokens_raw.get("input")),
output=_int(tokens_raw.get("output")),
reasoning=_int(tokens_raw.get("reasoning")),
cache_read=_int(tokens_raw.get("cache_read")),
cache_write=_int(tokens_raw.get("cache_write")),
)
cost = _float(metrics.get("cost"))
policy = result.get("policy")
reads_by_category: dict[str, list[str]] = {}
disallowed_reads: list[str] = []
escalated_to_product_code = False
opaque_shell_commands: list[str] = []
if isinstance(policy, dict):
reads_in = policy.get("reads_by_category")
if isinstance(reads_in, dict):
reads_by_category = {
k: list(v) if isinstance(v, (list, tuple)) else [str(v)]
for k, v in reads_in.items()
}
disallowed_reads = _list_str(policy.get("disallowed_reads"))
escalated_to_product_code = bool(policy.get("escalated_to_product_code", False))
opaque_shell_commands = _list_str(policy.get("opaque_shell_commands"))
return AutomaticEvidence(
step_count=step_count,
tool_call_count=tool_call_count,
failed_tool_call_count=failed_tool_call_count,
tool_counts=tool_counts,
tokens=tokens,
cost=cost,
unknown_event_count=unknown_event_count,
reads_by_category=reads_by_category,
escalated_to_product_code=escalated_to_product_code,
disallowed_reads=disallowed_reads,
opaque_shell_commands=opaque_shell_commands,
)
def _build_trial_report(
result: dict[str, object],
*,
audit: dict[str, object] | None,
raw_result_path: str,
workspace_path: str,
) -> TrialReport:
identity = _build_identity(result, raw_result_path, workspace_path)
outcome = _build_outcome(result)
commands_and_tools = _build_tool_briefs(result)
automatic_evidence = _build_automatic_evidence(result)
agent_self_report: dict[str, Any] | None = None
challenge_report = result.get("challenge_report")
if isinstance(challenge_report, dict):
agent_self_report = challenge_report
final_agent_answer: str | None = None
parsed = result.get("parsed")
if isinstance(parsed, dict):
text = parsed.get("text")
if isinstance(text, str) and text.strip():
final_agent_answer = text[:_MAX_FINAL_TEXT_CHARS]
policy_findings: list[str] = _build_policy_findings(result, automatic_evidence)
self_report_discrepancies: list[str] = _build_self_report_discrepancies(
result, agent_self_report
)
follow_up_notes: list[str] = _build_follow_up_notes(result, automatic_evidence)
manual_audit = _build_manual_audit(audit)
return TrialReport(
identity=identity,
outcome=outcome,
agent_self_report=agent_self_report,
final_agent_answer=final_agent_answer,
commands_and_tools=commands_and_tools,
automatic_evidence=automatic_evidence,
policy_findings=policy_findings,
self_report_discrepancies=self_report_discrepancies,
follow_up_notes=follow_up_notes,
manual_audit=manual_audit,
)
def _build_policy_findings(
result: dict[str, object], evidence: AutomaticEvidence
) -> list[str]:
findings: list[str] = []
policy = result.get("policy")
if isinstance(policy, dict):
validity = policy.get("validity")
if isinstance(validity, str) and validity != "clean":
findings.append(f"Evaluation validity: {validity}")
if evidence.disallowed_reads:
findings.append(f"Disallowed reads ({len(evidence.disallowed_reads)} paths)")
if evidence.opaque_shell_commands:
findings.append(
f"Opaque shell commands ({len(evidence.opaque_shell_commands)} commands)"
)
return findings
def _build_self_report_discrepancies(
result: dict[str, object], agent_self_report: dict[str, Any] | None
) -> list[str]:
discrepancies: list[str] = []
if agent_self_report is None:
return discrepancies
task_outcome = _str(result.get("task_outcome"))
agent_run_failed = agent_self_report.get("run_failed")
if agent_run_failed is True and task_outcome == "success":
discrepancies.append(
"Agent reported run_failed=true but task_outcome is 'success'"
)
elif agent_run_failed is False and task_outcome == "failed":
discrepancies.append(
"Agent reported run_failed=false but task_outcome is 'failed'"
)
elif agent_run_failed is False and task_outcome == "timeout":
discrepancies.append(
"Agent reported run_failed=false but task ended in timeout"
)
_check_escalation_discrepancy(result, agent_self_report, discrepancies)
return discrepancies
def _check_escalation_discrepancy(
result: dict[str, object],
agent_self_report: dict[str, Any],
discrepancies: list[str],
) -> None:
policy_raw = result.get("policy")
if not isinstance(policy_raw, dict):
return
escalated = policy_raw.get("escalated_to_product_code")
if escalated is not True:
return
agent_read_raw = agent_self_report.get("read")
if isinstance(agent_read_raw, dict):
for k, v in agent_read_raw.items():
if k == "product_code" and v is True:
return
discrepancies.append(
"Agent escalated to product code but did not report read.product_code"
)
def _build_follow_up_notes(
result: dict[str, object], evidence: AutomaticEvidence
) -> list[str]:
notes: list[str] = []
policy_raw = result.get("policy")
if isinstance(policy_raw, dict):
reads_by_cat_raw = policy_raw.get("reads_by_category")
if isinstance(reads_by_cat_raw, dict):
examples = reads_by_cat_raw.get("examples", [])
tests = reads_by_cat_raw.get("tests", [])
existing_solution_raw = reads_by_cat_raw.get("existing_solution", [])
if isinstance(examples, (list, tuple)) and examples:
notes.append(
f"Agent read {len(examples)} example file(s); verify whether "
"existing_solution applies"
)
if isinstance(tests, (list, tuple)) and tests:
notes.append(
f"Agent read {len(tests)} test file(s); verify whether "
"existing_solution applies"
)
if (
isinstance(existing_solution_raw, (list, tuple))
and existing_solution_raw
):
notes.append(
f"Agent found {len(existing_solution_raw)} existing solution "
"reference(s); verify self-report accuracy"
)
if evidence.disallowed_reads:
notes.append(
f"Disallowed reads ({len(evidence.disallowed_reads)} path(s)); "
"review for contamination impact"
)
return notes
def _build_manual_audit(
audit: dict[str, object] | None,
) -> ManualAuditSummary:
if audit is None:
return ManualAuditSummary()
manual = audit.get("manual_audit")
if isinstance(manual, dict):
return ManualAuditSummary(
status="complete",
official_outcome=_str_none(manual.get("official_outcome")),
auditor=_str_none(manual.get("auditor")),
audited_at=_str_none(manual.get("audited_at")),
corrections=_list_str(manual.get("corrections")),
notes=_str(manual.get("notes")),
read_flags=_dict_str_bool(manual.get("read_flags")),
evidence=_dict_any(manual.get("evidence")),
)
return ManualAuditSummary()
def build_trial_report(
result: dict[str, object],
*,
audit: dict[str, object] | None,
raw_result_path: str | None = None,
workspace_path: str | None = None,
) -> TrialReport:
if raw_result_path is None:
raw_result_path = _str(result.get("result_path"))
if workspace_path is None:
workspace_path = _str(result.get("workspace_path"))
if not raw_result_path:
raise ValueError("raw_result_path is required")
if not workspace_path:
raise ValueError("workspace_path is required")
return _build_trial_report(
result,
audit=audit,
raw_result_path=raw_result_path,
workspace_path=workspace_path,
)
def _str(value: object, *, default: str = "") -> str:
return value if isinstance(value, str) else default
def _str_none(value: object) -> str | None:
return value if isinstance(value, str) else None
def _int(value: object, *, default: int = 0) -> int:
if isinstance(value, (int, float)):
return int(value)
return default
def _int_none(value: object) -> int | None:
return value if isinstance(value, int) else None
def _float(value: object, *, default: float = 0.0) -> float:
if isinstance(value, (int, float)):
return float(value)
return default
def _bool_none(value: object) -> bool | None:
return value if isinstance(value, bool) else None
def _list_str(value: object) -> list[str]:
if isinstance(value, (list, tuple)):
return [v for v in value if isinstance(v, str)]
return []
def _dict_str_str(value: object) -> dict[str, str]:
if isinstance(value, dict):
return {
k: str(v)
for k, v in value.items()
if isinstance(k, str) and isinstance(v, str)
}
return {}
def _dict_str_int(value: object) -> dict[str, int]:
if isinstance(value, dict):
return {
k: int(v)
for k, v in value.items()
if isinstance(k, str) and isinstance(v, (int, float))
}
return {}
def _dict_str_bool(value: object) -> dict[str, bool]:
if isinstance(value, dict):
return {
k: bool(v)
for k, v in value.items()
if isinstance(k, str) and isinstance(v, bool)
}
return {}
def _dict_any(value: object) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _parse_errors(result: dict[str, object]) -> dict[str, dict[str, str]]:
errors: dict[str, dict[str, str]] = {}
parse_error = result.get("parse_error")
if isinstance(parse_error, dict):
err_type = parse_error.get("type")
err_msg = parse_error.get("message")
if isinstance(err_type, str) and isinstance(err_msg, str):
errors["parse_error"] = {"type": err_type, "message": err_msg}
report_parse_error = result.get("report_parse_error")
if isinstance(report_parse_error, dict):
err_type = report_parse_error.get("type")
err_msg = report_parse_error.get("message")
if isinstance(err_type, str) and isinstance(err_msg, str):
errors["report_parse_error"] = {"type": err_type, "message": err_msg}
return errors
+168
View File
@@ -3,10 +3,12 @@ from __future__ import annotations
import argparse
import json
import sys
from dataclasses import dataclass
from pathlib import Path
from examples.agent_challenges.classification import extract_challenge_report
from examples.agent_challenges.opencode_io import parse_opencode_output, result_text
from examples.agent_challenges.report_models import TrialReport
def save_report(
@@ -198,3 +200,169 @@ def report_from_v2_result(result: dict[str, object]) -> str:
lines.append("")
return "\n".join(lines)
@dataclass(frozen=True, slots=True)
class TrialReportPaths:
markdown: Path
machine: Path
def _atomic_write_text(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(f".{path.name}.tmp")
temporary.write_text(text, encoding="utf-8")
temporary.replace(path)
def render_trial_report_markdown(report: TrialReport) -> str:
lines: list[str] = []
lines.append("# Trial Report")
lines.append("")
lines.append("## Outcome")
lines.append("")
o = report.outcome
lines.append(f"- Task outcome: {o.task_outcome}")
lines.append(f"- Evaluation validity: {o.evaluation_validity}")
lines.append(f"- Duration: {o.duration_seconds}s")
if o.returncode is not None:
lines.append(f"- Return code: {o.returncode}")
if o.assertion_failures:
lines.append("- Assertion failures:")
for af in o.assertion_failures:
lines.append(f" - {af}")
if o.parse_errors:
for key, err in o.parse_errors.items():
lines.append(
f"- Parse error ({key}): {err.get('type', '')} - {err.get('message', '')}"
)
lines.append("")
lines.append("## Agent Self-Report")
lines.append("")
if report.agent_self_report is not None:
lines.append("```yaml")
for key, value in report.agent_self_report.items():
lines.append(f"{key}: {value}")
lines.append("```")
else:
lines.append("No agent self-report captured.")
if report.final_agent_answer:
lines.append("")
lines.append("Final agent answer:")
lines.append("")
lines.append(report.final_agent_answer)
lines.append("")
lines.append("## Commands And Tool Calls")
lines.append("")
if report.commands_and_tools:
for cmd in report.commands_and_tools:
parts = [
f"{cmd.ordinal}. **{cmd.tool}** ({cmd.status})",
f" - Title: {cmd.title}",
]
if cmd.detail:
parts.append(f" - Detail: `{cmd.detail}`")
parts.append(
f" - Output: {cmd.output_chars} chars, sha256: `{cmd.output_sha256}`"
)
lines.extend(parts)
else:
lines.append("No commands or tool calls recorded.")
lines.append("")
lines.append("## Automatic Evidence")
lines.append("")
ev = report.automatic_evidence
lines.append(f"- Steps: {ev.step_count}")
lines.append(
f"- Tool calls: {ev.tool_call_count} ({ev.failed_tool_call_count} failed)"
)
if ev.tool_counts:
for tool, count in sorted(ev.tool_counts.items()):
lines.append(f" - {tool}: {count}")
lines.append(
f"- Tokens: total={ev.tokens.total}, input={ev.tokens.input}, output={ev.tokens.output}, reasoning={ev.tokens.reasoning}, cache_read={ev.tokens.cache_read}, cache_write={ev.tokens.cache_write}"
)
lines.append(f"- Cost: {ev.cost}")
if ev.unknown_event_count:
lines.append(f"- Unknown events: {ev.unknown_event_count}")
if ev.reads_by_category:
for category, paths in sorted(ev.reads_by_category.items()):
lines.append(f"- {category}: {len(paths)} path(s)")
if ev.disallowed_reads:
lines.append(f"- Disallowed reads: {len(ev.disallowed_reads)} path(s)")
if ev.opaque_shell_commands:
lines.append(
f"- Opaque shell commands: {len(ev.opaque_shell_commands)} command(s)"
)
if ev.escalated_to_product_code:
lines.append("- Escalated to product code: yes")
lines.append("")
lines.append("## Policy Findings")
lines.append("")
if report.policy_findings:
for pf in report.policy_findings:
lines.append(f"- {pf}")
else:
lines.append("No policy findings.")
lines.append("")
lines.append("## Self-Report Discrepancies")
lines.append("")
if report.self_report_discrepancies:
for sd in report.self_report_discrepancies:
lines.append(f"- {sd}")
else:
lines.append("No discrepancies detected.")
lines.append("")
lines.append("## Manual Audit")
lines.append("")
ma = report.manual_audit
lines.append(f"- Status: {ma.status}")
if ma.official_outcome is not None:
lines.append(f"- Official outcome: {ma.official_outcome}")
if ma.auditor is not None:
lines.append(f"- Auditor: {ma.auditor}")
if ma.audited_at is not None:
lines.append(f"- Audited at: {ma.audited_at}")
if ma.corrections:
for c in ma.corrections:
lines.append(f"- Correction: {c}")
if ma.notes:
lines.append(f"- Notes: {ma.notes}")
if ma.read_flags:
for key, val in ma.read_flags.items():
lines.append(f"- read.{key}: {val}")
lines.append("")
lines.append("## Follow-Up Notes")
lines.append("")
if report.follow_up_notes:
for fn in report.follow_up_notes:
lines.append(f"- {fn}")
else:
lines.append("No follow-up notes.")
lines.append("")
return "\n".join(lines)
def write_trial_report_projections(
report: TrialReport,
*,
markdown_path: Path,
machine_path: Path,
) -> TrialReportPaths:
machine = (
json.dumps(report.model_dump(mode="json"), indent=2, sort_keys=True) + "\n"
)
markdown = render_trial_report_markdown(report).rstrip() + "\n"
_atomic_write_text(machine_path, machine)
_atomic_write_text(markdown_path, markdown)
return TrialReportPaths(markdown=markdown_path, machine=machine_path)
+2
View File
@@ -101,6 +101,8 @@ def main(argv: list[str] | None = None) -> int:
"task_outcome": result["task_outcome"],
"evaluation_validity": result["evaluation_validity"],
"duration_seconds": result["duration_seconds"],
"result_path": result.get("result_path"),
"report_paths": result.get("report_paths"),
}
)
print(json.dumps(summaries[-1], sort_keys=True))
+44 -5
View File
@@ -20,8 +20,12 @@ from examples.agent_challenges.opencode_io import ( # noqa: E402
parse_opencode_output,
result_text,
)
from examples.agent_challenges.report_models import ( # noqa: E402
build_trial_report,
)
from examples.agent_challenges.reports import ( # noqa: E402
save_report_from_result_payload,
write_trial_report_projections,
)
from examples.agent_challenges.workspace import ( # noqa: E402
ChallengeDef,
@@ -575,7 +579,23 @@ def run_v2_trial(
if assertion_failures:
task_outcome = "failed"
results_dir.mkdir(parents=True, exist_ok=True)
result_path = (
results_dir
/ f"{model.replace('/', '_').replace(':', '_')}-trial-{index:03d}.json"
)
workspace_root_str = str(workspace.root.resolve())
result_path_str = str(result_path.resolve())
markdown_path = workspace.root / "final-report.md"
machine_report_path = result_path.with_suffix(".report.json")
report_paths = {
"markdown": str(markdown_path.resolve()),
"machine": str(machine_report_path.resolve()),
}
result: dict[str, Any] = {
"challenge_id": challenge.manifest.id,
"instruction_profile": profile.value,
"task_outcome": task_outcome,
"evaluation_validity": policy.validity.value,
@@ -591,6 +611,9 @@ def run_v2_trial(
"disallowed_reads": list(policy.disallowed_reads),
"escalated_to_product_code": policy.escalated_to_product_code,
"opaque_shell_commands": list(policy.opaque_shell_commands),
"reads_by_category": {
k: list(v) for k, v in policy.reads_by_category.items()
},
},
"repository_commit": _get_git_commit(),
"repository_dirty": _get_git_dirty(),
@@ -598,11 +621,15 @@ def run_v2_trial(
"index": index,
"model": model,
"variant": variant,
"trial_index": index,
"duration_seconds": round(duration_seconds, 3),
"returncode": returncode,
"stdout": stdout,
"stderr": stderr,
"parsed": parsed_output,
"workspace_path": workspace_root_str,
"result_path": result_path_str,
"report_paths": report_paths,
}
if assertion_failures:
@@ -614,13 +641,25 @@ def run_v2_trial(
if challenge_report is not None:
result["challenge_report"] = challenge_report
results_dir.mkdir(parents=True, exist_ok=True)
result_path = (
results_dir
/ f"{model.replace('/', '_').replace(':', '_')}-trial-{index:03d}.json"
)
result_path.write_text(
json.dumps(result, indent=2, sort_keys=True), encoding="utf-8"
)
report_generation_error: str | None = None
try:
trial_report = build_trial_report(
result,
audit=None,
raw_result_path=result_path_str,
workspace_path=workspace_root_str,
)
write_trial_report_projections(
trial_report,
markdown_path=markdown_path,
machine_path=machine_report_path,
)
except Exception as exc:
report_generation_error = str(exc)
result["report_generation_error"] = report_generation_error
return result
+89 -1
View File
@@ -1,22 +1,110 @@
"""Central CLI for saving agent challenge manual audits."""
"""Central CLI for saving agent challenge manual audits.
Routes to V1 or V2 audit logic based on the harness version in the result file.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
try:
from .audit import main as audit_main
from .audit import save_v2_manual_audit
except ImportError:
_project_root = Path(__file__).resolve().parents[2]
if str(_project_root) not in sys.path:
sys.path.insert(0, str(_project_root))
from examples.agent_challenges.audit import main as audit_main
from examples.agent_challenges.audit import save_v2_manual_audit
def _is_v2_result(result_path: Path) -> bool:
try:
result = json.loads(result_path.read_text(encoding="utf-8"))
return isinstance(result, dict) and result.get("harness_version") == "v2"
except json.JSONDecodeError, OSError:
return False
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--from-result", type=Path, required=True)
parser.add_argument("--from-report", type=Path, default=None)
parser.add_argument("--manual-classification", required=True)
parser.add_argument("--auditor", default="human")
parser.add_argument("--audited-at", default=None)
parser.add_argument("--set-read", action="append", default=[])
parser.add_argument("--set-evidence", action="append", default=[])
parser.add_argument("--correction", action="append", default=[])
parser.add_argument("--notes", default="")
parser.add_argument("--output-name", default="manual-audit.yaml")
args = parser.parse_args(argv)
result_path = args.from_result
if _is_v2_result(result_path):
if args.from_report is not None:
parser.error("--from-report is not supported for V2 results")
read_overrides = dict(_parse_bool_assignment(item) for item in args.set_read)
evidence_overrides = dict(
_parse_value_assignment(item) for item in args.set_evidence
)
try:
paths = save_v2_manual_audit(
result_path,
official_outcome=args.manual_classification,
auditor=args.auditor,
audited_at=args.audited_at,
read_overrides=read_overrides,
evidence_overrides=evidence_overrides,
corrections=list(args.correction),
notes=args.notes,
)
print(
json.dumps(
{
"audit": paths.audit.as_posix(),
"markdown": paths.markdown.as_posix(),
"machine": paths.machine.as_posix(),
}
)
)
return 0
except ValueError as exc:
parser.error(str(exc))
return audit_main(argv)
def _parse_bool_assignment(value: str) -> tuple[str, bool]:
key, separator, raw = value.partition("=")
if separator != "=" or not key:
raise ValueError("expected KEY=true or KEY=false")
lowered = raw.lower()
if lowered == "true":
return key, True
if lowered == "false":
return key, False
raise ValueError("boolean override value must be true or false")
def _parse_value_assignment(value: str) -> tuple[str, object]:
key, separator, raw = value.partition("=")
if separator != "=" or not key:
raise ValueError("expected KEY=VALUE")
lowered = raw.lower()
if lowered == "true":
return key, True
if lowered == "false":
return key, False
try:
return key, int(raw)
except ValueError:
return key, raw
if __name__ == "__main__":
sys.exit(main())
@@ -1092,3 +1092,132 @@ def test_both_challenges_produce_different_challenge_hashes_but_same_base(
assert hashes["browser_none"] != hashes["report_none"]
for profile in InstructionProfile:
assert hashes[f"browser_{profile.value}"] != hashes[f"report_{profile.value}"]
def test_runner_to_report_success(tmp_path: Path) -> None:
from examples.agent_challenges.runner import run_v2_trial
challenge = load_challenge_manifest(_write_manifest(tmp_path / "challenge"))
bundle = ROOT / "examples/agent_challenges/instruction_bundles/workflow_cli.yaml"
workspaces_dir = tmp_path / "workspaces"
results_dir = tmp_path / "results"
results_dir.mkdir()
agent_answer = "Deployment dep_abc created."
stdout_jsonl = "\n".join(
[
json.dumps({"type": "step_start", "step": 1}),
json.dumps(
{
"type": "step_finish",
"tokens": {"total": 50, "input": 20, "output": 15},
"cost": 0.002,
}
),
json.dumps(
{
"type": "text",
"part": {"text": f"Final answer: {agent_answer}"},
}
),
]
)
def fake_run(
command: list[str],
*,
cwd: str,
text: bool,
capture_output: bool,
timeout: float | None,
check: bool,
) -> object:
return type(
"Result",
(),
{"returncode": 0, "stdout": stdout_jsonl, "stderr": ""},
)()
result = run_v2_trial(
challenge,
profile=InstructionProfile.NONE,
model="test-model",
variant="high",
index=1,
workspaces_dir=workspaces_dir,
results_dir=results_dir,
instruction_bundle=bundle,
run_fn=fake_run,
)
assert result["challenge_id"] == "fixture"
assert isinstance(result["workspace_path"], str)
assert isinstance(result["result_path"], str)
assert isinstance(result["report_paths"], dict)
assert "markdown" in result["report_paths"]
assert "machine" in result["report_paths"]
raw_path = Path(result["result_path"])
md_path = Path(result["report_paths"]["markdown"])
machine_path = Path(result["report_paths"]["machine"])
assert raw_path.is_file()
assert md_path.is_file()
assert machine_path.is_file()
machine = json.loads(machine_path.read_text(encoding="utf-8"))
assert machine["identity"]["challenge_id"] == "fixture"
md = md_path.read_text(encoding="utf-8")
assert agent_answer in md
def test_runner_to_report_timeout(tmp_path: Path) -> None:
import subprocess as sp
from examples.agent_challenges.runner import run_v2_trial
challenge = load_challenge_manifest(_write_manifest(tmp_path / "challenge"))
bundle = ROOT / "examples/agent_challenges/instruction_bundles/workflow_cli.yaml"
workspaces_dir = tmp_path / "workspaces"
results_dir = tmp_path / "results"
results_dir.mkdir()
def fake_run(
command: list[str],
*,
cwd: str,
text: bool,
capture_output: bool,
timeout: float | None,
check: bool,
) -> object:
raise sp.TimeoutExpired(cmd=command, timeout=3600)
result = run_v2_trial(
challenge,
profile=InstructionProfile.NONE,
model="test-model",
variant="high",
index=1,
workspaces_dir=workspaces_dir,
results_dir=results_dir,
instruction_bundle=bundle,
run_fn=fake_run,
)
assert result["task_outcome"] == "timeout"
assert isinstance(result.get("workspace_path"), str)
assert isinstance(result.get("report_paths"), dict)
BASE_PROMPT = ROOT / "examples/agent_challenges/base-prompt.md"
def test_base_prompt_mentions_self_report_rules() -> None:
text = BASE_PROMPT.read_text(encoding="utf-8")
assert "tests/" in text or "tests" in text
assert "examples/" in text or "examples" in text
assert "read.product_code" in text
assert "read.existing_solution" in text
assert "read.adjacent_attempts" in text
@@ -0,0 +1,412 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
def _raw_result(tmp_path: Path) -> dict[str, object]:
return {
"instruction_profile": "none",
"task_outcome": "success",
"evaluation_validity": "clean",
"challenge_id": "fixture",
"model": "test-model",
"variant": "high",
"trial_index": 1,
"prompt_hashes": {
"base": "abc",
"profile": "def",
"challenge": "ghi",
"rendered": "jkl",
},
"repository_commit": "abc123def",
"repository_dirty": False,
"result_path": str(tmp_path / "results" / "trial.json"),
"workspace_path": str(tmp_path),
"metrics": {
"step_count": 2,
"tool_call_count": 3,
"failed_tool_call_count": 0,
"tool_counts": {"bash": 2, "read": 1},
"tokens": {
"total": 500,
"input": 200,
"output": 200,
"reasoning": 50,
"cache_read": 50,
"cache_write": 0,
},
"cost": 0.025,
"unknown_event_count": 0,
"tool_calls": [
{
"ordinal": 1,
"call_id": "c1",
"tool": "read",
"status": "success",
"title": "Read workflow plan",
"input": {"path": str(tmp_path / "workflow.plan.json")},
"metadata": {},
"output_chars": 500,
"output_preview": "full tool output",
"output_sha256": "abc",
"failed": False,
},
{
"ordinal": 2,
"call_id": "c2",
"tool": "bash",
"status": "success",
"title": "Run workflow",
"input": {"command": "uv run wf status"},
"metadata": {},
"output_chars": 200,
"output_preview": "large raw stream",
"output_sha256": "def",
"failed": False,
},
],
},
"policy": {
"validity": "clean",
"disallowed_reads": [],
"escalated_to_product_code": False,
"opaque_shell_commands": [],
"reads_by_category": {
"workspace": [str(tmp_path / "workflow.plan.json")],
},
},
"stdout": "large raw stdout content that should not appear in bounded report\n"
* 1000,
"stderr": "",
"parsed": {"text": "The deployment succeeded with id dep_123."},
"assertion_failures": [],
"parse_errors": {},
}
def test_trial_report_is_bounded_machine_projection(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
payload = build_trial_report(_raw_result(tmp_path), audit=None).model_dump(
mode="json"
)
assert payload["schema_version"] == 1
assert payload["identity"]["challenge_id"] == "fixture"
assert payload["identity"]["raw_result_path"] == str(
tmp_path / "results" / "trial.json"
)
assert payload["identity"]["workspace_path"] == str(tmp_path)
assert payload["outcome"]["task_outcome"] == "success"
assert payload["commands_and_tools"][0]["detail"].endswith("workflow.plan.json")
serialized = json.dumps(payload)
assert "large raw stream" not in serialized
assert "full tool output" not in serialized
assert payload["commands_and_tools"][0]["output_chars"] == 500
assert payload["commands_and_tools"][0]["output_sha256"] == "abc"
assert payload["manual_audit"]["status"] == "pending"
def test_command_brief_supports_filepath(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
result = _raw_result(tmp_path)
metrics = result.get("metrics", {})
if isinstance(metrics, dict):
metrics["tool_calls"] = [
{
"ordinal": 1,
"call_id": "c3",
"tool": "read",
"status": "success",
"title": "Read source file",
"input": {"filePath": str(tmp_path / "src" / "app.py")},
"metadata": {},
"output_chars": 100,
"output_preview": "content",
"output_sha256": "xyz",
"failed": False,
},
]
payload = build_trial_report(result, audit=None).model_dump(mode="json")
cmd = payload["commands_and_tools"][0]
assert cmd["detail"] is not None
assert cmd["detail"].endswith("app.py")
def test_markdown_projection_has_stable_headings(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
from examples.agent_challenges.reports import render_trial_report_markdown
report = build_trial_report(_raw_result(tmp_path), audit=None)
md = render_trial_report_markdown(report)
expected_headings = [
"# Trial Report",
"## Outcome",
"## Agent Self-Report",
"## Commands And Tool Calls",
"## Automatic Evidence",
"## Policy Findings",
"## Self-Report Discrepancies",
"## Manual Audit",
"## Follow-Up Notes",
]
for heading in expected_headings:
assert heading in md, f"Missing heading: {heading}"
assert "Final agent answer" in md
assert "The deployment succeeded" in md
assert "large raw stream" not in md
assert "full tool output" not in md
def test_projections_write_both_files(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
from examples.agent_challenges.reports import (
TrialReportPaths,
write_trial_report_projections,
)
report = build_trial_report(_raw_result(tmp_path), audit=None)
markdown_path = tmp_path / "final-report.md"
machine_path = tmp_path / "trial.report.json"
paths = write_trial_report_projections(
report,
markdown_path=markdown_path,
machine_path=machine_path,
)
assert isinstance(paths, TrialReportPaths)
assert markdown_path.is_file()
assert machine_path.is_file()
md_content = markdown_path.read_text(encoding="utf-8")
assert "# Trial Report" in md_content
assert "## Outcome" in md_content
machine_content = json.loads(machine_path.read_text(encoding="utf-8"))
assert machine_content["schema_version"] == 1
assert machine_content["manual_audit"]["status"] == "pending"
assert list(tmp_path.iterdir()) == [markdown_path, machine_path]
def test_projections_exclude_raw_outputs(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
from examples.agent_challenges.reports import write_trial_report_projections
report = build_trial_report(_raw_result(tmp_path), audit=None)
markdown_path = tmp_path / "final-report.md"
machine_path = tmp_path / "trial.report.json"
write_trial_report_projections(
report,
markdown_path=markdown_path,
machine_path=machine_path,
)
md = markdown_path.read_text(encoding="utf-8")
assert "large raw stream" not in md
assert "full tool output" not in md
machine = json.loads(machine_path.read_text(encoding="utf-8"))
serialized = json.dumps(machine)
assert "large raw stream" not in serialized
assert "full tool output" not in serialized
def _write_v2_result(tmp_path: Path) -> Path:
result_dir = tmp_path / "results"
result_dir.mkdir(parents=True)
result = _raw_result(tmp_path)
result["harness_version"] = "v2"
result["result_path"] = str(tmp_path / "results" / "trial.json")
result["workspace_path"] = str(tmp_path)
path = result_dir / "trial.json"
path.write_text(json.dumps(result, indent=2, sort_keys=True), encoding="utf-8")
return path
def _write_v2_result_with_projections(tmp_path: Path) -> Path:
result_path = _write_v2_result(tmp_path)
from examples.agent_challenges.report_models import build_trial_report
from examples.agent_challenges.reports import write_trial_report_projections
result = json.loads(result_path.read_text(encoding="utf-8"))
report = build_trial_report(result, audit=None)
write_trial_report_projections(
report,
markdown_path=tmp_path / "final-report.md",
machine_path=tmp_path / "results" / "trial.report.json",
)
return result_path
def test_manual_audit_regenerates_projections(tmp_path: Path) -> None:
from examples.agent_challenges.audit import save_v2_manual_audit
result_path = _write_v2_result(tmp_path)
paths = save_v2_manual_audit(
result_path,
official_outcome="pass",
auditor="reviewer",
audited_at="2026-06-23T00:00:00Z",
read_overrides={"existing_solution": True},
corrections=["Agent inspected a ready-made workflow plan."],
notes="Technical run passed; self-report corrected.",
)
assert paths.audit.is_file()
assert paths.markdown.is_file()
assert paths.machine.is_file()
audit_yaml = paths.audit.read_text(encoding="utf-8")
assert "official_outcome: pass" in audit_yaml
assert "Agent inspected a ready-made workflow plan." in audit_yaml
md = paths.markdown.read_text(encoding="utf-8")
assert "Official outcome: pass" in md
assert "Agent inspected a ready-made workflow plan." in md
machine = json.loads(paths.machine.read_text(encoding="utf-8"))
assert machine["manual_audit"]["official_outcome"] == "pass"
assert machine["manual_audit"]["status"] == "complete"
def test_manual_audit_invalid_outcome_raises_and_preserves_projections(
tmp_path: Path,
) -> None:
from examples.agent_challenges.audit import save_v2_manual_audit
result_path = _write_v2_result_with_projections(tmp_path)
md_path = tmp_path / "final-report.md"
machine_path = tmp_path / "results" / "trial.report.json"
md_before = md_path.read_bytes()
machine_before = machine_path.read_bytes()
with pytest.raises(ValueError, match="official_outcome"):
save_v2_manual_audit(
result_path,
official_outcome="maybe",
auditor="reviewer",
)
assert md_path.read_bytes() == md_before
assert machine_path.read_bytes() == machine_before
def test_discrepancy_detects_run_failed_contradiction(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
result = _raw_result(tmp_path)
result["task_outcome"] = "failed"
result["challenge_report"] = {"run_failed": False, "used_product_path": True}
report = build_trial_report(result, audit=None)
assert any(
"run_failed=false" in d and "task_outcome is 'failed'" in d
for d in report.self_report_discrepancies
)
def test_discrepancy_detects_escalation_not_reported(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
result = _raw_result(tmp_path)
result["task_outcome"] = "success"
policy = result.get("policy", {})
if isinstance(policy, dict):
policy["escalated_to_product_code"] = True
result["challenge_report"] = {
"run_failed": False,
"read": {"product_code": False},
}
report = build_trial_report(result, audit=None)
assert any("read.product_code" in d for d in report.self_report_discrepancies)
def test_no_discrepancy_when_product_code_reported(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
result = _raw_result(tmp_path)
result["task_outcome"] = "success"
policy = result.get("policy", {})
if isinstance(policy, dict):
policy["escalated_to_product_code"] = True
result["challenge_report"] = {
"run_failed": False,
"read": {"product_code": True},
}
report = build_trial_report(result, audit=None)
assert not any("read.product_code" in d for d in report.self_report_discrepancies)
def test_example_read_does_not_create_discrepancy(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
result = _raw_result(tmp_path)
policy = result.get("policy", {})
if isinstance(policy, dict):
policy["reads_by_category"] = {
"examples": [str(tmp_path / "examples" / "solution.py")],
}
result["challenge_report"] = {
"run_failed": False,
"read": {"product_code": True, "existing_solution": False},
}
report = build_trial_report(result, audit=None)
assert not any("existing solution" in d for d in report.self_report_discrepancies)
assert any("example file(s)" in n for n in report.follow_up_notes)
def _policy_mut(result: dict[str, object]) -> dict[str, object]:
p = result.get("policy", {})
if not isinstance(p, dict):
p = {}
result["policy"] = p
return p
def test_follow_up_notes_for_example_reads(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
result = _raw_result(tmp_path)
_policy_mut(result)["reads_by_category"] = {
"examples": [str(tmp_path / "examples" / "solution.py")],
}
report = build_trial_report(result, audit=None)
assert any("example file(s)" in n for n in report.follow_up_notes)
def test_follow_up_notes_for_test_reads(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
result = _raw_result(tmp_path)
_policy_mut(result)["reads_by_category"] = {
"tests": [str(tmp_path / "tests" / "test_app.py")],
}
report = build_trial_report(result, audit=None)
assert any("test file(s)" in n for n in report.follow_up_notes)
def test_build_report_raises_on_missing_paths(tmp_path: Path) -> None:
from examples.agent_challenges.report_models import build_trial_report
result = _raw_result(tmp_path)
del result["result_path"]
del result["workspace_path"]
with pytest.raises(ValueError, match="raw_result_path"):
build_trial_report(result, audit=None)