Coverage for little_loops / cli / loop / lifecycle.py: 5%
442 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:12 -0500
1"""ll-loop lifecycle subcommands: status, stop, resume."""
3from __future__ import annotations
5import argparse
6import atexit
7import hashlib
8import json
9import os
10import signal
11import time
12from pathlib import Path
14from little_loops.cli.loop._helpers import (
15 load_loop,
16 register_loop_signal_handlers,
17 resolve_loop_path,
18 run_background,
19)
20from little_loops.fsm.concurrency import _process_alive
21from little_loops.fsm.persistence import (
22 LoopState,
23 StatePersistence,
24 _find_instances,
25 _read_pid_file,
26 _reconcile_stale_running,
27)
28from little_loops.logger import Logger
31def _format_relative_time(seconds: float) -> str:
32 """Format seconds as a human-readable relative time string (e.g., '3m ago').
34 Delegates to the shared ``format_relative_time`` in ``cli.output``.
35 """
36 from little_loops.cli.output import format_relative_time
38 return format_relative_time(seconds)
41def _format_log_label(running_dir: Path, stem: str) -> str:
42 """Return one of three Log: labels based on run-mode signals.
44 - log exists → path string (background run, normal)
45 - no log, no pid → foreground run (output went to terminal)
46 - no log, pid exists → background run whose log was deleted (something wrong)
47 """
48 log_file = running_dir / f"{stem}.log"
49 if log_file.exists():
50 return str(log_file)
51 pid_file = running_dir / f"{stem}.pid"
52 if pid_file.exists():
53 return f"(expected {log_file}, missing)"
54 return "(foreground run — output went to terminal)"
57def _get_events_info(running_dir: Path, stem: str) -> tuple[str | None, str | None]:
58 """Return (events_file_str, formatted_detail_line) for the events.jsonl file.
60 Returns (None, None) if the file doesn't exist.
61 Detail line format: 'Events: <path> (N events, last Xm ago)'
62 """
63 events_file = running_dir / f"{stem}.events.jsonl"
64 if not events_file.exists():
65 return None, None
66 try:
67 lines = [ln for ln in events_file.read_text().splitlines() if ln.strip()]
68 count = len(lines)
69 detail = f"({count} events)"
70 if lines:
71 try:
72 last_entry = json.loads(lines[-1])
73 last_ts = last_entry.get("ts")
74 if last_ts is not None:
75 from datetime import UTC, datetime
77 dt = datetime.fromisoformat(last_ts)
78 age = (datetime.now(tz=UTC) - dt).total_seconds()
79 detail = f"({count} events, last {_format_relative_time(age)})"
80 except (json.JSONDecodeError, AttributeError, ValueError, OverflowError):
81 pass
82 return str(events_file), f"Events: {events_file} {detail}"
83 except OSError:
84 return str(events_file), f"Events: {events_file}"
87def _kill_with_timeout(pid: int, label: str, logger: Logger) -> None:
88 """Kill pid and all descendants via process group; SIGTERM first, escalate to SIGKILL.
90 Uses os.killpg to atomically signal the entire process group, closing the
91 race window inherent in the old _get_descendant_pids() snapshot approach.
92 Since run_background and run_claude_command both launch with
93 start_new_session=True, the PID is a session leader with PGID == PID.
94 All children inherit this PGID unless they explicitly call setsid().
96 Falls back to os.kill on the root PID when os.killpg is unavailable (Windows).
97 """
98 try:
99 pgid = os.getpgid(pid)
100 except ProcessLookupError:
101 return # Already dead — nothing to do
103 # Atomically signal the entire process group
104 _signal_process_group(pgid, pid, signal.SIGTERM, label)
106 for _ in range(10):
107 time.sleep(1)
108 if not _process_alive(pid):
109 return
111 # Escalate to SIGKILL on the process group
112 _signal_process_group(pgid, pid, signal.SIGKILL, label)
113 logger.warning(f"Sent SIGKILL to {label} (PID: {pid})")
116def _signal_process_group(pgid: int, pid: int, sig: int, label: str) -> None:
117 """Send a signal to the process group, falling back to single-PID kill."""
118 try:
119 os.killpg(pgid, sig)
120 except AttributeError:
121 # os.killpg not available (Windows) — fall back to single-PID kill
122 try:
123 os.kill(pid, sig)
124 except OSError:
125 pass
126 except (ProcessLookupError, PermissionError):
127 pass # Group already gone or no permission — not actionable
130def _status_single(
131 instance_id: str | None,
132 state: LoopState,
133 loop_name: str,
134 running_dir: Path,
135 args: argparse.Namespace | None,
136) -> int:
137 """Render status for one instance (human-readable or JSON)."""
138 from little_loops.cli.output import print_json
140 stem = instance_id or loop_name
141 persistence = StatePersistence(loop_name, running_dir.parent, instance_id=instance_id)
142 state = _reconcile_stale_running(state, persistence, running_dir, stem)
144 pid_file = running_dir / f"{stem}.pid"
145 pid = _read_pid_file(pid_file)
146 pid_source: str | None = "pid_file" if pid is not None else None
148 if pid is None:
149 lock_file_path = running_dir / f"{stem}.lock"
150 if lock_file_path.exists():
151 try:
152 with open(lock_file_path) as _lf:
153 lock_data = json.load(_lf)
154 pid = lock_data.get("pid")
155 if pid is not None:
156 pid_source = "lock_file"
157 except (json.JSONDecodeError, KeyError, OSError):
158 pass
160 log_file = running_dir / f"{stem}.log"
161 log_file_str: str | None = None
162 log_updated_ago: str | None = None
163 last_event: str | None = None
164 if log_file.exists():
165 log_file_str = str(log_file)
166 age_seconds = time.time() - log_file.stat().st_mtime
167 log_updated_ago = _format_relative_time(age_seconds)
168 try:
169 lines = log_file.read_text().splitlines()
170 non_empty = [ln for ln in lines if ln.strip()]
171 last_event = non_empty[-1] if non_empty else None
172 except OSError:
173 last_event = None
175 events_file_str, events_detail_line = _get_events_info(running_dir, stem)
177 if getattr(args, "json", False):
178 d = state.to_dict()
179 d["pid"] = pid
180 d["pid_source"] = pid_source
181 d["log_file"] = log_file_str
182 d["log_updated_ago"] = log_updated_ago
183 d["last_event"] = last_event
184 d["events_file"] = events_file_str
185 print_json(d)
186 return 0
188 print(f"Loop: {state.loop_name}")
189 print(f"Status: {state.status}")
190 print(f"Current state: {state.current_state}")
191 print(f"Iteration: {state.iteration}")
192 print(f"Started: {state.started_at}")
193 print(f"Updated: {state.updated_at}")
195 if pid is not None:
196 if _process_alive(pid):
197 print(f"PID: {pid} (running)")
198 else:
199 print(f"PID: {pid} (not running - stale PID file)")
201 log_label = _format_log_label(running_dir, stem)
202 print(f"Log: {log_label}")
203 if log_file_str is not None:
204 print(f"Log updated: {log_updated_ago}")
205 if last_event:
206 print(f"Last event: {last_event}")
208 if events_detail_line is not None:
209 print(events_detail_line)
211 if state.continuation_prompt:
212 prompt_preview = state.continuation_prompt[:200]
213 if len(state.continuation_prompt) > 200:
214 prompt_preview += "..."
215 print(f"Continuation context: {prompt_preview}")
216 return 0
219def cmd_status(
220 loop_name: str,
221 loops_dir: Path,
222 logger: Logger,
223 args: argparse.Namespace | None = None,
224) -> int:
225 """Show loop status."""
226 from little_loops.cli.output import print_json
228 running_dir = loops_dir / ".running"
229 instances = _find_instances(loop_name, running_dir)
231 if not instances:
232 logger.error(f"No state found for: {loop_name}")
233 return 1
235 if len(instances) == 1:
236 instance_id, state = instances[0]
237 return _status_single(instance_id, state, loop_name, running_dir, args)
239 # Multiple instances: aggregate display
240 if getattr(args, "json", False):
241 result_list = []
242 for instance_id, state in instances:
243 stem = instance_id or loop_name
244 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
245 state = _reconcile_stale_running(state, persistence, running_dir, stem)
246 pid_file = running_dir / f"{stem}.pid"
247 pid = _read_pid_file(pid_file)
248 pid_source: str | None = "pid_file" if pid is not None else None
249 if pid is None:
250 lock_file_path = running_dir / f"{stem}.lock"
251 if lock_file_path.exists():
252 try:
253 with open(lock_file_path) as _lf:
254 lock_data = json.load(_lf)
255 pid = lock_data.get("pid")
256 if pid is not None:
257 pid_source = "lock_file"
258 except (json.JSONDecodeError, KeyError, OSError):
259 pass
260 log_file = running_dir / f"{stem}.log"
261 d = state.to_dict()
262 d["instance_id"] = instance_id
263 d["pid"] = pid
264 d["pid_source"] = pid_source
265 if log_file.exists():
266 d["log_file"] = str(log_file)
267 d["log_updated_ago"] = _format_relative_time(time.time() - log_file.stat().st_mtime)
268 else:
269 d["log_file"] = None
270 d["log_updated_ago"] = None
271 events_file_str, _ = _get_events_info(running_dir, stem)
272 d["events_file"] = events_file_str
273 result_list.append(d)
274 print_json(result_list)
275 return 0
277 print(f"{len(instances)} instances of '{loop_name}':")
278 for i, (instance_id, state) in enumerate(instances, 1):
279 stem = instance_id or loop_name
280 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
281 state = _reconcile_stale_running(state, persistence, running_dir, stem)
282 pid_file = running_dir / f"{stem}.pid"
283 pid = _read_pid_file(pid_file)
284 if pid is None:
285 lock_file_path = running_dir / f"{stem}.lock"
286 if lock_file_path.exists():
287 try:
288 with open(lock_file_path) as _lf:
289 lock_data = json.load(_lf)
290 pid = lock_data.get("pid")
291 except (json.JSONDecodeError, KeyError, OSError):
292 pass
293 log_file = running_dir / f"{stem}.log"
295 print()
296 print(f"[{i}] {stem}")
297 print(f" Status: {state.status}")
298 print(f" Current state: {state.current_state}")
299 print(f" Iteration: {state.iteration}")
300 if pid is not None:
301 if _process_alive(pid):
302 print(f" PID: {pid} (running)")
303 else:
304 print(f" PID: {pid} (not running - stale PID file)")
305 log_label = _format_log_label(running_dir, stem)
306 print(f" Log: {log_label}")
307 if log_file.exists():
308 age_seconds = time.time() - log_file.stat().st_mtime
309 print(f" Log updated: {_format_relative_time(age_seconds)}")
310 _, events_detail_line = _get_events_info(running_dir, stem)
311 if events_detail_line is not None:
312 print(f" {events_detail_line}")
313 return 0
316def cmd_stop(
317 loop_name: str,
318 loops_dir: Path,
319 logger: Logger,
320) -> int:
321 """Stop a running loop."""
322 from little_loops.fsm.persistence import StatePersistence # noqa: PLC0415
324 running_dir = loops_dir / ".running"
325 instances = _find_instances(loop_name, running_dir)
327 if not instances:
328 logger.error(f"No state found for: {loop_name}")
329 return 1
331 running_instances = [(iid, s) for iid, s in instances if s.status == "running"]
333 if not running_instances:
334 # Secondary check: an orphaned lock-file with a live PID blocks scope
335 # acquisition even when state is not "running". Kill the holder and release.
336 for instance_id, state in instances:
337 stem = instance_id or loop_name
338 lock_file = running_dir / f"{stem}.lock"
339 if lock_file.exists():
340 lock_pid: int | None = None
341 try:
342 with open(lock_file) as _lf:
343 lock_data = json.load(_lf)
344 lock_pid = lock_data.get("pid")
345 except (json.JSONDecodeError, KeyError, OSError):
346 pass
347 if lock_pid and _process_alive(lock_pid):
348 logger.warning(
349 f"Loop state is '{state.status}' but lock file holds live PID {lock_pid}. "
350 "Killing orphaned lock holder..."
351 )
352 _kill_with_timeout(lock_pid, stem, logger)
353 lock_file.unlink(missing_ok=True)
354 logger.success(f"Released orphaned scope lock for {stem}")
355 return 0
356 elif lock_pid is not None:
357 # Dead PID: stale lock file, just remove it
358 lock_file.unlink(missing_ok=True)
359 logger.info(f"Removed stale lock file for {stem}")
360 return 0
361 _, state = instances[0]
362 logger.error(f"Loop not running: {loop_name} (status: {state.status})")
363 return 1
365 for instance_id, state in running_instances:
366 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
367 stem = instance_id or loop_name
368 pid_file = running_dir / f"{stem}.pid"
369 pid = _read_pid_file(pid_file)
371 if pid is not None:
372 if _process_alive(pid):
373 _kill_with_timeout(pid, stem, logger)
374 state.status = "interrupted"
375 persistence.save_state(state)
376 pid_file.unlink(missing_ok=True)
377 logger.success(f"Stopped {stem} (PID: {pid})")
378 else:
379 # Process already exited: preserve its final status, only clean up PID file
380 logger.info(f"Process {pid} not running, cleaning up PID file")
381 pid_file.unlink(missing_ok=True)
382 else:
383 # No PID file: no background process tracked, update state only
384 state.status = "interrupted"
385 persistence.save_state(state)
386 logger.success(f"Marked {stem} as interrupted")
388 return 0
391def cmd_resume(
392 loop_name: str,
393 args: argparse.Namespace,
394 loops_dir: Path,
395 logger: Logger,
396) -> int:
397 """Resume an interrupted loop."""
398 import os
400 from little_loops.fsm.persistence import RESUMABLE_STATUSES, PersistentExecutor
402 running_dir = loops_dir / ".running"
403 running_dir.mkdir(parents=True, exist_ok=True)
405 # Discover all instances and resolve to a single resumable one.
406 instances = _find_instances(loop_name, running_dir)
407 resumable = [(iid, s) for iid, s in instances if s.status in RESUMABLE_STATUSES]
409 explicit_instance_id = getattr(args, "instance_id", None)
410 if explicit_instance_id:
411 filtered: list[tuple[str | None, LoopState]] = [
412 (iid, s) for iid, s in resumable if iid == explicit_instance_id
413 ]
414 if not filtered:
415 print(
416 f"Instance '{explicit_instance_id}' not found among resumable instances of '{loop_name}'."
417 )
418 if resumable:
419 print("Resumable instances:")
420 for iid, _ in resumable:
421 print(f" {iid or loop_name}")
422 return 1
423 resumable = filtered
424 elif len(resumable) > 1:
425 # Auto-select the most recent instance by sorted filename (matches cmd_monitor()).
426 selected_id = resumable[-1][0]
427 print(f"Auto-selected latest instance: {selected_id}")
428 resumable = [resumable[-1]]
430 # Use discovered instance_id (or fall back to args / None for no-state case)
431 if resumable:
432 instance_id: str | None = resumable[0][0]
433 state_for_display = resumable[0][1]
434 else:
435 instance_id = None
436 state_for_display = None
438 # Background mode: spawn detached process and return
439 if getattr(args, "background", False):
440 if instance_id is None:
441 print(f"No resumable instances of '{loop_name}'.")
442 return 1
443 return run_background(
444 loop_name, args, loops_dir, subcommand="resume", instance_id=instance_id
445 )
447 # Register PID file for all foreground runs so cmd_stop can send SIGTERM (BUG-639).
448 # Background-spawned processes (foreground_internal=True) have their PID written by the
449 # parent in run_background(); plain foreground runs must write their own PID here.
450 if instance_id is None:
451 instance_id = getattr(args, "instance_id", None)
453 pid_file = running_dir / f"{instance_id or loop_name}.pid"
454 foreground_pid_file: Path | None = pid_file
456 if not getattr(args, "foreground_internal", False):
457 pid_file.parent.mkdir(parents=True, exist_ok=True)
458 pid_file.write_text(str(os.getpid()))
460 def _cleanup_pid() -> None:
461 pid_file.unlink(missing_ok=True)
463 atexit.register(_cleanup_pid)
465 try:
466 fsm = load_loop(loop_name, loops_dir, logger)
467 except FileNotFoundError as e:
468 logger.error(str(e))
469 return 1
470 except ValueError as e:
471 logger.error(f"Validation error: {e}")
472 return 1
474 if getattr(args, "baseline", False):
475 logger.error(
476 "--baseline is not supported with resume. Start a fresh run with 'll-loop run'."
477 )
478 return 1
480 try:
481 loop_path = resolve_loop_path(loop_name, loops_dir)
482 except FileNotFoundError:
483 loop_path = None
485 for kv in getattr(args, "context", None) or []:
486 if "=" not in kv:
487 raise SystemExit(f"Invalid --context format: {kv!r} (expected KEY=VALUE)")
488 key, _, value = kv.partition("=")
489 fsm.context[key.strip()] = value.strip()
491 # Re-inject run_dir using the same instance_id as the original run so resumed
492 # loops write artifacts to the same directory they started with.
493 if "run_dir" not in fsm.context and instance_id is not None:
494 fsm.context["run_dir"] = str(loops_dir / "runs" / instance_id) + "/"
496 # Re-inject input hash for checkpoint fingerprinting during resumed runs.
497 if "input_hash" not in fsm.context and isinstance(fsm.context.get("input"), str):
498 fsm.context["input_hash"] = hashlib.sha256(fsm.context["input"].encode()).hexdigest()[:12]
500 if getattr(args, "delay", None) is not None:
501 fsm.backoff = args.delay
503 # Apply YAML loop config env-var overrides (CLI flags below overwrite these)
504 if fsm.config is not None and isinstance(fsm.config.handoff_threshold, int):
505 os.environ["LL_HANDOFF_THRESHOLD"] = str(fsm.config.handoff_threshold)
507 if getattr(args, "handoff_threshold", None) is not None:
508 if not (1 <= args.handoff_threshold <= 100):
509 raise SystemExit("--handoff-threshold must be between 1 and 100")
510 os.environ["LL_HANDOFF_THRESHOLD"] = str(args.handoff_threshold)
512 # Show context if resuming from a handoff
513 if state_for_display and state_for_display.status == "awaiting_continuation":
514 print(f"Resuming from context handoff (iteration {state_for_display.iteration})...")
515 if state_for_display.continuation_prompt:
516 prompt_preview = state_for_display.continuation_prompt[:500]
517 if len(state_for_display.continuation_prompt) > 500:
518 prompt_preview += "..."
519 print(f"Context: {prompt_preview}")
520 print()
522 from little_loops.config import BRConfig
523 from little_loops.design_tokens import load_design_tokens, render_as_prompt_context
524 from little_loops.extension import wire_extensions
525 from little_loops.fsm.rate_limit_circuit import RateLimitCircuit
526 from little_loops.transport import wire_transports
528 config = BRConfig(Path.cwd())
530 if not fsm.context.get("design_tokens_context"):
531 _tokens = load_design_tokens(config)
532 fsm.context["design_tokens_context"] = render_as_prompt_context(_tokens) if _tokens else ""
534 circuit = (
535 RateLimitCircuit(Path(config.commands.rate_limits.circuit_breaker_path))
536 if config.commands.rate_limits.circuit_breaker_enabled
537 else None
538 )
539 executor = PersistentExecutor(
540 fsm, loops_dir=loops_dir, circuit=circuit, instance_id=instance_id
541 )
543 # Register signal handlers for graceful shutdown (same as cmd_run)
544 register_loop_signal_handlers(executor, pid_file=foreground_pid_file)
546 wire_extensions(executor.event_bus, config.extensions, executor=executor)
547 wire_transports(executor.event_bus, config.events)
549 # Route through run_foreground (BUG-1645) so the resume path shares the
550 # display-callback wiring that cmd_run gets via run_foreground. Without this
551 # the FSM event bus has no subscriber on a resumed run, leaving the terminal
552 # silent for the entire duration (no per-iteration lines, no FSM diagram).
553 from little_loops.cli.loop._helpers import run_foreground
555 _edge_label_colors = config.cli.colors.fsm_edge_labels.to_dict()
556 _highlight_color = config.cli.colors.fsm_active_state
557 _badges = config.loops.glyphs.to_dict()
559 try:
560 return run_foreground(
561 executor,
562 fsm,
563 args,
564 highlight_color=_highlight_color,
565 edge_label_colors=_edge_label_colors,
566 badges=_badges,
567 mode="resume",
568 instance_id=instance_id,
569 running_dir=running_dir,
570 loop_path=loop_path,
571 model=fsm.llm.model,
572 )
573 finally:
574 executor.close_transports()
577def _print_last_state(state: LoopState) -> None:
578 """Print last-known state (used when not actively tailing)."""
579 print(f"Loop: {state.loop_name}")
580 print(f"Status: {state.status}")
581 print(f"Current state: {state.current_state}")
582 print(f"Iteration: {state.iteration}")
585def cmd_monitor(args: argparse.Namespace, loops_dir: Path) -> int:
586 """Attach to a running loop and render its FSM state in realtime.
588 Read-only attach: tails ``<stem>.events.jsonl`` and forwards events to a
589 ``StateFeedRenderer``. Ctrl-C detaches without sending any signal to the
590 loop process (FEAT-1764).
591 """
592 loop_name = args.loop
593 running_dir = loops_dir / ".running"
594 instances = _find_instances(loop_name, running_dir)
596 if not instances:
597 print(f"No instances of '{loop_name}' found")
598 return 1
600 instance_id, state = instances[-1]
601 stem = instance_id or loop_name
603 pid = _read_pid_file(running_dir / f"{stem}.pid")
604 if pid is None:
605 pid = getattr(state, "pid", None)
607 if pid is None or not _process_alive(pid):
608 _print_last_state(state)
609 return 0
611 # Fabricate missing Namespace attrs that StateFeedRenderer reads.
612 for attr, default in (
613 ("quiet", False),
614 ("verbose", False),
615 ("show_diagrams", None),
616 ("clear", True),
617 ("diagram_edge_labels", None),
618 ("diagram_state_detail", None),
619 ("diagram_scope", None),
620 ("follow", False),
621 ):
622 if not hasattr(args, attr):
623 setattr(args, attr, default)
625 try:
626 fsm = load_loop(loop_name, loops_dir, Logger())
627 except (FileNotFoundError, ValueError) as e:
628 print(f"Cannot load loop '{loop_name}': {e}")
629 return 1
631 try:
632 loop_path = resolve_loop_path(loop_name, loops_dir)
633 except FileNotFoundError:
634 loop_path = None
636 # Late import: tests patch StateFeedRenderer at its module-of-origin
637 # (little_loops.cli.loop._helpers.StateFeedRenderer); using a function-local
638 # import ensures the patch takes effect at call time.
639 from little_loops.cli.loop._helpers import (
640 StateFeedRenderer,
641 _install_sigwinch_handler,
642 _restore_sigwinch_handler,
643 )
645 renderer = StateFeedRenderer(
646 fsm, args, loops_dir=loops_dir, loop_path=loop_path, model=fsm.llm.model
647 )
649 events_file = running_dir / f"{stem}.events.jsonl"
651 if not events_file.exists():
652 _print_last_state(state)
653 return 0
655 sigwinch_installed = False
656 if renderer.in_pinned_mode:
657 _install_sigwinch_handler()
658 sigwinch_installed = True
660 try:
661 with open(events_file, encoding="utf-8") as ev_f:
662 ev_f.seek(0, 2)
663 while True:
664 progressed = False
665 try:
666 line = ev_f.readline()
667 except FileNotFoundError:
668 break
669 if line:
670 progressed = True
671 stripped = line.strip()
672 if stripped:
673 try:
674 event = json.loads(stripped)
675 except json.JSONDecodeError:
676 event = None
677 if event is not None:
678 renderer.handle_event(event)
679 if not progressed:
680 if not _process_alive(pid):
681 break
682 if not events_file.exists():
683 break
684 time.sleep(0.1)
685 except KeyboardInterrupt:
686 return 0
687 finally:
688 if sigwinch_installed:
689 _restore_sigwinch_handler()
691 return 0