Coverage for little_loops/fsm/persistence.py: 21%

1"""State persistence and event streaming for FSM loops.

3This module provides persistence capabilities for FSM loop execution:

4- LoopState: Dataclass representing loop execution state

5- StatePersistence: File I/O for state and events

6- PersistentExecutor: Wrapper that persists state during execution

7- Utility functions for listing running loops and reading history

9File structure:

10 .loops/

11 ├── fix-types.yaml # Loop definition

12 ├── .running/ # Runtime state (auto-managed)

13 │ ├── fix-types-20260503T122306.state.json

14 │ └── fix-types-20260503T122306.events.jsonl

15 └── .history/ # Archived run logs (auto-populated)

16 └── 2024-01-15T103000-fix-types/

17 ├── state.json

18 └── events.jsonl

19"""

21from __future__ import annotations

23import json

24import logging

25import os

26import re

27import shutil

28import tempfile

29import time

30from dataclasses import dataclass, field

31from datetime import UTC, datetime

32from pathlib import Path

33from typing import Any

35from little_loops.events import EventBus

36from little_loops.fsm.concurrency import _process_alive

37from little_loops.fsm.executor import EventCallback, ExecutionResult, FSMExecutor

38from little_loops.fsm.schema import FSMLoop

40RUNNING_DIR = ".running"

41HISTORY_DIR = ".history"

43logger = logging.getLogger(__name__)

45_RUN_FOLDER = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{6})-(.+)$")

46_INSTANCE_SUFFIX = re.compile(r"-\d{8}T\d{6}$")

49def _parse_run_folder(name: str) -> tuple[str, str] | None:

50 """Return (run_id, loop_name) from a flat history folder name, or None."""

51 m = _RUN_FOLDER.match(name)

52 return (m.group(1), m.group(2)) if m else None

55def _iso_now() -> str:

56 """Return current time as ISO 8601 string."""

57 return datetime.now(UTC).isoformat()

60def _now_ms() -> int:

61 """Return current time in milliseconds."""

62 return int(time.time() * 1000)

65@dataclass

66class LoopState:

67 """Persistent state for an FSM loop execution.

69 This captures all runtime state needed to resume a loop:

70 - Current state and iteration

71 - Captured variables and previous result

72 - Last evaluation result

73 - Timestamps and status

75 Attributes:

76 loop_name: Name of the loop

77 current_state: Current FSM state name

78 iteration: Current iteration count (1-based)

79 captured: Captured action outputs by variable name

80 prev_result: Previous state's result (output, exit_code, state)

81 last_result: Last evaluation result (verdict, details)

82 started_at: ISO timestamp when loop started

83 updated_at: ISO timestamp when state was last saved

84 status: Execution status (running, completed, failed, interrupted, awaiting_continuation, timed_out)

85 continuation_prompt: Continuation context from handoff signal (if status is awaiting_continuation)

86 accumulated_ms: Total milliseconds elapsed across all segments up to this save (used to restore

87 elapsed time correctly after resume, so duration_ms and ${loop.elapsed_ms} reflect the

88 full loop lifetime rather than only the most recent segment)

89 """

91 loop_name: str

92 current_state: str

93 iteration: int

94 captured: dict[str, dict[str, Any]]

95 prev_result: dict[str, Any] | None

96 last_result: dict[str, Any] | None

97 started_at: str

98 updated_at: str

99 status: (

100 str # "running", "completed", "failed", "interrupted", "awaiting_continuation", "timed_out"

101 )

102 continuation_prompt: str | None = None

103 accumulated_ms: int = 0 # total elapsed ms across all segments (for resume offset)

104 retry_counts: dict[str, int] = field(default_factory=dict) # per-state retry tracking

105 # Per-state rate-limit retry tracking (ENH-1133: dict-of-record).

106 # Each record: {"short_retries": int, "long_retries": int,

107 # "total_wait_seconds": float, "first_seen_at": float | None}.

108 # Legacy int values (dict[str, int]) are coerced in from_dict.

109 rate_limit_retries: dict[str, dict[str, Any]] = field(default_factory=dict)

110 # Count of consecutive rate_limit_exhausted emissions across states. Reset

111 # on any non-rate-limited state outcome. Persisted for resume durability.

112 consecutive_rate_limit_exhaustions: int = 0

113 # Per-edge revisit tracking for cycle detection.

114 edge_revisit_counts: dict[str, int] = field(default_factory=dict)

115 active_sub_loop: str | None = None # name of currently executing sub-loop (observability)

116 pid: int | None = None # OS PID of the process that started this run (for reconciliation sweep)

117

118 def to_dict(self) -> dict[str, Any]:

119 """Convert to dictionary for JSON serialization."""

120 result = {

121 "loop_name": self.loop_name,

122 "current_state": self.current_state,

123 "iteration": self.iteration,

124 "captured": self.captured,

125 "prev_result": self.prev_result,

126 "last_result": self.last_result,

127 "started_at": self.started_at,

128 "updated_at": self.updated_at,

129 "status": self.status,

130 "accumulated_ms": self.accumulated_ms,

131 }

132 if self.continuation_prompt is not None:

133 result["continuation_prompt"] = self.continuation_prompt

134 if self.retry_counts:

135 result["retry_counts"] = self.retry_counts

136 if self.rate_limit_retries:

137 result["rate_limit_retries"] = self.rate_limit_retries

138 if self.consecutive_rate_limit_exhaustions:

139 result["consecutive_rate_limit_exhaustions"] = self.consecutive_rate_limit_exhaustions

140 if self.edge_revisit_counts:

141 result["edge_revisit_counts"] = self.edge_revisit_counts

142 if self.active_sub_loop is not None:

143 result["active_sub_loop"] = self.active_sub_loop

144 if self.pid is not None:

145 result["pid"] = self.pid

146 return result

147

148 @classmethod

149 def from_dict(cls, data: dict[str, Any]) -> LoopState:

150 """Create LoopState from dictionary.

151

152 Migrates legacy ``rate_limit_retries`` values from ``dict[str, int]``

153 (BUG-1107 pre-ENH-1133 shape) to the dict-of-record shape. Integer

154 values are coerced to ``{"short_retries": <int>, "long_retries": 0,

155 "total_wait_seconds": 0.0, "first_seen_at": None}``.

156

157 Args:

158 data: Dictionary with loop state fields

159

160 Returns:

161 LoopState instance

162 """

163 raw_rl = data.get("rate_limit_retries", {}) or {}

164 migrated_rl: dict[str, dict[str, Any]] = {}

165 for state_name, value in raw_rl.items():

166 if isinstance(value, int):

167 migrated_rl[state_name] = {

168 "short_retries": value,

169 "long_retries": 0,

170 "total_wait_seconds": 0.0,

171 "first_seen_at": None,

172 }

173 elif isinstance(value, dict):

174 migrated_rl[state_name] = value

175 return cls(

176 loop_name=data["loop_name"],

177 current_state=data["current_state"],

178 iteration=data["iteration"],

179 captured=data.get("captured", {}),

180 prev_result=data.get("prev_result"),

181 last_result=data.get("last_result"),

182 started_at=data["started_at"],

183 updated_at=data.get("updated_at", ""),

184 status=data["status"],

185 continuation_prompt=data.get("continuation_prompt"),

186 accumulated_ms=data.get("accumulated_ms", 0),

187 retry_counts=data.get("retry_counts", {}),

188 rate_limit_retries=migrated_rl,

189 consecutive_rate_limit_exhaustions=data.get("consecutive_rate_limit_exhaustions", 0),

190 edge_revisit_counts=data.get("edge_revisit_counts", {}),

191 active_sub_loop=data.get("active_sub_loop"),

192 pid=data.get("pid"),

193 )

194

195

196class StatePersistence:

197 """Manage loop state persistence and event streaming.

198

199 Handles file I/O for:

200 - State file: JSON file with current execution state

201 - Events file: JSONL file with execution events (append-only)

202

203 Files are stored in .loops/.running/<instance_id>.*

204 """

205

206 def __init__(

207 self, loop_name: str, loops_dir: Path | None = None, instance_id: str | None = None

208 ) -> None:

209 """Initialize persistence for a loop.

210

211 Args:

212 loop_name: Name of the loop

213 loops_dir: Base directory for loops (default: .loops)

214 instance_id: Optional unique instance identifier; falls back to loop_name when None

215 """

216 self.loop_name = loop_name

217 self.loops_dir = loops_dir or Path(".loops")

218 self.running_dir = self.loops_dir / RUNNING_DIR

219 stem = instance_id or loop_name

220 self.state_file = self.running_dir / f"{stem}.state.json"

221 self.events_file = self.running_dir / f"{stem}.events.jsonl"

222

223 def initialize(self) -> None:

224 """Create running directory if needed."""

225 self.running_dir.mkdir(parents=True, exist_ok=True)

226

227 def save_state(self, state: LoopState) -> None:

228 """Save current state to file using an atomic write.

229

230 Updates the updated_at timestamp before saving. Writes to a temporary

231 file first, then renames it over the target to avoid leaving a corrupt

232 or empty state file if the process is killed mid-write.

233

234 Args:

235 state: LoopState to save

236 """

237 state.updated_at = _iso_now()

238 data = json.dumps(state.to_dict(), indent=2)

239 tmp_fd, tmp_path = tempfile.mkstemp(dir=self.state_file.parent, suffix=".tmp")

240 try:

241 with os.fdopen(tmp_fd, "w") as f:

242 f.write(data)

243 os.replace(tmp_path, self.state_file)

244 except Exception:

245 os.unlink(tmp_path)

246 raise

247

248 def load_state(self) -> LoopState | None:

249 """Load state from file, or None if not exists.

250

251 Returns:

252 LoopState if file exists and is valid, None otherwise

253 """

254 if not self.state_file.exists():

255 return None

256 try:

257 data = json.loads(self.state_file.read_text())

258 except json.JSONDecodeError:

259 return None

260 try:

261 return LoopState.from_dict(data)

262 except KeyError as e:

263 logger.warning("Corrupted state file %s: missing key %s", self.state_file, e)

264 return None

265

266 def clear_state(self) -> None:

267 """Remove state file."""

268 if self.state_file.exists():

269 self.state_file.unlink()

270

271 def append_event(self, event: dict[str, Any]) -> None:

272 """Append event to JSONL file.

273

274 Args:

275 event: Event dictionary to append

276 """

277 with open(self.events_file, "a", encoding="utf-8") as f:

278 f.write(json.dumps(event) + "\n")

279

280 def read_events(self) -> list[dict[str, Any]]:

281 """Read all events from file.

282

283 Returns:

284 List of event dictionaries, empty if file doesn't exist

285 """

286 if not self.events_file.exists():

287 return []

288 events: list[dict[str, Any]] = []

289 with open(self.events_file, encoding="utf-8") as f:

290 for line in f:

291 line = line.strip()

292 if line:

293 try:

294 events.append(json.loads(line))

295 except json.JSONDecodeError:

296 continue # Skip malformed lines

297 return events

298

299 def clear_events(self) -> None:

300 """Remove events file."""

301 if self.events_file.exists():

302 self.events_file.unlink()

303

304 def archive_run(self) -> Path | None:

305 """Archive current run files to .history/ before clearing.

306

307 Reads the current state to derive the run timestamp, then copies

308 both state.json and events.jsonl into:

309 <loops_dir>/.history/<run_id>-<loop_name>/

310

311 where run_id is a compact ISO timestamp derived from started_at

312 (e.g. "2024-01-15T103000" from "2024-01-15T10:30:00.123456+00:00").

313

314 Returns:

315 Path to the archive directory if files were archived, None if

316 there were no files to archive (fresh run).

317 """

318 has_state = self.state_file.exists()

319 has_events = self.events_file.exists()

320 if not has_state and not has_events:

321 return None

322

323 # Derive run ID from started_at in state file, or fall back to now

324 state = self.load_state()

325 if state is not None and state.started_at:

326 # Compact ISO: strip colons, dots, plus signs; take first 19 chars

327 # e.g. "2024-01-15T10:30:00.123+00:00" → "2024-01-15T103000"

328 run_id = state.started_at.replace(":", "").replace(".", "").replace("+", "")[:17]

329 else:

330 run_id = datetime.now(UTC).strftime("%Y-%m-%dT%H%M%S")

331

332 run_folder = f"{run_id}-{self.loop_name}"

333 archive_dir = self.loops_dir / HISTORY_DIR / run_folder

334 archive_dir.mkdir(parents=True, exist_ok=True)

335

336 if has_state:

337 shutil.copy2(self.state_file, archive_dir / "state.json")

338 if has_events:

339 shutil.copy2(self.events_file, archive_dir / "events.jsonl")

340

341 return archive_dir

342

343 def clear_all(self) -> None:

344 """Archive current run files then clear state and events (for new run)."""

345 self.archive_run()

346 self.clear_state()

347 self.clear_events()

348

349

350def _reconcile_stale_runs(loops_dir: Path) -> int:

351 """Archive state files in .running/ that belong to dead or terminal processes.

352

353 Called at loop startup to clean up files left by crashed or interrupted runs.

354 Returns the count of archived files.

355

356 Strategy (mirrors LockManager.find_conflict() stale-lock cleanup):

357 - Terminal-status files (completed/failed/interrupted/timed_out) are archived

358 unconditionally — they are definitionally stale by invariant.

359 - status="running" files are checked via their sibling .pid file; archived

360 only if the PID is confirmed dead. No .pid file → leave alone (can't confirm).

361 """

362 running_dir = loops_dir / RUNNING_DIR

363 if not running_dir.exists():

364 return 0

365

366 terminal_statuses = {"completed", "failed", "interrupted", "timed_out"}

367 archived = 0

368

369 for state_file in running_dir.glob("*.state.json"):

370 try:

371 data = json.loads(state_file.read_text())

372 state = LoopState.from_dict(data)

373 except (json.JSONDecodeError, KeyError, OSError):

374 continue

375

376 is_stale = state.status in terminal_statuses

377

378 if not is_stale and state.status == "running":

379 stem = state_file.name.removesuffix(".state.json")

380 pid_file = running_dir / f"{stem}.pid"

381 if pid_file.exists():

382 try:

383 pid = int(pid_file.read_text().strip())

384 is_stale = not _process_alive(pid)

385 except (OSError, ValueError):

386 pass

387

388 if not is_stale:

389 continue

390

391 stem = state_file.name.removesuffix(".state.json")

392 instance_id = stem if stem != state.loop_name else None

393 persistence = StatePersistence(

394 loop_name=state.loop_name,

395 loops_dir=loops_dir,

396 instance_id=instance_id,

397 )

398 try:

399 persistence.clear_all()

400 (running_dir / f"{stem}.pid").unlink(missing_ok=True)

401 archived += 1

402 logger.debug("Archived stale run: %s (status=%s)", stem, state.status)

403 except OSError as e:

404 logger.warning("Failed to archive stale run %s: %s", stem, e)

405

406 if archived:

407 logger.info("Reconciliation sweep archived %d stale run(s) from .running/", archived)

408

409 return archived

410

411

412class PersistentExecutor:

413 """FSM Executor with state persistence and event streaming.

414

415 Wraps FSMExecutor to:

416 - Save state after each state transition

417 - Append events to JSONL file as they occur

418 - Support resuming from saved state

419 - Support graceful shutdown via signal handling

420 """

421

422 def __init__(

423 self,

424 fsm: FSMLoop,

425 persistence: StatePersistence | None = None,

426 loops_dir: Path | None = None,

427 instance_id: str | None = None,

428 pid: int | None = None,

429 **executor_kwargs: Any,

430 ) -> None:

431 """Initialize persistent executor.

432

433 Args:

434 fsm: FSM loop definition

435 persistence: Optional pre-configured persistence (for testing)

436 loops_dir: Base directory for loops (default: .loops)

437 instance_id: Optional unique instance identifier for file path scoping

438 pid: OS PID of the running process; stored in saved state for reconciliation

439 **executor_kwargs: Additional kwargs for FSMExecutor

440 """

441 from little_loops.fsm.handoff_handler import HandoffBehavior, HandoffHandler

442 from little_loops.fsm.signal_detector import SignalDetector

443

444 self.fsm = fsm

445 self.loops_dir = loops_dir

446 self._run_pid = pid

447 self.persistence = persistence or StatePersistence(

448 fsm.name, loops_dir or Path(".loops"), instance_id=instance_id

449 )

450 self.persistence.initialize()

451

452 # Create signal detector and handler based on FSM config

453 signal_detector = SignalDetector()

454 handoff_handler = HandoffHandler(HandoffBehavior(fsm.on_handoff))

455

456 # Create base executor with event callback that persists

457 self._executor = FSMExecutor(

458 fsm,

459 event_callback=self._handle_event,

460 signal_detector=signal_detector,

461 handoff_handler=handoff_handler,

462 loops_dir=self.loops_dir,

463 **executor_kwargs,

464 )

465 self._last_result: dict[str, Any] | None = None

466 self._continuation_prompt: str | None = None

467 self.event_bus = EventBus()

468

469 @property

470 def _on_event(self) -> EventCallback | None:

471 """Backward-compatible access to the first observer on the event bus."""

472 return self.event_bus._observers[0][0] if self.event_bus._observers else None

473

474 @_on_event.setter

475 def _on_event(self, callback: EventCallback | None) -> None:

476 """Backward-compatible setter: replaces all observers with this one."""

477 self.event_bus._observers.clear()

478 if callback is not None:

479 self.event_bus.register(callback)

480

481 def close_transports(self) -> None:

482 """Close all transports registered on the underlying EventBus."""

483 self.event_bus.close_transports()

484

485 def request_shutdown(self) -> None:

486 """Request graceful shutdown of the executor.

487

488 Delegates to the underlying FSMExecutor's request_shutdown method.

489 The loop will exit cleanly after the current state completes,

490 saving state as "interrupted" so it can be resumed later.

491 """

492 self._executor.request_shutdown()

493

494 def _handle_event(self, event: dict[str, Any]) -> None:

495 """Handle event: persist to file and save state.

496

497 Args:

498 event: Event dictionary from executor

499 """

500 self.persistence.append_event(event)

501

502 # Save state after state transitions

503 event_type = event.get("event")

504 if event_type in ("state_enter", "loop_complete"):

505 self._save_state()

506

507 # Track evaluation results for state persistence

508 if event_type == "evaluate":

509 self._last_result = {

510 "verdict": event.get("verdict"),

511 "details": {

512 k: v for k, v in event.items() if k not in ("event", "ts", "type", "verdict")

513 },

514 }

515

516 # Track handoff events for continuation prompt

517 if event_type == "handoff_detected":

518 self._continuation_prompt = event.get("continuation")

519

520 # Delegate to registered observers (e.g. progress display, extensions)

521 self.event_bus.emit(event)

522

523 def _save_state(self) -> None:

524 """Save current executor state to file."""

525 status = "running"

526 if self._executor.current_state:

527 state_config = self.fsm.states.get(self._executor.current_state)

528 if state_config and state_config.terminal:

529 status = "completed"

530

531 state = LoopState(

532 loop_name=self.fsm.name,

533 current_state=self._executor.current_state,

534 iteration=self._executor.iteration,

535 captured=self._executor.captured,

536 prev_result=self._executor.prev_result,

537 last_result=self._last_result,

538 started_at=self._executor.started_at,

539 updated_at="", # Will be set by save_state

540 status=status,

541 accumulated_ms=_now_ms()

542 - self._executor.start_time_ms

543 + self._executor.elapsed_offset_ms,

544 retry_counts=dict(self._executor._retry_counts),

545 rate_limit_retries={k: dict(v) for k, v in self._executor._rate_limit_retries.items()},

546 consecutive_rate_limit_exhaustions=(self._executor._consecutive_rate_limit_exhaustions),

547 edge_revisit_counts=dict(self._executor._edge_revisit_counts),

548 pid=self._run_pid,

549 )

550 self.persistence.save_state(state)

551

552 def run(self, clear_previous: bool = True) -> ExecutionResult:

553 """Run the FSM with persistence.

554

555 Args:

556 clear_previous: If True, clear previous state/events before running

557

558 Returns:

559 ExecutionResult from the execution

560 """

561 if clear_previous:

562 self.persistence.clear_all()

563

564 result = self._executor.run()

565

566 # Update final state

567 final_status = "completed" if result.terminated_by == "terminal" else "failed"

568 if result.terminated_by in ("max_iterations", "signal"):

569 final_status = "interrupted"

570 if result.terminated_by == "handoff":

571 final_status = "awaiting_continuation"

572 if result.terminated_by == "timeout":

573 final_status = "timed_out"

574 if result.terminated_by == "cycle_detected":

575 final_status = "failed"

576

577 final_state = LoopState(

578 loop_name=self.fsm.name,

579 current_state=result.final_state,

580 iteration=result.iterations,

581 captured=result.captured,

582 prev_result=self._executor.prev_result,

583 last_result=self._last_result,

584 started_at=self._executor.started_at,

585 updated_at="",

586 status=final_status,

587 continuation_prompt=self._continuation_prompt,

588 accumulated_ms=result.duration_ms,

589 )

590 self.persistence.save_state(final_state)

591 self.persistence.archive_run()

592

593 return result

594

595 def resume(self) -> ExecutionResult | None:

596 """Resume from saved state, or None if no resumable state.

597

598 Resumable states are: "running" and "awaiting_continuation".

599

600 Returns:

601 ExecutionResult if resumed and completed, None if no resumable state

602 """

603 state = self.persistence.load_state()

604 if state is None:

605 return None

606

607 if state.status not in ("running", "awaiting_continuation"):

608 return None # Already completed/failed

609

610 # Restore executor state

611 self._executor.current_state = state.current_state

612 self._executor.iteration = state.iteration

613 self._executor.captured = state.captured

614 self._executor.prev_result = state.prev_result

615 self._executor.started_at = state.started_at

616 self._last_result = state.last_result

617 self._executor._retry_counts = dict(state.retry_counts)

618 self._executor._rate_limit_retries = {

619 k: dict(v) for k, v in state.rate_limit_retries.items()

620 }

621 self._executor._consecutive_rate_limit_exhaustions = (

622 state.consecutive_rate_limit_exhaustions

623 )

624 self._executor._edge_revisit_counts = dict(state.edge_revisit_counts)

625

626 # Restore accumulated elapsed time so duration_ms and ${loop.elapsed_ms} reflect

627 # the full loop lifetime (all segments), not just the resumed segment.

628 # FSMExecutor.run() will reset start_time_ms to _now_ms(), so we use elapsed_offset_ms

629 # to carry forward the time already spent before this resume.

630 self._executor.elapsed_offset_ms = state.accumulated_ms

631

632 # Clear any pending signals from previous run

633 self._executor._pending_handoff = None

634 self._executor._pending_error = None

635

636 # Emit resume event with continuation context if available

637 resume_event: dict[str, Any] = {

638 "event": "loop_resume",

639 "ts": _iso_now(),

640 "loop": self.fsm.name,

641 "from_state": state.current_state,

642 "iteration": state.iteration,

643 }

644 if state.status == "awaiting_continuation" and state.continuation_prompt:

645 resume_event["from_handoff"] = True

646 resume_event["continuation_prompt"] = state.continuation_prompt

647 self.persistence.append_event(resume_event)

648 self.event_bus.emit(resume_event)

649

650 # Continue execution (don't clear previous events)

651 return self.run(clear_previous=False)

652

653

654def _find_instances(loop_name: str, running_dir: Path) -> list[tuple[str | None, LoopState]]:

655 """Discover all state-file instances for *loop_name* in *running_dir*.

656

657 Globs ``{loop_name}-*.state.json`` for instance-scoped files and

658 ``{loop_name}.state.json`` for legacy bare-name files.

659

660 Returns:

661 List of ``(instance_id, LoopState)`` tuples sorted by file name.

662 *instance_id* is the file stem (e.g. ``"autodev-20260503T122306"``)

663 for instance-scoped files, or ``None`` for legacy bare-name files.

664 """

665 if not running_dir.exists():

666 return []

667

668 instances: list[tuple[str | None, LoopState]] = []

669

670 # Instance-scoped files: {loop_name}-YYYYMMDDTHHMMSS.state.json

671 # Use Path(stem).stem to strip both suffixes (.state.json → base stem).

672 for state_file in sorted(running_dir.glob(f"{loop_name}-*.state.json")):

673 base_stem = Path(state_file.stem).stem # e.g. "autodev-20260503T122306"

674 if not _INSTANCE_SUFFIX.search(base_stem):

675 continue # skip files like "loop-name-extra" that don't match timestamp pattern

676 try:

677 data = json.loads(state_file.read_text())

678 instances.append((base_stem, LoopState.from_dict(data)))

679 except (json.JSONDecodeError, KeyError):

680 continue

681

682 # Legacy bare-name file: {loop_name}.state.json

683 legacy_file = running_dir / f"{loop_name}.state.json"

684 if legacy_file.exists():

685 try:

686 data = json.loads(legacy_file.read_text())

687 instances.append((None, LoopState.from_dict(data)))

688 except (json.JSONDecodeError, KeyError):

689 pass

690

691 return instances

692

693

694def list_running_loops(loops_dir: Path | None = None) -> list[LoopState]:

695 """List all loops with saved state.

696

697 Args:

698 loops_dir: Base directory for loops (default: .loops)

699

700 Returns:

701 List of LoopState objects for all loops with state files

702 """

703 base_dir = loops_dir or Path(".loops")

704 running_dir = base_dir / RUNNING_DIR

705

706 if not running_dir.exists():

707 return []

708

709 states: list[LoopState] = []

710 for state_file in running_dir.glob("*.state.json"):

711 try:

712 data = json.loads(state_file.read_text())

713 states.append(LoopState.from_dict(data))

714 except (json.JSONDecodeError, KeyError):

715 continue # Skip malformed files

716

717 # Include loops that have a PID file but no state file yet (still starting up).

718 # Strip instance-ID timestamp suffix (e.g. "autodev-20240115T103000" → "autodev")

719 # before the known_names check to avoid spurious "starting" entries for loops

720 # that already have a state file under their logical name.

721 known_names = {s.loop_name for s in states}

722 for pid_file in running_dir.glob("*.pid"):

723 logical_name = _INSTANCE_SUFFIX.sub("", pid_file.stem)

724 if logical_name in known_names:

725 continue # state file already covers this loop

726 try:

727 pid = int(pid_file.read_text().strip())

728 except (ValueError, OSError):

729 continue

730 if _process_alive(pid):

731 states.append(

732 LoopState(

733 loop_name=logical_name,

734 current_state="(initializing)",

735 iteration=0,

736 captured={},

737 prev_result=None,

738 last_result=None,

739 started_at="",

740 updated_at="",

741 status="starting",

742 )

743 )

744

745 return states

746

747

748def list_run_history(loop_name: str, loops_dir: Path | None = None) -> list[LoopState]:

749 """List archived runs for a loop, newest first.

750

751 Reads state files from .loops/.history/<run_id>-<loop_name>/state.json and

752 returns them sorted by started_at descending (most recent run first).

753

754 Also checks the legacy nested layout .loops/.history/<loop_name>/*/state.json

755 for backward compatibility with existing history folders.

756

757 Args:

758 loop_name: Name of the loop

759 loops_dir: Base directory for loops (default: .loops)

760

761 Returns:

762 List of LoopState objects for all archived runs, newest first.

763 Returns an empty list if no history exists.

764 """

765 base_dir = loops_dir or Path(".loops")

766 history_dir = base_dir / HISTORY_DIR

767

768 if not history_dir.exists():

769 return []

770

771 states: list[LoopState] = []

772

773 # Flat layout: <run_id>-<loop_name>/state.json

774 for state_file in history_dir.glob(f"*-{loop_name}/state.json"):

775 try:

776 data = json.loads(state_file.read_text())

777 states.append(LoopState.from_dict(data))

778 except (json.JSONDecodeError, KeyError):

779 continue

780

781 # Backward compat: legacy nested layout <loop_name>/<run_id>/state.json

782 old_loop_dir = history_dir / loop_name

783 if old_loop_dir.exists():

784 logger.warning(

785 "Found legacy nested history at %s; migrate to flat layout by moving "

786 "each run to .history/<run_id>-%s/",

787 old_loop_dir,

788 loop_name,

789 )

790 for state_file in old_loop_dir.glob("*/state.json"):

791 try:

792 data = json.loads(state_file.read_text())

793 states.append(LoopState.from_dict(data))

794 except (json.JSONDecodeError, KeyError):

795 continue

796

797 states.sort(key=lambda s: s.started_at, reverse=True)

798 return states

799

800

801def get_archived_events(

802 loop_name: str, run_id: str, loops_dir: Path | None = None

803) -> list[dict[str, Any]]:

804 """Read events for a specific archived run.

805

806 Args:

807 loop_name: Name of the loop

808 run_id: The run directory name (compact timestamp)

809 loops_dir: Base directory for loops (default: .loops)

810

811 Returns:

812 List of event dictionaries, empty if not found.

813 """

814 base_dir = loops_dir or Path(".loops")

815 run_folder = f"{run_id}-{loop_name}"

816 events_file = base_dir / HISTORY_DIR / run_folder / "events.jsonl"

817

818 if not events_file.exists():

819 return []

820

821 events: list[dict[str, Any]] = []

822 with open(events_file, encoding="utf-8") as f:

823 for line in f:

824 line = line.strip()

825 if line:

826 try:

827 events.append(json.loads(line))

828 except json.JSONDecodeError:

829 continue

830 return events

831

832

833def get_loop_history(loop_name: str, loops_dir: Path | None = None) -> list[dict[str, Any]]:

834 """Get event history for a loop.

835

836 Args:

837 loop_name: Name of the loop

838 loops_dir: Base directory for loops (default: .loops)

839

840 Returns:

841 List of event dictionaries

842 """

843 persistence = StatePersistence(loop_name, loops_dir)

844 return persistence.read_events()

Coverage for little_loops / fsm / persistence.py: 21%

362 statements