Coverage for little_loops / cli / loop / next_loop.py: 82%
181 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-22 16:19 -0500
1"""ll-loop next-loop: Suggest the next loop to run from execution history."""
3from __future__ import annotations
5import argparse
6import json
7from dataclasses import dataclass
8from datetime import UTC, datetime
9from pathlib import Path
10from typing import Any
12from little_loops.logger import Logger
15@dataclass
16class LoopCandidate:
17 """A scored loop candidate for next-loop suggestions."""
19 loop: str
20 score: float
21 input: str | None
22 context: dict[str, str]
23 rationale: str
24 command: str
25 run_count: int = 0
26 last_run: str | None = None
27 success_rate: float = 1.0
29 def to_dict(self) -> dict[str, Any]:
30 return {
31 "loop": self.loop,
32 "input": self.input,
33 "context": self.context,
34 "score": round(self.score, 4),
35 "rationale": self.rationale,
36 "command": self.command,
37 }
40# ---------------------------------------------------------------------------
41# History scanning
42# ---------------------------------------------------------------------------
45def _scan_history(loops_dir: Path) -> dict[str, list[dict[str, Any]]]:
46 """Scan .loops/.history/ and return per-loop run metadata.
48 Returns dict mapping loop_name → list of {status, started_at, iterations}.
49 """
50 from little_loops.fsm.persistence import HISTORY_DIR, _parse_run_folder
52 history_base = loops_dir / HISTORY_DIR
53 if not history_base.exists():
54 return {}
56 per_loop: dict[str, list[dict[str, Any]]] = {}
57 for run_dir in sorted(history_base.iterdir()):
58 if not run_dir.is_dir():
59 continue
60 parsed = _parse_run_folder(run_dir.name)
61 if not parsed:
62 continue
63 run_id, loop_name = parsed
64 state_file = run_dir / "state.json"
65 entry: dict[str, Any] = {"run_id": run_id, "status": None, "started_at": None}
66 if state_file.exists():
67 try:
68 data = json.loads(state_file.read_text())
69 entry["status"] = data.get("status")
70 entry["started_at"] = data.get("started_at")
71 except (ValueError, OSError):
72 pass
73 per_loop.setdefault(loop_name, []).append(entry)
75 return per_loop
78# ---------------------------------------------------------------------------
79# Scoring
80# ---------------------------------------------------------------------------
82_SUCCESS_STATUSES = {"completed"}
83_DECAY_HALF_LIFE_DAYS = 7.0 # recency decay: halves every 7 days
86def _recency_score(started_at: str | None) -> float:
87 """Exponential decay score in [0, 1] based on days since last run."""
88 if not started_at:
89 return 0.0
90 try:
91 ts = datetime.fromisoformat(started_at.replace("Z", "+00:00"))
92 now = datetime.now(UTC)
93 days = (now - ts).total_seconds() / 86400.0
94 import math
96 return math.exp(-days * math.log(2) / _DECAY_HALF_LIFE_DAYS)
97 except (ValueError, TypeError):
98 return 0.0
101def _score_loop(runs: list[dict[str, Any]]) -> tuple[float, float, str | None]:
102 """Return (score, success_rate, last_started_at)."""
103 if not runs:
104 return 0.0, 1.0, None
106 count = len(runs)
107 successes = sum(1 for r in runs if r.get("status") in _SUCCESS_STATUSES)
108 success_rate = successes / count if count else 1.0
110 # Most recent run for recency
111 dated = [r for r in runs if r.get("started_at")]
112 dated.sort(key=lambda r: r["started_at"], reverse=True)
113 last_started_at = dated[0]["started_at"] if dated else None
115 recency = _recency_score(last_started_at)
116 # Weighted: 50% frequency (log-scale), 30% recency, 20% success rate
117 import math
119 freq_score = math.log1p(count) / math.log1p(50) # cap normalisation at 50 runs
120 score = 0.50 * freq_score + 0.30 * recency + 0.20 * success_rate
121 return score, success_rate, last_started_at
124# ---------------------------------------------------------------------------
125# Parameter resolver registry
126# ---------------------------------------------------------------------------
128_ParamResolver = dict[str, str | None] # {input: ..., **context_keys}
131def _resolve_autodev_params(_loops_dir: Path) -> _ParamResolver: # noqa: ARG001
132 """Resolve input params for autodev: return space-joined active issue IDs."""
133 try:
134 from pathlib import Path as _Path
136 from little_loops.cli.issues.search import _load_issues_with_status
137 from little_loops.config import BRConfig
139 config = BRConfig(_Path.cwd())
140 raw = _load_issues_with_status(
141 config, include_open=True, include_done=False, include_deferred=False
142 )
143 ids = [issue.issue_id for issue, _ in raw if issue.issue_id]
144 if ids:
145 return {"input": " ".join(ids)}
146 except Exception:
147 pass
148 return {}
151# Registry: loop name → resolver callable
152_PARAM_RESOLVERS: dict[str, Any] = {
153 "autodev": _resolve_autodev_params,
154}
157def _resolve_params(loop_name: str, loops_dir: Path) -> _ParamResolver:
158 """Return resolved params for a loop, falling back to empty dict."""
159 resolver = _PARAM_RESOLVERS.get(loop_name)
160 if resolver is not None:
161 try:
162 return resolver(loops_dir)
163 except Exception:
164 pass
165 return {}
168# ---------------------------------------------------------------------------
169# Command building
170# ---------------------------------------------------------------------------
173def _build_command(loop_name: str, params: _ParamResolver) -> str:
174 """Build a shell-ready ll-loop run command string."""
175 parts = ["ll-loop", "run", loop_name]
176 if params.get("input"):
177 parts.append(json.dumps(params["input"]))
178 for k, v in params.items():
179 if k != "input" and v is not None:
180 parts.extend(["--context", f"{k}={v}"])
181 return " ".join(parts)
184def _build_rationale(
185 run_count: int,
186 success_rate: float,
187 last_started_at: str | None,
188 param_note: str,
189) -> str:
190 parts = [f"{run_count} run{'s' if run_count != 1 else ''}"]
191 if last_started_at:
192 try:
193 ts = datetime.fromisoformat(last_started_at.replace("Z", "+00:00"))
194 ago = datetime.now(UTC) - ts
195 days = int(ago.total_seconds() / 86400)
196 if days == 0:
197 parts.append("last run today")
198 elif days == 1:
199 parts.append("last run yesterday")
200 else:
201 parts.append(f"last run {days}d ago")
202 except (ValueError, TypeError):
203 pass
204 parts.append(f"{int(success_rate * 100)}% success")
205 if param_note:
206 parts.append(param_note)
207 return "; ".join(parts)
210# ---------------------------------------------------------------------------
211# cmd_next_loop
212# ---------------------------------------------------------------------------
215def cmd_next_loop(
216 args: argparse.Namespace,
217 loops_dir: Path,
218 logger: Logger,
219) -> int:
220 """Suggest the next loop(s) to run based on execution history."""
221 from little_loops.cli.output import colorize, print_json
223 count = getattr(args, "count", 1)
224 as_json = getattr(args, "json", False)
225 execute = getattr(args, "execute", False)
226 exclude = set(getattr(args, "exclude", None) or [])
228 history = _scan_history(loops_dir)
229 if not history:
230 if as_json:
231 print_json([])
232 else:
233 print("No loop history available. Run some loops first.")
234 return 1
236 # Score each loop
237 scored: list[tuple[float, str, float, str | None, int]] = []
238 for loop_name, runs in history.items():
239 if loop_name in exclude:
240 continue
241 score, success_rate, last_started_at = _score_loop(runs)
242 scored.append((score, loop_name, success_rate, last_started_at, len(runs)))
244 if not scored:
245 if as_json:
246 print_json([])
247 else:
248 print("No candidates after applying exclusions.")
249 return 1
251 scored.sort(key=lambda t: t[0], reverse=True)
252 top = scored[:count]
254 candidates: list[LoopCandidate] = []
255 for score, loop_name, success_rate, last_started_at, run_count in top:
256 params = _resolve_params(loop_name, loops_dir)
257 param_note = ""
258 resolved_input = params.get("input")
259 if resolved_input:
260 item_count = len(resolved_input.split())
261 param_note = (
262 f"input resolved ({item_count} items)" if item_count > 1 else "input resolved"
263 )
264 elif loop_name in _PARAM_RESOLVERS:
265 param_note = "input resolver found no active items"
267 rationale = _build_rationale(run_count, success_rate, last_started_at, param_note)
268 command = _build_command(loop_name, params)
270 candidate = LoopCandidate(
271 loop=loop_name,
272 score=score,
273 input=params.get("input"),
274 context={k: v for k, v in params.items() if k != "input" and v is not None},
275 rationale=rationale,
276 command=command,
277 run_count=run_count,
278 last_run=last_started_at,
279 success_rate=success_rate,
280 )
281 candidates.append(candidate)
283 if as_json:
284 print_json([c.to_dict() for c in candidates])
285 return 0
287 # Text output
288 label = "suggestion" if len(candidates) == 1 else "suggestions"
289 print(colorize(f"Next loop {label}:", "1"))
290 print()
291 for i, c in enumerate(candidates, 1):
292 rank = colorize(f"#{i}", "36;1")
293 name = colorize(c.loop, "1")
294 score_str = colorize(f"score={c.score:.3f}", "2")
295 print(f" {rank} {name} {score_str}")
296 print(f" {colorize(c.rationale, '2')}")
297 print(f" {colorize('$', '32')} {c.command}")
298 print()
300 if execute:
301 top_candidate = candidates[0]
302 logger.info(f"Executing: {top_candidate.command}")
303 # Build minimal Namespace to call cmd_run
304 from little_loops.cli.loop.run import cmd_run
306 run_args = argparse.Namespace(
307 input=top_candidate.input,
308 max_iterations=None,
309 delay=None,
310 no_llm=False,
311 llm_model=None,
312 dry_run=False,
313 background=False,
314 foreground_internal=False,
315 instance_id=None,
316 quiet=False,
317 verbose=False,
318 show_diagrams=False,
319 clear=False,
320 queue=False,
321 context=[f"{k}={v}" for k, v in top_candidate.context.items()],
322 program_md=None,
323 builtin=False,
324 worktree=False,
325 handoff_threshold=None,
326 context_limit=None,
327 )
328 return cmd_run(top_candidate.loop, run_args, loops_dir, logger)
330 return 0