Coverage for little_loops / cli / loop / lifecycle.py: 0%
431 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -0500
1"""ll-loop lifecycle subcommands: status, stop, resume."""
3from __future__ import annotations
5import argparse
6import atexit
7import json
8import os
9import signal
10import time
11from pathlib import Path
13from little_loops.cli.loop._helpers import (
14 load_loop,
15 register_loop_signal_handlers,
16 run_background,
17)
18from little_loops.fsm.concurrency import _process_alive
19from little_loops.fsm.persistence import (
20 LoopState,
21 StatePersistence,
22 _find_instances,
23 _read_pid_file,
24 _reconcile_stale_running,
25)
26from little_loops.logger import Logger
29def _format_relative_time(seconds: float) -> str:
30 """Format seconds as a human-readable relative time string (e.g., '3m ago').
32 Delegates to the shared ``format_relative_time`` in ``cli.output``.
33 """
34 from little_loops.cli.output import format_relative_time
36 return format_relative_time(seconds)
39def _format_log_label(running_dir: Path, stem: str) -> str:
40 """Return one of three Log: labels based on run-mode signals.
42 - log exists → path string (background run, normal)
43 - no log, no pid → foreground run (output went to terminal)
44 - no log, pid exists → background run whose log was deleted (something wrong)
45 """
46 log_file = running_dir / f"{stem}.log"
47 if log_file.exists():
48 return str(log_file)
49 pid_file = running_dir / f"{stem}.pid"
50 if pid_file.exists():
51 return f"(expected {log_file}, missing)"
52 return "(foreground run — output went to terminal)"
55def _get_events_info(running_dir: Path, stem: str) -> tuple[str | None, str | None]:
56 """Return (events_file_str, formatted_detail_line) for the events.jsonl file.
58 Returns (None, None) if the file doesn't exist.
59 Detail line format: 'Events: <path> (N events, last Xm ago)'
60 """
61 events_file = running_dir / f"{stem}.events.jsonl"
62 if not events_file.exists():
63 return None, None
64 try:
65 lines = [ln for ln in events_file.read_text().splitlines() if ln.strip()]
66 count = len(lines)
67 detail = f"({count} events)"
68 if lines:
69 try:
70 last_entry = json.loads(lines[-1])
71 last_ts = last_entry.get("ts")
72 if last_ts is not None:
73 from datetime import UTC, datetime
75 dt = datetime.fromisoformat(last_ts)
76 age = (datetime.now(tz=UTC) - dt).total_seconds()
77 detail = f"({count} events, last {_format_relative_time(age)})"
78 except (json.JSONDecodeError, AttributeError, ValueError, OverflowError):
79 pass
80 return str(events_file), f"Events: {events_file} {detail}"
81 except OSError:
82 return str(events_file), f"Events: {events_file}"
85def _kill_with_timeout(pid: int, label: str, logger: Logger) -> None:
86 """Send SIGTERM to pid; escalate to SIGKILL after 10 s if still alive."""
87 os.kill(pid, signal.SIGTERM)
88 for _ in range(10):
89 time.sleep(1)
90 if not _process_alive(pid):
91 return
92 try:
93 os.kill(pid, signal.SIGKILL)
94 logger.warning(f"Sent SIGKILL to {label} (PID: {pid})")
95 except OSError:
96 pass
99def _status_single(
100 instance_id: str | None,
101 state: LoopState,
102 loop_name: str,
103 running_dir: Path,
104 args: argparse.Namespace | None,
105) -> int:
106 """Render status for one instance (human-readable or JSON)."""
107 from little_loops.cli.output import print_json
109 stem = instance_id or loop_name
110 persistence = StatePersistence(loop_name, running_dir.parent, instance_id=instance_id)
111 state = _reconcile_stale_running(state, persistence, running_dir, stem)
113 pid_file = running_dir / f"{stem}.pid"
114 pid = _read_pid_file(pid_file)
115 pid_source: str | None = "pid_file" if pid is not None else None
117 if pid is None:
118 lock_file_path = running_dir / f"{stem}.lock"
119 if lock_file_path.exists():
120 try:
121 with open(lock_file_path) as _lf:
122 lock_data = json.load(_lf)
123 pid = lock_data.get("pid")
124 if pid is not None:
125 pid_source = "lock_file"
126 except (json.JSONDecodeError, KeyError, OSError):
127 pass
129 log_file = running_dir / f"{stem}.log"
130 log_file_str: str | None = None
131 log_updated_ago: str | None = None
132 last_event: str | None = None
133 if log_file.exists():
134 log_file_str = str(log_file)
135 age_seconds = time.time() - log_file.stat().st_mtime
136 log_updated_ago = _format_relative_time(age_seconds)
137 try:
138 lines = log_file.read_text().splitlines()
139 non_empty = [ln for ln in lines if ln.strip()]
140 last_event = non_empty[-1] if non_empty else None
141 except OSError:
142 last_event = None
144 events_file_str, events_detail_line = _get_events_info(running_dir, stem)
146 if getattr(args, "json", False):
147 d = state.to_dict()
148 d["pid"] = pid
149 d["pid_source"] = pid_source
150 d["log_file"] = log_file_str
151 d["log_updated_ago"] = log_updated_ago
152 d["last_event"] = last_event
153 d["events_file"] = events_file_str
154 print_json(d)
155 return 0
157 print(f"Loop: {state.loop_name}")
158 print(f"Status: {state.status}")
159 print(f"Current state: {state.current_state}")
160 print(f"Iteration: {state.iteration}")
161 print(f"Started: {state.started_at}")
162 print(f"Updated: {state.updated_at}")
164 if pid is not None:
165 if _process_alive(pid):
166 print(f"PID: {pid} (running)")
167 else:
168 print(f"PID: {pid} (not running - stale PID file)")
170 log_label = _format_log_label(running_dir, stem)
171 print(f"Log: {log_label}")
172 if log_file_str is not None:
173 print(f"Log updated: {log_updated_ago}")
174 if last_event:
175 print(f"Last event: {last_event}")
177 if events_detail_line is not None:
178 print(events_detail_line)
180 if state.continuation_prompt:
181 prompt_preview = state.continuation_prompt[:200]
182 if len(state.continuation_prompt) > 200:
183 prompt_preview += "..."
184 print(f"Continuation context: {prompt_preview}")
185 return 0
188def cmd_status(
189 loop_name: str,
190 loops_dir: Path,
191 logger: Logger,
192 args: argparse.Namespace | None = None,
193) -> int:
194 """Show loop status."""
195 from little_loops.cli.output import print_json
197 running_dir = loops_dir / ".running"
198 instances = _find_instances(loop_name, running_dir)
200 if not instances:
201 logger.error(f"No state found for: {loop_name}")
202 return 1
204 if len(instances) == 1:
205 instance_id, state = instances[0]
206 return _status_single(instance_id, state, loop_name, running_dir, args)
208 # Multiple instances: aggregate display
209 if getattr(args, "json", False):
210 result_list = []
211 for instance_id, state in instances:
212 stem = instance_id or loop_name
213 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
214 state = _reconcile_stale_running(state, persistence, running_dir, stem)
215 pid_file = running_dir / f"{stem}.pid"
216 pid = _read_pid_file(pid_file)
217 pid_source: str | None = "pid_file" if pid is not None else None
218 if pid is None:
219 lock_file_path = running_dir / f"{stem}.lock"
220 if lock_file_path.exists():
221 try:
222 with open(lock_file_path) as _lf:
223 lock_data = json.load(_lf)
224 pid = lock_data.get("pid")
225 if pid is not None:
226 pid_source = "lock_file"
227 except (json.JSONDecodeError, KeyError, OSError):
228 pass
229 log_file = running_dir / f"{stem}.log"
230 d = state.to_dict()
231 d["instance_id"] = instance_id
232 d["pid"] = pid
233 d["pid_source"] = pid_source
234 if log_file.exists():
235 d["log_file"] = str(log_file)
236 d["log_updated_ago"] = _format_relative_time(time.time() - log_file.stat().st_mtime)
237 else:
238 d["log_file"] = None
239 d["log_updated_ago"] = None
240 events_file_str, _ = _get_events_info(running_dir, stem)
241 d["events_file"] = events_file_str
242 result_list.append(d)
243 print_json(result_list)
244 return 0
246 print(f"{len(instances)} instances of '{loop_name}':")
247 for i, (instance_id, state) in enumerate(instances, 1):
248 stem = instance_id or loop_name
249 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
250 state = _reconcile_stale_running(state, persistence, running_dir, stem)
251 pid_file = running_dir / f"{stem}.pid"
252 pid = _read_pid_file(pid_file)
253 if pid is None:
254 lock_file_path = running_dir / f"{stem}.lock"
255 if lock_file_path.exists():
256 try:
257 with open(lock_file_path) as _lf:
258 lock_data = json.load(_lf)
259 pid = lock_data.get("pid")
260 except (json.JSONDecodeError, KeyError, OSError):
261 pass
262 log_file = running_dir / f"{stem}.log"
264 print()
265 print(f"[{i}] {stem}")
266 print(f" Status: {state.status}")
267 print(f" Current state: {state.current_state}")
268 print(f" Iteration: {state.iteration}")
269 if pid is not None:
270 if _process_alive(pid):
271 print(f" PID: {pid} (running)")
272 else:
273 print(f" PID: {pid} (not running - stale PID file)")
274 log_label = _format_log_label(running_dir, stem)
275 print(f" Log: {log_label}")
276 if log_file.exists():
277 age_seconds = time.time() - log_file.stat().st_mtime
278 print(f" Log updated: {_format_relative_time(age_seconds)}")
279 _, events_detail_line = _get_events_info(running_dir, stem)
280 if events_detail_line is not None:
281 print(f" {events_detail_line}")
282 return 0
285def cmd_stop(
286 loop_name: str,
287 loops_dir: Path,
288 logger: Logger,
289) -> int:
290 """Stop a running loop."""
291 from little_loops.fsm.persistence import StatePersistence # noqa: PLC0415
293 running_dir = loops_dir / ".running"
294 instances = _find_instances(loop_name, running_dir)
296 if not instances:
297 logger.error(f"No state found for: {loop_name}")
298 return 1
300 running_instances = [(iid, s) for iid, s in instances if s.status == "running"]
302 if not running_instances:
303 # Secondary check: an orphaned lock-file with a live PID blocks scope
304 # acquisition even when state is not "running". Kill the holder and release.
305 for instance_id, state in instances:
306 stem = instance_id or loop_name
307 lock_file = running_dir / f"{stem}.lock"
308 if lock_file.exists():
309 lock_pid: int | None = None
310 try:
311 with open(lock_file) as _lf:
312 lock_data = json.load(_lf)
313 lock_pid = lock_data.get("pid")
314 except (json.JSONDecodeError, KeyError, OSError):
315 pass
316 if lock_pid and _process_alive(lock_pid):
317 logger.warning(
318 f"Loop state is '{state.status}' but lock file holds live PID {lock_pid}. "
319 "Killing orphaned lock holder..."
320 )
321 _kill_with_timeout(lock_pid, stem, logger)
322 lock_file.unlink(missing_ok=True)
323 logger.success(f"Released orphaned scope lock for {stem}")
324 return 0
325 elif lock_pid is not None:
326 # Dead PID: stale lock file, just remove it
327 lock_file.unlink(missing_ok=True)
328 logger.info(f"Removed stale lock file for {stem}")
329 return 0
330 _, state = instances[0]
331 logger.error(f"Loop not running: {loop_name} (status: {state.status})")
332 return 1
334 for instance_id, state in running_instances:
335 persistence = StatePersistence(loop_name, loops_dir, instance_id=instance_id)
336 stem = instance_id or loop_name
337 pid_file = running_dir / f"{stem}.pid"
338 pid = _read_pid_file(pid_file)
340 if pid is not None:
341 if _process_alive(pid):
342 _kill_with_timeout(pid, stem, logger)
343 state.status = "interrupted"
344 persistence.save_state(state)
345 pid_file.unlink(missing_ok=True)
346 logger.success(f"Stopped {stem} (PID: {pid})")
347 else:
348 # Process already exited: preserve its final status, only clean up PID file
349 logger.info(f"Process {pid} not running, cleaning up PID file")
350 pid_file.unlink(missing_ok=True)
351 else:
352 # No PID file: no background process tracked, update state only
353 state.status = "interrupted"
354 persistence.save_state(state)
355 logger.success(f"Marked {stem} as interrupted")
357 return 0
360def cmd_resume(
361 loop_name: str,
362 args: argparse.Namespace,
363 loops_dir: Path,
364 logger: Logger,
365) -> int:
366 """Resume an interrupted loop."""
367 from little_loops.fsm.persistence import PersistentExecutor
369 # Background mode: spawn detached process and return
370 if getattr(args, "background", False):
371 return run_background(loop_name, args, loops_dir, subcommand="resume")
373 # Register PID file for all foreground runs so cmd_stop can send SIGTERM (BUG-639).
374 # Background-spawned processes (foreground_internal=True) have their PID written by the
375 # parent in run_background(); plain foreground runs must write their own PID here.
376 import os
378 running_dir = loops_dir / ".running"
379 running_dir.mkdir(parents=True, exist_ok=True)
381 # Discover all instances and resolve to a single resumable one.
382 instances = _find_instances(loop_name, running_dir)
383 from little_loops.fsm.persistence import RESUMABLE_STATUSES
385 resumable = [(iid, s) for iid, s in instances if s.status in RESUMABLE_STATUSES]
387 explicit_instance_id = getattr(args, "instance_id", None)
388 if explicit_instance_id:
389 filtered: list[tuple[str | None, LoopState]] = [
390 (iid, s) for iid, s in resumable if iid == explicit_instance_id
391 ]
392 if not filtered:
393 print(
394 f"Instance '{explicit_instance_id}' not found among resumable instances of '{loop_name}'."
395 )
396 if resumable:
397 print("Resumable instances:")
398 for iid, _ in resumable:
399 print(f" {iid or loop_name}")
400 return 1
401 resumable = filtered
403 if len(resumable) > 1:
404 print(f"Multiple instances of '{loop_name}' are resumable:")
405 for iid, _ in resumable:
406 print(f" {iid or loop_name}")
407 print("Use --instance-id to select one.")
408 return 1
410 # Use discovered instance_id (or fall back to args / None for no-state case)
411 if resumable:
412 instance_id: str | None = resumable[0][0]
413 state_for_display = resumable[0][1]
414 else:
415 instance_id = getattr(args, "instance_id", None)
416 state_for_display = None
418 pid_file = running_dir / f"{instance_id or loop_name}.pid"
419 foreground_pid_file: Path | None = pid_file
421 if not getattr(args, "foreground_internal", False):
422 pid_file.write_text(str(os.getpid()))
424 def _cleanup_pid() -> None:
425 pid_file.unlink(missing_ok=True)
427 atexit.register(_cleanup_pid)
429 try:
430 fsm = load_loop(loop_name, loops_dir, logger)
431 except FileNotFoundError as e:
432 logger.error(str(e))
433 return 1
434 except ValueError as e:
435 logger.error(f"Validation error: {e}")
436 return 1
438 for kv in getattr(args, "context", None) or []:
439 if "=" not in kv:
440 raise SystemExit(f"Invalid --context format: {kv!r} (expected KEY=VALUE)")
441 key, _, value = kv.partition("=")
442 fsm.context[key.strip()] = value.strip()
444 # Re-inject run_dir using the same instance_id as the original run so resumed
445 # loops write artifacts to the same directory they started with.
446 if "run_dir" not in fsm.context and instance_id is not None:
447 fsm.context["run_dir"] = str(loops_dir / "runs" / instance_id) + "/"
449 if getattr(args, "delay", None) is not None:
450 fsm.backoff = args.delay
452 # Apply YAML loop config env-var overrides (CLI flags below overwrite these)
453 if fsm.config is not None and isinstance(fsm.config.handoff_threshold, int):
454 os.environ["LL_HANDOFF_THRESHOLD"] = str(fsm.config.handoff_threshold)
456 if getattr(args, "handoff_threshold", None) is not None:
457 if not (1 <= args.handoff_threshold <= 100):
458 raise SystemExit("--handoff-threshold must be between 1 and 100")
459 os.environ["LL_HANDOFF_THRESHOLD"] = str(args.handoff_threshold)
461 # Show context if resuming from a handoff
462 if state_for_display and state_for_display.status == "awaiting_continuation":
463 print(f"Resuming from context handoff (iteration {state_for_display.iteration})...")
464 if state_for_display.continuation_prompt:
465 prompt_preview = state_for_display.continuation_prompt[:500]
466 if len(state_for_display.continuation_prompt) > 500:
467 prompt_preview += "..."
468 print(f"Context: {prompt_preview}")
469 print()
471 from little_loops.config import BRConfig
472 from little_loops.design_tokens import load_design_tokens, render_as_prompt_context
473 from little_loops.extension import wire_extensions
474 from little_loops.fsm.rate_limit_circuit import RateLimitCircuit
475 from little_loops.transport import wire_transports
477 config = BRConfig(Path.cwd())
479 if "design_tokens_context" not in fsm.context:
480 _tokens = load_design_tokens(config)
481 fsm.context["design_tokens_context"] = render_as_prompt_context(_tokens) if _tokens else ""
483 circuit = (
484 RateLimitCircuit(Path(config.commands.rate_limits.circuit_breaker_path))
485 if config.commands.rate_limits.circuit_breaker_enabled
486 else None
487 )
488 executor = PersistentExecutor(
489 fsm, loops_dir=loops_dir, circuit=circuit, instance_id=instance_id
490 )
492 # Register signal handlers for graceful shutdown (same as cmd_run)
493 register_loop_signal_handlers(executor, pid_file=foreground_pid_file)
495 wire_extensions(executor.event_bus, config.extensions, executor=executor)
496 wire_transports(executor.event_bus, config.events)
498 # Route through run_foreground (BUG-1645) so the resume path shares the
499 # display-callback wiring that cmd_run gets via run_foreground. Without this
500 # the FSM event bus has no subscriber on a resumed run, leaving the terminal
501 # silent for the entire duration (no per-iteration lines, no FSM diagram).
502 from little_loops.cli.loop._helpers import run_foreground
504 _edge_label_colors = config.cli.colors.fsm_edge_labels.to_dict()
505 _highlight_color = config.cli.colors.fsm_active_state
506 _badges = config.loops.glyphs.to_dict()
508 try:
509 return run_foreground(
510 executor,
511 fsm,
512 args,
513 highlight_color=_highlight_color,
514 edge_label_colors=_edge_label_colors,
515 badges=_badges,
516 mode="resume",
517 instance_id=instance_id,
518 running_dir=running_dir,
519 )
520 finally:
521 executor.close_transports()
524def _print_last_state(state: LoopState) -> None:
525 """Print last-known state (used when not actively tailing)."""
526 print(f"Loop: {state.loop_name}")
527 print(f"Status: {state.status}")
528 print(f"Current state: {state.current_state}")
529 print(f"Iteration: {state.iteration}")
532def cmd_monitor(args: argparse.Namespace, loops_dir: Path) -> int:
533 """Attach to a running loop and render its FSM state in realtime.
535 Read-only attach: tails ``<stem>.events.jsonl`` and the log file from disk,
536 forwarding events to a ``StateFeedRenderer``. Ctrl-C detaches without
537 sending any signal to the loop process (FEAT-1764).
538 """
539 loop_name = args.loop
540 running_dir = loops_dir / ".running"
541 instances = _find_instances(loop_name, running_dir)
543 if not instances:
544 print(f"No instances of '{loop_name}' found")
545 return 1
547 instance_id, state = instances[-1]
548 stem = instance_id or loop_name
550 pid = _read_pid_file(running_dir / f"{stem}.pid")
551 if pid is None:
552 pid = getattr(state, "pid", None)
554 if pid is None or not _process_alive(pid):
555 _print_last_state(state)
556 return 0
558 # Fabricate missing Namespace attrs that StateFeedRenderer reads.
559 for attr, default in (
560 ("quiet", False),
561 ("verbose", False),
562 ("show_diagrams", None),
563 ("clear", False),
564 ("diagram_edge_labels", None),
565 ("diagram_state_detail", None),
566 ("diagram_scope", None),
567 ("follow", False),
568 ):
569 if not hasattr(args, attr):
570 setattr(args, attr, default)
572 try:
573 fsm = load_loop(loop_name, loops_dir, Logger())
574 except (FileNotFoundError, ValueError) as e:
575 print(f"Cannot load loop '{loop_name}': {e}")
576 return 1
578 # Late import: tests patch StateFeedRenderer at its module-of-origin
579 # (little_loops.cli.loop._helpers.StateFeedRenderer); using a function-local
580 # import ensures the patch takes effect at call time.
581 from little_loops.cli.loop._helpers import (
582 StateFeedRenderer,
583 _install_sigwinch_handler,
584 _restore_sigwinch_handler,
585 )
587 renderer = StateFeedRenderer(fsm, args, loops_dir=loops_dir)
589 log_override = getattr(args, "log_file", None)
590 log_path = Path(log_override) if log_override else running_dir / f"{stem}.log"
591 events_file = running_dir / f"{stem}.events.jsonl"
593 if not events_file.exists():
594 _print_last_state(state)
595 return 0
597 sigwinch_installed = False
598 if renderer.in_pinned_mode:
599 _install_sigwinch_handler()
600 sigwinch_installed = True
602 try:
603 with open(events_file, encoding="utf-8") as ev_f:
604 ev_f.seek(0, 2)
605 log_f = None
606 if log_path.exists():
607 log_f = open(log_path, encoding="utf-8")
608 log_f.seek(0, 2)
609 try:
610 while True:
611 progressed = False
612 try:
613 line = ev_f.readline()
614 except FileNotFoundError:
615 break
616 if line:
617 progressed = True
618 stripped = line.strip()
619 if stripped:
620 try:
621 event = json.loads(stripped)
622 except json.JSONDecodeError:
623 event = None
624 if event is not None:
625 renderer.handle_event(event)
626 if log_f is not None:
627 try:
628 log_line = log_f.readline()
629 except FileNotFoundError:
630 log_line = ""
631 if log_line:
632 progressed = True
633 print(log_line.rstrip())
634 if not progressed:
635 if not _process_alive(pid):
636 break
637 if not events_file.exists():
638 break
639 time.sleep(0.1)
640 finally:
641 if log_f is not None:
642 log_f.close()
643 except KeyboardInterrupt:
644 return 0
645 finally:
646 if sigwinch_installed:
647 _restore_sigwinch_handler()
649 return 0