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

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 shutil 

8import subprocess 

9import time 

10from datetime import UTC, datetime 

11from pathlib import Path 

12 

13__all__ = ["main_action"] 

14 

15 

16def _now_iso() -> str: 

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

18 

19 

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

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

22 

23 

24def _find_plugin_root() -> Path: 

25 from little_loops.skill_expander import _find_plugin_root as _fpr 

26 

27 return _fpr() 

28 

29 

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

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

32 try: 

33 content = skill_md.read_text() 

34 except OSError: 

35 return "" 

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

37 return "" 

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

39 if end == -1: 

40 return "" 

41 frontmatter = content[3:end] 

42 for line in frontmatter.splitlines(): 

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

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

45 return "" 

46 

47 

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

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

50 plugin_root = _find_plugin_root() 

51 skills_dir = plugin_root / "skills" 

52 skills = [] 

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

54 name = skill_md.parent.name 

55 description = _read_skill_description(skill_md) 

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

57 return skills 

58 

59 

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

61 from little_loops.subprocess_utils import run_claude_command 

62 

63 skill = args.skill 

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

65 timeout: int = args.timeout 

66 output_mode: str = args.output 

67 

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

69 if skill_args: 

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

71 

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

73 

74 if output_mode == "stream-json": 

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

76 

77 exit_code = 0 

78 

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

80 if not is_stderr: 

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

82 

83 try: 

84 result = run_claude_command( 

85 command=command, 

86 timeout=timeout, 

87 stream_callback=_stream_cb, 

88 ) 

89 exit_code = result.returncode 

90 except subprocess.TimeoutExpired: 

91 exit_code = 124 

92 

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

94 _emit( 

95 { 

96 "event": "action_complete", 

97 "ts": _now_iso(), 

98 "exit_code": exit_code, 

99 "duration_ms": duration_ms, 

100 } 

101 ) 

102 return exit_code 

103 

104 else: # --output json 

105 from little_loops.cli.output import print_json 

106 

107 output_lines: list[str] = [] 

108 stderr_lines: list[str] = [] 

109 

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

111 if is_stderr: 

112 stderr_lines.append(line) 

113 else: 

114 output_lines.append(line) 

115 

116 exit_code = 0 

117 try: 

118 result = run_claude_command( 

119 command=command, 

120 timeout=timeout, 

121 stream_callback=_stream_cb_json, 

122 ) 

123 exit_code = result.returncode 

124 except subprocess.TimeoutExpired: 

125 exit_code = 124 

126 

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

128 print_json( 

129 { 

130 "exit_code": exit_code, 

131 "duration_ms": duration_ms, 

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

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

134 } 

135 ) 

136 return exit_code 

137 

138 

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

140 from little_loops.cli.output import print_json 

141 

142 claude_path = shutil.which("claude") 

143 available = claude_path is not None 

144 

145 version = "" 

146 if available: 

147 try: 

148 version_result = subprocess.run( 

149 ["claude", "--version"], 

150 capture_output=True, 

151 text=True, 

152 timeout=10, 

153 ) 

154 version = version_result.stdout.strip() 

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

156 available = False 

157 

158 skills = _load_skills() 

159 supported_skills = [s["name"] for s in skills] 

160 

161 print_json( 

162 { 

163 "available": available, 

164 "version": version, 

165 "supported_skills": supported_skills, 

166 } 

167 ) 

168 return 0 

169 

170 

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

172 from little_loops.cli.output import print_json 

173 

174 skills = _load_skills() 

175 print_json(skills) 

176 return 0 

177 

178 

179def main_action() -> int: 

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

181 parser = argparse.ArgumentParser( 

182 prog="ll-action", 

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

184 formatter_class=argparse.RawDescriptionHelpFormatter, 

185 epilog=""" 

186Examples: 

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

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

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

190 ll-action capabilities 

191 ll-action list 

192""", 

193 ) 

194 

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

196 subparsers.required = True 

197 

198 # invoke subcommand 

199 invoke_parser = subparsers.add_parser( 

200 "invoke", 

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

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

203 ) 

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

205 invoke_parser.add_argument( 

206 "--args", 

207 nargs="+", 

208 metavar="ARG", 

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

210 ) 

211 invoke_parser.add_argument( 

212 "--timeout", 

213 type=int, 

214 default=300, 

215 metavar="SECONDS", 

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

217 ) 

218 invoke_parser.add_argument( 

219 "--output", 

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

221 default="stream-json", 

222 dest="output", 

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

224 ) 

225 

226 # capabilities subcommand 

227 cap_parser = subparsers.add_parser( 

228 "capabilities", 

229 help="Check Claude availability and list supported skills", 

230 description="Probe claude availability and return supported skill names as JSON", 

231 ) 

232 cap_parser.add_argument( 

233 "--output", 

234 choices=["json"], 

235 default="json", 

236 dest="output", 

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

238 ) 

239 

240 # list subcommand 

241 list_parser = subparsers.add_parser( 

242 "list", 

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

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

245 ) 

246 list_parser.add_argument( 

247 "--output", 

248 choices=["json"], 

249 default="json", 

250 dest="output", 

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

252 ) 

253 

254 parsed = parser.parse_args() 

255 

256 if parsed.command == "invoke": 

257 return cmd_invoke(parsed) 

258 elif parsed.command == "capabilities": 

259 return cmd_capabilities(parsed) 

260 elif parsed.command == "list": 

261 return cmd_list(parsed) 

262 else: 

263 parser.print_help() 

264 return 1