Coverage for little_loops / cli / action.py: 0%
123 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -0500
1"""ll-action: Thin CLI wrapper for invoking ll skills as one-shot commands."""
3from __future__ import annotations
5import argparse
6import json
7import subprocess
8import time
9from datetime import UTC, datetime
10from pathlib import Path
12from little_loops.host_runner import resolve_host
14__all__ = ["main_action"]
17def _now_iso() -> str:
18 return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
21def _emit(event: dict) -> None:
22 print(json.dumps(event), flush=True)
25def _find_plugin_root() -> Path:
26 from little_loops.skill_expander import _find_plugin_root as _fpr
28 return _fpr()
31def _read_skill_description(skill_md: Path) -> str:
32 """Extract description from SKILL.md YAML frontmatter."""
33 from little_loops.frontmatter import parse_skill_frontmatter
35 try:
36 content = skill_md.read_text()
37 except OSError:
38 return ""
39 fm = parse_skill_frontmatter(content)
40 return fm.get("description", "").strip().strip('"').strip("'")
43def _load_skills() -> list[dict[str, str | None]]:
44 """Return skill list with name, description, and args from skills/*/SKILL.md files."""
45 from little_loops.frontmatter import parse_skill_frontmatter
47 plugin_root = _find_plugin_root()
48 skills_dir = plugin_root / "skills"
49 skills = []
50 for skill_md in sorted(skills_dir.glob("*/SKILL.md")):
51 name = skill_md.parent.name
52 try:
53 content = skill_md.read_text()
54 except OSError:
55 content = ""
56 fm = parse_skill_frontmatter(content) if content else {}
57 description = str(fm.get("description", "") or "").strip().strip('"').strip("'")
58 # args: takes precedence over argument-hint: (aliasing for 19 existing skills)
59 raw_args = fm.get("args") or fm.get("argument-hint")
60 args: str | None = str(raw_args).strip().strip('"').strip("'") if raw_args else None
61 skills.append({"name": name, "description": description, "args": args})
62 return skills
65def cmd_invoke(args: argparse.Namespace) -> int:
66 from little_loops.subprocess_utils import run_claude_command
68 skill = args.skill
69 skill_args: list[str] = args.args or []
70 timeout: int = args.timeout
71 output_mode: str = args.output
73 command = f"/ll:{skill}"
74 if skill_args:
75 command += " " + " ".join(skill_args)
77 start_ms = int(time.time() * 1000)
79 if output_mode == "stream-json":
80 _emit({"event": "action_start", "ts": _now_iso(), "skill": skill, "args": skill_args})
82 exit_code = 0
84 def _stream_cb(line: str, is_stderr: bool) -> None:
85 if not is_stderr:
86 _emit({"event": "action_output", "ts": _now_iso(), "line": line})
88 try:
89 result = run_claude_command(
90 command=command,
91 timeout=timeout,
92 stream_callback=_stream_cb,
93 )
94 exit_code = result.returncode
95 except subprocess.TimeoutExpired:
96 exit_code = 124
98 duration_ms = int(time.time() * 1000) - start_ms
99 _emit(
100 {
101 "event": "action_complete",
102 "ts": _now_iso(),
103 "exit_code": exit_code,
104 "duration_ms": duration_ms,
105 }
106 )
107 return exit_code
109 else: # --output json
110 from little_loops.cli.output import print_json
112 output_lines: list[str] = []
113 stderr_lines: list[str] = []
115 def _stream_cb_json(line: str, is_stderr: bool) -> None:
116 if is_stderr:
117 stderr_lines.append(line)
118 else:
119 output_lines.append(line)
121 exit_code = 0
122 try:
123 result = run_claude_command(
124 command=command,
125 timeout=timeout,
126 stream_callback=_stream_cb_json,
127 )
128 exit_code = result.returncode
129 except subprocess.TimeoutExpired:
130 exit_code = 124
132 duration_ms = int(time.time() * 1000) - start_ms
133 print_json(
134 {
135 "exit_code": exit_code,
136 "duration_ms": duration_ms,
137 "output": "\n".join(output_lines),
138 "error": "\n".join(stderr_lines) if stderr_lines else None,
139 }
140 )
141 return exit_code
144def cmd_capabilities(args: argparse.Namespace) -> int:
145 from little_loops.cli.output import print_json
147 runner = resolve_host()
148 report = runner.describe_capabilities()
150 available = runner.detect()
151 version = ""
152 if available:
153 try:
154 invocation = runner.build_version_check()
155 version_result = subprocess.run(
156 [invocation.binary, *invocation.args],
157 capture_output=True,
158 text=True,
159 timeout=10,
160 )
161 version = version_result.stdout.strip()
162 except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
163 available = False
165 print_json(
166 {
167 "host": report.host,
168 "binary": report.binary,
169 "version": version,
170 "capabilities": [
171 {"name": c.name, "status": c.status, "note": c.note} for c in report.capabilities
172 ],
173 "hooks": [{"name": h.name, "status": h.status, "note": h.note} for h in report.hooks],
174 }
175 )
176 return 0
179def cmd_list(args: argparse.Namespace) -> int:
180 from little_loops.cli.output import print_json
182 skills = _load_skills()
183 print_json(skills)
184 return 0
187def main_action() -> int:
188 """Entry point for ll-action CLI."""
189 parser = argparse.ArgumentParser(
190 prog="ll-action",
191 description="Invoke ll skills as one-shot commands with JSON-structured output",
192 formatter_class=argparse.RawDescriptionHelpFormatter,
193 epilog="""
194Examples:
195 ll-action invoke refine-issue --args P2-ENH-1229
196 ll-action invoke confidence-check --args FEAT-042 --timeout 120
197 ll-action invoke refine-issue --args P2-ENH-1229 --output json
198 ll-action capabilities
199 ll-action list
200""",
201 )
203 subparsers = parser.add_subparsers(dest="command", metavar="COMMAND")
204 subparsers.required = True
206 # invoke subcommand
207 invoke_parser = subparsers.add_parser(
208 "invoke",
209 help="Invoke a skill and stream output as NDJSON events",
210 description="Invoke a skill and stream output as NDJSON events (default) or collect and print as JSON",
211 )
212 invoke_parser.add_argument("skill", help="Skill name (e.g. refine-issue, confidence-check)")
213 invoke_parser.add_argument(
214 "--args",
215 nargs="+",
216 metavar="ARG",
217 help="Arguments to pass to the skill",
218 )
219 invoke_parser.add_argument(
220 "--timeout",
221 type=int,
222 default=300,
223 metavar="SECONDS",
224 help="Timeout in seconds (default: 300)",
225 )
226 invoke_parser.add_argument(
227 "--output",
228 choices=["stream-json", "json"],
229 default="stream-json",
230 dest="output",
231 help="Output format: stream-json (default, streaming NDJSON) or json (collect then print)",
232 )
234 # capabilities subcommand
235 cap_parser = subparsers.add_parser(
236 "capabilities",
237 help="Emit full CapabilityReport as JSON (host, binary, version, capabilities, hooks)",
238 description="Call describe_capabilities() and serialize the full CapabilityReport to JSON",
239 )
240 cap_parser.add_argument(
241 "--output",
242 choices=["json"],
243 default="json",
244 dest="output",
245 help="Output format (json only)",
246 )
248 # list subcommand
249 list_parser = subparsers.add_parser(
250 "list",
251 help="List all available skills with descriptions",
252 description="List all available skills with names and descriptions from plugin manifest",
253 )
254 list_parser.add_argument(
255 "--output",
256 choices=["json"],
257 default="json",
258 dest="output",
259 help="Output format (json only)",
260 )
262 parsed = parser.parse_args()
264 if parsed.command == "invoke":
265 return cmd_invoke(parsed)
266 elif parsed.command == "capabilities":
267 return cmd_capabilities(parsed)
268 elif parsed.command == "list":
269 return cmd_list(parsed)
270 else:
271 parser.print_help()
272 return 1