Coverage for little_loops / cli / loop / run.py: 0%
257 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -0500
1"""ll-loop run subcommand."""
3from __future__ import annotations
5import argparse
6import atexit
7import json
8import os
9import re
10import time
11import uuid
12from datetime import UTC, datetime
13from pathlib import Path
15from little_loops.cli.loop._helpers import (
16 _is_earliest_waiter,
17 _make_instance_id,
18 get_builtin_loops_dir,
19 print_execution_plan,
20 register_loop_signal_handlers,
21 resolve_loop_path,
22 run_background,
23 run_foreground,
24)
25from little_loops.logger import Logger
28def _parse_program_md(path: Path) -> dict[str, str]:
29 """Parse .ll/program.md heading sections into context key-value pairs.
31 Sections mapped:
32 ## Directive → directive (prose)
33 ## Targets → targets (space-joined list items)
34 ## Benchmark → each key: value pair injected directly
35 ## Budget → budget (prose)
36 ## Constraints → constraints (prose)
37 """
38 if not path.exists():
39 return {}
40 try:
41 content = path.read_text()
42 except OSError:
43 return {}
45 def _extract(heading: str) -> str:
46 m = re.search(rf"^##\s+{re.escape(heading)}\s*$", content, re.MULTILINE | re.IGNORECASE)
47 if not m:
48 return ""
49 start = m.end()
50 nxt = re.search(r"^##\s", content[start:], re.MULTILINE)
51 return content[start : start + nxt.start()].strip() if nxt else content[start:].strip()
53 result: dict[str, str] = {}
55 directive = _extract("Directive")
56 if directive:
57 result["directive"] = directive
59 targets_text = _extract("Targets")
60 if targets_text:
61 items = [
62 line.lstrip("-* \t").strip()
63 for line in targets_text.splitlines()
64 if line.strip().startswith(("-", "*"))
65 ]
66 result["targets"] = " ".join(items) if items else targets_text
68 benchmark_text = _extract("Benchmark")
69 if benchmark_text:
70 for line in benchmark_text.splitlines():
71 if ":" in line:
72 k, _, v = line.partition(":")
73 k, v = k.strip(), v.strip()
74 if k and v:
75 result[k] = v
77 budget = _extract("Budget")
78 if budget:
79 result["budget"] = budget
81 constraints = _extract("Constraints")
82 if constraints:
83 result["constraints"] = constraints
85 return result
88def cmd_run(
89 loop_name: str,
90 args: argparse.Namespace,
91 loops_dir: Path,
92 logger: Logger,
93) -> int:
94 """Run a loop."""
95 from little_loops.fsm.concurrency import LockManager, resolve_scope
96 from little_loops.fsm.persistence import PersistentExecutor, _reconcile_stale_runs
97 from little_loops.fsm.rate_limit_circuit import RateLimitCircuit
98 from little_loops.fsm.validation import load_and_validate
100 try:
101 if getattr(args, "builtin", False):
102 path = get_builtin_loops_dir() / f"{loop_name}.yaml"
103 if not path.exists():
104 logger.error(f"Built-in loop not found: {loop_name!r}")
105 return 1
106 else:
107 path = resolve_loop_path(loop_name, loops_dir)
108 fsm, _ = load_and_validate(path)
109 except FileNotFoundError as e:
110 logger.error(str(e))
111 return 1
112 except ValueError as e:
113 logger.error(f"Validation error: {e}")
114 return 1
116 # Apply overrides
117 if args.max_iterations:
118 fsm.max_iterations = args.max_iterations
119 if args.delay is not None:
120 fsm.backoff = args.delay
121 if args.no_llm:
122 fsm.llm.enabled = False
123 if args.llm_model:
124 fsm.llm.model = args.llm_model
125 # Inject positional input arg before --context so --context can override
126 if getattr(args, "input", None) is not None:
127 raw = args.input
128 try:
129 parsed = json.loads(raw)
130 if isinstance(parsed, dict):
131 matched = {k: v for k, v in parsed.items() if k in fsm.context}
132 if matched:
133 fsm.context.update(matched)
134 else:
135 fsm.context[fsm.input_key] = raw
136 else:
137 fsm.context[fsm.input_key] = raw
138 except (json.JSONDecodeError, ValueError):
139 fsm.context[fsm.input_key] = raw
140 # Inject program.md fields before --context so --context can override
141 program_md_arg = getattr(args, "program_md", None)
142 program_md_path = (
143 program_md_arg if program_md_arg is not None else Path.cwd() / ".ll" / "program.md"
144 )
145 for key, value in _parse_program_md(program_md_path).items():
146 fsm.context[key] = value
147 for kv in getattr(args, "context", None) or []:
148 if "=" not in kv:
149 raise SystemExit(f"Invalid --context format: {kv!r} (expected KEY=VALUE)")
150 key, _, value = kv.partition("=")
151 fsm.context[key.strip()] = value.strip()
153 # Generate instance_id early so run_dir can be derived before the validation scan
154 if getattr(args, "foreground_internal", False):
155 _pre_instance_id: str | None = getattr(args, "instance_id", None)
156 else:
157 _pre_instance_id = _make_instance_id(loop_name)
159 # Inject run_dir into context before validation so ${context.run_dir} resolves.
160 # --context run_dir=VALUE (already applied above) takes precedence.
161 if "run_dir" not in fsm.context:
162 fsm.context["run_dir"] = str(loops_dir / "runs" / (_pre_instance_id or loop_name)) + "/"
164 # Apply YAML loop config env-var overrides (CLI flags below overwrite these)
165 if fsm.config is not None and isinstance(fsm.config.handoff_threshold, int):
166 os.environ["LL_HANDOFF_THRESHOLD"] = str(fsm.config.handoff_threshold)
168 if getattr(args, "handoff_threshold", None) is not None:
169 if not (1 <= args.handoff_threshold <= 100):
170 raise SystemExit("--handoff-threshold must be between 1 and 100")
171 os.environ["LL_HANDOFF_THRESHOLD"] = str(args.handoff_threshold)
173 if getattr(args, "context_limit", None) is not None:
174 os.environ["LL_CONTEXT_LIMIT"] = str(args.context_limit)
176 from little_loops.config import BRConfig
177 from little_loops.design_tokens import load_design_tokens, render_as_prompt_context
179 _config = BRConfig(Path.cwd())
181 if "design_tokens_context" not in fsm.context:
182 _tokens = load_design_tokens(_config)
183 fsm.context["design_tokens_context"] = render_as_prompt_context(_tokens) if _tokens else ""
185 _edge_label_colors = _config.cli.colors.fsm_edge_labels.to_dict()
186 _highlight_color = _config.cli.colors.fsm_active_state
187 _badges = _config.loops.glyphs.to_dict()
189 # Dry run
190 if args.dry_run:
191 from little_loops.cli.loop.diagram_modes import resolve_facets
192 from little_loops.cli.loop.layout import _render_fsm_diagram
193 from little_loops.cli.output import terminal_width
195 facets = resolve_facets(args)
196 if facets is not None:
197 tw = terminal_width()
198 header_text = f"== loop: {fsm.name} "
199 print(header_text + "=" * max(0, tw - len(header_text)))
200 diagram = _render_fsm_diagram(
201 fsm,
202 highlight_color=_highlight_color,
203 edge_label_colors=_edge_label_colors,
204 badges=_badges,
205 mode=facets.scope,
206 suppress_labels=not facets.edge_labels,
207 title_only=facets.state_detail == "title",
208 )
209 print(diagram)
210 print()
211 print_execution_plan(fsm, edge_label_colors=_edge_label_colors)
212 return 0
214 # Pre-run validation: check required context variables are present
215 _ctx_var_re = re.compile(r"\$\{context\.([^}.]+)")
216 missing_keys: set[str] = set()
217 for state in fsm.states.values():
218 templates = [state.action] if state.action else []
219 if state.evaluate and state.evaluate.prompt:
220 templates.append(state.evaluate.prompt)
221 for template in templates:
222 for m in _ctx_var_re.finditer(template):
223 key = m.group(1)
224 if key not in fsm.context:
225 missing_keys.add(key)
226 if missing_keys:
227 for key in sorted(missing_keys):
228 logger.error(
229 f"Missing required context variable: '{key}'. "
230 f"Run with: ll-loop run {loop_name} --context {key}=VALUE"
231 )
232 return 1
234 # Pre-run validation: check required_inputs are present and non-empty
235 for key in fsm.required_inputs:
236 if not fsm.context.get(key, ""):
237 logger.error(
238 f"loop '{loop_name}' requires input '{key}' but none was provided. "
239 f"Pass it via: ll-loop run {loop_name} <value>"
240 )
241 return 1
243 # Baseline mode: apply to context before background gate so flags forward correctly
244 baseline_enabled = getattr(args, "baseline", False)
245 if baseline_enabled:
246 if getattr(args, "worktree", False):
247 raise SystemExit("--baseline and --worktree cannot be combined")
248 fsm.context["_baseline"] = {
249 "enabled": True,
250 "skill": getattr(args, "baseline_skill", None),
251 "items": getattr(args, "items", None),
252 }
254 # Background mode: spawn detached process and return
255 if getattr(args, "background", False):
256 if getattr(args, "worktree", False):
257 raise SystemExit("--worktree and --background cannot be combined")
258 if getattr(args, "follow", False):
259 raise SystemExit(
260 "--follow and --background cannot be combined; use 'll-logs tail' to watch a background loop"
261 )
262 return run_background(loop_name, args, loops_dir)
264 # Register PID file for all foreground runs so cmd_stop can send SIGTERM (BUG-639).
265 # Background-spawned processes (foreground_internal=True) have their PID written by the
266 # parent in run_background(); plain foreground runs must write their own PID here.
267 running_dir = loops_dir / ".running"
268 running_dir.mkdir(parents=True, exist_ok=True)
269 _reconcile_stale_runs(loops_dir)
270 instance_id = _pre_instance_id
271 pid_file = running_dir / f"{instance_id or loop_name}.pid"
272 foreground_pid_file: Path | None = pid_file
274 if not getattr(args, "foreground_internal", False):
275 pid_file.parent.mkdir(parents=True, exist_ok=True)
276 pid_file.write_text(str(os.getpid()))
278 def _cleanup_pid() -> None:
279 pid_file.unlink(missing_ok=True)
281 atexit.register(_cleanup_pid)
283 # Scope-based locking
284 lock_manager = LockManager(loops_dir)
285 scope = resolve_scope(fsm.scope or ["."], fsm.context)
286 _queue_entry_file: Path | None = None
288 def _cleanup_queue_entry() -> None:
289 if _queue_entry_file is not None:
290 _queue_entry_file.unlink(missing_ok=True)
292 if not getattr(args, "no_lock", False):
293 if not lock_manager.acquire(fsm.name, scope, instance_id=instance_id):
294 conflict = lock_manager.find_conflict(scope)
295 if conflict and getattr(args, "queue", False):
296 # Write queue entry so dashboard shows the waiting loop
297 queue_dir = loops_dir / ".queue"
298 queue_dir.mkdir(parents=True, exist_ok=True)
299 entry_id = str(uuid.uuid4())
300 entry = {
301 "id": entry_id,
302 "loopName": loop_name,
303 "enqueuedAt": datetime.now(UTC).isoformat(),
304 "context": {
305 "waitingFor": conflict.loop_name,
306 "scope": conflict.scope,
307 "pid": os.getpid(),
308 },
309 }
310 _queue_entry_file = queue_dir / f"{entry_id}.json"
311 _queue_entry_file.write_text(json.dumps(entry, indent=2))
312 atexit.register(_cleanup_queue_entry)
314 logger.info(f"Waiting for conflicting loop '{conflict.loop_name}' to finish...")
315 # Retry loop: when N waiters are released simultaneously, only one wins
316 # acquire(); losers loop back and wait again rather than exiting (BUG-1281).
317 acquired = False
318 _wait_start = time.time()
319 _budget = _config.loops.queue_wait_timeout_seconds
320 while time.time() - _wait_start < _budget:
321 _remaining = _budget - (time.time() - _wait_start)
322 if not lock_manager.wait_for_scope(scope, timeout=int(_remaining)):
323 _cleanup_queue_entry()
324 logger.error("Timeout waiting for scope to become available")
325 return 1
326 if not _is_earliest_waiter(entry_id, queue_dir):
327 time.sleep(1)
328 continue
329 if lock_manager.acquire(fsm.name, scope, instance_id=instance_id):
330 acquired = True
331 break
332 if not acquired:
333 _cleanup_queue_entry()
334 logger.error("Failed to acquire lock after waiting")
335 return 1
336 # Lock acquired - no longer queued
337 _cleanup_queue_entry()
338 elif conflict:
339 logger.error(f"Scope conflict with running loop: {conflict.loop_name}")
340 logger.info(f" Conflicting scope: {conflict.scope}")
341 logger.info(" Use --queue to wait for it to finish")
342 return 1
343 else:
344 # Unexpected: find_conflict returned None but acquire failed
345 logger.error("Failed to acquire scope lock (unknown reason)")
346 return 1
348 executor: PersistentExecutor | None = None
349 try:
350 # Worktree isolation: create branch + directory before anything reads Path.cwd()
351 if getattr(args, "worktree", False):
352 from little_loops.config import BRConfig as _MainBRConfig
353 from little_loops.parallel.git_lock import GitLock
354 from little_loops.worktree_utils import cleanup_worktree, setup_worktree
356 _main_config = _MainBRConfig(Path.cwd())
357 _worktree_base = _main_config.get_worktree_base()
358 _copy_files = _main_config.parallel.worktree_copy_files
359 _repo_path = Path.cwd()
361 _timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
362 _safe_name = re.sub(r"[^a-zA-Z0-9-]", "-", loop_name)
363 _branch_name = f"{_timestamp}-{_safe_name}"
364 _worktree_path = _worktree_base / _branch_name
366 _worktree_base.mkdir(parents=True, exist_ok=True)
367 _git_lock = GitLock(logger)
369 setup_worktree(
370 repo_path=_repo_path,
371 worktree_path=_worktree_path,
372 branch_name=_branch_name,
373 copy_files=_copy_files,
374 logger=logger,
375 git_lock=_git_lock,
376 )
378 logger.info(f"Worktree: {_worktree_path}")
379 logger.info(f"Branch: {_branch_name}")
381 def _cleanup_worktree_on_exit() -> None:
382 cleanup_worktree(
383 worktree_path=_worktree_path,
384 repo_path=_repo_path,
385 logger=logger,
386 git_lock=_git_lock,
387 delete_branch=True,
388 )
390 atexit.register(_cleanup_worktree_on_exit)
392 os.environ["CLAUDE_BASH_MAINTAIN_PROJECT_WORKING_DIR"] = "1"
393 os.chdir(_worktree_path)
395 circuit = (
396 RateLimitCircuit(Path(_config.commands.rate_limits.circuit_breaker_path))
397 if _config.commands.rate_limits.circuit_breaker_enabled
398 else None
399 )
400 Path(fsm.context["run_dir"]).mkdir(parents=True, exist_ok=True)
401 executor = PersistentExecutor(
402 fsm, loops_dir=loops_dir, circuit=circuit, instance_id=instance_id, pid=os.getpid()
403 )
405 # Register signal handlers for graceful shutdown
406 register_loop_signal_handlers(executor, pid_file=foreground_pid_file)
408 from little_loops.extension import wire_extensions
409 from little_loops.transport import wire_transports
411 wire_extensions(executor.event_bus, _config.extensions, executor=executor)
412 wire_transports(executor.event_bus, _config.events)
413 return run_foreground(
414 executor,
415 fsm,
416 args,
417 highlight_color=_highlight_color,
418 edge_label_colors=_edge_label_colors,
419 badges=_badges,
420 instance_id=instance_id,
421 loop_path=path,
422 running_dir=running_dir,
423 model=fsm.llm.model,
424 )
425 finally:
426 if executor is not None:
427 executor.close_transports()
428 if not getattr(args, "no_lock", False):
429 lock_manager.release(fsm.name, instance_id=instance_id)