Coverage for little_loops / fsm / concurrency.py: 26%

1"""Scope-based concurrency control for FSM loops. 

2 

3Prevents concurrent loops from conflicting when operating on 

4the same files or directories through file-based locking. 

5 

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""" 

11 

12from __future__ import annotations 

13 

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 

26 

27logger = logging.getLogger(__name__) 

28 

29RUNNING_DIR = ".running" 

30 

31# Match ${context.<var>} templates in scope paths 

32_CONTEXT_VAR_RE = re.compile(r"\$\{context\.([^}]+)\}") 

33 

34 

35def resolve_scope(scope: list[str], context: dict[str, Any]) -> list[str]: 

36 """Resolve ${context.<var>} templates in scope paths. 

37 

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). 

42 

43 Static paths (no templates) pass through unchanged. 

44 """ 

45 resolved: list[str] = [] 

46 for path in scope: 

47 

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) 

51 

52 resolved.append(_CONTEXT_VAR_RE.sub(_replace, path)) 

53 return resolved 

54 

55 

56def _process_alive(pid: int) -> bool: 

57 """Check if a process is still running. 

58 

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 

69 

70 

71def _iso_now() -> str: 

72 """Return current time as ISO8601 string.""" 

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

74 

75 

76@dataclass 

77class ScopeLock: 

78 """Represents a lock on a set of paths for a running loop. 

79 

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 """ 

86 

87 loop_name: str 

88 scope: list[str] 

89 pid: int 

90 started_at: str 

91 

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 } 

100 

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 ) 

110 

111 

112class LockManager: 

113 """Manage scope-based locks for concurrent loop execution. 

114 

115 Lock files are stored in .loops/.running/<instance_id>.lock 

116 and contain JSON with ScopeLock data. 

117 """ 

118 

119 _cached_ancestry: set[int] 

120 

121 def __init__(self, loops_dir: Path | None = None) -> None: 

122 """Initialize the lock manager. 

123 

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 

129 

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. 

132 

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 

137 

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] 

145 

146 # Ensure running directory exists before opening sentinel lock 

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

148 

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) 

157 

158 # Check for conflicts (now atomic with write below) 

159 conflict = self.find_conflict(scope) 

160 if conflict: 

161 return False 

162 

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) 

174 

175 return True 

176 

177 def release(self, loop_name: str, instance_id: str | None = None) -> None: 

178 """Release lock for a loop. 

179 

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) 

186 

187 def find_conflict(self, scope: list[str]) -> ScopeLock | None: 

188 """Find any running loop with overlapping scope. 

189 

190 Also cleans up stale locks from dead processes. 

191 

192 Args: 

193 scope: Scope to check for conflicts 

194 

195 Returns: 

196 ScopeLock of conflicting loop, or None if no conflict 

197 """ 

198 if not self.running_dir.exists(): 

199 return None 

200 

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] 

203 

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) 

209 

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 

215 

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 

231 

232 except (json.JSONDecodeError, KeyError, FileNotFoundError): 

233 # Malformed or deleted lock file, skip 

234 continue 

235 

236 return None 

237 

238 def list_locks(self) -> list[ScopeLock]: 

239 """List all active locks. 

240 

241 Cleans up stale locks as a side effect. 

242 

243 Returns: 

244 List of active ScopeLock objects 

245 """ 

246 locks: list[ScopeLock] = [] 

247 if not self.running_dir.exists(): 

248 return locks 

249 

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) 

255 

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 

263 

264 return locks 

265 

266 def wait_for_scope(self, scope: list[str], timeout: int = 300) -> bool: 

267 """Wait until scope is available. 

268 

269 Args: 

270 scope: Scope to wait for 

271 timeout: Maximum time to wait in seconds 

272 

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) 

282 

283 return False 

284 

285 def _get_ancestry(self) -> set[int]: 

286 """Walk the process tree to collect ancestor PIDs. 

287 

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). 

292 

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 

299 

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 ) 

319 

320 self._cached_ancestry = ancestors 

321 return ancestors 

322 

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 

330 

331 def _paths_overlap(self, path1: str, path2: str) -> bool: 

332 """Check if two paths overlap (same, or one contains the other). 

333 

334 Assumes paths are already normalized (pre-resolved absolute strings). 

335 """ 

336 p1 = Path(path1) 

337 p2 = Path(path2) 

338 

339 # Same path 

340 if p1 == p2: 

341 return True 

342 

343 # One is parent of other 

344 try: 

345 p1.relative_to(p2) 

346 return True 

347 except ValueError: 

348 pass 

349 

350 try: 

351 p2.relative_to(p1) 

352 return True 

353 except ValueError: 

354 pass 

355 

356 return False 

357 

358 def _normalize_path(self, path: str) -> str: 

359 """Normalize path for consistent comparison.""" 

360 return str(Path(path).resolve()) 

361 

362 def _process_alive(self, pid: int) -> bool: 

363 """Check if process is still running.""" 

364 return _process_alive(pid)