Coverage for little_loops / fsm / concurrency.py: 27%
133 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -0500
1"""Scope-based concurrency control for FSM loops.
3Prevents concurrent loops from conflicting when operating on
4the same files or directories through file-based locking.
6Public exports:
7 ScopeLock: Dataclass representing a scope lock
8 LockManager: Manager for acquiring/releasing scope locks
9 resolve_scope: Resolve ${context.<var>} templates in scope paths
10"""
12from __future__ import annotations
14import errno
15import fcntl
16import json
17import os
18import re
19import time
20from dataclasses import dataclass
21from datetime import UTC, datetime
22from pathlib import Path
23from typing import Any
25RUNNING_DIR = ".running"
27# Match ${context.<var>} templates in scope paths
28_CONTEXT_VAR_RE = re.compile(r"\$\{context\.([^}]+)\}")
31def resolve_scope(scope: list[str], context: dict[str, Any]) -> list[str]:
32 """Resolve ${context.<var>} templates in scope paths.
34 Each template expression referencing a context variable is replaced
35 with the variable's value. Templates referencing unknown variables
36 are left as-is (the literal string becomes the path, which
37 Path.resolve() will turn into an absolute path).
39 Static paths (no templates) pass through unchanged.
40 """
41 resolved: list[str] = []
42 for path in scope:
44 def _replace(m: re.Match[str]) -> str:
45 var = m.group(1)
46 return str(context[var]) if var in context else m.group(0)
48 resolved.append(_CONTEXT_VAR_RE.sub(_replace, path))
49 return resolved
52def _process_alive(pid: int) -> bool:
53 """Check if a process is still running.
55 Returns True if alive (or alive but unreadable due to EPERM),
56 False only if process does not exist (ESRCH).
57 """
58 try:
59 os.kill(pid, 0)
60 return True
61 except OSError as e:
62 if e.errno == errno.ESRCH:
63 return False # No such process
64 return True # EPERM or other: process exists, no permission
67def _iso_now() -> str:
68 """Return current time as ISO8601 string."""
69 return datetime.now(UTC).isoformat()
72@dataclass
73class ScopeLock:
74 """Represents a lock on a set of paths for a running loop.
76 Attributes:
77 loop_name: Name of the loop holding the lock
78 scope: List of paths this loop operates on
79 pid: Process ID of the lock holder
80 started_at: ISO timestamp when lock was acquired
81 """
83 loop_name: str
84 scope: list[str]
85 pid: int
86 started_at: str
88 def to_dict(self) -> dict[str, Any]:
89 """Convert to dictionary for JSON serialization."""
90 return {
91 "loop_name": self.loop_name,
92 "scope": self.scope,
93 "pid": self.pid,
94 "started_at": self.started_at,
95 }
97 @classmethod
98 def from_dict(cls, data: dict[str, Any]) -> ScopeLock:
99 """Create from dictionary (JSON deserialization)."""
100 return cls(
101 loop_name=str(data["loop_name"]),
102 scope=list(data["scope"]) if isinstance(data["scope"], list) else [str(data["scope"])],
103 pid=int(data["pid"]),
104 started_at=str(data["started_at"]),
105 )
108class LockManager:
109 """Manage scope-based locks for concurrent loop execution.
111 Lock files are stored in .loops/.running/<instance_id>.lock
112 and contain JSON with ScopeLock data.
113 """
115 def __init__(self, loops_dir: Path | None = None) -> None:
116 """Initialize the lock manager.
118 Args:
119 loops_dir: Base directory for loops (default: .loops)
120 """
121 self.loops_dir = loops_dir or Path(".loops")
122 self.running_dir = self.loops_dir / RUNNING_DIR
124 def acquire(self, loop_name: str, scope: list[str], instance_id: str | None = None) -> bool:
125 """Attempt to acquire lock for the given scope.
127 Args:
128 loop_name: Name of the loop to acquire lock for
129 scope: List of paths the loop operates on
130 instance_id: Optional unique instance identifier; falls back to loop_name when None
132 Returns:
133 True if lock acquired, False if conflict exists
134 """
135 # Normalize scope - empty means whole project
136 if not scope:
137 scope = ["."]
138 scope = [self._normalize_path(p) for p in scope]
140 # Ensure running directory exists before opening sentinel lock
141 self.running_dir.mkdir(parents=True, exist_ok=True)
143 # Serialize the check-and-create sequence across processes using a
144 # sentinel file. This eliminates the TOCTOU window between
145 # find_conflict() (read) and lock-file creation (write).
146 # .acquire.lock is a dotfile so Path.glob("*.lock") will not match it
147 # and stale-lock cleanup in find_conflict/list_locks ignores it.
148 dir_lock_path = self.running_dir / ".acquire.lock"
149 with open(dir_lock_path, "w") as dir_lock:
150 fcntl.flock(dir_lock, fcntl.LOCK_EX)
152 # Check for conflicts (now atomic with write below)
153 conflict = self.find_conflict(scope)
154 if conflict:
155 return False
157 # Create lock file
158 lock_file = self.running_dir / f"{instance_id or loop_name}.lock"
159 lock_file.parent.mkdir(parents=True, exist_ok=True)
160 lock = ScopeLock(
161 loop_name=loop_name,
162 scope=scope,
163 pid=os.getpid(),
164 started_at=_iso_now(),
165 )
166 with open(lock_file, "w") as f:
167 json.dump(lock.to_dict(), f)
169 return True
171 def release(self, loop_name: str, instance_id: str | None = None) -> None:
172 """Release lock for a loop.
174 Args:
175 loop_name: Name of the loop to release lock for
176 instance_id: Optional unique instance identifier; falls back to loop_name when None
177 """
178 lock_file = self.running_dir / f"{instance_id or loop_name}.lock"
179 lock_file.unlink(missing_ok=True)
181 def find_conflict(self, scope: list[str]) -> ScopeLock | None:
182 """Find any running loop with overlapping scope.
184 Also cleans up stale locks from dead processes.
186 Args:
187 scope: Scope to check for conflicts
189 Returns:
190 ScopeLock of conflicting loop, or None if no conflict
191 """
192 if not self.running_dir.exists():
193 return None
195 # Normalize once before the comparison loop to avoid O(n*m) stat calls
196 normalized_scope = [self._normalize_path(p) for p in scope]
198 for lock_file in self.running_dir.glob("*.lock"):
199 try:
200 with open(lock_file) as f:
201 data = json.load(f)
202 lock = ScopeLock.from_dict(data)
204 # Check if process is still alive
205 if not self._process_alive(lock.pid):
206 # Stale lock, remove it
207 lock_file.unlink(missing_ok=True)
208 continue
210 # Normalize lock scope (lock files from acquire() are already
211 # absolute, but normalize defensively in case of legacy files)
212 lock_scope = [self._normalize_path(p) for p in lock.scope]
213 if self._scopes_overlap(normalized_scope, lock_scope):
214 return lock
216 except (json.JSONDecodeError, KeyError, FileNotFoundError):
217 # Malformed or deleted lock file, skip
218 continue
220 return None
222 def list_locks(self) -> list[ScopeLock]:
223 """List all active locks.
225 Cleans up stale locks as a side effect.
227 Returns:
228 List of active ScopeLock objects
229 """
230 locks: list[ScopeLock] = []
231 if not self.running_dir.exists():
232 return locks
234 for lock_file in self.running_dir.glob("*.lock"):
235 try:
236 with open(lock_file) as f:
237 data = json.load(f)
238 lock = ScopeLock.from_dict(data)
240 if self._process_alive(lock.pid):
241 locks.append(lock)
242 else:
243 # Stale lock, remove it
244 lock_file.unlink(missing_ok=True)
245 except (json.JSONDecodeError, KeyError, FileNotFoundError):
246 continue
248 return locks
250 def wait_for_scope(self, scope: list[str], timeout: int = 300) -> bool:
251 """Wait until scope is available.
253 Args:
254 scope: Scope to wait for
255 timeout: Maximum time to wait in seconds
257 Returns:
258 True if scope became available, False if timeout
259 """
260 start = time.time()
261 while time.time() - start < timeout:
262 conflict = self.find_conflict(scope)
263 if conflict is None:
264 return True
265 time.sleep(1)
267 return False
269 def _scopes_overlap(self, scope1: list[str], scope2: list[str]) -> bool:
270 """Check if two scopes have any overlapping paths."""
271 for p1 in scope1:
272 for p2 in scope2:
273 if self._paths_overlap(p1, p2):
274 return True
275 return False
277 def _paths_overlap(self, path1: str, path2: str) -> bool:
278 """Check if two paths overlap (same, or one contains the other).
280 Assumes paths are already normalized (pre-resolved absolute strings).
281 """
282 p1 = Path(path1)
283 p2 = Path(path2)
285 # Same path
286 if p1 == p2:
287 return True
289 # One is parent of other
290 try:
291 p1.relative_to(p2)
292 return True
293 except ValueError:
294 pass
296 try:
297 p2.relative_to(p1)
298 return True
299 except ValueError:
300 pass
302 return False
304 def _normalize_path(self, path: str) -> str:
305 """Normalize path for consistent comparison."""
306 return str(Path(path).resolve())
308 def _process_alive(self, pid: int) -> bool:
309 """Check if process is still running."""
310 return _process_alive(pid)