Coverage for little_loops / cli / harness.py: 21%
211 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -0500
1"""ll-harness: One-shot runner evaluation CLI (FEAT-1851)."""
3from __future__ import annotations
5import argparse
6import json
7import os
8import subprocess
9import sys
10import threading
11from dataclasses import dataclass
12from pathlib import Path
13from typing import Any
15import yaml
17from little_loops.cli.output import configure_output, print_json, status_block, use_color_enabled
18from little_loops.fsm.evaluators import evaluate_llm_structured
19from little_loops.host_runner import resolve_host
20from little_loops.logger import Logger
21from little_loops.mcp_call import call_mcp_tool
22from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context
25@dataclass
26class RunnerResult:
27 """Captured output from a runner invocation."""
29 stdout: str
30 stderr: str
31 exit_code: int
32 timed_out: bool = False
33 error: str | None = None
36@dataclass
37class DslTask:
38 """A single DSL evaluation task loaded from a task YAML file."""
40 prompt: str
41 blanks: list[str]
42 expected: dict[str, str]
43 source_dsl: str
44 task_type: str
45 source_file: str = ""
46 generated_at: str = ""
48 @classmethod
49 def from_dict(cls, data: dict[str, Any]) -> DslTask:
50 return cls(
51 prompt=data["prompt"],
52 blanks=data.get("blanks") or [],
53 expected=data.get("expected") or {},
54 source_dsl=data.get("source_dsl", ""),
55 task_type=data.get("task_type", ""),
56 source_file=data.get("source_file", ""),
57 generated_at=data.get("generated_at", ""),
58 )
61def _build_harness_parser() -> argparse.ArgumentParser:
62 """Build the ll-harness argument parser (exposed for testing)."""
63 parser = argparse.ArgumentParser(
64 prog="ll-harness",
65 description="One-shot runner evaluation for little-loops skills and commands",
66 formatter_class=argparse.RawDescriptionHelpFormatter,
67 epilog="""
68Examples:
69 ll-harness skill check-code
70 ll-harness cmd "echo hello" --exit-code 0
71 ll-harness mcp my-server:my-tool --args '{"key": "val"}' --semantic "tool returned results"
72 ll-harness prompt "What is 2+2?" --semantic "response contains a number"
74Exit codes:
75 0 PASS
76 1 FAIL
77 2 Internal error / timeout
78""",
79 )
81 subparsers = parser.add_subparsers(dest="runner", metavar="RUNNER")
82 subparsers.required = True
84 def _add_evaluator_flags(p: argparse.ArgumentParser) -> None:
85 p.add_argument(
86 "--exit-code",
87 dest="exit_code",
88 type=int,
89 default=None,
90 metavar="INT",
91 help="Expected exit code (default: not checked)",
92 )
93 p.add_argument(
94 "--semantic",
95 type=str,
96 default=None,
97 metavar="TEXT",
98 help="Natural-language criterion for output evaluation",
99 )
100 p.add_argument(
101 "--timeout",
102 type=int,
103 default=120,
104 metavar="SECONDS",
105 help="Runner timeout in seconds (default: 120)",
106 )
107 p.add_argument(
108 "--output",
109 choices=["text", "json"],
110 default="text",
111 help="Output format (default: text)",
112 )
113 p.add_argument(
114 "--verbose",
115 action="store_true",
116 help="Show full captured output even on pass",
117 )
119 skill_p = subparsers.add_parser(
120 "skill",
121 help="Invoke a little-loops skill",
122 description="Invoke a little-loops skill via the active host CLI",
123 )
124 skill_p.add_argument("target", help="Skill name (e.g. check-code, refine-issue)")
125 skill_p.add_argument(
126 "runner_args",
127 nargs="*",
128 help="Additional arguments passed to the skill",
129 )
130 _add_evaluator_flags(skill_p)
132 cmd_p = subparsers.add_parser(
133 "cmd",
134 help="Run a shell command",
135 description="Run a shell command and capture its output",
136 )
137 cmd_p.add_argument("target", help="Shell command to execute")
138 _add_evaluator_flags(cmd_p)
140 mcp_p = subparsers.add_parser(
141 "mcp",
142 help="Call an MCP tool",
143 description="Call an MCP tool via JSON-RPC",
144 )
145 mcp_p.add_argument("target", help="MCP server and tool (format: server:tool)")
146 mcp_p.add_argument(
147 "--args",
148 dest="mcp_args",
149 type=str,
150 default="{}",
151 metavar="JSON",
152 help="JSON arguments to pass to the MCP tool (default: {})",
153 )
154 _add_evaluator_flags(mcp_p)
156 prompt_p = subparsers.add_parser(
157 "prompt",
158 help="Send a raw prompt to Claude",
159 description="Send a raw prompt to Claude via the active host CLI",
160 )
161 prompt_p.add_argument("target", help="Prompt text to send")
162 prompt_p.add_argument(
163 "--model",
164 default=None,
165 metavar="MODEL",
166 help="Override Claude model (e.g. claude-haiku-4-5-20251001)",
167 )
168 _add_evaluator_flags(prompt_p)
170 dsl_p = subparsers.add_parser(
171 "dsl",
172 help="Run a DSL task set and report pass rates by model",
173 description="Load and run DSL eval task YAML files, reporting pass rate with Wilson CI",
174 )
175 dsl_p.add_argument("path", help="DSL task file or directory of .yaml task files")
176 dsl_p.add_argument(
177 "--model",
178 default=None,
179 metavar="MODEL",
180 help="Override Claude model (e.g. claude-haiku-4-5-20251001)",
181 )
182 _add_evaluator_flags(dsl_p)
184 return parser
187def _parse_harness_args(argv: list[str] | None) -> argparse.Namespace:
188 """Parse argv into a Namespace (exposed for testing)."""
189 return _build_harness_parser().parse_args(argv)
192def _evaluate_and_report(
193 runner_label: str,
194 result: RunnerResult,
195 args: argparse.Namespace,
196) -> int:
197 """Evaluate result against criteria and print the report. Returns exit code."""
198 if result.timed_out:
199 _report(runner_label, result, args, error_msg="timeout")
200 return 2
201 if result.error is not None:
202 _report(runner_label, result, args, error_msg=result.error)
203 return 2
205 passed = True
206 exit_code_display = str(result.exit_code)
207 semantic_display = "[not checked]"
209 if args.exit_code is not None:
210 if result.exit_code != args.exit_code:
211 passed = False
212 exit_code_display = f"{result.exit_code} (expected {args.exit_code})"
214 if args.semantic is not None:
215 eval_result = evaluate_llm_structured(output=result.stdout, prompt=args.semantic)
216 semantic_display = eval_result.verdict
217 if eval_result.verdict != "yes":
218 passed = False
220 overall = "PASS" if passed else "FAIL"
221 show_output = not passed or args.verbose
223 if args.output == "json":
224 print_json(
225 {
226 "runner": runner_label,
227 "exit_code": result.exit_code,
228 "exit_code_check": exit_code_display,
229 "semantic": semantic_display,
230 "result": overall,
231 "stdout": result.stdout,
232 "stderr": result.stderr,
233 }
234 )
235 else:
236 print(
237 status_block(
238 {
239 "Runner": runner_label,
240 "Exit": exit_code_display,
241 "Semantic": semantic_display,
242 "Result": overall,
243 }
244 )
245 )
246 if show_output and result.stdout:
247 print("---")
248 sys.stdout.write(result.stdout)
249 if not result.stdout.endswith("\n"):
250 print()
252 return 0 if passed else 1
255def _report(
256 runner_label: str,
257 result: RunnerResult,
258 args: argparse.Namespace,
259 error_msg: str,
260) -> None:
261 """Print an error/timeout report."""
262 if args.output == "json":
263 print_json(
264 {
265 "runner": runner_label,
266 "result": "ERROR",
267 "error": error_msg,
268 "exit_code": result.exit_code,
269 "stdout": result.stdout,
270 "stderr": result.stderr,
271 }
272 )
273 else:
274 print(status_block({"Runner": runner_label, "Result": f"ERROR ({error_msg})"}))
277def cmd_skill(args: argparse.Namespace) -> int:
278 """Invoke a little-loops skill via the active host CLI."""
279 runner_args: list[str] = getattr(args, "runner_args", None) or []
280 parts = [f"/ll:{args.target}"] + runner_args
281 prompt = " ".join(parts)
283 inv = resolve_host().build_streaming(prompt=prompt)
284 runner_label = f"skill {args.target}"
286 try:
287 proc = subprocess.run(
288 [inv.binary, *inv.args],
289 capture_output=True,
290 text=True,
291 timeout=args.timeout,
292 env={**os.environ, **inv.env},
293 )
294 result = RunnerResult(stdout=proc.stdout, stderr=proc.stderr, exit_code=proc.returncode)
295 except subprocess.TimeoutExpired:
296 result = RunnerResult(stdout="", stderr="", exit_code=2, timed_out=True)
297 except FileNotFoundError as e:
298 result = RunnerResult(stdout="", stderr="", exit_code=2, error=str(e))
300 return _evaluate_and_report(runner_label, result, args)
303def cmd_cmd(args: argparse.Namespace) -> int:
304 """Run a shell command with deadlock-safe stderr draining."""
305 runner_label = f"cmd {args.target}"
307 process = subprocess.Popen(
308 ["bash", "-c", args.target],
309 stdout=subprocess.PIPE,
310 stderr=subprocess.PIPE,
311 text=True,
312 )
314 stdout_chunks: list[str] = []
315 stderr_chunks: list[str] = []
317 def _drain_stderr() -> None:
318 assert process.stderr is not None
319 for line in process.stderr:
320 stderr_chunks.append(line)
322 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True)
323 stderr_thread.start()
325 try:
326 assert process.stdout is not None
327 for line in process.stdout:
328 stdout_chunks.append(line)
329 process.wait(timeout=args.timeout)
330 except subprocess.TimeoutExpired:
331 process.kill()
332 process.wait()
333 stderr_thread.join(timeout=5)
334 return _evaluate_and_report(
335 runner_label,
336 RunnerResult(
337 stdout="".join(stdout_chunks),
338 stderr="".join(stderr_chunks),
339 exit_code=2,
340 timed_out=True,
341 ),
342 args,
343 )
345 stderr_thread.join(timeout=5)
346 return _evaluate_and_report(
347 runner_label,
348 RunnerResult(
349 stdout="".join(stdout_chunks),
350 stderr="".join(stderr_chunks),
351 exit_code=process.returncode,
352 ),
353 args,
354 )
357def cmd_mcp(args: argparse.Namespace) -> int:
358 """Call an MCP tool and evaluate the result."""
359 if ":" not in args.target:
360 print(
361 f"Error: MCP target must be 'server:tool', got: {args.target!r}",
362 file=sys.stderr,
363 )
364 return 2
366 server, tool = args.target.split(":", 1)
367 runner_label = f"mcp {args.target}"
369 try:
370 params: dict[str, Any] = json.loads(args.mcp_args)
371 except json.JSONDecodeError as e:
372 print(f"Error: --args is not valid JSON: {e}", file=sys.stderr)
373 return 2
375 response, exit_code = call_mcp_tool(server, tool, params, timeout=args.timeout)
376 result = RunnerResult(stdout=json.dumps(response), stderr="", exit_code=exit_code)
377 return _evaluate_and_report(runner_label, result, args)
380def cmd_prompt(args: argparse.Namespace) -> int:
381 """Send a raw prompt to Claude and evaluate the response."""
382 label_text = args.target[:40] + ("..." if len(args.target) > 40 else "")
383 runner_label = f"prompt {label_text}"
384 inv = resolve_host().build_blocking_json(prompt=args.target, model=args.model)
386 try:
387 proc = subprocess.run(
388 [inv.binary, *inv.args],
389 capture_output=True,
390 text=True,
391 timeout=args.timeout,
392 )
393 result = RunnerResult(stdout=proc.stdout, stderr=proc.stderr, exit_code=proc.returncode)
394 except subprocess.TimeoutExpired:
395 result = RunnerResult(stdout="", stderr="", exit_code=2, timed_out=True)
396 except FileNotFoundError as e:
397 result = RunnerResult(stdout="", stderr="", exit_code=2, error=str(e))
399 return _evaluate_and_report(runner_label, result, args)
402def cmd_dsl(args: argparse.Namespace) -> int:
403 """Run a DSL task set and report pass rates with Wilson CI."""
404 from little_loops.stats import wilson_ci
406 path = Path(args.path)
407 if path.is_dir():
408 task_files = sorted(path.glob("*.yaml"))
409 elif path.is_file():
410 task_files = [path]
411 else:
412 print(f"Error: DSL path not found: {path}", file=sys.stderr)
413 return 2
415 if not task_files:
416 print(f"Error: no .yaml task files found in {path}", file=sys.stderr)
417 return 2
419 pass_count = 0
420 total = 0
422 for task_file in task_files:
423 with open(task_file) as f:
424 data = yaml.safe_load(f)
425 task = DslTask.from_dict(data)
427 prompt_text = task.prompt
428 if task.blanks:
429 prompt_text += f"\n\nBlanks to fill: {task.blanks}"
431 task_args = argparse.Namespace(
432 target=prompt_text,
433 exit_code=args.exit_code,
434 semantic=args.semantic,
435 timeout=args.timeout,
436 output=args.output,
437 verbose=args.verbose,
438 model=args.model,
439 )
440 rc = cmd_prompt(task_args)
441 total += 1
442 if rc == 0:
443 pass_count += 1
445 lo, hi = wilson_ci(pass_count, total)
446 print(f"\nDSL pass-rate: {pass_count}/{total} [{lo:.2f}, {hi:.2f}] (95% CI)")
447 return 0 if pass_count == total else 1
450def main_harness(argv: list[str] | None = None) -> int:
451 """Entry point for ll-harness CLI."""
452 with cli_event_context(DEFAULT_DB_PATH, "ll-harness", sys.argv[1:]):
453 args = _parse_harness_args(argv)
454 configure_output()
455 Logger(use_color=use_color_enabled())
457 if args.runner == "skill":
458 return cmd_skill(args)
459 elif args.runner == "cmd":
460 return cmd_cmd(args)
461 elif args.runner == "mcp":
462 return cmd_mcp(args)
463 elif args.runner == "prompt":
464 return cmd_prompt(args)
465 elif args.runner == "dsl":
466 return cmd_dsl(args)
467 else:
468 print(f"Unknown runner: {args.runner}", file=sys.stderr)
469 return 2