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