test: harden agent challenge evidence reports

This commit is contained in:
lda
2026-06-24 17:52:29 +07:00 Verified
parent e4d163a0cf
commit 9c68bcd75f
15 changed files with 307 additions and 8 deletions
+5
View File
@@ -15,6 +15,11 @@ 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.
Do not read files under other `workspaces/*` trial directories. Prior trial
workspaces may contain complete answers. If you do read another trial workspace,
report `read.adjacent_attempts: true`; if the file is a workflow plan, patch, or
report for this challenge, also report `read.existing_solution: true`.
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
@@ -11,6 +11,15 @@ Discover the `local.browser_click` source capabilities through `wf cap list`,
`wf cap inspect`, and `wf schema`. Repository implementation inspection is
profile-controlled; do not read source files unless your profile permits it.
The writable trial workspace includes a safe fixture input file:
- `run-input.json` -- run input with `button_label`, `open_browser`,
`simulate`, and `timeout_seconds`.
You may use `run-input.json` with `wf run start --input-file`. Do not inspect
source implementation files to learn input behavior. Use `wf cap inspect` for
node contracts.
## Workflow Authoring Paths
Two product-facing authoring paths are acceptable:
@@ -1,2 +1,5 @@
.wf_browser_click_store/
*
!.gitignore
!run-input.json
!TASK_FILES.md
@@ -0,0 +1,11 @@
# Task Files
This workspace contains safe fixture inputs for the browser-click challenge.
These files are inputs, not workflow solutions.
- `run-input.json` contains the workflow input for a deterministic simulated
click run.
Use these files with `wf` commands from the rendered prompt. Do not inspect
source implementation files to learn the workflow behavior unless your
instruction profile permits it.
@@ -0,0 +1,6 @@
{
"button_label": "Launch Workflow",
"open_browser": false,
"simulate": true,
"timeout_seconds": 2
}
+30 -1
View File
@@ -32,6 +32,31 @@ class PolicyEvidence:
reads_by_category: dict[str, tuple[str, ...]]
SOLUTION_FILE_NAMES = {
"final-report.md",
"workflow.plan.json",
}
SOLUTION_FILE_SUFFIXES = (
".plan.json",
".report.json",
".report.md",
)
SOLUTION_FILE_PREFIXES = ("patch",)
def _looks_like_solution_artifact(path: Path) -> bool:
"""Return true for prior-trial files likely to contain a full answer."""
name = path.name
return (
name in SOLUTION_FILE_NAMES
or name.endswith(SOLUTION_FILE_SUFFIXES)
or any(
name.startswith(prefix) and name.endswith(".json")
for prefix in SOLUTION_FILE_PREFIXES
)
)
def _classify_path(
path_str: str,
*,
@@ -64,6 +89,8 @@ def _classify_path(
return "workspace"
if p.is_relative_to(workspaces):
if _looks_like_solution_artifact(p):
return "existing_solution"
return "adjacent_attempts"
if p.is_relative_to(repository):
@@ -161,7 +188,9 @@ def evaluate_policy(
)
reads_by_category.setdefault(category, []).append(path_str)
if profile == InstructionProfile.NONE:
if category == "existing_solution":
disallowed_reads.append(path_str)
elif profile == InstructionProfile.NONE:
if category not in ("workspace", "unknown"):
disallowed_reads.append(path_str)
elif profile == InstructionProfile.SKILLS:
@@ -317,6 +317,7 @@ def _build_self_report_discrepancies(
)
_check_escalation_discrepancy(result, agent_self_report, discrepancies)
_check_existing_solution_discrepancy(result, agent_self_report, discrepancies)
return discrepancies
@@ -342,6 +343,31 @@ def _check_escalation_discrepancy(
)
def _check_existing_solution_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
reads_raw = policy_raw.get("reads_by_category")
if not isinstance(reads_raw, dict):
return
existing_solution = reads_raw.get("existing_solution")
if not isinstance(existing_solution, (list, tuple)) or not existing_solution:
return
agent_read_raw = agent_self_report.get("read")
if isinstance(agent_read_raw, dict):
for key, value in agent_read_raw.items():
if key == "existing_solution" and value is True:
return
discrepancies.append(
"Agent read an existing solution reference but did not report "
"read.existing_solution"
)
def _build_follow_up_notes(
result: dict[str, object], evidence: AutomaticEvidence
) -> list[str]:
+4
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import argparse
import json
import sys
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path
@@ -362,6 +363,7 @@ def write_trial_report_projections(
*,
markdown_path: Path,
machine_path: Path,
extra_markdown_paths: Sequence[Path] = (),
) -> TrialReportPaths:
machine = (
json.dumps(report.model_dump(mode="json"), indent=2, sort_keys=True) + "\n"
@@ -369,4 +371,6 @@ def write_trial_report_projections(
markdown = render_trial_report_markdown(report).rstrip() + "\n"
_atomic_write_text(machine_path, machine)
_atomic_write_text(markdown_path, markdown)
for extra_path in extra_markdown_paths:
_atomic_write_text(extra_path, markdown)
return TrialReportPaths(markdown=markdown_path, machine=machine_path)
+67
View File
@@ -0,0 +1,67 @@
param(
[string]$AttachUrl = "http://127.0.0.1:8192",
[int]$Trials = 5,
[int]$TimeoutSeconds = 3600,
[object[]]$models = @( # object[] because ModelProfile is not known yet
[ModelProfile]::new("opencode/deepseek-v4-flash-free", "max"),
[ModelProfile]::new("opencode/mimo-v2.5-free", "high"),
[ModelProfile]::new("opencode/nemotron-3-ultra-free", "high")
)
)
class ModelProfile {
[string]$Model
[string]$Variant
ModelProfile([string]$model, [string]$variant) {
$this.Model = $model
$this.Variant = $variant
}
[string]ToString() {
return "$($this.Model) ($($this.Variant))"
}
ModelProfile() {}
}
function New-ModelProfile([string]$model, [string]$variant) {
return [ModelProfile]::new($model, $variant)
}
$ErrorActionPreference = "Stop"
[ModelProfile[]]$models = $models # cast should fail if there are any non-ModelProfile objects in the array
# Run from the repository root no matter where the script is invoked.
Set-Location (Get-Item $PSScriptRoot).Parent.Parent.FullName
$profiles = @(
"none",
"skills",
"all"
)
$challenges = @(
"examples/agent_challenges/browser_click_challenge/challenge.yaml",
"examples/agent_challenges/report_workflow_challenge/challenge.yaml"
)
foreach ($challenge in $challenges) {
foreach ($challengeProfile in $profiles) {
foreach ($model in $models) {
Write-Host ""
Write-Host "==> challenge=$challenge profile=$challengeProfile model=$model trials=$Trials"
uv run python examples/agent_challenges/run_trials.py `
--challenge $challenge `
--instruction-profile $challengeProfile `
--model $model.Model `
--variant $model.Variant `
--trials $Trials `
--attach $AttachUrl `
--timeout-seconds $TimeoutSeconds
}
}
}
Pop-Location
+3
View File
@@ -588,8 +588,10 @@ def run_v2_trial(
markdown_path = workspace.root / "final-report.md"
machine_report_path = result_path.with_suffix(".report.json")
results_markdown_path = result_path.with_suffix(".report.md")
report_paths = {
"markdown": str(markdown_path.resolve()),
"results_markdown": str(results_markdown_path.resolve()),
"machine": str(machine_report_path.resolve()),
}
@@ -657,6 +659,7 @@ def run_v2_trial(
trial_report,
markdown_path=markdown_path,
machine_path=machine_report_path,
extra_markdown_paths=[results_markdown_path],
)
except Exception as exc:
report_generation_error = str(exc)
+33 -6
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
from copy import deepcopy
from enum import StrEnum
from typing import Any
@@ -110,12 +111,12 @@ def wrapper_hints_for_capability(
output_schema, output_properties
)
output_map = {name: f"state.{name}" for name in sorted(output_map_properties)}
mapped_output_schema = {
"type": "object",
"properties": {
mapped_output_schema = _schema_with_local_definitions(
output_schema,
properties={
name: schema for name, schema in sorted(output_map_properties.items())
},
}
)
# The minimal wrapper stores mapped outputs in state and returns the same
# fields. Split this later only when wrappers support separate return shape.
state_schema = mapped_output_schema
@@ -175,7 +176,10 @@ def workflow_output_schema_for_authoring(output_schema: JsonObject) -> JsonObjec
``content[0].text`` or handle resources/images, but the authoring surface
must not invent that decision as a top-level schema field.
"""
return {"type": "object", "properties": _object_properties(output_schema)}
return _schema_with_local_definitions(
output_schema,
properties=_object_properties(output_schema),
)
def _object_properties(schema: JsonObject) -> dict[str, JsonObject]:
@@ -184,12 +188,35 @@ def _object_properties(schema: JsonObject) -> dict[str, JsonObject]:
if not isinstance(properties, dict):
return {}
return {
str(name): value
str(name): deepcopy(value)
for name, value in properties.items()
if isinstance(value, dict)
}
def _schema_with_local_definitions(
schema: JsonObject,
*,
properties: dict[str, JsonObject],
) -> JsonObject:
"""Build an object schema while preserving local reusable definitions.
Capability schemas commonly use refs such as ``{"$ref":
"#/$defs/Snapshot"}``. Wrapper hints project selected top-level fields into
workflow state/output schemas, but those refs become invalid unless the
local definition block is copied with them.
"""
projected: JsonObject = {
"type": "object",
"properties": deepcopy(properties),
}
for definition_key in ("$defs", "definitions"):
definitions = schema.get(definition_key)
if isinstance(definitions, dict):
projected[definition_key] = deepcopy(definitions)
return projected
def _has_raw_mcp_content(schema: JsonObject) -> bool:
"""Return true when schema exposes MCP's raw content-block envelope."""
properties = _object_properties(schema)
@@ -494,6 +494,47 @@ def test_policy_evidence_classifies_reads(tmp_path: Path) -> None:
assert wf_policy.opaque_shell_commands == ()
def test_policy_classifies_sibling_workspace_solution_reads(tmp_path: Path) -> None:
from examples.agent_challenges.metrics import ToolCallEvidence
from examples.agent_challenges.policy import evaluate_policy
workspace_root = tmp_path / "workspaces" / "trial-002"
workspace_root.mkdir(parents=True)
repository_root = tmp_path / "repo"
repository_root.mkdir()
workspaces_root = tmp_path / "workspaces"
solution_path = workspaces_root / "trial-001" / "workflow.plan.json"
notes_path = workspaces_root / "trial-001" / "notes.txt"
def _read(path: Path) -> ToolCallEvidence:
return ToolCallEvidence(
ordinal=1,
call_id="c1",
tool="read",
status="success",
title="read",
input={"path": str(path)},
metadata={},
output_chars=100,
output_preview="",
output_sha256="abc",
failed=False,
)
policy = evaluate_policy(
"all",
[_read(solution_path), _read(notes_path)],
workspace_root=workspace_root,
repository_root=repository_root,
workspaces_root=workspaces_root,
)
assert policy.validity.value == "contaminated"
assert policy.disallowed_reads == (str(solution_path),)
assert policy.reads_by_category["existing_solution"] == (str(solution_path),)
assert policy.reads_by_category["adjacent_attempts"] == (str(notes_path),)
def test_policy_reads_real_filepath_input(tmp_path: Path) -> None:
from examples.agent_challenges.metrics import ToolCallEvidence
from examples.agent_challenges.policy import evaluate_policy
@@ -1376,14 +1417,17 @@ def test_runner_to_report_success(tmp_path: Path) -> None:
assert isinstance(result["result_path"], str)
assert isinstance(result["report_paths"], dict)
assert "markdown" in result["report_paths"]
assert "results_markdown" in result["report_paths"]
assert "machine" in result["report_paths"]
raw_path = Path(result["result_path"])
md_path = Path(result["report_paths"]["markdown"])
results_md_path = Path(result["report_paths"]["results_markdown"])
machine_path = Path(result["report_paths"]["machine"])
assert raw_path.is_file()
assert md_path.is_file()
assert results_md_path.is_file()
assert machine_path.is_file()
machine = json.loads(machine_path.read_text(encoding="utf-8"))
@@ -1391,6 +1435,7 @@ def test_runner_to_report_success(tmp_path: Path) -> None:
md = md_path.read_text(encoding="utf-8")
assert agent_answer in md
assert results_md_path.read_text(encoding="utf-8") == md
def test_runner_to_report_timeout(tmp_path: Path) -> None:
@@ -396,6 +396,28 @@ def test_example_read_does_not_create_discrepancy(tmp_path: Path) -> None:
assert any("example file(s)" in n for n in report.follow_up_notes)
def test_existing_solution_read_creates_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"] = {
"existing_solution": [
str(tmp_path / "workspaces" / "trial-001" / "workflow.plan.json")
],
}
result["challenge_report"] = {
"run_failed": False,
"read": {"existing_solution": False},
}
report = build_trial_report(result, audit=None)
assert any("read.existing_solution" in d for d in report.self_report_discrepancies)
assert any("existing solution" 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):
@@ -1087,6 +1087,16 @@ def test_browser_click_workspace_uses_generic_profile_copy(tmp_path: Path) -> No
assert workspace.config_path.is_file()
assert (workspace.root / ".agent/skills/wf-cli/SKILL.md").is_file()
run_input = json.loads((workspace.root / "run-input.json").read_text())
assert run_input == {
"button_label": "Launch Workflow",
"open_browser": False,
"simulate": True,
"timeout_seconds": 2,
}
assert "safe fixture inputs" in (workspace.root / "TASK_FILES.md").read_text(
encoding="utf-8"
)
def test_central_runner_accepts_browser_challenge() -> None:
@@ -154,6 +154,38 @@ def test_wrapper_hints_mark_nested_outputs_as_low_confidence() -> None:
assert dumped["output_map"] == {"results": "state.results"}
def test_wrapper_hints_preserve_local_defs_for_mapped_output_refs() -> None:
hints = wrapper_hints_for_capability(
capability_name="local.browser_click.open_click_page",
input_schema={"type": "object", "properties": {}},
output_schema={
"type": "object",
"$defs": {
"Snapshot": {
"type": "object",
"properties": {"clicked": {"type": "boolean"}},
}
},
"properties": {
"before": {"$ref": "#/$defs/Snapshot"},
"session_id": {"type": "string"},
},
"required": ["before", "session_id"],
},
outcomes=["ok"],
)
dumped = hints.model_dump(mode="json")
assert dumped["state_schema"]["properties"]["before"] == {
"$ref": "#/$defs/Snapshot"
}
assert dumped["state_schema"]["$defs"]["Snapshot"]["properties"]["clicked"] == {
"type": "boolean"
}
assert dumped["output_schema"]["$defs"] == dumped["state_schema"]["$defs"]
def test_wrapper_hints_do_not_auto_map_raw_mcp_content_blocks() -> None:
hints = wrapper_hints_for_capability(
capability_name="everything.default.echo",