Coverage for little_loops / cli / harness.py: 19%

1"""ll-harness: One-shot runner evaluation CLI (FEAT-1851).""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import json 

7import subprocess 

8import sys 

9import threading 

10from dataclasses import dataclass 

11from typing import Any 

12 

13from little_loops.cli.output import configure_output, print_json, status_block, use_color_enabled 

14from little_loops.fsm.evaluators import evaluate_llm_structured 

15from little_loops.host_runner import resolve_host 

16from little_loops.logger import Logger 

17from little_loops.mcp_call import call_mcp_tool 

18from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context 

19 

20 

21@dataclass 

22class RunnerResult: 

23 """Captured output from a runner invocation.""" 

24 

25 stdout: str 

26 stderr: str 

27 exit_code: int 

28 timed_out: bool = False 

29 error: str | None = None 

30 

31 

32def _build_harness_parser() -> argparse.ArgumentParser: 

33 """Build the ll-harness argument parser (exposed for testing).""" 

34 parser = argparse.ArgumentParser( 

35 prog="ll-harness", 

36 description="One-shot runner evaluation for little-loops skills and commands", 

37 formatter_class=argparse.RawDescriptionHelpFormatter, 

38 epilog=""" 

39Examples: 

40 ll-harness skill check-code 

41 ll-harness cmd "echo hello" --exit-code 0 

42 ll-harness mcp my-server:my-tool --args '{"key": "val"}' --semantic "tool returned results" 

43 ll-harness prompt "What is 2+2?" --semantic "response contains a number" 

44 

45Exit codes: 

46 0 PASS 

47 1 FAIL 

48 2 Internal error / timeout 

49""", 

50 ) 

51 

52 subparsers = parser.add_subparsers(dest="runner", metavar="RUNNER") 

53 subparsers.required = True 

54 

55 def _add_evaluator_flags(p: argparse.ArgumentParser) -> None: 

56 p.add_argument( 

57 "--exit-code", 

58 dest="exit_code", 

59 type=int, 

60 default=None, 

61 metavar="INT", 

62 help="Expected exit code (default: not checked)", 

63 ) 

64 p.add_argument( 

65 "--semantic", 

66 type=str, 

67 default=None, 

68 metavar="TEXT", 

69 help="Natural-language criterion for output evaluation", 

70 ) 

71 p.add_argument( 

72 "--timeout", 

73 type=int, 

74 default=120, 

75 metavar="SECONDS", 

76 help="Runner timeout in seconds (default: 120)", 

77 ) 

78 p.add_argument( 

79 "--output", 

80 choices=["text", "json"], 

81 default="text", 

82 help="Output format (default: text)", 

83 ) 

84 p.add_argument( 

85 "--verbose", 

86 action="store_true", 

87 help="Show full captured output even on pass", 

88 ) 

89 

90 skill_p = subparsers.add_parser( 

91 "skill", 

92 help="Invoke a little-loops skill", 

93 description="Invoke a little-loops skill via the active host CLI", 

94 ) 

95 skill_p.add_argument("target", help="Skill name (e.g. check-code, refine-issue)") 

96 skill_p.add_argument( 

97 "runner_args", 

98 nargs="*", 

99 help="Additional arguments passed to the skill", 

100 ) 

101 _add_evaluator_flags(skill_p) 

102 

103 cmd_p = subparsers.add_parser( 

104 "cmd", 

105 help="Run a shell command", 

106 description="Run a shell command and capture its output", 

107 ) 

108 cmd_p.add_argument("target", help="Shell command to execute") 

109 _add_evaluator_flags(cmd_p) 

110 

111 mcp_p = subparsers.add_parser( 

112 "mcp", 

113 help="Call an MCP tool", 

114 description="Call an MCP tool via JSON-RPC", 

115 ) 

116 mcp_p.add_argument("target", help="MCP server and tool (format: server:tool)") 

117 mcp_p.add_argument( 

118 "--args", 

119 dest="mcp_args", 

120 type=str, 

121 default="{}", 

122 metavar="JSON", 

123 help="JSON arguments to pass to the MCP tool (default: {})", 

124 ) 

125 _add_evaluator_flags(mcp_p) 

126 

127 prompt_p = subparsers.add_parser( 

128 "prompt", 

129 help="Send a raw prompt to Claude", 

130 description="Send a raw prompt to Claude via the active host CLI", 

131 ) 

132 prompt_p.add_argument("target", help="Prompt text to send") 

133 _add_evaluator_flags(prompt_p) 

134 

135 return parser 

136 

137 

138def _parse_harness_args(argv: list[str] | None) -> argparse.Namespace: 

139 """Parse argv into a Namespace (exposed for testing).""" 

140 return _build_harness_parser().parse_args(argv) 

141 

142 

143def _evaluate_and_report( 

144 runner_label: str, 

145 result: RunnerResult, 

146 args: argparse.Namespace, 

147) -> int: 

148 """Evaluate result against criteria and print the report. Returns exit code.""" 

149 if result.timed_out: 

150 _report(runner_label, result, args, error_msg="timeout") 

151 return 2 

152 if result.error is not None: 

153 _report(runner_label, result, args, error_msg=result.error) 

154 return 2 

155 

156 passed = True 

157 exit_code_display = str(result.exit_code) 

158 semantic_display = "[not checked]" 

159 

160 if args.exit_code is not None: 

161 if result.exit_code != args.exit_code: 

162 passed = False 

163 exit_code_display = f"{result.exit_code} (expected {args.exit_code})" 

164 

165 if args.semantic is not None: 

166 eval_result = evaluate_llm_structured(output=result.stdout, prompt=args.semantic) 

167 semantic_display = eval_result.verdict 

168 if eval_result.verdict != "yes": 

169 passed = False 

170 

171 overall = "PASS" if passed else "FAIL" 

172 show_output = not passed or args.verbose 

173 

174 if args.output == "json": 

175 print_json( 

176 { 

177 "runner": runner_label, 

178 "exit_code": result.exit_code, 

179 "exit_code_check": exit_code_display, 

180 "semantic": semantic_display, 

181 "result": overall, 

182 "stdout": result.stdout, 

183 "stderr": result.stderr, 

184 } 

185 ) 

186 else: 

187 print( 

188 status_block( 

189 { 

190 "Runner": runner_label, 

191 "Exit": exit_code_display, 

192 "Semantic": semantic_display, 

193 "Result": overall, 

194 } 

195 ) 

196 ) 

197 if show_output and result.stdout: 

198 print("---") 

199 sys.stdout.write(result.stdout) 

200 if not result.stdout.endswith("\n"): 

201 print() 

202 

203 return 0 if passed else 1 

204 

205 

206def _report( 

207 runner_label: str, 

208 result: RunnerResult, 

209 args: argparse.Namespace, 

210 error_msg: str, 

211) -> None: 

212 """Print an error/timeout report.""" 

213 if args.output == "json": 

214 print_json( 

215 { 

216 "runner": runner_label, 

217 "result": "ERROR", 

218 "error": error_msg, 

219 "exit_code": result.exit_code, 

220 "stdout": result.stdout, 

221 "stderr": result.stderr, 

222 } 

223 ) 

224 else: 

225 print(status_block({"Runner": runner_label, "Result": f"ERROR ({error_msg})"})) 

226 

227 

228def cmd_skill(args: argparse.Namespace) -> int: 

229 """Invoke a little-loops skill via the active host CLI.""" 

230 runner_args: list[str] = getattr(args, "runner_args", None) or [] 

231 parts = [f"/ll:{args.target}"] + runner_args 

232 prompt = " ".join(parts) 

233 

234 inv = resolve_host().build_streaming(prompt=prompt) 

235 runner_label = f"skill {args.target}" 

236 

237 try: 

238 proc = subprocess.run( 

239 [inv.binary, *inv.args], 

240 capture_output=True, 

241 text=True, 

242 timeout=args.timeout, 

243 ) 

244 result = RunnerResult(stdout=proc.stdout, stderr=proc.stderr, exit_code=proc.returncode) 

245 except subprocess.TimeoutExpired: 

246 result = RunnerResult(stdout="", stderr="", exit_code=2, timed_out=True) 

247 except FileNotFoundError as e: 

248 result = RunnerResult(stdout="", stderr="", exit_code=2, error=str(e)) 

249 

250 return _evaluate_and_report(runner_label, result, args) 

251 

252 

253def cmd_cmd(args: argparse.Namespace) -> int: 

254 """Run a shell command with deadlock-safe stderr draining.""" 

255 runner_label = f"cmd {args.target}" 

256 

257 process = subprocess.Popen( 

258 ["bash", "-c", args.target], 

259 stdout=subprocess.PIPE, 

260 stderr=subprocess.PIPE, 

261 text=True, 

262 ) 

263 

264 stdout_chunks: list[str] = [] 

265 stderr_chunks: list[str] = [] 

266 

267 def _drain_stderr() -> None: 

268 assert process.stderr is not None 

269 for line in process.stderr: 

270 stderr_chunks.append(line) 

271 

272 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True) 

273 stderr_thread.start() 

274 

275 try: 

276 assert process.stdout is not None 

277 for line in process.stdout: 

278 stdout_chunks.append(line) 

279 process.wait(timeout=args.timeout) 

280 except subprocess.TimeoutExpired: 

281 process.kill() 

282 process.wait() 

283 stderr_thread.join(timeout=5) 

284 return _evaluate_and_report( 

285 runner_label, 

286 RunnerResult( 

287 stdout="".join(stdout_chunks), 

288 stderr="".join(stderr_chunks), 

289 exit_code=2, 

290 timed_out=True, 

291 ), 

292 args, 

293 ) 

294 

295 stderr_thread.join(timeout=5) 

296 return _evaluate_and_report( 

297 runner_label, 

298 RunnerResult( 

299 stdout="".join(stdout_chunks), 

300 stderr="".join(stderr_chunks), 

301 exit_code=process.returncode, 

302 ), 

303 args, 

304 ) 

305 

306 

307def cmd_mcp(args: argparse.Namespace) -> int: 

308 """Call an MCP tool and evaluate the result.""" 

309 if ":" not in args.target: 

310 print( 

311 f"Error: MCP target must be 'server:tool', got: {args.target!r}", 

312 file=sys.stderr, 

313 ) 

314 return 2 

315 

316 server, tool = args.target.split(":", 1) 

317 runner_label = f"mcp {args.target}" 

318 

319 try: 

320 params: dict[str, Any] = json.loads(args.mcp_args) 

321 except json.JSONDecodeError as e: 

322 print(f"Error: --args is not valid JSON: {e}", file=sys.stderr) 

323 return 2 

324 

325 response, exit_code = call_mcp_tool(server, tool, params, timeout=args.timeout) 

326 result = RunnerResult(stdout=json.dumps(response), stderr="", exit_code=exit_code) 

327 return _evaluate_and_report(runner_label, result, args) 

328 

329 

330def cmd_prompt(args: argparse.Namespace) -> int: 

331 """Send a raw prompt to Claude and evaluate the response.""" 

332 label_text = args.target[:40] + ("..." if len(args.target) > 40 else "") 

333 runner_label = f"prompt {label_text}" 

334 inv = resolve_host().build_blocking_json(prompt=args.target) 

335 

336 try: 

337 proc = subprocess.run( 

338 [inv.binary, *inv.args], 

339 capture_output=True, 

340 text=True, 

341 timeout=args.timeout, 

342 ) 

343 result = RunnerResult(stdout=proc.stdout, stderr=proc.stderr, exit_code=proc.returncode) 

344 except subprocess.TimeoutExpired: 

345 result = RunnerResult(stdout="", stderr="", exit_code=2, timed_out=True) 

346 except FileNotFoundError as e: 

347 result = RunnerResult(stdout="", stderr="", exit_code=2, error=str(e)) 

348 

349 return _evaluate_and_report(runner_label, result, args) 

350 

351 

352def main_harness(argv: list[str] | None = None) -> int: 

353 """Entry point for ll-harness CLI.""" 

354 with cli_event_context(DEFAULT_DB_PATH, "ll-harness", sys.argv[1:]): 

355 args = _parse_harness_args(argv) 

356 configure_output() 

357 Logger(use_color=use_color_enabled()) 

358 

359 if args.runner == "skill": 

360 return cmd_skill(args) 

361 elif args.runner == "cmd": 

362 return cmd_cmd(args) 

363 elif args.runner == "mcp": 

364 return cmd_mcp(args) 

365 elif args.runner == "prompt": 

366 return cmd_prompt(args) 

367 else: 

368 print(f"Unknown runner: {args.runner}", file=sys.stderr) 

369 return 2