coordination.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Multi-agent coordination layer for the Muse VCS. |
| 2 | |
| 3 | Coordination data lives under ``.muse/coordination/``. It is purely advisory — |
| 4 | the VCS engine never reads it for correctness decisions. Its purpose is to |
| 5 | enable agents working in parallel to announce their intentions, detect likely |
| 6 | conflicts *before* they happen, and plan merges without writing to the repo. |
| 7 | |
| 8 | Layout:: |
| 9 | |
| 10 | .muse/coordination/ |
| 11 | reservations/<uuid>.json write-once advisory symbol lease |
| 12 | intents/<uuid>.json write-once operation declaration |
| 13 | releases/<reservation-id>.json write-once tombstone (agent done / cancelled) |
| 14 | heartbeats/<reservation-id>.json mutable keep-alive, atomically updated |
| 15 | |
| 16 | Reservation schema:: |
| 17 | |
| 18 | { |
| 19 | "schema_version": "<muse package version>", |
| 20 | "reservation_id": "<uuid>", |
| 21 | "run_id": "<agent-supplied ID>", |
| 22 | "branch": "<branch name>", |
| 23 | "addresses": ["src/billing.py::compute_total", ...], |
| 24 | "created_at": "2026-03-18T12:00:00+00:00", |
| 25 | "expires_at": "2026-03-18T13:00:00+00:00", |
| 26 | "operation": null | "rename" | "move" | "extract" | "modify" | "delete" |
| 27 | } |
| 28 | |
| 29 | Release schema:: |
| 30 | |
| 31 | { |
| 32 | "schema_version": "<muse package version>", |
| 33 | "reservation_id": "<uuid>", |
| 34 | "run_id": "<agent-supplied ID>", |
| 35 | "released_at": "2026-03-18T12:05:00+00:00", |
| 36 | "reason": "completed" | "cancelled" | "superseded" |
| 37 | } |
| 38 | |
| 39 | Heartbeat schema:: |
| 40 | |
| 41 | { |
| 42 | "schema_version": "<muse package version>", |
| 43 | "reservation_id": "<uuid>", |
| 44 | "run_id": "<agent-supplied ID>", |
| 45 | "last_beat_at": "2026-03-18T12:30:00+00:00", |
| 46 | "extended_expires_at": "2026-03-18T13:30:00+00:00" |
| 47 | } |
| 48 | |
| 49 | Intent schema:: |
| 50 | |
| 51 | { |
| 52 | "schema_version": "<muse package version>", |
| 53 | "intent_id": "<uuid>", |
| 54 | "reservation_id": "<uuid>", |
| 55 | "run_id": "<agent-supplied ID>", |
| 56 | "branch": "<branch name>", |
| 57 | "addresses": ["src/billing.py::compute_total"], |
| 58 | "operation": "rename", |
| 59 | "created_at": "2026-03-18T12:00:00+00:00", |
| 60 | "detail": "rename to compute_invoice_total" |
| 61 | } |
| 62 | |
| 63 | Lifecycle |
| 64 | --------- |
| 65 | Reservations are write-once (immutable audit trail). Releases and intents |
| 66 | are also write-once. Heartbeats are the *only* mutable state — each call to |
| 67 | :func:`create_heartbeat` atomically rewrites the single heartbeat file for |
| 68 | that reservation, extending its effective TTL. |
| 69 | |
| 70 | A reservation is *active* when **all** of the following hold: |
| 71 | |
| 72 | 1. No release tombstone exists for its ID. |
| 73 | 2. The current time is before ``max(reservation.expires_at, |
| 74 | heartbeat.extended_expires_at)`` (heartbeat extends the TTL when present). |
| 75 | |
| 76 | :func:`active_reservations` enforces this check. :func:`filter_reservations` |
| 77 | accepts optional ``released_ids`` and ``heartbeat_expires`` kwargs so callers |
| 78 | that have already loaded those structures avoid redundant I/O. |
| 79 | |
| 80 | Filtering |
| 81 | --------- |
| 82 | :func:`filter_reservations` and :func:`filter_intents` accept keyword-only |
| 83 | arguments for in-memory filtering without additional I/O. Address matching |
| 84 | uses :mod:`fnmatch` glob syntax (e.g. ``"billing.py::*"``), which is safe |
| 85 | because it operates only on strings — no filesystem access occurs. |
| 86 | |
| 87 | Security |
| 88 | -------- |
| 89 | All functions that accept a ``reservation_id`` from external input call |
| 90 | :func:`_validate_reservation_id` before constructing any file path. A valid |
| 91 | UUID4 string cannot contain ``/`` or ``..``, preventing directory traversal |
| 92 | attacks. Path containment is verified with ``resolve().relative_to()`` as a |
| 93 | second line of defence. |
| 94 | """ |
| 95 | |
| 96 | from __future__ import annotations |
| 97 | |
| 98 | import dataclasses |
| 99 | import datetime |
| 100 | import fnmatch |
| 101 | import json |
| 102 | import logging |
| 103 | import pathlib |
| 104 | import re |
| 105 | import time as _time_mod |
| 106 | import uuid as _uuid_mod |
| 107 | |
| 108 | from muse._version import __version__ as _SCHEMA_VERSION |
| 109 | from muse.core.store import write_text_atomic |
| 110 | from typing import TypedDict |
| 111 | |
| 112 | logger = logging.getLogger(__name__) |
| 113 | |
| 114 | type _ReservationVal = str | int | list[str] | None |
| 115 | type _ReservationDict = dict[str, _ReservationVal] # serialized Reservation |
| 116 | type _IntentVal = str | int | list[str] |
| 117 | type _IntentDict = dict[str, _IntentVal] # serialized Intent |
| 118 | type HeartbeatMap = dict[str, datetime.datetime] # reservation_id → expiry |
| 119 | type HeartbeatRecordMap = dict[str, "Heartbeat"] # reservation_id → Heartbeat |
| 120 | type ReleaseMap = dict[str, "Release"] # reservation_id → Release |
| 121 | |
| 122 | |
| 123 | class _ReleaseDict(TypedDict): |
| 124 | """JSON-serialisable form of a :class:`Release`.""" |
| 125 | |
| 126 | schema_version: str |
| 127 | reservation_id: str |
| 128 | run_id: str |
| 129 | released_at: str |
| 130 | reason: str |
| 131 | |
| 132 | |
| 133 | class _HeartbeatDict(TypedDict): |
| 134 | """JSON-serialisable form of a :class:`Heartbeat`.""" |
| 135 | |
| 136 | schema_version: str |
| 137 | reservation_id: str |
| 138 | run_id: str |
| 139 | last_beat_at: str |
| 140 | agent_id: str |
| 141 | |
| 142 | |
| 143 | # --------------------------------------------------------------------------- |
| 144 | # Directory helpers |
| 145 | # --------------------------------------------------------------------------- |
| 146 | |
| 147 | |
| 148 | def _coord_dir(root: pathlib.Path) -> pathlib.Path: |
| 149 | return root / ".muse" / "coordination" |
| 150 | |
| 151 | |
| 152 | def _reservations_dir(root: pathlib.Path) -> pathlib.Path: |
| 153 | return _coord_dir(root) / "reservations" |
| 154 | |
| 155 | |
| 156 | def _intents_dir(root: pathlib.Path) -> pathlib.Path: |
| 157 | return _coord_dir(root) / "intents" |
| 158 | |
| 159 | |
| 160 | def _releases_dir(root: pathlib.Path) -> pathlib.Path: |
| 161 | return _coord_dir(root) / "releases" |
| 162 | |
| 163 | |
| 164 | def _heartbeats_dir(root: pathlib.Path) -> pathlib.Path: |
| 165 | return _coord_dir(root) / "heartbeats" |
| 166 | |
| 167 | |
| 168 | def _ensure_coord_dirs(root: pathlib.Path) -> None: |
| 169 | _reservations_dir(root).mkdir(parents=True, exist_ok=True) |
| 170 | _intents_dir(root).mkdir(parents=True, exist_ok=True) |
| 171 | _releases_dir(root).mkdir(parents=True, exist_ok=True) |
| 172 | _heartbeats_dir(root).mkdir(parents=True, exist_ok=True) |
| 173 | |
| 174 | |
| 175 | def _now_utc() -> datetime.datetime: |
| 176 | return datetime.datetime.now(datetime.timezone.utc) |
| 177 | |
| 178 | |
| 179 | def _parse_dt(s: str) -> datetime.datetime: |
| 180 | return datetime.datetime.fromisoformat(s) |
| 181 | |
| 182 | |
| 183 | # Matches any UUID variant (not just v4) — sufficient for path-safety validation. |
| 184 | _UUID_RE = re.compile( |
| 185 | r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', |
| 186 | re.IGNORECASE, |
| 187 | ) |
| 188 | |
| 189 | |
| 190 | def _validate_reservation_id(reservation_id: str) -> None: |
| 191 | """Raise ``ValueError`` if *reservation_id* is not a well-formed UUID. |
| 192 | |
| 193 | Security note |
| 194 | ------------- |
| 195 | ``reservation_id`` is used to construct file paths inside |
| 196 | ``.muse/coordination/``. A valid UUID cannot contain ``/`` or ``..``, |
| 197 | which prevents directory traversal attacks where a crafted ID such as |
| 198 | ``../../etc/passwd`` would escape the coordination directory. |
| 199 | """ |
| 200 | if not _UUID_RE.match(str(reservation_id)): |
| 201 | raise ValueError( |
| 202 | f"reservation_id must be a valid UUID: {reservation_id!r}" |
| 203 | ) |
| 204 | |
| 205 | |
| 206 | # --------------------------------------------------------------------------- |
| 207 | # Reservation |
| 208 | # --------------------------------------------------------------------------- |
| 209 | |
| 210 | |
| 211 | class Reservation: |
| 212 | """An advisory lock on a set of symbol addresses.""" |
| 213 | |
| 214 | def __init__( |
| 215 | self, |
| 216 | reservation_id: str, |
| 217 | run_id: str, |
| 218 | branch: str, |
| 219 | addresses: list[str], |
| 220 | created_at: datetime.datetime, |
| 221 | expires_at: datetime.datetime, |
| 222 | operation: str | None, |
| 223 | ) -> None: |
| 224 | self.reservation_id = reservation_id |
| 225 | self.run_id = run_id |
| 226 | self.branch = branch |
| 227 | self.addresses = addresses |
| 228 | self.created_at = created_at |
| 229 | self.expires_at = expires_at |
| 230 | self.operation = operation |
| 231 | |
| 232 | def is_active(self) -> bool: |
| 233 | """Return True if this reservation has not yet expired.""" |
| 234 | return _now_utc() < self.expires_at |
| 235 | |
| 236 | def ttl_remaining_seconds(self) -> float: |
| 237 | """Seconds remaining until expiry. |
| 238 | |
| 239 | Returns a positive float when active, zero or negative when expired. |
| 240 | Callers should use :meth:`is_active` for boolean checks; this method |
| 241 | is intended for display (e.g. "expires in 42m 17s"). |
| 242 | """ |
| 243 | return (self.expires_at - _now_utc()).total_seconds() |
| 244 | |
| 245 | def to_dict(self) -> _ReservationDict: |
| 246 | return { |
| 247 | "schema_version": _SCHEMA_VERSION, |
| 248 | "reservation_id": self.reservation_id, |
| 249 | "run_id": self.run_id, |
| 250 | "branch": self.branch, |
| 251 | "addresses": self.addresses, |
| 252 | "created_at": self.created_at.isoformat(), |
| 253 | "expires_at": self.expires_at.isoformat(), |
| 254 | "operation": self.operation, |
| 255 | } |
| 256 | |
| 257 | @classmethod |
| 258 | def from_dict(cls, d: _ReservationDict) -> "Reservation": |
| 259 | expires_at_raw = d.get("expires_at") |
| 260 | created_at_raw = d.get("created_at") |
| 261 | expires_at = _parse_dt(str(expires_at_raw)) if expires_at_raw else _now_utc() |
| 262 | created_at = _parse_dt(str(created_at_raw)) if created_at_raw else _now_utc() |
| 263 | addrs_raw = d.get("addresses", []) |
| 264 | addrs = list(addrs_raw) if isinstance(addrs_raw, list) else [] |
| 265 | op_raw = d.get("operation") |
| 266 | return cls( |
| 267 | reservation_id=str(d.get("reservation_id", "")), |
| 268 | run_id=str(d.get("run_id", "")), |
| 269 | branch=str(d.get("branch", "")), |
| 270 | addresses=addrs, |
| 271 | created_at=created_at, |
| 272 | expires_at=expires_at, |
| 273 | operation=str(op_raw) if op_raw is not None else None, |
| 274 | ) |
| 275 | |
| 276 | |
| 277 | def create_reservation( |
| 278 | root: pathlib.Path, |
| 279 | run_id: str, |
| 280 | branch: str, |
| 281 | addresses: list[str], |
| 282 | ttl_seconds: int = 3600, |
| 283 | operation: str | None = None, |
| 284 | ) -> Reservation: |
| 285 | """Write a new advisory reservation and return the :class:`Reservation` object. |
| 286 | |
| 287 | The reservation file is written atomically to |
| 288 | ``.muse/coordination/reservations/<uuid>.json`` via |
| 289 | :func:`~muse.core.store.write_text_atomic` (mkstemp → fsync → rename). |
| 290 | Concurrent calls from separate processes are safe — each call generates a |
| 291 | distinct UUID4 filename, so there is no write-write race. |
| 292 | |
| 293 | The written file is **immutable** after creation. To cancel a reservation |
| 294 | call :func:`create_release`; to extend its TTL call :func:`create_heartbeat`. |
| 295 | |
| 296 | Args: |
| 297 | root: Repository root (directory containing ``.muse/``). |
| 298 | run_id: Opaque agent/pipeline identifier stored verbatim. |
| 299 | The caller is responsible for length-limiting this value |
| 300 | before passing it (see ``_MAX_RUN_ID_LEN`` in the CLI). |
| 301 | branch: Current branch name, stored for filtering by branch. |
| 302 | addresses: Symbol addresses this reservation covers. Stored as-is; |
| 303 | the caller is responsible for count-limiting this list. |
| 304 | ttl_seconds: How long (in seconds) the reservation is valid. |
| 305 | Default 3600 (1 hour). The caller must clamp this to a |
| 306 | sensible range before passing it. |
| 307 | operation: Optional declared operation type (e.g. ``"rename"``). |
| 308 | The CLI restricts this to a known set; the core layer |
| 309 | stores whatever string is passed for forward-compatibility. |
| 310 | |
| 311 | Returns: |
| 312 | The newly created :class:`Reservation` with ``reservation_id`` set to |
| 313 | a fresh UUID4 and ``expires_at`` set to ``now + ttl_seconds``. |
| 314 | """ |
| 315 | _ensure_coord_dirs(root) |
| 316 | now = _now_utc() |
| 317 | res = Reservation( |
| 318 | reservation_id=str(_uuid_mod.uuid4()), |
| 319 | run_id=run_id, |
| 320 | branch=branch, |
| 321 | addresses=addresses, |
| 322 | created_at=now, |
| 323 | expires_at=now + datetime.timedelta(seconds=ttl_seconds), |
| 324 | operation=operation, |
| 325 | ) |
| 326 | path = _reservations_dir(root) / f"{res.reservation_id}.json" |
| 327 | write_text_atomic(path, json.dumps(res.to_dict(), indent=2) + "\n") |
| 328 | logger.debug("✅ Created reservation %s for %d addresses", res.reservation_id[:8], len(addresses)) |
| 329 | return res |
| 330 | |
| 331 | |
| 332 | def load_all_reservations(root: pathlib.Path) -> list[Reservation]: |
| 333 | """Load every reservation file from ``.muse/coordination/reservations/``. |
| 334 | |
| 335 | Returns *all* reservations — active, expired, and released alike. Use |
| 336 | :func:`active_reservations` when you only want currently-live ones, or |
| 337 | :func:`filter_reservations` for in-memory filtering without additional I/O. |
| 338 | |
| 339 | Corrupt or unreadable files are skipped with a ``WARNING`` log entry; they |
| 340 | never cause the function to raise. This makes the coordination layer |
| 341 | resilient to partial writes (e.g. a process killed during an older, |
| 342 | non-atomic write path) without sacrificing the valid records. |
| 343 | |
| 344 | Args: |
| 345 | root: Repository root (directory containing ``.muse/``). |
| 346 | |
| 347 | Returns: |
| 348 | List of :class:`Reservation` objects in filesystem-glob order |
| 349 | (not sorted). Returns ``[]`` if the reservations directory does not |
| 350 | exist yet (fresh repo). |
| 351 | """ |
| 352 | rdir = _reservations_dir(root) |
| 353 | if not rdir.exists(): |
| 354 | return [] |
| 355 | reservations: list[Reservation] = [] |
| 356 | for path in rdir.glob("*.json"): |
| 357 | try: |
| 358 | raw = json.loads(path.read_text()) |
| 359 | reservations.append(Reservation.from_dict(raw)) |
| 360 | except (json.JSONDecodeError, KeyError) as exc: |
| 361 | logger.warning("⚠️ Corrupt reservation %s: %s", path.name, exc) |
| 362 | return reservations |
| 363 | |
| 364 | |
| 365 | def active_reservations(root: pathlib.Path) -> list[Reservation]: |
| 366 | """Return reservations that are currently live. |
| 367 | |
| 368 | A reservation is live when **all** of the following hold: |
| 369 | |
| 370 | * No release tombstone exists for its ID. |
| 371 | * The current time is before its effective expiry — which is |
| 372 | ``max(reservation.expires_at, heartbeat.extended_expires_at)`` when a |
| 373 | heartbeat file exists for this reservation, or simply |
| 374 | ``reservation.expires_at`` otherwise. |
| 375 | |
| 376 | This is the authoritative liveness check. All swarm-coordination logic |
| 377 | should use this function rather than calling ``r.is_active()`` directly, |
| 378 | because the bare ``is_active()`` method cannot see releases or heartbeats. |
| 379 | """ |
| 380 | released = load_released_ids(root) |
| 381 | hb_map = load_heartbeat_map(root) |
| 382 | now = _now_utc() |
| 383 | result: list[Reservation] = [] |
| 384 | for r in load_all_reservations(root): |
| 385 | if r.reservation_id in released: |
| 386 | continue |
| 387 | hb = hb_map.get(r.reservation_id) |
| 388 | effective_expires = ( |
| 389 | max(r.expires_at, hb.extended_expires_at) |
| 390 | if hb is not None |
| 391 | else r.expires_at |
| 392 | ) |
| 393 | if now < effective_expires: |
| 394 | result.append(r) |
| 395 | return result |
| 396 | |
| 397 | |
| 398 | # --------------------------------------------------------------------------- |
| 399 | # Intent |
| 400 | # --------------------------------------------------------------------------- |
| 401 | |
| 402 | |
| 403 | class Intent: |
| 404 | """A declared operational intent extending a reservation. |
| 405 | |
| 406 | Whereas a :class:`Reservation` says "I will touch these symbols", an |
| 407 | ``Intent`` says "I will *rename* src/billing.py::compute_total". This |
| 408 | extra specificity allows ``muse forecast`` and ``muse plan-merge`` to |
| 409 | classify conflicts by operation type (a rename conflicts differently from |
| 410 | a delete) and gives post-mortem tooling a structured audit trail. |
| 411 | |
| 412 | Intents are write-once records with no TTL — they are permanent until |
| 413 | explicitly removed by ``muse coord gc --include-intents``. |
| 414 | |
| 415 | Attributes |
| 416 | ---------- |
| 417 | intent_id: |
| 418 | UUID uniquely identifying this intent record. |
| 419 | reservation_id: |
| 420 | UUID of the linked :class:`Reservation`, or ``""`` for a standalone |
| 421 | intent (not attached to any reservation). |
| 422 | run_id: |
| 423 | Agent or pipeline identifier that declared this intent. |
| 424 | branch: |
| 425 | Branch on which the intent was declared. |
| 426 | addresses: |
| 427 | List of symbol addresses the operation targets. |
| 428 | operation: |
| 429 | Declared operation type (e.g. ``"rename"``, ``"delete"``). |
| 430 | created_at: |
| 431 | UTC timestamp when the intent was written. |
| 432 | detail: |
| 433 | Optional human-readable description of the intended change. |
| 434 | """ |
| 435 | |
| 436 | def __init__( |
| 437 | self, |
| 438 | intent_id: str, |
| 439 | reservation_id: str, |
| 440 | run_id: str, |
| 441 | branch: str, |
| 442 | addresses: list[str], |
| 443 | operation: str, |
| 444 | created_at: datetime.datetime, |
| 445 | detail: str, |
| 446 | ) -> None: |
| 447 | self.intent_id = intent_id |
| 448 | self.reservation_id = reservation_id |
| 449 | self.run_id = run_id |
| 450 | self.branch = branch |
| 451 | self.addresses = addresses |
| 452 | self.operation = operation |
| 453 | self.created_at = created_at |
| 454 | self.detail = detail |
| 455 | |
| 456 | def to_dict(self) -> _IntentDict: |
| 457 | return { |
| 458 | "schema_version": _SCHEMA_VERSION, |
| 459 | "intent_id": self.intent_id, |
| 460 | "reservation_id": self.reservation_id, |
| 461 | "run_id": self.run_id, |
| 462 | "branch": self.branch, |
| 463 | "addresses": self.addresses, |
| 464 | "operation": self.operation, |
| 465 | "created_at": self.created_at.isoformat(), |
| 466 | "detail": self.detail, |
| 467 | } |
| 468 | |
| 469 | @classmethod |
| 470 | def from_dict(cls, d: _IntentDict) -> "Intent": |
| 471 | created_raw = d.get("created_at") |
| 472 | created_at = _parse_dt(str(created_raw)) if created_raw else _now_utc() |
| 473 | addrs_raw = d.get("addresses", []) |
| 474 | addrs = list(addrs_raw) if isinstance(addrs_raw, list) else [] |
| 475 | return cls( |
| 476 | intent_id=str(d.get("intent_id", "")), |
| 477 | reservation_id=str(d.get("reservation_id", "")), |
| 478 | run_id=str(d.get("run_id", "")), |
| 479 | branch=str(d.get("branch", "")), |
| 480 | addresses=addrs, |
| 481 | operation=str(d.get("operation", "")), |
| 482 | created_at=created_at, |
| 483 | detail=str(d.get("detail", "")), |
| 484 | ) |
| 485 | |
| 486 | |
| 487 | def create_intent( |
| 488 | root: pathlib.Path, |
| 489 | reservation_id: str, |
| 490 | run_id: str, |
| 491 | branch: str, |
| 492 | addresses: list[str], |
| 493 | operation: str, |
| 494 | detail: str = "", |
| 495 | ) -> Intent: |
| 496 | """Write and return a new intent record. |
| 497 | |
| 498 | Atomically writes ``.muse/coordination/intents/<uuid>.json`` and returns |
| 499 | the populated :class:`Intent` object. Each call produces a unique UUID, |
| 500 | so multiple intents for the same address and operation are additive rather |
| 501 | than replacing each other. |
| 502 | |
| 503 | Args: |
| 504 | root: Repository root (the directory containing ``.muse/``). |
| 505 | reservation_id: UUID of a linked reservation, or ``""`` for a |
| 506 | standalone intent. Not validated here — callers should validate |
| 507 | before calling (see :func:`~muse.core.coordination._validate_reservation_id`). |
| 508 | run_id: Agent or pipeline identifier for the audit trail. |
| 509 | branch: Branch on which the intent is being declared. |
| 510 | addresses: Symbol addresses the operation targets. |
| 511 | operation: One of the recognised operation strings (e.g. ``"rename"``, |
| 512 | ``"delete"``). Not validated here — callers are responsible for |
| 513 | restricting to the supported set. |
| 514 | detail: Optional human-readable description of the intended change. |
| 515 | |
| 516 | Returns: |
| 517 | The written :class:`Intent` with a freshly generated ``intent_id`` |
| 518 | and ``created_at`` timestamp. |
| 519 | """ |
| 520 | _ensure_coord_dirs(root) |
| 521 | now = _now_utc() |
| 522 | intent = Intent( |
| 523 | intent_id=str(_uuid_mod.uuid4()), |
| 524 | reservation_id=reservation_id, |
| 525 | run_id=run_id, |
| 526 | branch=branch, |
| 527 | addresses=addresses, |
| 528 | operation=operation, |
| 529 | created_at=now, |
| 530 | detail=detail, |
| 531 | ) |
| 532 | path = _intents_dir(root) / f"{intent.intent_id}.json" |
| 533 | write_text_atomic(path, json.dumps(intent.to_dict(), indent=2) + "\n") |
| 534 | logger.debug("✅ Created intent %s (%s)", intent.intent_id[:8], operation) |
| 535 | return intent |
| 536 | |
| 537 | |
| 538 | def load_all_intents(root: pathlib.Path) -> list[Intent]: |
| 539 | """Load and return every intent record in the repository. |
| 540 | |
| 541 | Reads ``.muse/coordination/intents/*.json``. Intents have no TTL — |
| 542 | they are permanent audit records until explicitly purged by |
| 543 | ``muse coord gc --include-intents``. All stored intents are returned |
| 544 | regardless of age. Use :func:`filter_intents` to narrow by run_id, |
| 545 | branch, operation, or address glob after loading. |
| 546 | |
| 547 | Corrupt or unreadable files are skipped with a warning rather than |
| 548 | raising, so a single damaged file never blocks the rest. |
| 549 | |
| 550 | Args: |
| 551 | root: Repository root (the directory containing ``.muse/``). |
| 552 | |
| 553 | Returns: |
| 554 | A list of :class:`Intent` objects in filesystem iteration order |
| 555 | (not sorted). Returns an empty list when the intents directory |
| 556 | does not yet exist. |
| 557 | """ |
| 558 | idir = _intents_dir(root) |
| 559 | if not idir.exists(): |
| 560 | return [] |
| 561 | intents: list[Intent] = [] |
| 562 | for path in idir.glob("*.json"): |
| 563 | try: |
| 564 | raw = json.loads(path.read_text()) |
| 565 | intents.append(Intent.from_dict(raw)) |
| 566 | except (json.JSONDecodeError, KeyError) as exc: |
| 567 | logger.warning("⚠️ Corrupt intent %s: %s", path.name, exc) |
| 568 | return intents |
| 569 | |
| 570 | |
| 571 | # --------------------------------------------------------------------------- |
| 572 | # Filtering helpers |
| 573 | # --------------------------------------------------------------------------- |
| 574 | |
| 575 | |
| 576 | def filter_reservations( |
| 577 | reservations: list[Reservation], |
| 578 | *, |
| 579 | run_id: str | None = None, |
| 580 | branch: str | None = None, |
| 581 | address_glob: str | None = None, |
| 582 | operation: str | None = None, |
| 583 | include_expired: bool = False, |
| 584 | released_ids: frozenset[str] | None = None, |
| 585 | heartbeat_expires: HeartbeatMap | None = None, |
| 586 | ) -> list[Reservation]: |
| 587 | """Return a filtered subset of *reservations*. |
| 588 | |
| 589 | All filters are applied with AND semantics — a reservation must satisfy |
| 590 | every supplied criterion to be included. |
| 591 | |
| 592 | Parameters |
| 593 | ---------- |
| 594 | reservations: |
| 595 | Source list, typically from :func:`load_all_reservations`. |
| 596 | run_id: |
| 597 | Exact-match filter on :attr:`~Reservation.run_id`. |
| 598 | branch: |
| 599 | Exact-match filter on :attr:`~Reservation.branch`. |
| 600 | address_glob: |
| 601 | :mod:`fnmatch` glob applied to each address in the reservation. |
| 602 | A reservation is included when *any* of its addresses match. |
| 603 | Example: ``"billing.py::*"`` matches all symbols in ``billing.py``. |
| 604 | This is a string operation — no filesystem access occurs. |
| 605 | operation: |
| 606 | Exact-match filter on :attr:`~Reservation.operation` |
| 607 | (e.g. ``"rename"``, ``"modify"``). ``None`` matches reservations |
| 608 | with any operation, including those with no declared operation. |
| 609 | include_expired: |
| 610 | When ``False`` (default) expired and released reservations are |
| 611 | excluded. When ``True`` all reservations pass the liveness check. |
| 612 | released_ids: |
| 613 | Optional pre-loaded set of released reservation IDs (from |
| 614 | :func:`load_released_ids`). When supplied, released reservations |
| 615 | are excluded even when they have not yet expired by TTL. Pass this |
| 616 | to avoid redundant filesystem I/O when the caller has already loaded |
| 617 | the releases directory. |
| 618 | heartbeat_expires: |
| 619 | Optional mapping of ``reservation_id`` → ``extended_expires_at`` |
| 620 | (from :func:`load_heartbeat_map`). When supplied, the effective |
| 621 | expiry is ``max(reservation.expires_at, extended_expires_at)``, |
| 622 | keeping heartbeat-extended reservations alive past their original TTL. |
| 623 | |
| 624 | Returns |
| 625 | ------- |
| 626 | list[Reservation] |
| 627 | Filtered list, preserving the order of *reservations*. |
| 628 | """ |
| 629 | now = _now_utc() |
| 630 | result: list[Reservation] = [] |
| 631 | for res in reservations: |
| 632 | if not include_expired: |
| 633 | # Released check. |
| 634 | if released_ids is not None and res.reservation_id in released_ids: |
| 635 | continue |
| 636 | # Effective TTL check (with optional heartbeat extension). |
| 637 | hb_exp = ( |
| 638 | heartbeat_expires.get(res.reservation_id) |
| 639 | if heartbeat_expires is not None |
| 640 | else None |
| 641 | ) |
| 642 | effective_expires = ( |
| 643 | max(res.expires_at, hb_exp) |
| 644 | if hb_exp is not None |
| 645 | else res.expires_at |
| 646 | ) |
| 647 | if now >= effective_expires: |
| 648 | continue |
| 649 | if run_id is not None and res.run_id != run_id: |
| 650 | continue |
| 651 | if branch is not None and res.branch != branch: |
| 652 | continue |
| 653 | if address_glob is not None: |
| 654 | if not any(fnmatch.fnmatch(addr, address_glob) for addr in res.addresses): |
| 655 | continue |
| 656 | if operation is not None and res.operation != operation: |
| 657 | continue |
| 658 | result.append(res) |
| 659 | return result |
| 660 | |
| 661 | |
| 662 | # --------------------------------------------------------------------------- |
| 663 | # Release — write-once tombstone |
| 664 | # --------------------------------------------------------------------------- |
| 665 | |
| 666 | _VALID_REASONS = frozenset({"completed", "cancelled", "superseded"}) |
| 667 | |
| 668 | |
| 669 | class Release: |
| 670 | """A write-once tombstone that marks a reservation as no longer active. |
| 671 | |
| 672 | Once a release record exists for a ``reservation_id``, that reservation |
| 673 | is excluded from :func:`active_reservations` regardless of its TTL. |
| 674 | Releases are never deleted by normal operation — only :func:`run_coord_gc` |
| 675 | removes them after the grace period. |
| 676 | """ |
| 677 | |
| 678 | def __init__( |
| 679 | self, |
| 680 | reservation_id: str, |
| 681 | run_id: str, |
| 682 | released_at: datetime.datetime, |
| 683 | reason: str, |
| 684 | ) -> None: |
| 685 | self.reservation_id = reservation_id |
| 686 | self.run_id = run_id |
| 687 | self.released_at = released_at |
| 688 | self.reason = reason |
| 689 | |
| 690 | def to_dict(self) -> _ReleaseDict: |
| 691 | return { |
| 692 | "schema_version": _SCHEMA_VERSION, |
| 693 | "reservation_id": self.reservation_id, |
| 694 | "run_id": self.run_id, |
| 695 | "released_at": self.released_at.isoformat(), |
| 696 | "reason": self.reason, |
| 697 | } |
| 698 | |
| 699 | @classmethod |
| 700 | def from_dict(cls, d: _ReleaseDict) -> "Release": |
| 701 | released_raw = d.get("released_at") |
| 702 | released_at = _parse_dt(str(released_raw)) if released_raw else _now_utc() |
| 703 | return cls( |
| 704 | reservation_id=str(d.get("reservation_id", "")), |
| 705 | run_id=str(d.get("run_id", "")), |
| 706 | released_at=released_at, |
| 707 | reason=str(d.get("reason", "completed")), |
| 708 | ) |
| 709 | |
| 710 | |
| 711 | def create_release( |
| 712 | root: pathlib.Path, |
| 713 | reservation_id: str, |
| 714 | run_id: str, |
| 715 | reason: str = "completed", |
| 716 | ) -> Release: |
| 717 | """Write and return a release tombstone for *reservation_id*. |
| 718 | |
| 719 | Parameters |
| 720 | ---------- |
| 721 | root: |
| 722 | Repository root (the directory containing ``.muse/``). |
| 723 | reservation_id: |
| 724 | The UUID of the reservation being released. Must be a valid UUID — |
| 725 | invalid values raise ``ValueError`` before any file I/O occurs. |
| 726 | run_id: |
| 727 | The agent releasing the reservation (for audit trail). |
| 728 | reason: |
| 729 | One of ``"completed"``, ``"cancelled"``, or ``"superseded"``. |
| 730 | |
| 731 | Raises |
| 732 | ------ |
| 733 | ValueError |
| 734 | If *reservation_id* is not a valid UUID or *reason* is not recognised. |
| 735 | FileExistsError |
| 736 | If a release tombstone already exists for this reservation. Callers |
| 737 | should check :func:`load_released_ids` first when idempotency is needed. |
| 738 | """ |
| 739 | _validate_reservation_id(reservation_id) |
| 740 | if reason not in _VALID_REASONS: |
| 741 | raise ValueError( |
| 742 | f"reason must be one of {sorted(_VALID_REASONS)}: {reason!r}" |
| 743 | ) |
| 744 | _ensure_coord_dirs(root) |
| 745 | path = _releases_dir(root) / f"{reservation_id}.json" |
| 746 | # Containment check — belt-and-suspenders after UUID validation. |
| 747 | try: |
| 748 | path.resolve().relative_to(_releases_dir(root).resolve()) |
| 749 | except ValueError: |
| 750 | raise ValueError(f"reservation_id produces a path outside the releases directory") |
| 751 | if path.exists(): |
| 752 | raise FileExistsError( |
| 753 | f"reservation {reservation_id!r} is already released" |
| 754 | ) |
| 755 | now = _now_utc() |
| 756 | rel = Release( |
| 757 | reservation_id=reservation_id, |
| 758 | run_id=run_id, |
| 759 | released_at=now, |
| 760 | reason=reason, |
| 761 | ) |
| 762 | write_text_atomic(path, json.dumps(rel.to_dict(), indent=2) + "\n") |
| 763 | logger.debug("✅ Released reservation %s (%s)", reservation_id[:8], reason) |
| 764 | return rel |
| 765 | |
| 766 | |
| 767 | def load_all_releases(root: pathlib.Path) -> list[Release]: |
| 768 | """Load and return every release tombstone in the repository. |
| 769 | |
| 770 | Reads ``.muse/coordination/releases/*.json``. Corrupt or unreadable |
| 771 | files are skipped with a warning rather than raising, so a single |
| 772 | damaged file never prevents other records from being read. |
| 773 | |
| 774 | Args: |
| 775 | root: Repository root (the directory containing ``.muse/``). |
| 776 | |
| 777 | Returns: |
| 778 | A list of :class:`Release` objects in filesystem iteration order |
| 779 | (not sorted). Returns an empty list when the releases directory |
| 780 | does not yet exist. |
| 781 | """ |
| 782 | rdir = _releases_dir(root) |
| 783 | if not rdir.exists(): |
| 784 | return [] |
| 785 | releases: list[Release] = [] |
| 786 | for path in rdir.glob("*.json"): |
| 787 | try: |
| 788 | raw = json.loads(path.read_text()) |
| 789 | releases.append(Release.from_dict(raw)) |
| 790 | except (json.JSONDecodeError, KeyError) as exc: |
| 791 | logger.warning("⚠️ Corrupt release record %s: %s", path.name, exc) |
| 792 | return releases |
| 793 | |
| 794 | |
| 795 | def load_released_ids(root: pathlib.Path) -> frozenset[str]: |
| 796 | """Return the set of reservation IDs that have been released. |
| 797 | |
| 798 | This is a lightweight alternative to :func:`load_all_releases` when the |
| 799 | caller only needs to know *whether* a reservation has been released, not |
| 800 | the full release metadata. Scans only file stems (no JSON parsing) when |
| 801 | the releases directory contains no corrupt files — but falls back to |
| 802 | parsing when needed. |
| 803 | """ |
| 804 | rdir = _releases_dir(root) |
| 805 | if not rdir.exists(): |
| 806 | return frozenset() |
| 807 | # Use file stems as reservation IDs — avoids JSON parsing entirely. |
| 808 | return frozenset(p.stem for p in rdir.glob("*.json")) |
| 809 | |
| 810 | |
| 811 | # --------------------------------------------------------------------------- |
| 812 | # Heartbeat — mutable keep-alive (one file per reservation, atomically updated) |
| 813 | # --------------------------------------------------------------------------- |
| 814 | |
| 815 | |
| 816 | class Heartbeat: |
| 817 | """A keep-alive record that extends a reservation's effective TTL. |
| 818 | |
| 819 | Unlike reservations and releases, heartbeat files are *mutable* — each |
| 820 | call to :func:`create_heartbeat` atomically rewrites the single heartbeat |
| 821 | file for a reservation. Only the most recent heartbeat has meaning. |
| 822 | """ |
| 823 | |
| 824 | def __init__( |
| 825 | self, |
| 826 | reservation_id: str, |
| 827 | run_id: str, |
| 828 | last_beat_at: datetime.datetime, |
| 829 | extended_expires_at: datetime.datetime, |
| 830 | ) -> None: |
| 831 | self.reservation_id = reservation_id |
| 832 | self.run_id = run_id |
| 833 | self.last_beat_at = last_beat_at |
| 834 | self.extended_expires_at = extended_expires_at |
| 835 | |
| 836 | def to_dict(self) -> _HeartbeatDict: |
| 837 | return { |
| 838 | "schema_version": _SCHEMA_VERSION, |
| 839 | "reservation_id": self.reservation_id, |
| 840 | "run_id": self.run_id, |
| 841 | "last_beat_at": self.last_beat_at.isoformat(), |
| 842 | "extended_expires_at": self.extended_expires_at.isoformat(), |
| 843 | } |
| 844 | |
| 845 | @classmethod |
| 846 | def from_dict(cls, d: _HeartbeatDict) -> "Heartbeat": |
| 847 | beat_raw = d.get("last_beat_at") |
| 848 | exp_raw = d.get("extended_expires_at") |
| 849 | last_beat_at = _parse_dt(str(beat_raw)) if beat_raw else _now_utc() |
| 850 | extended_expires_at = _parse_dt(str(exp_raw)) if exp_raw else _now_utc() |
| 851 | return cls( |
| 852 | reservation_id=str(d.get("reservation_id", "")), |
| 853 | run_id=str(d.get("run_id", "")), |
| 854 | last_beat_at=last_beat_at, |
| 855 | extended_expires_at=extended_expires_at, |
| 856 | ) |
| 857 | |
| 858 | |
| 859 | def create_heartbeat( |
| 860 | root: pathlib.Path, |
| 861 | reservation_id: str, |
| 862 | run_id: str, |
| 863 | extension_seconds: int = 3600, |
| 864 | ) -> Heartbeat: |
| 865 | """Write or update the heartbeat file for *reservation_id*. |
| 866 | |
| 867 | The heartbeat file at ``.muse/coordination/heartbeats/<reservation-id>.json`` |
| 868 | is atomically rewritten with a new ``extended_expires_at`` equal to |
| 869 | ``now + extension_seconds``. This extends the reservation's effective TTL |
| 870 | (as seen by :func:`active_reservations` and :func:`filter_reservations`) |
| 871 | without modifying the immutable reservation record itself. |
| 872 | |
| 873 | Parameters |
| 874 | ---------- |
| 875 | root: |
| 876 | Repository root. |
| 877 | reservation_id: |
| 878 | The UUID of the reservation to keep alive. Validated as a UUID before |
| 879 | any file I/O. |
| 880 | run_id: |
| 881 | The agent sending the heartbeat (for audit). |
| 882 | extension_seconds: |
| 883 | How far into the future the new ``extended_expires_at`` is set |
| 884 | (default: 3600 s = 1 h). |
| 885 | |
| 886 | Raises |
| 887 | ------ |
| 888 | ValueError |
| 889 | If *reservation_id* is not a valid UUID or *extension_seconds* ≤ 0. |
| 890 | """ |
| 891 | _validate_reservation_id(reservation_id) |
| 892 | if extension_seconds <= 0: |
| 893 | raise ValueError(f"extension_seconds must be > 0: {extension_seconds!r}") |
| 894 | _ensure_coord_dirs(root) |
| 895 | path = _heartbeats_dir(root) / f"{reservation_id}.json" |
| 896 | try: |
| 897 | path.resolve().relative_to(_heartbeats_dir(root).resolve()) |
| 898 | except ValueError: |
| 899 | raise ValueError("reservation_id produces a path outside the heartbeats directory") |
| 900 | now = _now_utc() |
| 901 | hb = Heartbeat( |
| 902 | reservation_id=reservation_id, |
| 903 | run_id=run_id, |
| 904 | last_beat_at=now, |
| 905 | extended_expires_at=now + datetime.timedelta(seconds=extension_seconds), |
| 906 | ) |
| 907 | write_text_atomic(path, json.dumps(hb.to_dict(), indent=2) + "\n") |
| 908 | logger.debug( |
| 909 | "💓 Heartbeat %s → expires %s", |
| 910 | reservation_id[:8], |
| 911 | hb.extended_expires_at.isoformat()[:19], |
| 912 | ) |
| 913 | return hb |
| 914 | |
| 915 | |
| 916 | def load_heartbeat_map(root: pathlib.Path) -> HeartbeatRecordMap: |
| 917 | """Return a ``reservation_id`` → :class:`Heartbeat` mapping for all live heartbeats. |
| 918 | |
| 919 | Reads ``.muse/coordination/heartbeats/*.json``. Because heartbeat files |
| 920 | are mutable (each :func:`create_heartbeat` call atomically replaces the |
| 921 | file), only the most recent heartbeat per reservation ID has meaning. |
| 922 | Corrupt or unreadable files are skipped with a warning rather than |
| 923 | raising, so a single damaged file never blocks the rest. |
| 924 | |
| 925 | Args: |
| 926 | root: Repository root (the directory containing ``.muse/``). |
| 927 | |
| 928 | Returns: |
| 929 | A dict mapping each ``reservation_id`` to its most recent |
| 930 | :class:`Heartbeat`. Returns an empty dict when the heartbeats |
| 931 | directory does not yet exist. |
| 932 | """ |
| 933 | hdir = _heartbeats_dir(root) |
| 934 | if not hdir.exists(): |
| 935 | return {} |
| 936 | hb_map: HeartbeatRecordMap = {} |
| 937 | for path in hdir.glob("*.json"): |
| 938 | try: |
| 939 | raw = json.loads(path.read_text()) |
| 940 | hb = Heartbeat.from_dict(raw) |
| 941 | hb_map[hb.reservation_id] = hb |
| 942 | except (json.JSONDecodeError, KeyError) as exc: |
| 943 | logger.warning("⚠️ Corrupt heartbeat %s: %s", path.name, exc) |
| 944 | return hb_map |
| 945 | |
| 946 | |
| 947 | # --------------------------------------------------------------------------- |
| 948 | # Coordination GC |
| 949 | # --------------------------------------------------------------------------- |
| 950 | |
| 951 | |
| 952 | @dataclasses.dataclass |
| 953 | class CoordGcResult: |
| 954 | """Structured result returned by :func:`run_coord_gc`. |
| 955 | |
| 956 | All ``*_removed`` counters reflect records that were (or would be, in |
| 957 | dry-run mode) deleted. ``*_removed_bytes`` is the sum of ``st_size`` |
| 958 | values taken *before* deletion. ``removed_ids`` lists reservation UUIDs |
| 959 | whose reservation file was removed (or would be). |
| 960 | |
| 961 | Attributes |
| 962 | ---------- |
| 963 | reservations_removed: |
| 964 | Count of reservation JSON files deleted. |
| 965 | reservations_removed_bytes: |
| 966 | Total bytes freed from reservation files. |
| 967 | releases_removed: |
| 968 | Count of release tombstone files deleted (matched and orphaned). |
| 969 | releases_removed_bytes: |
| 970 | Total bytes freed from release files. |
| 971 | heartbeats_removed: |
| 972 | Count of heartbeat files deleted (matched and orphaned). |
| 973 | heartbeats_removed_bytes: |
| 974 | Total bytes freed from heartbeat files. |
| 975 | intents_removed: |
| 976 | Count of intent files deleted (only when ``include_intents=True``). |
| 977 | intents_removed_bytes: |
| 978 | Total bytes freed from intent files. |
| 979 | total_removed: |
| 980 | Sum of all ``*_removed`` counters. |
| 981 | total_removed_bytes: |
| 982 | Sum of all ``*_removed_bytes`` counters. |
| 983 | elapsed_seconds: |
| 984 | Wall-clock time for the full GC pass (monotonic). |
| 985 | grace_period_seconds: |
| 986 | Grace-period value used for this run (echoed for audit). |
| 987 | dry_run: |
| 988 | ``True`` when no files were actually deleted. |
| 989 | include_intents: |
| 990 | ``True`` when intent cleanup was enabled. |
| 991 | removed_ids: |
| 992 | Reservation UUIDs collected this pass. Populated in dry-run mode |
| 993 | too so callers can preview what would be removed. |
| 994 | """ |
| 995 | |
| 996 | reservations_removed: int = 0 |
| 997 | reservations_removed_bytes: int = 0 |
| 998 | releases_removed: int = 0 |
| 999 | releases_removed_bytes: int = 0 |
| 1000 | heartbeats_removed: int = 0 |
| 1001 | heartbeats_removed_bytes: int = 0 |
| 1002 | intents_removed: int = 0 |
| 1003 | intents_removed_bytes: int = 0 |
| 1004 | total_removed: int = 0 |
| 1005 | total_removed_bytes: int = 0 |
| 1006 | elapsed_seconds: float = 0.0 |
| 1007 | grace_period_seconds: int = 300 |
| 1008 | dry_run: bool = False |
| 1009 | include_intents: bool = False |
| 1010 | removed_ids: list[str] = dataclasses.field(default_factory=list) |
| 1011 | |
| 1012 | |
| 1013 | def run_coord_gc( |
| 1014 | root: pathlib.Path, |
| 1015 | *, |
| 1016 | dry_run: bool = False, |
| 1017 | grace_period_seconds: int = 300, |
| 1018 | include_intents: bool = False, |
| 1019 | max_intent_age_seconds: int = 604800, # 7 days |
| 1020 | ) -> CoordGcResult: |
| 1021 | """Remove stale coordination records from ``.muse/coordination/``. |
| 1022 | |
| 1023 | What gets collected |
| 1024 | ------------------- |
| 1025 | * **Expired reservations** — TTL exhausted (considering heartbeat |
| 1026 | extensions), past the grace period. |
| 1027 | * **Released reservations** — release tombstone present and the release |
| 1028 | is older than the grace period. |
| 1029 | * **Corresponding release tombstones** — removed alongside their reservation. |
| 1030 | * **Corresponding heartbeat files** — removed when the reservation is removed. |
| 1031 | * **Orphaned heartbeats** — heartbeat file exists but the reservation file |
| 1032 | does not (e.g. a previous partial GC or corrupted write). |
| 1033 | * **Orphaned releases** — tombstone exists but the reservation file does not. |
| 1034 | * **Old intents** (opt-in) — ``created_at`` older than *max_intent_age_seconds* |
| 1035 | when ``include_intents=True``. |
| 1036 | |
| 1037 | What is never collected |
| 1038 | ----------------------- |
| 1039 | * Active reservations (not released, effective expiry in the future). |
| 1040 | * Records within the grace period of expiry or release. |
| 1041 | |
| 1042 | Parameters |
| 1043 | ---------- |
| 1044 | root: |
| 1045 | Repository root. |
| 1046 | dry_run: |
| 1047 | When ``True``, compute what would be removed but delete nothing. |
| 1048 | grace_period_seconds: |
| 1049 | Records expired or released within the last N seconds are skipped. |
| 1050 | Protects against races with agents that are still reading coordination |
| 1051 | state. Default: 300 s (5 min). |
| 1052 | include_intents: |
| 1053 | When ``True``, intents older than *max_intent_age_seconds* are also |
| 1054 | collected. Default: ``False`` (intents are permanent audit records |
| 1055 | unless explicitly cleaned). |
| 1056 | max_intent_age_seconds: |
| 1057 | Age threshold for intent cleanup when ``include_intents=True``. |
| 1058 | Default: 604 800 s (7 days). |
| 1059 | """ |
| 1060 | t0 = _time_mod.monotonic() |
| 1061 | result = CoordGcResult( |
| 1062 | grace_period_seconds=grace_period_seconds, |
| 1063 | dry_run=dry_run, |
| 1064 | include_intents=include_intents, |
| 1065 | ) |
| 1066 | |
| 1067 | now = _now_utc() |
| 1068 | grace = datetime.timedelta(seconds=grace_period_seconds) |
| 1069 | |
| 1070 | # ── Load all coordination state ─────────────────────────────────────────── |
| 1071 | all_reservations = load_all_reservations(root) |
| 1072 | released_ids = load_released_ids(root) |
| 1073 | hb_map = load_heartbeat_map(root) |
| 1074 | |
| 1075 | # Build a set of reservation IDs that currently exist on disk. |
| 1076 | existing_res_ids = {r.reservation_id for r in all_reservations} |
| 1077 | |
| 1078 | # Map reservation_id → Release for tombstones that match a reservation. |
| 1079 | release_map: ReleaseMap = {} |
| 1080 | for rel in load_all_releases(root): |
| 1081 | release_map[rel.reservation_id] = rel |
| 1082 | |
| 1083 | # ── Determine which reservations to collect ─────────────────────────────── |
| 1084 | collectable_res_ids: set[str] = set() |
| 1085 | |
| 1086 | for res in all_reservations: |
| 1087 | hb = hb_map.get(res.reservation_id) |
| 1088 | effective_expires = ( |
| 1089 | max(res.expires_at, hb.extended_expires_at) |
| 1090 | if hb is not None |
| 1091 | else res.expires_at |
| 1092 | ) |
| 1093 | |
| 1094 | if res.reservation_id in released_ids: |
| 1095 | # Released: collect if release is old enough. |
| 1096 | rel = release_map.get(res.reservation_id) |
| 1097 | cutoff = rel.released_at if rel is not None else effective_expires |
| 1098 | if (now - cutoff) >= grace: |
| 1099 | collectable_res_ids.add(res.reservation_id) |
| 1100 | else: |
| 1101 | # Expired: collect if past grace period. |
| 1102 | if effective_expires < now and (now - effective_expires) >= grace: |
| 1103 | collectable_res_ids.add(res.reservation_id) |
| 1104 | |
| 1105 | # ── Delete reservation files ────────────────────────────────────────────── |
| 1106 | # Each stat+unlink pair is wrapped in a try/except FileNotFoundError so |
| 1107 | # that a concurrent GC pass that already deleted the file does not crash |
| 1108 | # this one. The counters reflect what *this* pass actually removed. |
| 1109 | res_dir = _reservations_dir(root) |
| 1110 | if res_dir.exists(): |
| 1111 | for path in res_dir.glob("*.json"): |
| 1112 | if path.stem in collectable_res_ids: |
| 1113 | try: |
| 1114 | size = path.stat().st_size |
| 1115 | except FileNotFoundError: |
| 1116 | continue # already removed by a concurrent GC pass |
| 1117 | result.reservations_removed += 1 |
| 1118 | result.reservations_removed_bytes += size |
| 1119 | result.removed_ids.append(path.stem) |
| 1120 | if not dry_run: |
| 1121 | path.unlink(missing_ok=True) |
| 1122 | |
| 1123 | # ── Delete corresponding release tombstones ─────────────────────────────── |
| 1124 | rel_dir = _releases_dir(root) |
| 1125 | if rel_dir.exists(): |
| 1126 | for path in rel_dir.glob("*.json"): |
| 1127 | stem = path.stem |
| 1128 | if stem in collectable_res_ids or stem not in existing_res_ids: |
| 1129 | # Collect if reservation is being removed, or orphaned. |
| 1130 | try: |
| 1131 | size = path.stat().st_size |
| 1132 | except FileNotFoundError: |
| 1133 | continue |
| 1134 | result.releases_removed += 1 |
| 1135 | result.releases_removed_bytes += size |
| 1136 | if not dry_run: |
| 1137 | path.unlink(missing_ok=True) |
| 1138 | |
| 1139 | # ── Delete corresponding heartbeat files ────────────────────────────────── |
| 1140 | hb_dir = _heartbeats_dir(root) |
| 1141 | if hb_dir.exists(): |
| 1142 | for path in hb_dir.glob("*.json"): |
| 1143 | stem = path.stem |
| 1144 | if stem in collectable_res_ids or stem not in existing_res_ids: |
| 1145 | try: |
| 1146 | size = path.stat().st_size |
| 1147 | except FileNotFoundError: |
| 1148 | continue |
| 1149 | result.heartbeats_removed += 1 |
| 1150 | result.heartbeats_removed_bytes += size |
| 1151 | if not dry_run: |
| 1152 | path.unlink(missing_ok=True) |
| 1153 | |
| 1154 | # ── Optionally collect old intents ──────────────────────────────────────── |
| 1155 | if include_intents: |
| 1156 | intent_dir = _intents_dir(root) |
| 1157 | max_age = datetime.timedelta(seconds=max_intent_age_seconds) |
| 1158 | if intent_dir.exists(): |
| 1159 | for path in intent_dir.glob("*.json"): |
| 1160 | try: |
| 1161 | raw = json.loads(path.read_text()) |
| 1162 | created_raw = raw.get("created_at") |
| 1163 | if not created_raw: |
| 1164 | continue |
| 1165 | created_at = _parse_dt(str(created_raw)) |
| 1166 | if (now - created_at) >= max_age: |
| 1167 | size = path.stat().st_size |
| 1168 | result.intents_removed += 1 |
| 1169 | result.intents_removed_bytes += size |
| 1170 | if not dry_run: |
| 1171 | path.unlink(missing_ok=True) |
| 1172 | except FileNotFoundError: |
| 1173 | continue # concurrent GC removed it first |
| 1174 | except (json.JSONDecodeError, KeyError, ValueError): |
| 1175 | pass # Corrupt intent — skip silently. |
| 1176 | |
| 1177 | result.total_removed = ( |
| 1178 | result.reservations_removed |
| 1179 | + result.releases_removed |
| 1180 | + result.heartbeats_removed |
| 1181 | + result.intents_removed |
| 1182 | ) |
| 1183 | result.total_removed_bytes = ( |
| 1184 | result.reservations_removed_bytes |
| 1185 | + result.releases_removed_bytes |
| 1186 | + result.heartbeats_removed_bytes |
| 1187 | + result.intents_removed_bytes |
| 1188 | ) |
| 1189 | result.elapsed_seconds = round(_time_mod.monotonic() - t0, 4) |
| 1190 | return result |
| 1191 | |
| 1192 | |
| 1193 | # --------------------------------------------------------------------------- |
| 1194 | # filter_intents (placed after all classes are defined) |
| 1195 | # --------------------------------------------------------------------------- |
| 1196 | |
| 1197 | |
| 1198 | def filter_intents( |
| 1199 | intents: list[Intent], |
| 1200 | *, |
| 1201 | run_id: str | None = None, |
| 1202 | branch: str | None = None, |
| 1203 | address_glob: str | None = None, |
| 1204 | operation: str | None = None, |
| 1205 | ) -> list[Intent]: |
| 1206 | """Return a filtered subset of *intents*. |
| 1207 | |
| 1208 | All filters are applied with AND semantics. |
| 1209 | |
| 1210 | Parameters |
| 1211 | ---------- |
| 1212 | intents: |
| 1213 | Source list, typically from :func:`load_all_intents`. |
| 1214 | run_id: |
| 1215 | Exact-match filter on :attr:`~Intent.run_id`. |
| 1216 | branch: |
| 1217 | Exact-match filter on :attr:`~Intent.branch`. |
| 1218 | address_glob: |
| 1219 | :mod:`fnmatch` glob applied to each address. An intent is included |
| 1220 | when *any* of its addresses match. String-only — no filesystem I/O. |
| 1221 | operation: |
| 1222 | Exact-match filter on :attr:`~Intent.operation` |
| 1223 | (e.g. ``"rename"``, ``"delete"``). ``None`` returns all intents |
| 1224 | regardless of operation. |
| 1225 | |
| 1226 | Returns |
| 1227 | ------- |
| 1228 | list[Intent] |
| 1229 | Filtered list, preserving the order of *intents*. |
| 1230 | """ |
| 1231 | result: list[Intent] = [] |
| 1232 | for intent in intents: |
| 1233 | if run_id is not None and intent.run_id != run_id: |
| 1234 | continue |
| 1235 | if branch is not None and intent.branch != branch: |
| 1236 | continue |
| 1237 | if address_glob is not None: |
| 1238 | if not any(fnmatch.fnmatch(addr, address_glob) for addr in intent.addresses): |
| 1239 | continue |
| 1240 | if operation is not None and intent.operation != operation: |
| 1241 | continue |
| 1242 | result.append(intent) |
| 1243 | return result |