Coverage for little_loops / cli / loop / lifecycle.py: 0%
431 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -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 """Send SIGTERM to pid; escalate to SIGKILL after 10 s if still alive."""
89 os.kill(pid, signal.SIGTERM)
90 for _ in range(10):
91 time.sleep(1)
92 if not _process_alive(pid):
93 return
94 try:
95 os.kill(pid, signal.SIGKILL)
96 logger.warning(f"Sent SIGKILL to {label} (PID: {pid})")
97 except OSError:
98 pass
101def _status_single(
102 instance_id: str | None,
103 state: LoopState,
104 loop_name: str,
105 running_dir: Path,
106 args: argparse.Namespace | None,
107) -> int:
108 """Render status for one instance (human-readable or JSON)."""
109 from little_loops.cli.output import print_json
111 stem = instance_id or loop_name
112 persistence = StatePersistence(loop_name, running_dir.parent, instance_id=instance_id)
113 state = _reconcile_stale_running(state, persistence, running_dir, stem)
115 pid_file = running_dir / f"{stem}.pid"
116 pid = _read_pid_file(pid_file)
117 pid_source: str | None = "pid_file" if pid is not None else None
119 if pid is None:
120 lock_file_path = running_dir / f"{stem}.lock"
121 if lock_file_path.exists():
122 try:
123 with open(lock_file_path) as _lf:
124 lock_data = json.load(_lf)
125 pid = lock_data.get("pid")
126 if pid is not None:
127 pid_source = "lock_file"
128 except (json.JSONDecodeError, KeyError, OSError):
129 pass
131 log_file = running_dir / f"{stem}.log"
132 log_file_str: str | None = None
133 log_updated_ago: str | None = None
134 last_event: str | None = None
135 if log_file.exists():
136 log_file_str = str(log_file)
137 age_seconds = time.time() - log_file.stat().st_mtime
138 log_updated_ago = _format_relative_time(age_seconds)
139 try:
140 lines = log_file.read_text().splitlines()
141 non_empty = [ln for ln in lines if ln.strip()]
142 last_event = non_empty[-1] if non_empty else None
143 except OSError:
144 last_event = None
146 events_file_str, events_detail_line = _get_events_info(running_dir, stem)
148 if getattr(args, "json", False):
149 d = state.to_dict()
150 d["pid"] = pid
151 d["pid_source"] = pid_source
152 d["log_file"] = log_file_str
153 d["log_updated_ago"] = log_updated_ago
154 d["last_event"] = last_event
155 d["events_file"] = events_file_str
156 print_json(d)
157 return 0
159 print(f"Loop: {state.loop_name}")
160 print(f"Status: {state.status}")
161 print(f"Current state: {state.current_state}")
162 print(f"Iteration: {state.iteration}")
163 print(f"Started: {state.started_at}")
164 print(f"Updated: {state.updated_at}")
166 if pid is not None:
167 if _process_alive(pid):
168 print(f"PID: {pid} (running)")
169 else:
170 print(f"PID: {pid} (not running - stale PID file)")
172 log_label = _format_log_label(running_dir, stem)
173 print(f"Log: {log_label}")
174 if log_file_str is not None:
175 print(f"Log updated: {log_updated_ago}")
176 if last_event:
177 print(f"Last event: {last_event}")
179 if events_detail_line is not None:
180 print(events_detail_line)
182 if state.continuation_prompt:
183 prompt_preview = state.continuation_prompt[:200]
184 if len(state.continuation_prompt) > 200:
185 prompt_preview += "..."
186 print(f"Continuation context: {prompt_preview}")
187 return 0
190def cmd_status(
191 loop_name: str,
192 loops_dir: Path,
193 logger: Logger,
194 args: argparse.Namespace | None = None,
195) -> int:
196 """Show loop status."""
197 from little_loops.cli.output import print_json
199 running_dir = loops_dir / ".running"
200 instances = _find_instances(loop_name, running_dir)
202 if not instances:
203 logger.error(f"No state found for: {loop_name}")
204 return 1
206 if len(instances) == 1:
207 instance_id, state = instances[0]
208 return _status_single(instance_id, state, loop_name, running_dir, args)
210 # Multiple instances: aggregate display
211 if getattr(args, "json", False):
212 result_list = []
213 for instance_id, state in instances:
214 stem = instance_id or loop_name
215 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
216 state = _reconcile_stale_running(state, persistence, running_dir, stem)
217 pid_file = running_dir / f"{stem}.pid"
218 pid = _read_pid_file(pid_file)
219 pid_source: str | None = "pid_file" if pid is not None else None
220 if pid is None:
221 lock_file_path = running_dir / f"{stem}.lock"
222 if lock_file_path.exists():
223 try:
224 with open(lock_file_path) as _lf:
225 lock_data = json.load(_lf)
226 pid = lock_data.get("pid")
227 if pid is not None:
228 pid_source = "lock_file"
229 except (json.JSONDecodeError, KeyError, OSError):
230 pass
231 log_file = running_dir / f"{stem}.log"
232 d = state.to_dict()
233 d["instance_id"] = instance_id
234 d["pid"] = pid
235 d["pid_source"] = pid_source
236 if log_file.exists():
237 d["log_file"] = str(log_file)
238 d["log_updated_ago"] = _format_relative_time(time.time() - log_file.stat().st_mtime)
239 else:
240 d["log_file"] = None
241 d["log_updated_ago"] = None
242 events_file_str, _ = _get_events_info(running_dir, stem)
243 d["events_file"] = events_file_str
244 result_list.append(d)
245 print_json(result_list)
246 return 0
248 print(f"{len(instances)} instances of '{loop_name}':")
249 for i, (instance_id, state) in enumerate(instances, 1):
250 stem = instance_id or loop_name
251 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
252 state = _reconcile_stale_running(state, persistence, running_dir, stem)
253 pid_file = running_dir / f"{stem}.pid"
254 pid = _read_pid_file(pid_file)
255 if pid is None:
256 lock_file_path = running_dir / f"{stem}.lock"
257 if lock_file_path.exists():
258 try:
259 with open(lock_file_path) as _lf:
260 lock_data = json.load(_lf)
261 pid = lock_data.get("pid")
262 except (json.JSONDecodeError, KeyError, OSError):
263 pass
264 log_file = running_dir / f"{stem}.log"
266 print()
267 print(f"[{i}] {stem}")
268 print(f" Status: {state.status}")
269 print(f" Current state: {state.current_state}")
270 print(f" Iteration: {state.iteration}")
271 if pid is not None:
272 if _process_alive(pid):
273 print(f" PID: {pid} (running)")
274 else:
275 print(f" PID: {pid} (not running - stale PID file)")
276 log_label = _format_log_label(running_dir, stem)
277 print(f" Log: {log_label}")
278 if log_file.exists():
279 age_seconds = time.time() - log_file.stat().st_mtime
280 print(f" Log updated: {_format_relative_time(age_seconds)}")
281 _, events_detail_line = _get_events_info(running_dir, stem)
282 if events_detail_line is not None:
283 print(f" {events_detail_line}")
284 return 0
287def cmd_stop(
288 loop_name: str,
289 loops_dir: Path,
290 logger: Logger,
291) -> int:
292 """Stop a running loop."""
293 from little_loops.fsm.persistence import StatePersistence # noqa: PLC0415
295 running_dir = loops_dir / ".running"
296 instances = _find_instances(loop_name, running_dir)
298 if not instances:
299 logger.error(f"No state found for: {loop_name}")
300 return 1
302 running_instances = [(iid, s) for iid, s in instances if s.status == "running"]
304 if not running_instances:
305 # Secondary check: an orphaned lock-file with a live PID blocks scope
306 # acquisition even when state is not "running". Kill the holder and release.
307 for instance_id, state in instances:
308 stem = instance_id or loop_name
309 lock_file = running_dir / f"{stem}.lock"
310 if lock_file.exists():
311 lock_pid: int | None = None
312 try:
313 with open(lock_file) as _lf:
314 lock_data = json.load(_lf)
315 lock_pid = lock_data.get("pid")
316 except (json.JSONDecodeError, KeyError, OSError):
317 pass
318 if lock_pid and _process_alive(lock_pid):
319 logger.warning(
320 f"Loop state is '{state.status}' but lock file holds live PID {lock_pid}. "
321 "Killing orphaned lock holder..."
322 )
323 _kill_with_timeout(lock_pid, stem, logger)
324 lock_file.unlink(missing_ok=True)
325 logger.success(f"Released orphaned scope lock for {stem}")
326 return 0
327 elif lock_pid is not None:
328 # Dead PID: stale lock file, just remove it
329 lock_file.unlink(missing_ok=True)
330 logger.info(f"Removed stale lock file for {stem}")
331 return 0
332 _, state = instances[0]
333 logger.error(f"Loop not running: {loop_name} (status: {state.status})")
334 return 1
336 for instance_id, state in running_instances:
337 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
338 stem = instance_id or loop_name
339 pid_file = running_dir / f"{stem}.pid"
340 pid = _read_pid_file(pid_file)
342 if pid is not None:
343 if _process_alive(pid):
344 _kill_with_timeout(pid, stem, logger)
345 state.status = "interrupted"
346 persistence.save_state(state)
347 pid_file.unlink(missing_ok=True)
348 logger.success(f"Stopped {stem} (PID: {pid})")
349 else:
350 # Process already exited: preserve its final status, only clean up PID file
351 logger.info(f"Process {pid} not running, cleaning up PID file")
352 pid_file.unlink(missing_ok=True)
353 else:
354 # No PID file: no background process tracked, update state only
355 state.status = "interrupted"
356 persistence.save_state(state)
357 logger.success(f"Marked {stem} as interrupted")
359 return 0
362def cmd_resume(
363 loop_name: str,
364 args: argparse.Namespace,
365 loops_dir: Path,
366 logger: Logger,
367) -> int:
368 """Resume an interrupted loop."""
369 import os
371 from little_loops.fsm.persistence import RESUMABLE_STATUSES, PersistentExecutor
373 running_dir = loops_dir / ".running"
374 running_dir.mkdir(parents=True, exist_ok=True)
376 # Discover all instances and resolve to a single resumable one.
377 instances = _find_instances(loop_name, running_dir)
378 resumable = [(iid, s) for iid, s in instances if s.status in RESUMABLE_STATUSES]
380 explicit_instance_id = getattr(args, "instance_id", None)
381 if explicit_instance_id:
382 filtered: list[tuple[str | None, LoopState]] = [
383 (iid, s) for iid, s in resumable if iid == explicit_instance_id
384 ]
385 if not filtered:
386 print(
387 f"Instance '{explicit_instance_id}' not found among resumable instances of '{loop_name}'."
388 )
389 if resumable:
390 print("Resumable instances:")
391 for iid, _ in resumable:
392 print(f" {iid or loop_name}")
393 return 1
394 resumable = filtered
395 elif len(resumable) > 1:
396 # Auto-select the most recent instance by sorted filename (matches cmd_monitor()).
397 selected_id = resumable[-1][0]
398 print(f"Auto-selected latest instance: {selected_id}")
399 resumable = [resumable[-1]]
401 # Use discovered instance_id (or fall back to args / None for no-state case)
402 if resumable:
403 instance_id: str | None = resumable[0][0]
404 state_for_display = resumable[0][1]
405 else:
406 instance_id = None
407 state_for_display = None
409 # Background mode: spawn detached process and return
410 if getattr(args, "background", False):
411 if instance_id is None:
412 print(f"No resumable instances of '{loop_name}'.")
413 return 1
414 return run_background(
415 loop_name, args, loops_dir, subcommand="resume", instance_id=instance_id
416 )
418 # Register PID file for all foreground runs so cmd_stop can send SIGTERM (BUG-639).
419 # Background-spawned processes (foreground_internal=True) have their PID written by the
420 # parent in run_background(); plain foreground runs must write their own PID here.
421 if instance_id is None:
422 instance_id = getattr(args, "instance_id", None)
424 pid_file = running_dir / f"{instance_id or loop_name}.pid"
425 foreground_pid_file: Path | None = pid_file
427 if not getattr(args, "foreground_internal", False):
428 pid_file.parent.mkdir(parents=True, exist_ok=True)
429 pid_file.write_text(str(os.getpid()))
431 def _cleanup_pid() -> None:
432 pid_file.unlink(missing_ok=True)
434 atexit.register(_cleanup_pid)
436 try:
437 fsm = load_loop(loop_name, loops_dir, logger)
438 except FileNotFoundError as e:
439 logger.error(str(e))
440 return 1
441 except ValueError as e:
442 logger.error(f"Validation error: {e}")
443 return 1
445 if getattr(args, "baseline", False):
446 logger.error(
447 "--baseline is not supported with resume. Start a fresh run with 'll-loop run'."
448 )
449 return 1
451 try:
452 loop_path = resolve_loop_path(loop_name, loops_dir)
453 except FileNotFoundError:
454 loop_path = None
456 for kv in getattr(args, "context", None) or []:
457 if "=" not in kv:
458 raise SystemExit(f"Invalid --context format: {kv!r} (expected KEY=VALUE)")
459 key, _, value = kv.partition("=")
460 fsm.context[key.strip()] = value.strip()
462 # Re-inject run_dir using the same instance_id as the original run so resumed
463 # loops write artifacts to the same directory they started with.
464 if "run_dir" not in fsm.context and instance_id is not None:
465 fsm.context["run_dir"] = str(loops_dir / "runs" / instance_id) + "/"
467 # Re-inject input hash for checkpoint fingerprinting during resumed runs.
468 if "input_hash" not in fsm.context and isinstance(fsm.context.get("input"), str):
469 fsm.context["input_hash"] = hashlib.sha256(fsm.context["input"].encode()).hexdigest()[:12]
471 if getattr(args, "delay", None) is not None:
472 fsm.backoff = args.delay
474 # Apply YAML loop config env-var overrides (CLI flags below overwrite these)
475 if fsm.config is not None and isinstance(fsm.config.handoff_threshold, int):
476 os.environ["LL_HANDOFF_THRESHOLD"] = str(fsm.config.handoff_threshold)
478 if getattr(args, "handoff_threshold", None) is not None:
479 if not (1 <= args.handoff_threshold <= 100):
480 raise SystemExit("--handoff-threshold must be between 1 and 100")
481 os.environ["LL_HANDOFF_THRESHOLD"] = str(args.handoff_threshold)
483 # Show context if resuming from a handoff
484 if state_for_display and state_for_display.status == "awaiting_continuation":
485 print(f"Resuming from context handoff (iteration {state_for_display.iteration})...")
486 if state_for_display.continuation_prompt:
487 prompt_preview = state_for_display.continuation_prompt[:500]
488 if len(state_for_display.continuation_prompt) > 500:
489 prompt_preview += "..."
490 print(f"Context: {prompt_preview}")
491 print()
493 from little_loops.config import BRConfig
494 from little_loops.design_tokens import load_design_tokens, render_as_prompt_context
495 from little_loops.extension import wire_extensions
496 from little_loops.fsm.rate_limit_circuit import RateLimitCircuit
497 from little_loops.transport import wire_transports
499 config = BRConfig(Path.cwd())
501 if not fsm.context.get("design_tokens_context"):
502 _tokens = load_design_tokens(config)
503 fsm.context["design_tokens_context"] = render_as_prompt_context(_tokens) if _tokens else ""
505 circuit = (
506 RateLimitCircuit(Path(config.commands.rate_limits.circuit_breaker_path))
507 if config.commands.rate_limits.circuit_breaker_enabled
508 else None
509 )
510 executor = PersistentExecutor(
511 fsm, loops_dir=loops_dir, circuit=circuit, instance_id=instance_id
512 )
514 # Register signal handlers for graceful shutdown (same as cmd_run)
515 register_loop_signal_handlers(executor, pid_file=foreground_pid_file)
517 wire_extensions(executor.event_bus, config.extensions, executor=executor)
518 wire_transports(executor.event_bus, config.events)
520 # Route through run_foreground (BUG-1645) so the resume path shares the
521 # display-callback wiring that cmd_run gets via run_foreground. Without this
522 # the FSM event bus has no subscriber on a resumed run, leaving the terminal
523 # silent for the entire duration (no per-iteration lines, no FSM diagram).
524 from little_loops.cli.loop._helpers import run_foreground
526 _edge_label_colors = config.cli.colors.fsm_edge_labels.to_dict()
527 _highlight_color = config.cli.colors.fsm_active_state
528 _badges = config.loops.glyphs.to_dict()
530 try:
531 return run_foreground(
532 executor,
533 fsm,
534 args,
535 highlight_color=_highlight_color,
536 edge_label_colors=_edge_label_colors,
537 badges=_badges,
538 mode="resume",
539 instance_id=instance_id,
540 running_dir=running_dir,
541 loop_path=loop_path,
542 model=fsm.llm.model,
543 )
544 finally:
545 executor.close_transports()
548def _print_last_state(state: LoopState) -> None:
549 """Print last-known state (used when not actively tailing)."""
550 print(f"Loop: {state.loop_name}")
551 print(f"Status: {state.status}")
552 print(f"Current state: {state.current_state}")
553 print(f"Iteration: {state.iteration}")
556def cmd_monitor(args: argparse.Namespace, loops_dir: Path) -> int:
557 """Attach to a running loop and render its FSM state in realtime.
559 Read-only attach: tails ``<stem>.events.jsonl`` and forwards events to a
560 ``StateFeedRenderer``. Ctrl-C detaches without sending any signal to the
561 loop process (FEAT-1764).
562 """
563 loop_name = args.loop
564 running_dir = loops_dir / ".running"
565 instances = _find_instances(loop_name, running_dir)
567 if not instances:
568 print(f"No instances of '{loop_name}' found")
569 return 1
571 instance_id, state = instances[-1]
572 stem = instance_id or loop_name
574 pid = _read_pid_file(running_dir / f"{stem}.pid")
575 if pid is None:
576 pid = getattr(state, "pid", None)
578 if pid is None or not _process_alive(pid):
579 _print_last_state(state)
580 return 0
582 # Fabricate missing Namespace attrs that StateFeedRenderer reads.
583 for attr, default in (
584 ("quiet", False),
585 ("verbose", False),
586 ("show_diagrams", None),
587 ("clear", True),
588 ("diagram_edge_labels", None),
589 ("diagram_state_detail", None),
590 ("diagram_scope", None),
591 ("follow", False),
592 ):
593 if not hasattr(args, attr):
594 setattr(args, attr, default)
596 try:
597 fsm = load_loop(loop_name, loops_dir, Logger())
598 except (FileNotFoundError, ValueError) as e:
599 print(f"Cannot load loop '{loop_name}': {e}")
600 return 1
602 try:
603 loop_path = resolve_loop_path(loop_name, loops_dir)
604 except FileNotFoundError:
605 loop_path = None
607 # Late import: tests patch StateFeedRenderer at its module-of-origin
608 # (little_loops.cli.loop._helpers.StateFeedRenderer); using a function-local
609 # import ensures the patch takes effect at call time.
610 from little_loops.cli.loop._helpers import (
611 StateFeedRenderer,
612 _install_sigwinch_handler,
613 _restore_sigwinch_handler,
614 )
616 renderer = StateFeedRenderer(
617 fsm, args, loops_dir=loops_dir, loop_path=loop_path, model=fsm.llm.model
618 )
620 events_file = running_dir / f"{stem}.events.jsonl"
622 if not events_file.exists():
623 _print_last_state(state)
624 return 0
626 sigwinch_installed = False
627 if renderer.in_pinned_mode:
628 _install_sigwinch_handler()
629 sigwinch_installed = True
631 try:
632 with open(events_file, encoding="utf-8") as ev_f:
633 ev_f.seek(0, 2)
634 while True:
635 progressed = False
636 try:
637 line = ev_f.readline()
638 except FileNotFoundError:
639 break
640 if line:
641 progressed = True
642 stripped = line.strip()
643 if stripped:
644 try:
645 event = json.loads(stripped)
646 except json.JSONDecodeError:
647 event = None
648 if event is not None:
649 renderer.handle_event(event)
650 if not progressed:
651 if not _process_alive(pid):
652 break
653 if not events_file.exists():
654 break
655 time.sleep(0.1)
656 except KeyboardInterrupt:
657 return 0
658 finally:
659 if sigwinch_installed:
660 _restore_sigwinch_handler()
662 return 0