Coverage for little_loops / fsm / concurrency.py: 48%
157 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:12 -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 logging
18import os
19import re
20import subprocess
21import time
22from dataclasses import dataclass
23from datetime import UTC, datetime
24from pathlib import Path
25from typing import Any
27logger = logging.getLogger(__name__)
29RUNNING_DIR = ".running"
31# Match ${context.<var>} templates in scope paths
32_CONTEXT_VAR_RE = re.compile(r"\$\{context\.([^}]+)\}")
35def resolve_scope(scope: list[str], context: dict[str, Any]) -> list[str]:
36 """Resolve ${context.<var>} templates in scope paths.
38 Each template expression referencing a context variable is replaced
39 with the variable's value. Templates referencing unknown variables
40 are left as-is (the literal string becomes the path, which
41 Path.resolve() will turn into an absolute path).
43 Static paths (no templates) pass through unchanged.
44 """
45 resolved: list[str] = []
46 for path in scope:
48 def _replace(m: re.Match[str]) -> str:
49 var = m.group(1)
50 return str(context[var]) if var in context else m.group(0)
52 resolved.append(_CONTEXT_VAR_RE.sub(_replace, path))
53 return resolved
56def _process_alive(pid: int) -> bool:
57 """Check if a process is still running.
59 Returns True if alive (or alive but unreadable due to EPERM),
60 False only if process does not exist (ESRCH).
61 """
62 try:
63 os.kill(pid, 0)
64 return True
65 except OSError as e:
66 if e.errno == errno.ESRCH:
67 return False # No such process
68 return True # EPERM or other: process exists, no permission
71def _iso_now() -> str:
72 """Return current time as ISO8601 string."""
73 return datetime.now(UTC).isoformat()
76@dataclass
77class ScopeLock:
78 """Represents a lock on a set of paths for a running loop.
80 Attributes:
81 loop_name: Name of the loop holding the lock
82 scope: List of paths this loop operates on
83 pid: Process ID of the lock holder
84 started_at: ISO timestamp when lock was acquired
85 """
87 loop_name: str
88 scope: list[str]
89 pid: int
90 started_at: str
92 def to_dict(self) -> dict[str, Any]:
93 """Convert to dictionary for JSON serialization."""
94 return {
95 "loop_name": self.loop_name,
96 "scope": self.scope,
97 "pid": self.pid,
98 "started_at": self.started_at,
99 }
101 @classmethod
102 def from_dict(cls, data: dict[str, Any]) -> ScopeLock:
103 """Create from dictionary (JSON deserialization)."""
104 return cls(
105 loop_name=str(data["loop_name"]),
106 scope=list(data["scope"]) if isinstance(data["scope"], list) else [str(data["scope"])],
107 pid=int(data["pid"]),
108 started_at=str(data["started_at"]),
109 )
112class LockManager:
113 """Manage scope-based locks for concurrent loop execution.
115 Lock files are stored in .loops/.running/<instance_id>.lock
116 and contain JSON with ScopeLock data.
117 """
119 _cached_ancestry: set[int]
121 def __init__(self, loops_dir: Path | None = None) -> None:
122 """Initialize the lock manager.
124 Args:
125 loops_dir: Base directory for loops (default: .loops)
126 """
127 self.loops_dir = loops_dir or Path(".loops")
128 self.running_dir = self.loops_dir / RUNNING_DIR
130 def acquire(self, loop_name: str, scope: list[str], instance_id: str | None = None) -> bool:
131 """Attempt to acquire lock for the given scope.
133 Args:
134 loop_name: Name of the loop to acquire lock for
135 scope: List of paths the loop operates on
136 instance_id: Optional unique instance identifier; falls back to loop_name when None
138 Returns:
139 True if lock acquired, False if conflict exists
140 """
141 # Normalize scope - empty means whole project
142 if not scope:
143 scope = ["."]
144 scope = [self._normalize_path(p) for p in scope]
146 # Ensure running directory exists before opening sentinel lock
147 self.running_dir.mkdir(parents=True, exist_ok=True)
149 # Serialize the check-and-create sequence across processes using a
150 # sentinel file. This eliminates the TOCTOU window between
151 # find_conflict() (read) and lock-file creation (write).
152 # .acquire.lock is a dotfile so Path.glob("*.lock") will not match it
153 # and stale-lock cleanup in find_conflict/list_locks ignores it.
154 dir_lock_path = self.running_dir / ".acquire.lock"
155 with open(dir_lock_path, "w") as dir_lock:
156 fcntl.flock(dir_lock, fcntl.LOCK_EX)
158 # Check for conflicts (now atomic with write below)
159 conflict = self.find_conflict(scope)
160 if conflict:
161 return False
163 # Create lock file
164 lock_file = self.running_dir / f"{instance_id or loop_name}.lock"
165 lock_file.parent.mkdir(parents=True, exist_ok=True)
166 lock = ScopeLock(
167 loop_name=loop_name,
168 scope=scope,
169 pid=os.getpid(),
170 started_at=_iso_now(),
171 )
172 with open(lock_file, "w") as f:
173 json.dump(lock.to_dict(), f)
175 return True
177 def release(self, loop_name: str, instance_id: str | None = None) -> None:
178 """Release lock for a loop.
180 Args:
181 loop_name: Name of the loop to release lock for
182 instance_id: Optional unique instance identifier; falls back to loop_name when None
183 """
184 lock_file = self.running_dir / f"{instance_id or loop_name}.lock"
185 lock_file.unlink(missing_ok=True)
187 def find_conflict(self, scope: list[str]) -> ScopeLock | None:
188 """Find any running loop with overlapping scope.
190 Also cleans up stale locks from dead processes.
192 Args:
193 scope: Scope to check for conflicts
195 Returns:
196 ScopeLock of conflicting loop, or None if no conflict
197 """
198 if not self.running_dir.exists():
199 return None
201 # Normalize once before the comparison loop to avoid O(n*m) stat calls
202 normalized_scope = [self._normalize_path(p) for p in scope]
204 for lock_file in self.running_dir.glob("*.lock"):
205 try:
206 with open(lock_file) as f:
207 data = json.load(f)
208 lock = ScopeLock.from_dict(data)
210 # Check if process is still alive
211 if not self._process_alive(lock.pid):
212 # Stale lock, remove it
213 lock_file.unlink(missing_ok=True)
214 continue
216 # Normalize lock scope (lock files from acquire() are already
217 # absolute, but normalize defensively in case of legacy files)
218 lock_scope = [self._normalize_path(p) for p in lock.scope]
219 if self._scopes_overlap(normalized_scope, lock_scope):
220 # If the lock holder is an ancestor of this process, it's a
221 # self-reference: a parent loop spawned this child via shell.
222 # Treat as non-conflict so nested ll-loop invocations work.
223 if lock.pid in self._get_ancestry():
224 logger.debug(
225 "Ignoring ancestor lock: pid=%d loop=%s",
226 lock.pid,
227 lock.loop_name,
228 )
229 continue
230 return lock
232 except (json.JSONDecodeError, KeyError, FileNotFoundError):
233 # Malformed or deleted lock file, skip
234 continue
236 return None
238 def list_locks(self) -> list[ScopeLock]:
239 """List all active locks.
241 Cleans up stale locks as a side effect.
243 Returns:
244 List of active ScopeLock objects
245 """
246 locks: list[ScopeLock] = []
247 if not self.running_dir.exists():
248 return locks
250 for lock_file in self.running_dir.glob("*.lock"):
251 try:
252 with open(lock_file) as f:
253 data = json.load(f)
254 lock = ScopeLock.from_dict(data)
256 if self._process_alive(lock.pid):
257 locks.append(lock)
258 else:
259 # Stale lock, remove it
260 lock_file.unlink(missing_ok=True)
261 except (json.JSONDecodeError, KeyError, FileNotFoundError):
262 continue
264 return locks
266 def wait_for_scope(self, scope: list[str], timeout: int = 300) -> bool:
267 """Wait until scope is available.
269 Args:
270 scope: Scope to wait for
271 timeout: Maximum time to wait in seconds
273 Returns:
274 True if scope became available, False if timeout
275 """
276 start = time.time()
277 while time.time() - start < timeout:
278 conflict = self.find_conflict(scope)
279 if conflict is None:
280 return True
281 time.sleep(1)
283 return False
285 def _get_ancestry(self) -> set[int]:
286 """Walk the process tree to collect ancestor PIDs.
288 Uses ps(1) to walk up from the current process to PID 1.
289 Returns the set of ancestor PIDs (empty on any error — the
290 ancestry check degrades gracefully so unrelated processes are
291 never incorrectly treated as self).
293 The ancestry set is cached on the instance for the lifetime of
294 a single acquire()/release() cycle — it won't change between
295 find_conflict and lock-file creation.
296 """
297 if hasattr(self, "_cached_ancestry"):
298 return self._cached_ancestry
300 ancestors: set[int] = set()
301 try:
302 current = os.getpid()
303 while current > 1:
304 result = subprocess.check_output(
305 ["ps", "-o", "ppid=", "-p", str(current)],
306 text=True,
307 stderr=subprocess.DEVNULL,
308 ).strip()
309 ppid = int(result)
310 if ppid <= 1 or ppid in ancestors:
311 break
312 ancestors.add(ppid)
313 current = ppid
314 except (subprocess.CalledProcessError, ValueError, FileNotFoundError, OSError):
315 logger.debug(
316 "Could not walk process ancestry; falling back to no-ancestor-check",
317 exc_info=True,
318 )
320 self._cached_ancestry = ancestors
321 return ancestors
323 def _scopes_overlap(self, scope1: list[str], scope2: list[str]) -> bool:
324 """Check if two scopes have any overlapping paths."""
325 for p1 in scope1:
326 for p2 in scope2:
327 if self._paths_overlap(p1, p2):
328 return True
329 return False
331 def _paths_overlap(self, path1: str, path2: str) -> bool:
332 """Check if two paths overlap (same, or one contains the other).
334 Assumes paths are already normalized (pre-resolved absolute strings).
335 """
336 p1 = Path(path1)
337 p2 = Path(path2)
339 # Same path
340 if p1 == p2:
341 return True
343 # One is parent of other
344 try:
345 p1.relative_to(p2)
346 return True
347 except ValueError:
348 pass
350 try:
351 p2.relative_to(p1)
352 return True
353 except ValueError:
354 pass
356 return False
358 def _normalize_path(self, path: str) -> str:
359 """Normalize path for consistent comparison."""
360 return str(Path(path).resolve())
362 def _process_alive(self, pid: int) -> bool:
363 """Check if process is still running."""
364 return _process_alive(pid)