lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: summarize agent challenge matrix runs

Browse Source
This commit is contained in:
lda
2026-06-25 04:03:40 +07:00 Verified
parent 6a14d0fd03
commit 8e93e3dd66
7 changed files with 425 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/current_roadmap.md
+3
View File
@@ -233,6 +233,9 @@ stable.
execution, instruction profiles, manual audit, and the distinction between
evaluation validity and policy coverage:
[`agent challenge evaluation`](runbooks/agent-challenge-evaluation.md).
- Completed: challenge matrix operations now have compact OpenCode thread
titles, policy handling for canonical skill-document reads, and a central
`summarize_trials.py` command for audited result tables.
## Historical References
docs/runbooks/agent-challenge-evaluation.md
+16
View File
@@ -186,6 +186,22 @@ disqualifying evidence is found. Use `fail` when the task was not completed.
Use `invalid` when the workflow ran but the evaluation is contaminated,
bypassed, or otherwise not usable as clean benchmark evidence.
## Summarize Audited Results
After manual audits, generate a compact matrix table from the bounded report
projections:
```powershell
uv run python examples/agent_challenges/summarize_trials.py `
examples/agent_challenges/browser_click_challenge `
examples/agent_challenges/report_workflow_challenge
```
The table uses `manual_audit.official_outcome` when present, while keeping the
automatic task outcome, policy validity, duration, token count, attempt count,
and read flags visible. Use it as a working operator summary, not as a
statistical claim by itself.
## Common Invalid Patterns
Mark the trial invalid or at least contaminated when any of these happen:
examples/agent_challenges/policy.py
+5
View File
@@ -101,8 +101,12 @@ def _classify_path(
if p.is_relative_to(repository):
rel = p.relative_to(repository)
parts = rel.parts
if not parts:
return "repository_index"
if parts and parts[0] == ".wf_store":
return "prior_store"
if parts and parts[0] in (".agent", "skills"):
return "supplied_skills"
if parts and parts[0] == "tests":
return "tests"
if parts and parts[0] == "src":
@@ -175,6 +179,7 @@ def evaluate_policy(
reads_by_category: dict[str, list[str]] = {}
escalated_to_product_code = False
allowed_skills_categories = {
"repository_index",
"workspace",
"supplied_skills",
"search_intent",
examples/agent_challenges/runner.py
+20
View File
@@ -40,6 +40,19 @@ from examples.agent_challenges.workspace import ( # noqa: E402
)
def _opencode_trial_title(
*, challenge_id: str, model: str, profile: str, index: int
) -> str:
"""Build a compact OpenCode session title for crowded trial matrices."""
challenge_name = {
"browser_click": "browser",
"report_workflow": "report",
}.get(challenge_id, challenge_id)
model_name = model.rsplit("/", 1)[-1].replace("-v4-flash-free", "")
model_name = model_name.replace("-v2.5-free", "").replace("-3-ultra-free", "")
return f"{challenge_name} {model_name} {profile} {index:03d}"
@dataclass(slots=True)
class ManagedServer:
process: subprocess.Popen[str]
@@ -449,6 +462,13 @@ def run_v2_trial(
command.extend(
[
rendered.text,
"--title",
_opencode_trial_title(
challenge_id=challenge.manifest.id,
model=model,
profile=profile.value,
index=index,
),
"--format",
"json",
"--model",
examples/agent_challenges/summarize_trials.py
+208
View File
@@ -0,0 +1,208 @@
from __future__ import annotations
import argparse
import json
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[2]
DEFAULT_CHALLENGES_ROOT = ROOT / "examples" / "agent_challenges"
@dataclass(frozen=True, slots=True)
class TrialSummary:
challenge: str
model: str
profile: str
trial: int
manual: str
task: str
validity: str
duration_seconds: float
tokens_total: int
attempts: str
read_flags: str
notes: str
def _short_model(model: str) -> str:
name = model.rsplit("/", 1)[-1]
for suffix in ("-v4-flash-free", "-v2.5-free", "-3-ultra-free"):
name = name.replace(suffix, "")
return name
def _short_challenge(challenge: str) -> str:
return {
"browser_click": "browser",
"report_workflow": "report",
}.get(challenge, challenge)
def _string(value: object, default: str = "") -> str:
return value if isinstance(value, str) else default
def _int(value: object, default: int = 0) -> int:
return value if isinstance(value, int) else default
def _float(value: object, default: float = 0.0) -> float:
return float(value) if isinstance(value, int | float) else default
def _dict(value: object) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _attempts(agent_self_report: dict[str, Any]) -> str:
attempts = _dict(agent_self_report.get("attempts"))
total = attempts.get("total")
failed = attempts.get("failed")
if isinstance(total, int) and isinstance(failed, int):
return f"{failed}/{total}"
return ""
def _read_flags(agent_self_report: dict[str, Any]) -> str:
reads = _dict(agent_self_report.get("read"))
enabled = [key for key, value in sorted(reads.items()) if value is True]
return ",".join(enabled)
def load_trial_summary(path: Path) -> TrialSummary:
data = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(data, dict):
raise ValueError(f"report must be a JSON object: {path}")
identity = _dict(data.get("identity"))
outcome = _dict(data.get("outcome"))
manual_audit = _dict(data.get("manual_audit"))
evidence = _dict(data.get("automatic_evidence"))
tokens = _dict(evidence.get("tokens"))
self_report = _dict(data.get("agent_self_report"))
challenge = _short_challenge(_string(identity.get("challenge_id")))
model = _short_model(_string(identity.get("model")))
manual = _string(manual_audit.get("official_outcome"), "pending")
notes = _string(manual_audit.get("notes"))
return TrialSummary(
challenge=challenge,
model=model,
profile=_string(identity.get("instruction_profile")),
trial=_int(identity.get("trial_index")),
manual=manual or "pending",
task=_string(outcome.get("task_outcome")),
validity=_string(outcome.get("evaluation_validity")),
duration_seconds=_float(outcome.get("duration_seconds")),
tokens_total=_int(tokens.get("total")),
attempts=_attempts(self_report),
read_flags=_read_flags(self_report),
notes=" ".join(notes.split()),
)
def find_report_files(paths: list[Path]) -> list[Path]:
roots = paths or sorted(DEFAULT_CHALLENGES_ROOT.glob("*_challenge"))
reports: list[Path] = []
for root in roots:
if root.is_file():
reports.append(root)
continue
results_dir = root / "results"
if results_dir.is_dir():
reports.extend(sorted(results_dir.glob("*.report.json")))
continue
reports.extend(sorted(root.glob("*.report.json")))
return sorted(set(reports))
def _markdown_escape(value: object) -> str:
text = str(value)
return text.replace("|", "\\|").replace("\n", " ")
def render_markdown(summaries: list[TrialSummary]) -> str:
headers = [
"challenge",
"model",
"profile",
"trial",
"manual",
"task",
"validity",
"minutes",
"tokens",
"attempts",
"reads",
"notes",
]
rows = [
[
item.challenge,
item.model,
item.profile,
f"{item.trial:03d}",
item.manual,
item.task,
item.validity,
f"{item.duration_seconds / 60:.1f}",
str(item.tokens_total),
item.attempts,
item.read_flags,
item.notes,
]
for item in summaries
]
lines = [
"| " + " | ".join(headers) + " |",
"| " + " | ".join("---" for _ in headers) + " |",
]
lines.extend(
"| " + " | ".join(_markdown_escape(cell) for cell in row) + " |" for row in rows
)
return "\n".join(lines) + "\n"
def render_json(summaries: list[TrialSummary]) -> str:
return json.dumps(
[asdict(item) for item in summaries],
indent=2,
sort_keys=True,
)
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Summarize agent challenge report projections."
)
parser.add_argument(
"paths",
nargs="*",
type=Path,
help="Challenge directories, results directories, or .report.json files.",
)
parser.add_argument(
"--format",
choices=("markdown", "json"),
default="markdown",
help="Output format.",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = _parse_args(argv)
reports = find_report_files(args.paths)
summaries = [load_trial_summary(path) for path in reports]
if args.format == "json":
print(render_json(summaries))
else:
print(render_markdown(summaries), end="")
return 0
if __name__ == "__main__":
raise SystemExit(main())
tests/examples/test_agent_challenge_harness_v2.py
+72 -2
View File
@@ -447,10 +447,15 @@ def test_policy_evidence_classifies_reads(tmp_path: Path) -> None:
)
source_read = _tc("read", str(repository_root / "src" / "app.py"))
repository_index_read = _tc("read", str(repository_root))
skills_read = _tc(
"read",
str(workspace_root / ".agent" / "skills" / "wf-cli" / "SKILL.md"),
)
canonical_skills_read = _tc(
"read",
str(repository_root / "skills" / "wf-cli" / "SKILL.md"),
)
workspace_read = _tc("read", str(workspace_root / "attempt.md"))
test_read = _tc("read", str(repository_root / "tests" / "test_app.py"))
@@ -464,19 +469,58 @@ def test_policy_evidence_classifies_reads(tmp_path: Path) -> None:
assert none_policy.validity.value == "contaminated"
assert any("app.py" in p for p in none_policy.disallowed_reads)
none_skills_policy = evaluate_policy(
"none",
[repository_index_read, canonical_skills_read],
workspace_root=workspace_root,
repository_root=repository_root,
workspaces_root=workspaces_root,
)
assert none_skills_policy.validity.value == "contaminated"
skills_only_policy = evaluate_policy(
"skills",
[workspace_read, repository_index_read, skills_read, canonical_skills_read],
workspace_root=workspace_root,
repository_root=repository_root,
workspaces_root=workspaces_root,
)
assert skills_only_policy.validity.value == "clean"
assert not skills_only_policy.escalated_to_product_code
skills_policy = evaluate_policy(
"skills",
[workspace_read, skills_read, source_read, test_read],
[
workspace_read,
repository_index_read,
skills_read,
canonical_skills_read,
source_read,
test_read,
],
workspace_root=workspace_root,
repository_root=repository_root,
workspaces_root=workspaces_root,
)
assert skills_policy.validity.value == "contaminated"
assert not skills_policy.escalated_to_product_code
assert skills_policy.reads_by_category["repository_index"] == (
str(repository_root),
)
assert str(repository_root / "skills" / "wf-cli" / "SKILL.md") not in (
skills_policy.disallowed_reads
)
all_policy = evaluate_policy(
"all",
[workspace_read, skills_read, source_read, test_read],
[
workspace_read,
repository_index_read,
skills_read,
canonical_skills_read,
source_read,
test_read,
],
workspace_root=workspace_root,
repository_root=repository_root,
workspaces_root=workspaces_root,
@@ -915,6 +959,9 @@ def test_v2_runner_default_timeout_and_workspace_cwd(tmp_path: Path) -> None:
assert captured["timeout"] == 3600
assert isinstance(captured["cwd"], str)
command = captured["command"]
assert isinstance(command, list)
assert command[command.index("--title") + 1] == "fixture test-model none 001"
assert result["instruction_profile"] == "none"
assert "prompt_hashes" in result
assert "metrics" in result
@@ -933,6 +980,29 @@ def test_safe_model_name_replaces_windows_path_separators() -> None:
assert ".." not in safe
def test_opencode_trial_title_uses_short_matrix_labels() -> None:
from examples.agent_challenges.runner import _opencode_trial_title
assert (
_opencode_trial_title(
challenge_id="browser_click",
model="opencode/deepseek-v4-flash-free",
profile="skills",
index=4,
)
== "browser deepseek skills 004"
)
assert (
_opencode_trial_title(
challenge_id="report_workflow",
model="opencode/nemotron-3-ultra-free",
profile="all",
index=12,
)
== "report nemotron all 012"
)
def test_instruction_bundle_rejects_path_traversal(tmp_path: Path) -> None:
import yaml
tests/examples/test_agent_challenge_summary.py
+101
View File
@@ -0,0 +1,101 @@
from __future__ import annotations
import json
from pathlib import Path
def _write_report(
path: Path,
*,
challenge: str = "browser_click",
model: str = "opencode/deepseek-v4-flash-free",
profile: str = "skills",
trial: int = 4,
manual: str | None = "pass",
) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(
{
"identity": {
"challenge_id": challenge,
"model": model,
"variant": "high",
"instruction_profile": profile,
"trial_index": trial,
"raw_result_path": str(path.with_suffix(".json")),
"workspace_path": str(
path.parent.parent / "workspaces" / path.stem
),
},
"outcome": {
"task_outcome": "success",
"evaluation_validity": "clean",
"duration_seconds": 125.0,
"returncode": 0,
},
"automatic_evidence": {
"tokens": {"total": 12345},
},
"agent_self_report": {
"attempts": {"failed": 1, "total": 2},
"read": {
"skills": True,
"docs": False,
"product_code": False,
},
},
"manual_audit": {
"official_outcome": manual,
"notes": "Pass | with newline\nand spacing.",
},
}
),
encoding="utf-8",
)
return path
def test_load_trial_summary_uses_short_labels_and_manual_outcome(
tmp_path: Path,
) -> None:
from examples.agent_challenges.summarize_trials import load_trial_summary
report = _write_report(tmp_path / "results" / "trial.report.json")
summary = load_trial_summary(report)
assert summary.challenge == "browser"
assert summary.model == "deepseek"
assert summary.profile == "skills"
assert summary.trial == 4
assert summary.manual == "pass"
assert summary.duration_seconds == 125.0
assert summary.tokens_total == 12345
assert summary.attempts == "1/2"
assert summary.read_flags == "skills"
assert summary.notes == "Pass | with newline and spacing."
def test_render_markdown_escapes_table_cells(tmp_path: Path) -> None:
from examples.agent_challenges.summarize_trials import (
load_trial_summary,
render_markdown,
)
summary = load_trial_summary(_write_report(tmp_path / "one.report.json"))
markdown = render_markdown([summary])
assert "| browser | deepseek | skills | 004 | pass |" in markdown
assert "Pass \\| with newline and spacing." in markdown
def test_find_report_files_accepts_challenge_and_results_dirs(tmp_path: Path) -> None:
from examples.agent_challenges.summarize_trials import find_report_files
challenge = tmp_path / "browser_click_challenge"
report = _write_report(challenge / "results" / "trial.report.json")
assert find_report_files([challenge]) == [report]
assert find_report_files([challenge / "results"]) == [report]