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