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