-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathevaluate_command.py
More file actions
173 lines (150 loc) · 6.45 KB
/
Copy pathevaluate_command.py
File metadata and controls
173 lines (150 loc) · 6.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
"""Evaluate command - run criteria against a directory without an agent."""
import asyncio
from pathlib import Path
import typer
from ..logging_config import setup_logging
from ..models import (
AgentKind,
EvaluationResult,
FinalStatus,
PreservationMode,
TemplateDirSource,
parse_agent_config,
)
from ..orchestration.task_loader import load_task
from ..orchestrator import Orchestrator
from ..sandbox import Sandbox
from .console import console
from .run_helpers import prepare_run_directory
def evaluate_command(
task_file: Path = typer.Argument( # noqa: B008
...,
help="Path to task YAML file",
exists=True,
),
work_dir: Path = typer.Argument( # noqa: B008
...,
help="Directory containing the code to evaluate",
exists=True,
file_okay=False,
dir_okay=True,
),
verbose: bool = typer.Option(
False,
"--verbose",
"-v",
help="Enable verbose (DEBUG level) logging",
),
preserve: bool = typer.Option(
True,
"--preserve/--no-preserve",
"-p/-P",
help="Move sandbox artifacts to run directory (default: preserve). The temp sandbox is always removed.",
),
run_dir: Path | None = typer.Option( # noqa: B008
None,
"--run-dir",
help="Custom run directory (default: auto-generated timestamped directory in runs/)",
),
) -> None:
"""Evaluate criteria against a directory without running an agent.
Runs the success criteria defined in a task against a work directory.
Artifacts are saved to a run directory when --preserve is used.
Examples:
coder-eval evaluate tasks/hello.yaml ./my_solution
coder-eval evaluate tasks/test.yaml /path/to/code --preserve
coder-eval evaluate tasks/test.yaml /path/to/code --run-dir ./my_eval_run
"""
setup_logging(verbose=verbose)
console.print("\n[bold]Evaluating Criteria[/bold]\n")
try:
task, source_yaml = load_task(task_file)
except Exception as e:
console.print(f"[red]✗ Failed to load task:[/red] {e}")
raise typer.Exit(1) from e
# Evaluate-only mode bypasses experiment resolution + CLI overrides, so
# `agent` may be None or `agent.type` may be unset for tasks that defer
# those to the experiment / CLI layers. The orchestrator only uses
# `agent.type` for result labeling here (no agent is created), so a
# default is safe.
if task.agent is None:
task.agent = parse_agent_config(type=AgentKind.CLAUDE_CODE)
elif task.agent.type is None:
task.agent = parse_agent_config(**{**task.agent.model_dump(exclude_unset=True), "type": AgentKind.CLAUDE_CODE})
if not work_dir.is_dir():
console.print(f"[red]✗ Work directory is not a directory:[/red] {work_dir}")
raise typer.Exit(1)
try:
prepared_run_dir = prepare_run_directory(run_dir)
except Exception as e:
console.print(f"[red]✗ Failed to prepare run directory:[/red] {e}")
raise typer.Exit(1) from e
# Build a sandbox pre-loaded with the work_dir contents, then run evaluate-only
sandbox_config = task.sandbox.model_copy(deep=True)
template_source = TemplateDirSource(path=str(work_dir.resolve()))
if sandbox_config.template_sources:
sandbox_config.template_sources = [template_source, *sandbox_config.template_sources]
else:
sandbox_config.template_sources = [template_source]
task_dir = task_file.parent.resolve()
sandbox = Sandbox(sandbox_config, task_id=task.task_id, task_dir=task_dir)
async def _setup_and_run() -> EvaluationResult:
await asyncio.to_thread(sandbox.setup)
orchestrator = Orchestrator(
task=task,
run_dir=prepared_run_dir,
preservation_mode=PreservationMode.MOVE_ON_WRITE if preserve else PreservationMode.NONE,
task_file=task_file,
sandbox=sandbox,
variant_id="evaluate",
source_yaml=source_yaml,
)
return await orchestrator.run()
result = asyncio.run(_setup_and_run())
# Display results
console.print("[bold]Criteria Results:[/bold]\n")
criteria_results = result.success_criteria_results or []
if len(criteria_results) != len(task.success_criteria):
console.print(
f"[red]✗ Result count mismatch: got {len(criteria_results)}, expected {len(task.success_criteria)}[/red]"
)
raise typer.Exit(1)
for criterion, cr in zip(task.success_criteria, criteria_results, strict=True):
if not criterion.is_gating:
# weight=0 is informational: it cannot pass/fail the task, so don't
# render it as ✓/✗ (that would contradict the gate and the exit code).
status = "[dim]○[/dim]"
else:
status = "[green]✓[/green]" if cr.score >= criterion.pass_threshold else "[red]✗[/red]"
console.print(f"{status} {cr.criterion_type}")
console.print(f" [dim]{cr.description}[/dim]")
console.print(f" [dim]Score: {cr.score:.2f}[/dim]")
if cr.details:
console.print(f" [dim]Details: {cr.details}[/dim]")
if cr.error:
console.print(f" [red]Error: {cr.error}[/red]")
console.print()
# Gate over gating criteria only (weight=0 is informational and cannot fail
# the task) so this summary + the exit code below match final_status.
gating = [(cr, c) for cr, c in zip(criteria_results, task.success_criteria, strict=True) if c.is_gating]
passed = sum(1 for cr, c in gating if cr.score >= c.pass_threshold)
total = len(gating)
failed = total - passed
informational = len(task.success_criteria) - total
console.print("[bold]Summary:[/bold]")
console.print(f" Passed: {passed}/{total}")
console.print(f" Failed: {failed}/{total}")
if informational:
console.print(f" [dim]Informational (weight=0, not gated): {informational}[/dim]")
console.print(f"\n[dim]Run directory: {prepared_run_dir}[/dim]")
if result.sandbox_path:
console.print(f"[dim]Artifacts: {result.sandbox_path}[/dim]")
if result.final_status == FinalStatus.ERROR:
console.print(f"\n[red]✗ Evaluation error: {result.error_message}[/red]")
raise typer.Exit(1)
elif failed == 0:
console.print("\n[green]All criteria passed! ✓[/green]")
raise typer.Exit(0)
else:
console.print(f"\n[red]{failed} criterion/criteria failed.[/red]")
raise typer.Exit(1)