2101 lines
64 KiB
Python
2101 lines
64 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from examples.agent_challenges.manifests import load_challenge_manifest
|
|
from examples.agent_challenges.models import InstructionProfile
|
|
from examples.agent_challenges.prompts import compose_trial_prompt
|
|
from examples.agent_challenges.workspace import prepare_v2_trial_workspace
|
|
|
|
|
|
def _write_manifest(root: Path) -> Path:
|
|
(root / "workspace_template").mkdir(parents=True)
|
|
(root / "challenge-prompt.md").write_text("Build it.\n", encoding="utf-8")
|
|
path = root / "challenge.yaml"
|
|
path.write_text(
|
|
"""\
|
|
version: 1
|
|
id: fixture
|
|
prompt: challenge-prompt.md
|
|
workspace_template: workspace_template
|
|
source:
|
|
id: local.fixture
|
|
root: source
|
|
module: ops
|
|
registry: registry
|
|
store_root: .wf_fixture_store
|
|
server:
|
|
config: wf.config.json
|
|
default_port: 8779
|
|
report:
|
|
required_fields: [value, run_failed]
|
|
success_assertions:
|
|
value: expected
|
|
run_failed: false
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def test_load_challenge_manifest_resolves_paths(tmp_path: Path) -> None:
|
|
manifest_path = _write_manifest(tmp_path)
|
|
|
|
loaded = load_challenge_manifest(manifest_path)
|
|
|
|
assert loaded.manifest.id == "fixture"
|
|
assert loaded.root == tmp_path.resolve()
|
|
assert loaded.prompt_path == (tmp_path / "challenge-prompt.md").resolve()
|
|
assert loaded.workspace_template == (tmp_path / "workspace_template").resolve()
|
|
assert loaded.manifest.report.success_assertions == {
|
|
"value": "expected",
|
|
"run_failed": False,
|
|
}
|
|
|
|
|
|
def test_instruction_profiles_are_exactly_the_supported_conditions() -> None:
|
|
assert [profile.value for profile in InstructionProfile] == [
|
|
"none",
|
|
"skills",
|
|
"all",
|
|
"debug",
|
|
]
|
|
|
|
|
|
def test_invalid_manifest_rejects_parent_traversal(tmp_path: Path) -> None:
|
|
path = _write_manifest(tmp_path)
|
|
text = path.read_text(encoding="utf-8").replace(
|
|
"workspace_template: workspace_template",
|
|
"workspace_template: ../outside",
|
|
)
|
|
path.write_text(text, encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError, match="workspace_template"):
|
|
load_challenge_manifest(path)
|
|
|
|
|
|
def test_invalid_manifest_rejects_source_root_escape(tmp_path: Path) -> None:
|
|
path = _write_manifest(tmp_path)
|
|
text = path.read_text(encoding="utf-8").replace(
|
|
" root: source",
|
|
" root: ../source",
|
|
)
|
|
path.write_text(text, encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError, match="source.root"):
|
|
load_challenge_manifest(path)
|
|
|
|
|
|
def test_invalid_manifest_rejects_server_config_escape(tmp_path: Path) -> None:
|
|
path = _write_manifest(tmp_path)
|
|
text = path.read_text(encoding="utf-8").replace(
|
|
" config: wf.config.json",
|
|
" config: ../wf.config.json",
|
|
)
|
|
path.write_text(text, encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError, match="server.config"):
|
|
load_challenge_manifest(path)
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def test_challenge_prompt_is_identical_across_profiles(tmp_path: Path) -> None:
|
|
challenge = load_challenge_manifest(_write_manifest(tmp_path / "challenge"))
|
|
rendered = {
|
|
profile: compose_trial_prompt(
|
|
challenge,
|
|
profile=profile,
|
|
wf_command_prefix="uv run wf --config wf.config.json --local",
|
|
server_context="Local mode.",
|
|
workspace_path=tmp_path / profile.value,
|
|
)
|
|
for profile in InstructionProfile
|
|
}
|
|
|
|
assert {value.challenge_sha256 for value in rendered.values()} == {
|
|
rendered[InstructionProfile.NONE].challenge_sha256
|
|
}
|
|
assert len({value.rendered_sha256 for value in rendered.values()}) == 4
|
|
assert "report the exact blocker" in rendered[InstructionProfile.NONE].text.replace(
|
|
"\n", " "
|
|
)
|
|
assert ".agent/skills" in rendered[InstructionProfile.SKILLS].text
|
|
assert "inspect broader repository" in rendered[InstructionProfile.ALL].text
|
|
assert "genuinely blocked" in rendered[InstructionProfile.DEBUG].text
|
|
assert "debug profile only" in rendered[InstructionProfile.DEBUG].text
|
|
assert "ux_issues_found" in rendered[InstructionProfile.DEBUG].text
|
|
for profile in (
|
|
InstructionProfile.NONE,
|
|
InstructionProfile.SKILLS,
|
|
InstructionProfile.ALL,
|
|
):
|
|
assert "debug profile only" not in rendered[profile].text
|
|
assert "ux_issues_found" not in rendered[profile].text
|
|
none_prompt = rendered[InstructionProfile.NONE].text.replace("\n", " ")
|
|
assert "inline in your" in none_prompt
|
|
assert "without an inline self-report is invalid" in none_prompt
|
|
|
|
|
|
def test_skills_profile_copies_bundle_but_none_does_not(tmp_path: Path) -> None:
|
|
challenge = load_challenge_manifest(_write_manifest(tmp_path / "challenge"))
|
|
bundle = ROOT / "examples/agent_challenges/instruction_bundles/workflow_cli.yaml"
|
|
|
|
none_workspace = prepare_v2_trial_workspace(
|
|
challenge,
|
|
profile=InstructionProfile.NONE,
|
|
model="model",
|
|
index=1,
|
|
workspaces_dir=tmp_path / "workspaces",
|
|
instruction_bundle=bundle,
|
|
)
|
|
skills_workspace = prepare_v2_trial_workspace(
|
|
challenge,
|
|
profile=InstructionProfile.SKILLS,
|
|
model="model",
|
|
index=2,
|
|
workspaces_dir=tmp_path / "workspaces",
|
|
instruction_bundle=bundle,
|
|
)
|
|
|
|
assert not (none_workspace.root / ".agent/skills").exists()
|
|
assert (skills_workspace.root / ".agent/skills/wf-cli/SKILL.md").is_file()
|
|
assert skills_workspace.instruction_files
|
|
|
|
|
|
def test_extract_trial_metrics_parses_jsonl_events() -> None:
|
|
from examples.agent_challenges.metrics import extract_trial_metrics
|
|
|
|
stdout = "\n".join(
|
|
[
|
|
json.dumps({"type": "step_start", "step": 1}),
|
|
json.dumps(
|
|
{
|
|
"type": "tool_use",
|
|
"tool": "read",
|
|
"status": "success",
|
|
"title": "Read file",
|
|
"input": {"path": "foo.py"},
|
|
"metadata": {},
|
|
"output": "x" * 4000,
|
|
}
|
|
),
|
|
json.dumps(
|
|
{
|
|
"type": "tool_use",
|
|
"tool": "bash",
|
|
"status": "error",
|
|
"title": "Run command",
|
|
"input": {"command": "ls"},
|
|
"metadata": {},
|
|
"output": "error occurred",
|
|
}
|
|
),
|
|
json.dumps(
|
|
{
|
|
"type": "step_finish",
|
|
"tokens": {
|
|
"total": 120,
|
|
"input": 20,
|
|
"output": 30,
|
|
"reasoning": 10,
|
|
"cache": {"read": 60, "write": 0},
|
|
},
|
|
"cost": 0.01,
|
|
}
|
|
),
|
|
]
|
|
)
|
|
|
|
metrics = extract_trial_metrics(stdout)
|
|
|
|
assert metrics.step_count == 1
|
|
assert metrics.tool_call_count == 2
|
|
assert metrics.failed_tool_call_count == 1
|
|
assert metrics.tool_counts == {"bash": 1, "read": 1}
|
|
assert metrics.tokens.total == 120
|
|
assert metrics.tokens.input == 20
|
|
assert metrics.tokens.output == 30
|
|
assert metrics.tokens.reasoning == 10
|
|
assert metrics.tokens.cache_read == 60
|
|
assert metrics.cost == 0.01
|
|
assert metrics.tool_calls[0].tool == "read"
|
|
assert metrics.tool_calls[0].output_chars == 4000
|
|
assert len(metrics.tool_calls[0].output_preview) <= 500
|
|
|
|
|
|
def test_extract_trial_metrics_accepts_missing_stdout() -> None:
|
|
from examples.agent_challenges.metrics import extract_trial_metrics
|
|
|
|
metrics = extract_trial_metrics(None)
|
|
|
|
assert metrics.step_count == 0
|
|
assert metrics.tool_call_count == 0
|
|
|
|
|
|
def test_extract_trial_metrics_sums_tokens_across_steps() -> None:
|
|
from examples.agent_challenges.metrics import extract_trial_metrics
|
|
|
|
stdout = "\n".join(
|
|
[
|
|
json.dumps({"type": "step_start", "step": 1}),
|
|
json.dumps(
|
|
{
|
|
"type": "step_finish",
|
|
"tokens": {
|
|
"total": 50,
|
|
"input": 20,
|
|
"output": 15,
|
|
"reasoning": 5,
|
|
"cache": {"read": 10, "write": 0},
|
|
},
|
|
"cost": 0.003,
|
|
}
|
|
),
|
|
json.dumps({"type": "step_start", "step": 2}),
|
|
json.dumps(
|
|
{
|
|
"type": "step_finish",
|
|
"tokens": {
|
|
"total": 70,
|
|
"input": 30,
|
|
"output": 25,
|
|
"reasoning": 10,
|
|
"cache": {"read": 5, "write": 0},
|
|
},
|
|
"cost": 0.004,
|
|
}
|
|
),
|
|
]
|
|
)
|
|
|
|
metrics = extract_trial_metrics(stdout)
|
|
|
|
assert metrics.step_count == 2
|
|
assert metrics.tokens.total == 120
|
|
assert metrics.tokens.input == 50
|
|
assert metrics.tokens.output == 40
|
|
assert metrics.tokens.reasoning == 15
|
|
assert metrics.tokens.cache_read == 15
|
|
assert metrics.cost == 0.007
|
|
|
|
|
|
def test_extract_trial_metrics_handles_nested_part_state_format() -> None:
|
|
from examples.agent_challenges.metrics import extract_trial_metrics
|
|
|
|
stdout = "\n".join(
|
|
[
|
|
json.dumps({"type": "step_start", "step": 1}),
|
|
json.dumps(
|
|
{
|
|
"type": "tool_use",
|
|
"part": {
|
|
"tool": "read",
|
|
"callID": "call-abc",
|
|
"state": {
|
|
"status": "success",
|
|
"title": "Read file",
|
|
"input": {"path": "src/app.py"},
|
|
"output": "file content here",
|
|
"metadata": {"size": 100},
|
|
},
|
|
},
|
|
}
|
|
),
|
|
json.dumps(
|
|
{
|
|
"type": "tool_use",
|
|
"part": {
|
|
"tool": "bash",
|
|
"callID": "call-def",
|
|
"state": {
|
|
"status": "error",
|
|
"title": "Run command",
|
|
"input": {"command": "ls nonexistent"},
|
|
"output": "command failed",
|
|
},
|
|
},
|
|
}
|
|
),
|
|
json.dumps(
|
|
{
|
|
"type": "step_finish",
|
|
"tokens": {
|
|
"total": 100,
|
|
"input": 50,
|
|
"output": 30,
|
|
"reasoning": 10,
|
|
"cache": {"read": 10, "write": 0},
|
|
},
|
|
"cost": 0.005,
|
|
}
|
|
),
|
|
]
|
|
)
|
|
|
|
metrics = extract_trial_metrics(stdout)
|
|
|
|
assert metrics.step_count == 1
|
|
assert metrics.tool_call_count == 2
|
|
assert metrics.failed_tool_call_count == 1
|
|
assert metrics.tool_counts == {"bash": 1, "read": 1}
|
|
assert metrics.tool_calls[0].tool == "read"
|
|
assert metrics.tool_calls[0].status == "success"
|
|
assert metrics.tool_calls[0].call_id == "call-abc"
|
|
assert metrics.tool_calls[0].output_chars == 17
|
|
assert metrics.tool_calls[0].input == {"path": "src/app.py"}
|
|
assert metrics.tool_calls[1].tool == "bash"
|
|
assert metrics.tool_calls[1].status == "error"
|
|
assert metrics.tool_calls[1].failed is True
|
|
assert metrics.tool_calls[1].input == {"command": "ls nonexistent"}
|
|
assert metrics.tokens.total == 100
|
|
assert metrics.cost == 0.005
|
|
|
|
|
|
def test_extract_trial_metrics_handles_nested_step_finish_format() -> None:
|
|
from examples.agent_challenges.metrics import extract_trial_metrics
|
|
|
|
stdout = "\n".join(
|
|
[
|
|
json.dumps(
|
|
{
|
|
"type": "step_finish",
|
|
"part": {
|
|
"tokens": {
|
|
"total": 100,
|
|
"input": 40,
|
|
"output": 30,
|
|
"reasoning": 10,
|
|
"cache": {"read": 20, "write": 0},
|
|
},
|
|
"cost": 0.005,
|
|
},
|
|
}
|
|
),
|
|
json.dumps(
|
|
{
|
|
"type": "step_finish",
|
|
"part": {
|
|
"tokens": {
|
|
"total": 50,
|
|
"input": 20,
|
|
"output": 20,
|
|
"reasoning": 5,
|
|
"cache": {"read": 5, "write": 0},
|
|
},
|
|
"cost": 0.002,
|
|
},
|
|
}
|
|
),
|
|
]
|
|
)
|
|
|
|
metrics = extract_trial_metrics(stdout)
|
|
|
|
assert metrics.tokens.total == 150
|
|
assert metrics.tokens.input == 60
|
|
assert metrics.tokens.output == 50
|
|
assert metrics.tokens.reasoning == 15
|
|
assert metrics.tokens.cache_read == 25
|
|
assert metrics.cost == 0.007
|
|
|
|
|
|
def test_opencode_text_results_preserve_report_before_later_summary() -> None:
|
|
from examples.agent_challenges.opencode_io import opencode_text_results
|
|
|
|
report = "```yaml\nchallenge_report:\n run_failed: false\n```"
|
|
stdout = "\n".join(
|
|
[
|
|
json.dumps({"type": "text", "part": {"text": report}}),
|
|
json.dumps(
|
|
{
|
|
"type": "text",
|
|
"part": {"text": "Challenge completed successfully."},
|
|
}
|
|
),
|
|
]
|
|
)
|
|
|
|
results = opencode_text_results(stdout)
|
|
|
|
assert [result["text"] for result in results] == [
|
|
report,
|
|
"Challenge completed successfully.",
|
|
]
|
|
|
|
|
|
def test_opencode_text_results_accept_json_array_output() -> None:
|
|
from examples.agent_challenges.opencode_io import opencode_text_results
|
|
|
|
stdout = json.dumps(
|
|
[
|
|
{"type": "text", "part": {"text": "first"}},
|
|
{"type": "text", "part": {"text": "second"}},
|
|
]
|
|
)
|
|
|
|
results = opencode_text_results(stdout)
|
|
|
|
assert [result["text"] for result in results] == ["first", "second"]
|
|
|
|
|
|
def test_policy_evidence_classifies_reads(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
workspace_root.mkdir()
|
|
repository_root = tmp_path / "repo"
|
|
repository_root.mkdir()
|
|
workspaces_root = tmp_path / "workspaces"
|
|
workspaces_root.mkdir()
|
|
|
|
def _tc(tool: str, path: str) -> ToolCallEvidence:
|
|
return ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="c1",
|
|
tool=tool,
|
|
status="success",
|
|
title="read",
|
|
input={"path": path},
|
|
metadata={},
|
|
output_chars=100,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
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"))
|
|
|
|
none_policy = evaluate_policy(
|
|
"none",
|
|
[workspace_read, source_read],
|
|
workspace_root=workspace_root,
|
|
repository_root=repository_root,
|
|
workspaces_root=workspaces_root,
|
|
)
|
|
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,
|
|
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,
|
|
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 all_policy.validity.value == "clean"
|
|
assert all_policy.escalated_to_product_code is True
|
|
|
|
debug_policy = evaluate_policy(
|
|
"debug",
|
|
[
|
|
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 debug_policy.validity.value == "clean"
|
|
assert debug_policy.escalated_to_product_code is True
|
|
|
|
bash_tc = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="c1",
|
|
tool="bash",
|
|
status="success",
|
|
title="run",
|
|
input={"command": "cat /etc/passwd"},
|
|
metadata={},
|
|
output_chars=100,
|
|
output_preview="",
|
|
output_sha256="def",
|
|
failed=False,
|
|
)
|
|
bash_policy = evaluate_policy(
|
|
"none",
|
|
[bash_tc],
|
|
workspace_root=workspace_root,
|
|
repository_root=repository_root,
|
|
workspaces_root=workspaces_root,
|
|
)
|
|
assert bash_policy.validity.value == "clean"
|
|
assert bash_policy.coverage.value == "partial"
|
|
assert len(bash_policy.opaque_shell_commands) == 1
|
|
|
|
wf_tc = ToolCallEvidence(
|
|
ordinal=2,
|
|
call_id="c2",
|
|
tool="bash",
|
|
status="success",
|
|
title="run workflow",
|
|
input={"command": "uv run wf --config trial/wf.config.json --local status"},
|
|
metadata={},
|
|
output_chars=100,
|
|
output_preview="",
|
|
output_sha256="ghi",
|
|
failed=False,
|
|
)
|
|
wf_policy = evaluate_policy(
|
|
"all",
|
|
[wf_tc],
|
|
workspace_root=workspace_root,
|
|
repository_root=repository_root,
|
|
workspaces_root=workspaces_root,
|
|
)
|
|
assert wf_policy.validity.value == "clean"
|
|
assert wf_policy.coverage.value == "complete"
|
|
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
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = repository / "examples" / "challenge" / "workspaces" / "trial"
|
|
workspace.mkdir(parents=True)
|
|
source_path = repository / "src" / "app.py"
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="read-1",
|
|
tool="read",
|
|
status="completed",
|
|
title="Read source",
|
|
input={"filePath": str(source_path)},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
policy = evaluate_policy(
|
|
"all",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
|
|
assert policy.escalated_to_product_code is True
|
|
assert policy.reads_by_category["source"] == (str(source_path),)
|
|
|
|
|
|
def test_policy_allows_supplied_skill_globs_for_skills_profile(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = repository / "examples" / "challenge" / "workspaces" / "trial"
|
|
workspace.mkdir(parents=True)
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="glob-1",
|
|
tool="glob",
|
|
status="completed",
|
|
title="Find supplied skills",
|
|
input={"pattern": ".agent/skills/**/*"},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
policy = evaluate_policy(
|
|
"skills",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
|
|
assert policy.validity.value == "clean"
|
|
assert policy.reads_by_category["supplied_skills"] == (".agent/skills/**/*",)
|
|
assert policy.disallowed_reads == ()
|
|
|
|
|
|
def test_policy_treats_workspace_challenge_paths_as_workspace(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = (
|
|
repository
|
|
/ "examples"
|
|
/ "agent_challenges"
|
|
/ "report_workflow_challenge"
|
|
/ "workspaces"
|
|
/ "trial-001"
|
|
)
|
|
workspace.mkdir(parents=True)
|
|
workspace_file = workspace / "rendered-prompt.md"
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="read-1",
|
|
tool="read",
|
|
status="completed",
|
|
title="Read prompt",
|
|
input={"filePath": str(workspace_file)},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
policy = evaluate_policy(
|
|
"skills",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
|
|
assert policy.validity.value == "clean"
|
|
assert policy.reads_by_category["workspace"] == (str(workspace_file),)
|
|
assert policy.disallowed_reads == ()
|
|
|
|
|
|
def test_policy_records_broad_globs_without_contaminating(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = repository / "examples" / "challenge" / "workspaces" / "trial"
|
|
workspace.mkdir(parents=True)
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="glob-1",
|
|
tool="glob",
|
|
status="completed",
|
|
title="Search report workflow files",
|
|
input={"pattern": "**/report_workflow/**"},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
policy = evaluate_policy(
|
|
"skills",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
|
|
assert policy.validity.value == "clean"
|
|
assert policy.reads_by_category["search_intent"] == ("**/report_workflow/**",)
|
|
assert policy.disallowed_reads == ()
|
|
|
|
|
|
def test_policy_none_records_broad_globs_without_contaminating(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = repository / "examples" / "challenge" / "workspaces" / "trial"
|
|
workspace.mkdir(parents=True)
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="glob-1",
|
|
tool="glob",
|
|
status="completed",
|
|
title="Search workspace files",
|
|
input={"pattern": "*.json"},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
policy = evaluate_policy(
|
|
"none",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
|
|
assert policy.validity.value == "clean"
|
|
assert policy.reads_by_category["search_intent"] == ("*.json",)
|
|
assert policy.disallowed_reads == ()
|
|
|
|
|
|
def test_policy_classifies_example_implementation_reads_separately(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = repository / "examples" / "challenge" / "workspaces" / "trial"
|
|
workspace.mkdir(parents=True)
|
|
ops_path = repository / "examples" / "report_workflow" / "ops.py"
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="read-1",
|
|
tool="read",
|
|
status="completed",
|
|
title="Read example implementation",
|
|
input={"filePath": str(ops_path)},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
policy = evaluate_policy(
|
|
"skills",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
|
|
assert policy.validity.value == "contaminated"
|
|
assert policy.escalated_to_product_code is True
|
|
assert policy.reads_by_category["example_implementation"] == (str(ops_path),)
|
|
assert policy.disallowed_reads == (str(ops_path),)
|
|
|
|
|
|
def test_policy_treats_current_workspace_store_as_store_access(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = repository / "examples" / "challenge" / "workspaces" / "trial"
|
|
store_file = (
|
|
workspace / ".wf_browser_click_store" / "draft_workspaces" / "draft.json"
|
|
)
|
|
store_file.parent.mkdir(parents=True)
|
|
store_file.write_text("{}", encoding="utf-8")
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="read-1",
|
|
tool="read",
|
|
status="completed",
|
|
title="Read draft store internals",
|
|
input={"filePath": str(store_file)},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
none_policy = evaluate_policy(
|
|
"none",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
assert none_policy.validity.value == "contaminated"
|
|
assert none_policy.reads_by_category["prior_store"] == (str(store_file),)
|
|
assert none_policy.disallowed_reads == (str(store_file),)
|
|
|
|
skills_policy = evaluate_policy(
|
|
"skills",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
assert skills_policy.validity.value == "contaminated"
|
|
assert skills_policy.disallowed_reads == (str(store_file),)
|
|
|
|
all_policy = evaluate_policy(
|
|
"all",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
assert all_policy.validity.value == "clean"
|
|
assert all_policy.disallowed_reads == ()
|
|
|
|
|
|
def test_policy_anchors_relative_paths_to_workspace(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = repository / "examples" / "challenge" / "workspaces" / "trial"
|
|
workspace.mkdir(parents=True)
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="read-1",
|
|
tool="read",
|
|
status="completed",
|
|
title="Read workspace plan",
|
|
input={"path": "workflow.plan.json"},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
policy = evaluate_policy(
|
|
"none",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
|
|
assert policy.validity.value == "clean"
|
|
assert policy.reads_by_category["workspace"] == ("workflow.plan.json",)
|
|
|
|
|
|
def test_policy_classifies_ready_made_example_plans_as_existing_solution(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from examples.agent_challenges.metrics import ToolCallEvidence
|
|
from examples.agent_challenges.policy import evaluate_policy
|
|
|
|
repository = tmp_path / "repo"
|
|
workspace = repository / "examples" / "challenge" / "workspaces" / "trial"
|
|
workspace.mkdir(parents=True)
|
|
plan_path = repository / "examples" / "report_workflow" / "workflow.plan.json"
|
|
call = ToolCallEvidence(
|
|
ordinal=1,
|
|
call_id="read-1",
|
|
tool="read",
|
|
status="completed",
|
|
title="Read ready-made plan",
|
|
input={"filePath": str(plan_path)},
|
|
metadata={},
|
|
output_chars=10,
|
|
output_preview="",
|
|
output_sha256="abc",
|
|
failed=False,
|
|
)
|
|
|
|
policy = evaluate_policy(
|
|
"skills",
|
|
[call],
|
|
workspace_root=workspace,
|
|
repository_root=repository,
|
|
workspaces_root=workspace.parent,
|
|
)
|
|
|
|
assert policy.validity.value == "contaminated"
|
|
assert policy.reads_by_category["existing_solution"] == (str(plan_path),)
|
|
assert policy.disallowed_reads == (str(plan_path),)
|
|
|
|
|
|
def test_v2_runner_default_timeout_and_workspace_cwd(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()
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_run(
|
|
command: list[str],
|
|
*,
|
|
cwd: str,
|
|
text: bool,
|
|
capture_output: bool,
|
|
timeout: float | None,
|
|
check: bool,
|
|
) -> object:
|
|
captured["cwd"] = cwd
|
|
captured["timeout"] = timeout
|
|
captured["command"] = command
|
|
return type(
|
|
"Result",
|
|
(),
|
|
{
|
|
"returncode": 0,
|
|
"stdout": json.dumps(
|
|
{
|
|
"type": "step_finish",
|
|
"tokens": {"total": 10, "input": 5, "output": 5},
|
|
"cost": 0.001,
|
|
}
|
|
),
|
|
"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 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
|
|
assert "policy" in result
|
|
assert "repository_commit" in result
|
|
|
|
|
|
def test_safe_model_name_replaces_windows_path_separators() -> None:
|
|
from examples.agent_challenges.workspace import _safe_model_name
|
|
|
|
safe = _safe_model_name(r"..\bad/model:name")
|
|
|
|
assert "\\" not in safe
|
|
assert "/" not in safe
|
|
assert ":" not in safe
|
|
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
|
|
|
|
from examples.agent_challenges.workspace import _load_instruction_bundle
|
|
|
|
bundle = tmp_path / "bundle.yaml"
|
|
bundle.write_text(
|
|
yaml.safe_dump(
|
|
{
|
|
"files": [
|
|
{
|
|
"source": "../outside.md",
|
|
"destination": "wf-cli/SKILL.md",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="escapes project root"):
|
|
_load_instruction_bundle(bundle)
|
|
|
|
|
|
def test_v2_runner_timeout_preserves_partial_evidence(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 result["returncode"] == -1
|
|
assert "metrics" in result
|
|
|
|
|
|
def test_v2_runner_handles_missing_completed_stdout(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()
|
|
|
|
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": None, "stderr": None},
|
|
)()
|
|
|
|
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["stdout"] == ""
|
|
assert result["stderr"] == ""
|
|
assert result["task_outcome"] == "failed"
|
|
assert "metrics" in result
|
|
|
|
|
|
def test_v2_report_contains_expected_sections(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.reports import report_from_v2_result
|
|
|
|
result_payload = {
|
|
"instruction_profile": "skills",
|
|
"task_outcome": "success",
|
|
"evaluation_validity": "contaminated",
|
|
"prompt_hashes": {
|
|
"base": "abc123",
|
|
"profile": "def456",
|
|
"challenge": "ghi789",
|
|
"rendered": "jkl012",
|
|
},
|
|
"metrics": {
|
|
"step_count": 1,
|
|
"tool_call_count": 2,
|
|
"failed_tool_call_count": 0,
|
|
"tool_counts": {"bash": 1, "read": 1},
|
|
"tokens": {
|
|
"total": 100,
|
|
"input": 50,
|
|
"output": 30,
|
|
"reasoning": 10,
|
|
"cache_read": 10,
|
|
"cache_write": 0,
|
|
},
|
|
"cost": 0.005,
|
|
"unknown_event_count": 0,
|
|
"tool_calls": [
|
|
{
|
|
"ordinal": 1,
|
|
"call_id": "c1",
|
|
"tool": "read",
|
|
"status": "success",
|
|
"title": "Read file",
|
|
"input": {"path": "foo.py"},
|
|
"metadata": {},
|
|
"output_chars": 100,
|
|
"output_preview": "file content...",
|
|
"output_sha256": "abc",
|
|
"failed": False,
|
|
},
|
|
],
|
|
},
|
|
"policy": {
|
|
"validity": "contaminated",
|
|
"disallowed_reads": ["src/app.py"],
|
|
"escalated_to_product_code": False,
|
|
"opaque_shell_commands": [],
|
|
},
|
|
"repository_commit": "abc123",
|
|
"repository_dirty": False,
|
|
"harness_version": "v2",
|
|
"index": 1,
|
|
"model": "test-model",
|
|
"variant": "high",
|
|
"duration_seconds": 10.5,
|
|
"returncode": 0,
|
|
"stdout": "test stdout",
|
|
"stderr": "",
|
|
"parsed": {"text": "Agent answer here"},
|
|
}
|
|
|
|
report_text = report_from_v2_result(result_payload)
|
|
|
|
assert "Instruction profile: skills" in report_text
|
|
assert "Task outcome: success" in report_text
|
|
assert "Evaluation validity: contaminated" in report_text
|
|
assert "Duration" in report_text
|
|
assert "Observed token metrics" in report_text
|
|
assert "Tool calls by tool" in report_text
|
|
assert "Disallowed reads" in report_text
|
|
assert "Agent self-report discrepancies" in report_text
|
|
assert "Final agent answer" in report_text
|
|
assert "Manual audit: pending" in report_text
|
|
assert "file content..." in report_text
|
|
assert "src/app.py" in report_text
|
|
|
|
|
|
def test_v2_manual_audit_includes_automatic_evidence(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.audit import manual_audit_from_v2_result
|
|
|
|
result_payload = {
|
|
"instruction_profile": "none",
|
|
"task_outcome": "success",
|
|
"evaluation_validity": "contaminated",
|
|
"policy": {
|
|
"validity": "contaminated",
|
|
"disallowed_reads": ["src/app.py"],
|
|
"escalated_to_product_code": False,
|
|
"opaque_shell_commands": [],
|
|
},
|
|
"metrics": {"tokens": {"total": 100}},
|
|
}
|
|
|
|
workspace, audit = manual_audit_from_v2_result(
|
|
result_payload,
|
|
official_outcome="pass",
|
|
auditor_notes="Reviewed and passed.",
|
|
output_name=str(tmp_path / "manual-audit.yaml"),
|
|
)
|
|
|
|
assert workspace == tmp_path / "manual-audit.yaml"
|
|
assert workspace.is_file()
|
|
assert audit["manual_audit"]["task_outcome"] == "success"
|
|
assert audit["manual_audit"]["evaluation_validity"] == "contaminated"
|
|
assert audit["manual_audit"]["official_outcome"] == "pass"
|
|
assert audit["manual_audit"]["auditor_notes"] == "Reviewed and passed."
|
|
assert "automatic_evidence" in audit["manual_audit"]
|
|
|
|
|
|
def test_v2_runner_assertions_pass_on_matching_report(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()
|
|
|
|
report_yaml = (
|
|
"```yaml\nchallenge_report:\n value: expected\n run_failed: false\n```\n"
|
|
)
|
|
stdout_jsonl = json.dumps({"text": report_yaml})
|
|
|
|
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["task_outcome"] == "success"
|
|
assert "assertion_failures" not in result
|
|
assert result.get("challenge_report") is not None
|
|
|
|
|
|
def test_v2_runner_debug_profile_requires_ux_issues_found(
|
|
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()
|
|
|
|
report_yaml = (
|
|
"```yaml\nchallenge_report:\n value: expected\n run_failed: false\n```\n"
|
|
)
|
|
stdout_jsonl = json.dumps({"text": report_yaml})
|
|
|
|
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.DEBUG,
|
|
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"] == "failed"
|
|
assert any("ux_issues_found" in f for f in result["assertion_failures"])
|
|
|
|
|
|
def test_v2_runner_debug_profile_requires_challenge_report(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.runner import run_v2_trial
|
|
|
|
manifest_path = _write_manifest(tmp_path / "challenge")
|
|
manifest_path.write_text(
|
|
manifest_path.read_text(encoding="utf-8").replace(
|
|
" required_fields: [value, run_failed]\n"
|
|
" success_assertions:\n"
|
|
" value: expected\n"
|
|
" run_failed: false\n",
|
|
" required_fields: []\n success_assertions: {}\n",
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
challenge = load_challenge_manifest(manifest_path)
|
|
bundle = ROOT / "examples/agent_challenges/instruction_bundles/workflow_cli.yaml"
|
|
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:
|
|
return type(
|
|
"Result",
|
|
(),
|
|
{
|
|
"returncode": 0,
|
|
"stdout": json.dumps({"text": "Completed without a report."}),
|
|
"stderr": "",
|
|
},
|
|
)()
|
|
|
|
result = run_v2_trial(
|
|
challenge,
|
|
profile=InstructionProfile.DEBUG,
|
|
model="test-model",
|
|
variant="high",
|
|
index=1,
|
|
workspaces_dir=tmp_path / "workspaces",
|
|
results_dir=results_dir,
|
|
instruction_bundle=bundle,
|
|
run_fn=fake_run,
|
|
)
|
|
|
|
assert result["task_outcome"] == "failed"
|
|
assert any("challenge_report" in f for f in result["assertion_failures"])
|
|
|
|
|
|
def test_v2_runner_assertions_fail_on_mismatched_report(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()
|
|
|
|
report_yaml = (
|
|
"```yaml\nchallenge_report:\n value: wrong_value\n run_failed: true\n```\n"
|
|
)
|
|
stdout_jsonl = json.dumps({"text": report_yaml})
|
|
|
|
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["task_outcome"] == "failed"
|
|
assert "assertion_failures" in result
|
|
assert len(result["assertion_failures"]) == 2
|
|
assert any("value" in f for f in result["assertion_failures"])
|
|
assert any("run_failed" in f for f in result["assertion_failures"])
|
|
|
|
|
|
def test_v2_runner_required_fields_missing(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()
|
|
|
|
report_yaml = "```yaml\nchallenge_report:\n other_field: true\n```\n"
|
|
stdout_jsonl = json.dumps({"text": report_yaml})
|
|
|
|
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["task_outcome"] == "failed"
|
|
assert "assertion_failures" in result
|
|
assert any("required field missing" in f for f in result["assertion_failures"])
|
|
|
|
|
|
def test_v2_runner_preserves_evidence_on_parse_failure(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()
|
|
|
|
def fake_run(
|
|
command: list[str],
|
|
*,
|
|
cwd: str,
|
|
text: bool,
|
|
capture_output: bool,
|
|
timeout: float | None,
|
|
check: bool,
|
|
) -> object:
|
|
raise RuntimeError("subprocess exploded")
|
|
|
|
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"] == "runner_error"
|
|
assert result["returncode"] == -2
|
|
assert "parse_error" in result
|
|
assert result["parse_error"]["type"] == "RuntimeError"
|
|
assert "subprocess exploded" in result["parse_error"]["message"]
|
|
assert result["duration_seconds"] >= 0
|
|
|
|
|
|
def test_v2_runner_to_report_shows_final_answer(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.reports import report_from_v2_result
|
|
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 = "The deployment succeeded with id dep_123."
|
|
challenge_report_yaml = (
|
|
"```yaml\nchallenge_report:\n value: expected\n run_failed: false\n```\n"
|
|
)
|
|
stdout_jsonl = "\n".join(
|
|
[
|
|
json.dumps({"type": "step_start", "step": 1}),
|
|
json.dumps({"type": "step_finish", "tokens": {"total": 50}, "cost": 0.001}),
|
|
json.dumps(
|
|
{"text": f"Final answer: {agent_answer}\n\n{challenge_report_yaml}"}
|
|
),
|
|
json.dumps(
|
|
{"text": "Challenge completed successfully after the full report."}
|
|
),
|
|
]
|
|
)
|
|
|
|
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["parsed"] is not None
|
|
assert result["task_outcome"] == "success"
|
|
|
|
report_text = report_from_v2_result(result)
|
|
assert "Final agent answer" in report_text
|
|
assert agent_answer in report_text
|
|
|
|
|
|
def test_v2_runner_preserves_report_parse_error_on_malformed_yaml(
|
|
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()
|
|
|
|
malformed_yaml = (
|
|
"```yaml\nchallenge_report:\n value: expected\n run_failed: [unclosed\n```\n"
|
|
)
|
|
stdout_jsonl = "\n".join(
|
|
[
|
|
json.dumps({"type": "step_start", "step": 1}),
|
|
json.dumps({"type": "step_finish", "tokens": {"total": 50}, "cost": 0.001}),
|
|
json.dumps({"text": f"Some output.\n\n{malformed_yaml}"}),
|
|
]
|
|
)
|
|
|
|
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["parsed"] is not None
|
|
assert result.get("report_parse_error") is not None
|
|
assert result["report_parse_error"]["type"] in (
|
|
"ParserError",
|
|
"ScannerError",
|
|
"YAMLError",
|
|
)
|
|
assert result.get("challenge_report") is None
|
|
|
|
|
|
BROWSER_CHALLENGE = (
|
|
ROOT
|
|
/ "examples"
|
|
/ "agent_challenges"
|
|
/ "browser_click_challenge"
|
|
/ "challenge.yaml"
|
|
)
|
|
REPORT_CHALLENGE = (
|
|
ROOT
|
|
/ "examples"
|
|
/ "agent_challenges"
|
|
/ "report_workflow_challenge"
|
|
/ "challenge.yaml"
|
|
)
|
|
INSTRUCTION_BUNDLE = (
|
|
ROOT / "examples" / "agent_challenges" / "instruction_bundles" / "workflow_cli.yaml"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"manifest_path",
|
|
[BROWSER_CHALLENGE, REPORT_CHALLENGE],
|
|
ids=["browser_click", "report_workflow"],
|
|
)
|
|
def test_both_challenges_load_through_same_manifest(manifest_path: Path) -> None:
|
|
loaded = load_challenge_manifest(manifest_path)
|
|
|
|
assert loaded.manifest.version == 1
|
|
assert loaded.prompt_path.is_file()
|
|
assert loaded.workspace_template.is_dir()
|
|
assert loaded.manifest.report.required_fields
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"manifest_path, expected_id, expected_source_id",
|
|
[
|
|
(BROWSER_CHALLENGE, "browser_click", "local.browser_click"),
|
|
(REPORT_CHALLENGE, "report_workflow", "local.report"),
|
|
],
|
|
ids=["browser_click", "report_workflow"],
|
|
)
|
|
def test_both_challenges_prepare_workspaces_under_each_profile(
|
|
manifest_path: Path,
|
|
expected_id: str,
|
|
expected_source_id: str,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
loaded = load_challenge_manifest(manifest_path)
|
|
|
|
assert loaded.manifest.id == expected_id
|
|
assert loaded.manifest.source.id == expected_source_id
|
|
|
|
for profile in InstructionProfile:
|
|
workspace = prepare_v2_trial_workspace(
|
|
loaded,
|
|
profile=profile,
|
|
model="test-model",
|
|
index=1,
|
|
workspaces_dir=tmp_path / profile.value,
|
|
instruction_bundle=INSTRUCTION_BUNDLE,
|
|
)
|
|
|
|
assert workspace.config_path.is_file()
|
|
config = json.loads(workspace.config_path.read_text(encoding="utf-8"))
|
|
assert config["client"]["target"] == {"kind": "local"}
|
|
assert config["server"]["store"]["root"] == loaded.manifest.store_root
|
|
|
|
if profile in (
|
|
InstructionProfile.SKILLS,
|
|
InstructionProfile.ALL,
|
|
InstructionProfile.DEBUG,
|
|
):
|
|
assert (workspace.root / ".agent/skills/wf-cli/SKILL.md").is_file()
|
|
else:
|
|
assert not (workspace.root / ".agent").exists()
|
|
|
|
|
|
def test_both_challenges_produce_different_challenge_hashes_but_same_base(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from examples.agent_challenges.prompts import compose_trial_prompt
|
|
|
|
browser = load_challenge_manifest(BROWSER_CHALLENGE)
|
|
report = load_challenge_manifest(REPORT_CHALLENGE)
|
|
|
|
hashes: dict[str, str] = {}
|
|
for name, challenge in [("browser", browser), ("report", report)]:
|
|
for profile in InstructionProfile:
|
|
rendered = compose_trial_prompt(
|
|
challenge,
|
|
profile=profile,
|
|
wf_command_prefix="uv run wf --config wf.config.json --local",
|
|
server_context="Local mode.",
|
|
workspace_path=tmp_path / f"{name}_{profile.value}",
|
|
)
|
|
hashes[f"{name}_{profile.value}"] = rendered.challenge_sha256
|
|
|
|
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 "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"))
|
|
assert machine["identity"]["challenge_id"] == "fixture"
|
|
|
|
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:
|
|
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 result["returncode"] == -1
|
|
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
|
|
|
|
|
|
def test_v2_runner_stores_opencode_resume_metadata(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.models import InstructionProfile
|
|
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"
|
|
stdout = (
|
|
'{"type":"step_start","sessionID":"ses_runner"}\n'
|
|
'{"type":"text","sessionID":"ses_runner","text":"```yaml\\n'
|
|
"challenge_report:\\n"
|
|
" used_product_path: true\\n"
|
|
" used_helper_script: false\\n"
|
|
" before_clicked: false\\n"
|
|
" after_clicked: true\\n"
|
|
" run_failed: false\\n"
|
|
" leftover_processes: false\\n"
|
|
'```"}\n'
|
|
)
|
|
|
|
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, "stderr": ""},
|
|
)()
|
|
|
|
result = run_v2_trial(
|
|
challenge,
|
|
profile=InstructionProfile.NONE,
|
|
model="opencode/deepseek-v4-flash-free",
|
|
variant="max",
|
|
index=1,
|
|
results_dir=tmp_path / "results",
|
|
workspaces_dir=tmp_path / "workspaces",
|
|
instruction_bundle=bundle,
|
|
attach_url="http://127.0.0.1:8192/",
|
|
run_fn=fake_run,
|
|
)
|
|
|
|
opencode = result["opencode"]
|
|
assert opencode["session_id"] == "ses_runner"
|
|
assert opencode["attach_url"] == "http://127.0.0.1:8192/"
|
|
assert opencode["model"] == "opencode/deepseek-v4-flash-free"
|
|
assert opencode["variant"] == "max"
|
|
assert opencode["resume_command"][0:4] == [
|
|
"opencode",
|
|
"run",
|
|
"--session",
|
|
"ses_runner",
|
|
]
|
|
assert "challenge_report" in opencode["resume_prompt"]
|
|
|
|
|
|
def test_v2_runner_resume_prompt_sees_assertion_failures(tmp_path: Path) -> None:
|
|
from examples.agent_challenges.models import InstructionProfile
|
|
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"
|
|
stdout = (
|
|
'{"type":"step_start","sessionID":"ses_runner"}\n'
|
|
'{"type":"text","sessionID":"ses_runner","text":"run completed"}\n'
|
|
)
|
|
|
|
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, "stderr": ""},
|
|
)()
|
|
|
|
result = run_v2_trial(
|
|
challenge,
|
|
profile=InstructionProfile.NONE,
|
|
model="opencode/deepseek-v4-flash-free",
|
|
variant="max",
|
|
index=1,
|
|
results_dir=tmp_path / "results",
|
|
workspaces_dir=tmp_path / "workspaces",
|
|
instruction_bundle=bundle,
|
|
run_fn=fake_run,
|
|
)
|
|
|
|
opencode = result["opencode"]
|
|
assert result["task_outcome"] == "failed"
|
|
assert "assertion_failures" in result
|
|
assert "do not continue coding" in opencode["resume_prompt"].lower()
|
|
|
|
|
|
def test_v2_runner_stores_null_session_when_stdout_has_no_session(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from examples.agent_challenges.models import InstructionProfile
|
|
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"
|
|
|
|
def fake_run(
|
|
command: list[str],
|
|
*,
|
|
cwd: str,
|
|
text: bool,
|
|
capture_output: bool,
|
|
timeout: float | None,
|
|
check: bool,
|
|
) -> object:
|
|
return type(
|
|
"Result",
|
|
(),
|
|
{"returncode": 1, "stdout": "", "stderr": "boom"},
|
|
)()
|
|
|
|
result = run_v2_trial(
|
|
challenge,
|
|
profile=InstructionProfile.NONE,
|
|
model="opencode/deepseek-v4-flash-free",
|
|
variant="max",
|
|
index=1,
|
|
results_dir=tmp_path / "results",
|
|
workspaces_dir=tmp_path / "workspaces",
|
|
instruction_bundle=bundle,
|
|
run_fn=fake_run,
|
|
)
|
|
|
|
opencode = result["opencode"]
|
|
assert opencode["session_id"] is None
|
|
assert opencode["resume_command"] is None
|