Coverage for little_loops / cli / action.py: 96%

1"""ll-action: Thin CLI wrapper for invoking ll skills as one-shot commands.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import json 

7import subprocess 

8import time 

9from datetime import UTC, datetime 

10from pathlib import Path 

11 

12from little_loops.host_runner import resolve_host 

13 

14__all__ = ["main_action"] 

15 

16 

17def _now_iso() -> str: 

18 return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") 

19 

20 

21def _emit(event: dict) -> None: 

22 print(json.dumps(event), flush=True) 

23 

24 

25def _find_plugin_root() -> Path: 

26 from little_loops.skill_expander import _find_plugin_root as _fpr 

27 

28 return _fpr() 

29 

30 

31def _read_skill_description(skill_md: Path) -> str: 

32 """Extract description from SKILL.md YAML frontmatter.""" 

33 try: 

34 content = skill_md.read_text() 

35 except OSError: 

36 return "" 

37 if not content.startswith("---"): 

38 return "" 

39 end = content.find("---", 3) 

40 if end == -1: 

41 return "" 

42 frontmatter = content[3:end] 

43 for line in frontmatter.splitlines(): 

44 if line.startswith("description:"): 

45 return line[len("description:") :].strip().strip('"').strip("'") 

46 return "" 

47 

48 

49def _load_skills() -> list[dict[str, str]]: 

50 """Return skill list with name and description from skills/*/SKILL.md files.""" 

51 plugin_root = _find_plugin_root() 

52 skills_dir = plugin_root / "skills" 

53 skills = [] 

54 for skill_md in sorted(skills_dir.glob("*/SKILL.md")): 

55 name = skill_md.parent.name 

56 description = _read_skill_description(skill_md) 

57 skills.append({"name": name, "description": description}) 

58 return skills 

59 

60 

61def cmd_invoke(args: argparse.Namespace) -> int: 

62 from little_loops.subprocess_utils import run_claude_command 

63 

64 skill = args.skill 

65 skill_args: list[str] = args.args or [] 

66 timeout: int = args.timeout 

67 output_mode: str = args.output 

68 

69 command = f"/ll:{skill}" 

70 if skill_args: 

71 command += " " + " ".join(skill_args) 

72 

73 start_ms = int(time.time() * 1000) 

74 

75 if output_mode == "stream-json": 

76 _emit({"event": "action_start", "ts": _now_iso(), "skill": skill, "args": skill_args}) 

77 

78 exit_code = 0 

79 

80 def _stream_cb(line: str, is_stderr: bool) -> None: 

81 if not is_stderr: 

82 _emit({"event": "action_output", "ts": _now_iso(), "line": line}) 

83 

84 try: 

85 result = run_claude_command( 

86 command=command, 

87 timeout=timeout, 

88 stream_callback=_stream_cb, 

89 ) 

90 exit_code = result.returncode 

91 except subprocess.TimeoutExpired: 

92 exit_code = 124 

93 

94 duration_ms = int(time.time() * 1000) - start_ms 

95 _emit( 

96 { 

97 "event": "action_complete", 

98 "ts": _now_iso(), 

99 "exit_code": exit_code, 

100 "duration_ms": duration_ms, 

101 } 

102 ) 

103 return exit_code 

104 

105 else: # --output json 

106 from little_loops.cli.output import print_json 

107 

108 output_lines: list[str] = [] 

109 stderr_lines: list[str] = [] 

110 

111 def _stream_cb_json(line: str, is_stderr: bool) -> None: 

112 if is_stderr: 

113 stderr_lines.append(line) 

114 else: 

115 output_lines.append(line) 

116 

117 exit_code = 0 

118 try: 

119 result = run_claude_command( 

120 command=command, 

121 timeout=timeout, 

122 stream_callback=_stream_cb_json, 

123 ) 

124 exit_code = result.returncode 

125 except subprocess.TimeoutExpired: 

126 exit_code = 124 

127 

128 duration_ms = int(time.time() * 1000) - start_ms 

129 print_json( 

130 { 

131 "exit_code": exit_code, 

132 "duration_ms": duration_ms, 

133 "output": "\n".join(output_lines), 

134 "error": "\n".join(stderr_lines) if stderr_lines else None, 

135 } 

136 ) 

137 return exit_code 

138 

139 

140def cmd_capabilities(args: argparse.Namespace) -> int: 

141 from little_loops.cli.output import print_json 

142 

143 runner = resolve_host() 

144 report = runner.describe_capabilities() 

145 

146 available = runner.detect() 

147 version = "" 

148 if available: 

149 try: 

150 invocation = runner.build_version_check() 

151 version_result = subprocess.run( 

152 [invocation.binary, *invocation.args], 

153 capture_output=True, 

154 text=True, 

155 timeout=10, 

156 ) 

157 version = version_result.stdout.strip() 

158 except (subprocess.TimeoutExpired, FileNotFoundError, OSError): 

159 available = False 

160 

161 print_json( 

162 { 

163 "host": report.host, 

164 "binary": report.binary, 

165 "version": version, 

166 "capabilities": [ 

167 {"name": c.name, "status": c.status, "note": c.note} for c in report.capabilities 

168 ], 

169 "hooks": [{"name": h.name, "status": h.status, "note": h.note} for h in report.hooks], 

170 } 

171 ) 

172 return 0 

173 

174 

175def cmd_list(args: argparse.Namespace) -> int: 

176 from little_loops.cli.output import print_json 

177 

178 skills = _load_skills() 

179 print_json(skills) 

180 return 0 

181 

182 

183def main_action() -> int: 

184 """Entry point for ll-action CLI.""" 

185 parser = argparse.ArgumentParser( 

186 prog="ll-action", 

187 description="Invoke ll skills as one-shot commands with JSON-structured output", 

188 formatter_class=argparse.RawDescriptionHelpFormatter, 

189 epilog=""" 

190Examples: 

191 ll-action invoke refine-issue --args P2-ENH-1229 

192 ll-action invoke confidence-check --args FEAT-042 --timeout 120 

193 ll-action invoke refine-issue --args P2-ENH-1229 --output json 

194 ll-action capabilities 

195 ll-action list 

196""", 

197 ) 

198 

199 subparsers = parser.add_subparsers(dest="command", metavar="COMMAND") 

200 subparsers.required = True 

201 

202 # invoke subcommand 

203 invoke_parser = subparsers.add_parser( 

204 "invoke", 

205 help="Invoke a skill and stream output as NDJSON events", 

206 description="Invoke a skill and stream output as NDJSON events (default) or collect and print as JSON", 

207 ) 

208 invoke_parser.add_argument("skill", help="Skill name (e.g. refine-issue, confidence-check)") 

209 invoke_parser.add_argument( 

210 "--args", 

211 nargs="+", 

212 metavar="ARG", 

213 help="Arguments to pass to the skill", 

214 ) 

215 invoke_parser.add_argument( 

216 "--timeout", 

217 type=int, 

218 default=300, 

219 metavar="SECONDS", 

220 help="Timeout in seconds (default: 300)", 

221 ) 

222 invoke_parser.add_argument( 

223 "--output", 

224 choices=["stream-json", "json"], 

225 default="stream-json", 

226 dest="output", 

227 help="Output format: stream-json (default, streaming NDJSON) or json (collect then print)", 

228 ) 

229 

230 # capabilities subcommand 

231 cap_parser = subparsers.add_parser( 

232 "capabilities", 

233 help="Emit full CapabilityReport as JSON (host, binary, version, capabilities, hooks)", 

234 description="Call describe_capabilities() and serialize the full CapabilityReport to JSON", 

235 ) 

236 cap_parser.add_argument( 

237 "--output", 

238 choices=["json"], 

239 default="json", 

240 dest="output", 

241 help="Output format (json only)", 

242 ) 

243 

244 # list subcommand 

245 list_parser = subparsers.add_parser( 

246 "list", 

247 help="List all available skills with descriptions", 

248 description="List all available skills with names and descriptions from plugin manifest", 

249 ) 

250 list_parser.add_argument( 

251 "--output", 

252 choices=["json"], 

253 default="json", 

254 dest="output", 

255 help="Output format (json only)", 

256 ) 

257 

258 parsed = parser.parse_args() 

259 

260 if parsed.command == "invoke": 

261 return cmd_invoke(parsed) 

262 elif parsed.command == "capabilities": 

263 return cmd_capabilities(parsed) 

264 elif parsed.command == "list": 

265 return cmd_list(parsed) 

266 else: 

267 parser.print_help() 

268 return 1