"""Multi-agent coordination layer for the Muse VCS. Coordination data lives under ``.muse/coordination/``. It is purely advisory — the VCS engine never reads it for correctness decisions. Its purpose is to enable agents working in parallel to announce their intentions, detect likely conflicts *before* they happen, and plan merges without writing to the repo. Layout:: .muse/coordination/ reservations/.json write-once advisory symbol lease intents/.json write-once operation declaration releases/.json write-once tombstone (agent done / cancelled) heartbeats/.json mutable keep-alive, atomically updated Reservation schema:: { "schema_version": "", "reservation_id": "", "run_id": "", "branch": "", "addresses": ["src/billing.py::compute_total", ...], "created_at": "2026-03-18T12:00:00+00:00", "expires_at": "2026-03-18T13:00:00+00:00", "operation": null | "rename" | "move" | "extract" | "modify" | "delete" } Release schema:: { "schema_version": "", "reservation_id": "", "run_id": "", "released_at": "2026-03-18T12:05:00+00:00", "reason": "completed" | "cancelled" | "superseded" } Heartbeat schema:: { "schema_version": "", "reservation_id": "", "run_id": "", "last_beat_at": "2026-03-18T12:30:00+00:00", "extended_expires_at": "2026-03-18T13:30:00+00:00" } Intent schema:: { "schema_version": "", "intent_id": "", "reservation_id": "", "run_id": "", "branch": "", "addresses": ["src/billing.py::compute_total"], "operation": "rename", "created_at": "2026-03-18T12:00:00+00:00", "detail": "rename to compute_invoice_total" } Lifecycle --------- Reservations are write-once (immutable audit trail). Releases and intents are also write-once. Heartbeats are the *only* mutable state — each call to :func:`create_heartbeat` atomically rewrites the single heartbeat file for that reservation, extending its effective TTL. A reservation is *active* when **all** of the following hold: 1. No release tombstone exists for its ID. 2. The current time is before ``max(reservation.expires_at, heartbeat.extended_expires_at)`` (heartbeat extends the TTL when present). :func:`active_reservations` enforces this check. :func:`filter_reservations` accepts optional ``released_ids`` and ``heartbeat_expires`` kwargs so callers that have already loaded those structures avoid redundant I/O. Filtering --------- :func:`filter_reservations` and :func:`filter_intents` accept keyword-only arguments for in-memory filtering without additional I/O. Address matching uses :mod:`fnmatch` glob syntax (e.g. ``"billing.py::*"``), which is safe because it operates only on strings — no filesystem access occurs. Security -------- All functions that accept a ``reservation_id`` from external input call :func:`_validate_reservation_id` before constructing any file path. A valid UUID4 string cannot contain ``/`` or ``..``, preventing directory traversal attacks. Path containment is verified with ``resolve().relative_to()`` as a second line of defence. """ from __future__ import annotations import dataclasses import datetime import fnmatch import json import logging import pathlib import re import time as _time_mod import uuid as _uuid_mod from muse._version import __version__ as _SCHEMA_VERSION from muse.core.store import write_text_atomic from typing import TypedDict logger = logging.getLogger(__name__) type _ReservationVal = str | int | list[str] | None type _ReservationDict = dict[str, _ReservationVal] # serialized Reservation type _IntentVal = str | int | list[str] type _IntentDict = dict[str, _IntentVal] # serialized Intent type HeartbeatMap = dict[str, datetime.datetime] # reservation_id → expiry type HeartbeatRecordMap = dict[str, "Heartbeat"] # reservation_id → Heartbeat type ReleaseMap = dict[str, "Release"] # reservation_id → Release class _ReleaseDict(TypedDict): """JSON-serialisable form of a :class:`Release`.""" schema_version: str reservation_id: str run_id: str released_at: str reason: str class _HeartbeatDict(TypedDict): """JSON-serialisable form of a :class:`Heartbeat`.""" schema_version: str reservation_id: str run_id: str last_beat_at: str agent_id: str # --------------------------------------------------------------------------- # Directory helpers # --------------------------------------------------------------------------- def _coord_dir(root: pathlib.Path) -> pathlib.Path: return root / ".muse" / "coordination" def _reservations_dir(root: pathlib.Path) -> pathlib.Path: return _coord_dir(root) / "reservations" def _intents_dir(root: pathlib.Path) -> pathlib.Path: return _coord_dir(root) / "intents" def _releases_dir(root: pathlib.Path) -> pathlib.Path: return _coord_dir(root) / "releases" def _heartbeats_dir(root: pathlib.Path) -> pathlib.Path: return _coord_dir(root) / "heartbeats" def _ensure_coord_dirs(root: pathlib.Path) -> None: _reservations_dir(root).mkdir(parents=True, exist_ok=True) _intents_dir(root).mkdir(parents=True, exist_ok=True) _releases_dir(root).mkdir(parents=True, exist_ok=True) _heartbeats_dir(root).mkdir(parents=True, exist_ok=True) def _now_utc() -> datetime.datetime: return datetime.datetime.now(datetime.timezone.utc) def _parse_dt(s: str) -> datetime.datetime: return datetime.datetime.fromisoformat(s) # Matches any UUID variant (not just v4) — sufficient for path-safety validation. _UUID_RE = re.compile( r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE, ) def _validate_reservation_id(reservation_id: str) -> None: """Raise ``ValueError`` if *reservation_id* is not a well-formed UUID. Security note ------------- ``reservation_id`` is used to construct file paths inside ``.muse/coordination/``. A valid UUID cannot contain ``/`` or ``..``, which prevents directory traversal attacks where a crafted ID such as ``../../etc/passwd`` would escape the coordination directory. """ if not _UUID_RE.match(str(reservation_id)): raise ValueError( f"reservation_id must be a valid UUID: {reservation_id!r}" ) # --------------------------------------------------------------------------- # Reservation # --------------------------------------------------------------------------- class Reservation: """An advisory lock on a set of symbol addresses.""" def __init__( self, reservation_id: str, run_id: str, branch: str, addresses: list[str], created_at: datetime.datetime, expires_at: datetime.datetime, operation: str | None, ) -> None: self.reservation_id = reservation_id self.run_id = run_id self.branch = branch self.addresses = addresses self.created_at = created_at self.expires_at = expires_at self.operation = operation def is_active(self) -> bool: """Return True if this reservation has not yet expired.""" return _now_utc() < self.expires_at def ttl_remaining_seconds(self) -> float: """Seconds remaining until expiry. Returns a positive float when active, zero or negative when expired. Callers should use :meth:`is_active` for boolean checks; this method is intended for display (e.g. "expires in 42m 17s"). """ return (self.expires_at - _now_utc()).total_seconds() def to_dict(self) -> _ReservationDict: return { "schema_version": _SCHEMA_VERSION, "reservation_id": self.reservation_id, "run_id": self.run_id, "branch": self.branch, "addresses": self.addresses, "created_at": self.created_at.isoformat(), "expires_at": self.expires_at.isoformat(), "operation": self.operation, } @classmethod def from_dict(cls, d: _ReservationDict) -> "Reservation": expires_at_raw = d.get("expires_at") created_at_raw = d.get("created_at") expires_at = _parse_dt(str(expires_at_raw)) if expires_at_raw else _now_utc() created_at = _parse_dt(str(created_at_raw)) if created_at_raw else _now_utc() addrs_raw = d.get("addresses", []) addrs = list(addrs_raw) if isinstance(addrs_raw, list) else [] op_raw = d.get("operation") return cls( reservation_id=str(d.get("reservation_id", "")), run_id=str(d.get("run_id", "")), branch=str(d.get("branch", "")), addresses=addrs, created_at=created_at, expires_at=expires_at, operation=str(op_raw) if op_raw is not None else None, ) def create_reservation( root: pathlib.Path, run_id: str, branch: str, addresses: list[str], ttl_seconds: int = 3600, operation: str | None = None, ) -> Reservation: """Write a new advisory reservation and return the :class:`Reservation` object. The reservation file is written atomically to ``.muse/coordination/reservations/.json`` via :func:`~muse.core.store.write_text_atomic` (mkstemp → fsync → rename). Concurrent calls from separate processes are safe — each call generates a distinct UUID4 filename, so there is no write-write race. The written file is **immutable** after creation. To cancel a reservation call :func:`create_release`; to extend its TTL call :func:`create_heartbeat`. Args: root: Repository root (directory containing ``.muse/``). run_id: Opaque agent/pipeline identifier stored verbatim. The caller is responsible for length-limiting this value before passing it (see ``_MAX_RUN_ID_LEN`` in the CLI). branch: Current branch name, stored for filtering by branch. addresses: Symbol addresses this reservation covers. Stored as-is; the caller is responsible for count-limiting this list. ttl_seconds: How long (in seconds) the reservation is valid. Default 3600 (1 hour). The caller must clamp this to a sensible range before passing it. operation: Optional declared operation type (e.g. ``"rename"``). The CLI restricts this to a known set; the core layer stores whatever string is passed for forward-compatibility. Returns: The newly created :class:`Reservation` with ``reservation_id`` set to a fresh UUID4 and ``expires_at`` set to ``now + ttl_seconds``. """ _ensure_coord_dirs(root) now = _now_utc() res = Reservation( reservation_id=str(_uuid_mod.uuid4()), run_id=run_id, branch=branch, addresses=addresses, created_at=now, expires_at=now + datetime.timedelta(seconds=ttl_seconds), operation=operation, ) path = _reservations_dir(root) / f"{res.reservation_id}.json" write_text_atomic(path, json.dumps(res.to_dict(), indent=2) + "\n") logger.debug("✅ Created reservation %s for %d addresses", res.reservation_id[:8], len(addresses)) return res def load_all_reservations(root: pathlib.Path) -> list[Reservation]: """Load every reservation file from ``.muse/coordination/reservations/``. Returns *all* reservations — active, expired, and released alike. Use :func:`active_reservations` when you only want currently-live ones, or :func:`filter_reservations` for in-memory filtering without additional I/O. Corrupt or unreadable files are skipped with a ``WARNING`` log entry; they never cause the function to raise. This makes the coordination layer resilient to partial writes (e.g. a process killed during an older, non-atomic write path) without sacrificing the valid records. Args: root: Repository root (directory containing ``.muse/``). Returns: List of :class:`Reservation` objects in filesystem-glob order (not sorted). Returns ``[]`` if the reservations directory does not exist yet (fresh repo). """ rdir = _reservations_dir(root) if not rdir.exists(): return [] reservations: list[Reservation] = [] for path in rdir.glob("*.json"): try: raw = json.loads(path.read_text()) reservations.append(Reservation.from_dict(raw)) except (json.JSONDecodeError, KeyError) as exc: logger.warning("⚠️ Corrupt reservation %s: %s", path.name, exc) return reservations def active_reservations(root: pathlib.Path) -> list[Reservation]: """Return reservations that are currently live. A reservation is live when **all** of the following hold: * No release tombstone exists for its ID. * The current time is before its effective expiry — which is ``max(reservation.expires_at, heartbeat.extended_expires_at)`` when a heartbeat file exists for this reservation, or simply ``reservation.expires_at`` otherwise. This is the authoritative liveness check. All swarm-coordination logic should use this function rather than calling ``r.is_active()`` directly, because the bare ``is_active()`` method cannot see releases or heartbeats. """ released = load_released_ids(root) hb_map = load_heartbeat_map(root) now = _now_utc() result: list[Reservation] = [] for r in load_all_reservations(root): if r.reservation_id in released: continue hb = hb_map.get(r.reservation_id) effective_expires = ( max(r.expires_at, hb.extended_expires_at) if hb is not None else r.expires_at ) if now < effective_expires: result.append(r) return result # --------------------------------------------------------------------------- # Intent # --------------------------------------------------------------------------- class Intent: """A declared operational intent extending a reservation. Whereas a :class:`Reservation` says "I will touch these symbols", an ``Intent`` says "I will *rename* src/billing.py::compute_total". This extra specificity allows ``muse forecast`` and ``muse plan-merge`` to classify conflicts by operation type (a rename conflicts differently from a delete) and gives post-mortem tooling a structured audit trail. Intents are write-once records with no TTL — they are permanent until explicitly removed by ``muse coord gc --include-intents``. Attributes ---------- intent_id: UUID uniquely identifying this intent record. reservation_id: UUID of the linked :class:`Reservation`, or ``""`` for a standalone intent (not attached to any reservation). run_id: Agent or pipeline identifier that declared this intent. branch: Branch on which the intent was declared. addresses: List of symbol addresses the operation targets. operation: Declared operation type (e.g. ``"rename"``, ``"delete"``). created_at: UTC timestamp when the intent was written. detail: Optional human-readable description of the intended change. """ def __init__( self, intent_id: str, reservation_id: str, run_id: str, branch: str, addresses: list[str], operation: str, created_at: datetime.datetime, detail: str, ) -> None: self.intent_id = intent_id self.reservation_id = reservation_id self.run_id = run_id self.branch = branch self.addresses = addresses self.operation = operation self.created_at = created_at self.detail = detail def to_dict(self) -> _IntentDict: return { "schema_version": _SCHEMA_VERSION, "intent_id": self.intent_id, "reservation_id": self.reservation_id, "run_id": self.run_id, "branch": self.branch, "addresses": self.addresses, "operation": self.operation, "created_at": self.created_at.isoformat(), "detail": self.detail, } @classmethod def from_dict(cls, d: _IntentDict) -> "Intent": created_raw = d.get("created_at") created_at = _parse_dt(str(created_raw)) if created_raw else _now_utc() addrs_raw = d.get("addresses", []) addrs = list(addrs_raw) if isinstance(addrs_raw, list) else [] return cls( intent_id=str(d.get("intent_id", "")), reservation_id=str(d.get("reservation_id", "")), run_id=str(d.get("run_id", "")), branch=str(d.get("branch", "")), addresses=addrs, operation=str(d.get("operation", "")), created_at=created_at, detail=str(d.get("detail", "")), ) def create_intent( root: pathlib.Path, reservation_id: str, run_id: str, branch: str, addresses: list[str], operation: str, detail: str = "", ) -> Intent: """Write and return a new intent record. Atomically writes ``.muse/coordination/intents/.json`` and returns the populated :class:`Intent` object. Each call produces a unique UUID, so multiple intents for the same address and operation are additive rather than replacing each other. Args: root: Repository root (the directory containing ``.muse/``). reservation_id: UUID of a linked reservation, or ``""`` for a standalone intent. Not validated here — callers should validate before calling (see :func:`~muse.core.coordination._validate_reservation_id`). run_id: Agent or pipeline identifier for the audit trail. branch: Branch on which the intent is being declared. addresses: Symbol addresses the operation targets. operation: One of the recognised operation strings (e.g. ``"rename"``, ``"delete"``). Not validated here — callers are responsible for restricting to the supported set. detail: Optional human-readable description of the intended change. Returns: The written :class:`Intent` with a freshly generated ``intent_id`` and ``created_at`` timestamp. """ _ensure_coord_dirs(root) now = _now_utc() intent = Intent( intent_id=str(_uuid_mod.uuid4()), reservation_id=reservation_id, run_id=run_id, branch=branch, addresses=addresses, operation=operation, created_at=now, detail=detail, ) path = _intents_dir(root) / f"{intent.intent_id}.json" write_text_atomic(path, json.dumps(intent.to_dict(), indent=2) + "\n") logger.debug("✅ Created intent %s (%s)", intent.intent_id[:8], operation) return intent def load_all_intents(root: pathlib.Path) -> list[Intent]: """Load and return every intent record in the repository. Reads ``.muse/coordination/intents/*.json``. Intents have no TTL — they are permanent audit records until explicitly purged by ``muse coord gc --include-intents``. All stored intents are returned regardless of age. Use :func:`filter_intents` to narrow by run_id, branch, operation, or address glob after loading. Corrupt or unreadable files are skipped with a warning rather than raising, so a single damaged file never blocks the rest. Args: root: Repository root (the directory containing ``.muse/``). Returns: A list of :class:`Intent` objects in filesystem iteration order (not sorted). Returns an empty list when the intents directory does not yet exist. """ idir = _intents_dir(root) if not idir.exists(): return [] intents: list[Intent] = [] for path in idir.glob("*.json"): try: raw = json.loads(path.read_text()) intents.append(Intent.from_dict(raw)) except (json.JSONDecodeError, KeyError) as exc: logger.warning("⚠️ Corrupt intent %s: %s", path.name, exc) return intents # --------------------------------------------------------------------------- # Filtering helpers # --------------------------------------------------------------------------- def filter_reservations( reservations: list[Reservation], *, run_id: str | None = None, branch: str | None = None, address_glob: str | None = None, operation: str | None = None, include_expired: bool = False, released_ids: frozenset[str] | None = None, heartbeat_expires: HeartbeatMap | None = None, ) -> list[Reservation]: """Return a filtered subset of *reservations*. All filters are applied with AND semantics — a reservation must satisfy every supplied criterion to be included. Parameters ---------- reservations: Source list, typically from :func:`load_all_reservations`. run_id: Exact-match filter on :attr:`~Reservation.run_id`. branch: Exact-match filter on :attr:`~Reservation.branch`. address_glob: :mod:`fnmatch` glob applied to each address in the reservation. A reservation is included when *any* of its addresses match. Example: ``"billing.py::*"`` matches all symbols in ``billing.py``. This is a string operation — no filesystem access occurs. operation: Exact-match filter on :attr:`~Reservation.operation` (e.g. ``"rename"``, ``"modify"``). ``None`` matches reservations with any operation, including those with no declared operation. include_expired: When ``False`` (default) expired and released reservations are excluded. When ``True`` all reservations pass the liveness check. released_ids: Optional pre-loaded set of released reservation IDs (from :func:`load_released_ids`). When supplied, released reservations are excluded even when they have not yet expired by TTL. Pass this to avoid redundant filesystem I/O when the caller has already loaded the releases directory. heartbeat_expires: Optional mapping of ``reservation_id`` → ``extended_expires_at`` (from :func:`load_heartbeat_map`). When supplied, the effective expiry is ``max(reservation.expires_at, extended_expires_at)``, keeping heartbeat-extended reservations alive past their original TTL. Returns ------- list[Reservation] Filtered list, preserving the order of *reservations*. """ now = _now_utc() result: list[Reservation] = [] for res in reservations: if not include_expired: # Released check. if released_ids is not None and res.reservation_id in released_ids: continue # Effective TTL check (with optional heartbeat extension). hb_exp = ( heartbeat_expires.get(res.reservation_id) if heartbeat_expires is not None else None ) effective_expires = ( max(res.expires_at, hb_exp) if hb_exp is not None else res.expires_at ) if now >= effective_expires: continue if run_id is not None and res.run_id != run_id: continue if branch is not None and res.branch != branch: continue if address_glob is not None: if not any(fnmatch.fnmatch(addr, address_glob) for addr in res.addresses): continue if operation is not None and res.operation != operation: continue result.append(res) return result # --------------------------------------------------------------------------- # Release — write-once tombstone # --------------------------------------------------------------------------- _VALID_REASONS = frozenset({"completed", "cancelled", "superseded"}) class Release: """A write-once tombstone that marks a reservation as no longer active. Once a release record exists for a ``reservation_id``, that reservation is excluded from :func:`active_reservations` regardless of its TTL. Releases are never deleted by normal operation — only :func:`run_coord_gc` removes them after the grace period. """ def __init__( self, reservation_id: str, run_id: str, released_at: datetime.datetime, reason: str, ) -> None: self.reservation_id = reservation_id self.run_id = run_id self.released_at = released_at self.reason = reason def to_dict(self) -> _ReleaseDict: return { "schema_version": _SCHEMA_VERSION, "reservation_id": self.reservation_id, "run_id": self.run_id, "released_at": self.released_at.isoformat(), "reason": self.reason, } @classmethod def from_dict(cls, d: _ReleaseDict) -> "Release": released_raw = d.get("released_at") released_at = _parse_dt(str(released_raw)) if released_raw else _now_utc() return cls( reservation_id=str(d.get("reservation_id", "")), run_id=str(d.get("run_id", "")), released_at=released_at, reason=str(d.get("reason", "completed")), ) def create_release( root: pathlib.Path, reservation_id: str, run_id: str, reason: str = "completed", ) -> Release: """Write and return a release tombstone for *reservation_id*. Parameters ---------- root: Repository root (the directory containing ``.muse/``). reservation_id: The UUID of the reservation being released. Must be a valid UUID — invalid values raise ``ValueError`` before any file I/O occurs. run_id: The agent releasing the reservation (for audit trail). reason: One of ``"completed"``, ``"cancelled"``, or ``"superseded"``. Raises ------ ValueError If *reservation_id* is not a valid UUID or *reason* is not recognised. FileExistsError If a release tombstone already exists for this reservation. Callers should check :func:`load_released_ids` first when idempotency is needed. """ _validate_reservation_id(reservation_id) if reason not in _VALID_REASONS: raise ValueError( f"reason must be one of {sorted(_VALID_REASONS)}: {reason!r}" ) _ensure_coord_dirs(root) path = _releases_dir(root) / f"{reservation_id}.json" # Containment check — belt-and-suspenders after UUID validation. try: path.resolve().relative_to(_releases_dir(root).resolve()) except ValueError: raise ValueError(f"reservation_id produces a path outside the releases directory") if path.exists(): raise FileExistsError( f"reservation {reservation_id!r} is already released" ) now = _now_utc() rel = Release( reservation_id=reservation_id, run_id=run_id, released_at=now, reason=reason, ) write_text_atomic(path, json.dumps(rel.to_dict(), indent=2) + "\n") logger.debug("✅ Released reservation %s (%s)", reservation_id[:8], reason) return rel def load_all_releases(root: pathlib.Path) -> list[Release]: """Load and return every release tombstone in the repository. Reads ``.muse/coordination/releases/*.json``. Corrupt or unreadable files are skipped with a warning rather than raising, so a single damaged file never prevents other records from being read. Args: root: Repository root (the directory containing ``.muse/``). Returns: A list of :class:`Release` objects in filesystem iteration order (not sorted). Returns an empty list when the releases directory does not yet exist. """ rdir = _releases_dir(root) if not rdir.exists(): return [] releases: list[Release] = [] for path in rdir.glob("*.json"): try: raw = json.loads(path.read_text()) releases.append(Release.from_dict(raw)) except (json.JSONDecodeError, KeyError) as exc: logger.warning("⚠️ Corrupt release record %s: %s", path.name, exc) return releases def load_released_ids(root: pathlib.Path) -> frozenset[str]: """Return the set of reservation IDs that have been released. This is a lightweight alternative to :func:`load_all_releases` when the caller only needs to know *whether* a reservation has been released, not the full release metadata. Scans only file stems (no JSON parsing) when the releases directory contains no corrupt files — but falls back to parsing when needed. """ rdir = _releases_dir(root) if not rdir.exists(): return frozenset() # Use file stems as reservation IDs — avoids JSON parsing entirely. return frozenset(p.stem for p in rdir.glob("*.json")) # --------------------------------------------------------------------------- # Heartbeat — mutable keep-alive (one file per reservation, atomically updated) # --------------------------------------------------------------------------- class Heartbeat: """A keep-alive record that extends a reservation's effective TTL. Unlike reservations and releases, heartbeat files are *mutable* — each call to :func:`create_heartbeat` atomically rewrites the single heartbeat file for a reservation. Only the most recent heartbeat has meaning. """ def __init__( self, reservation_id: str, run_id: str, last_beat_at: datetime.datetime, extended_expires_at: datetime.datetime, ) -> None: self.reservation_id = reservation_id self.run_id = run_id self.last_beat_at = last_beat_at self.extended_expires_at = extended_expires_at def to_dict(self) -> _HeartbeatDict: return { "schema_version": _SCHEMA_VERSION, "reservation_id": self.reservation_id, "run_id": self.run_id, "last_beat_at": self.last_beat_at.isoformat(), "extended_expires_at": self.extended_expires_at.isoformat(), } @classmethod def from_dict(cls, d: _HeartbeatDict) -> "Heartbeat": beat_raw = d.get("last_beat_at") exp_raw = d.get("extended_expires_at") last_beat_at = _parse_dt(str(beat_raw)) if beat_raw else _now_utc() extended_expires_at = _parse_dt(str(exp_raw)) if exp_raw else _now_utc() return cls( reservation_id=str(d.get("reservation_id", "")), run_id=str(d.get("run_id", "")), last_beat_at=last_beat_at, extended_expires_at=extended_expires_at, ) def create_heartbeat( root: pathlib.Path, reservation_id: str, run_id: str, extension_seconds: int = 3600, ) -> Heartbeat: """Write or update the heartbeat file for *reservation_id*. The heartbeat file at ``.muse/coordination/heartbeats/.json`` is atomically rewritten with a new ``extended_expires_at`` equal to ``now + extension_seconds``. This extends the reservation's effective TTL (as seen by :func:`active_reservations` and :func:`filter_reservations`) without modifying the immutable reservation record itself. Parameters ---------- root: Repository root. reservation_id: The UUID of the reservation to keep alive. Validated as a UUID before any file I/O. run_id: The agent sending the heartbeat (for audit). extension_seconds: How far into the future the new ``extended_expires_at`` is set (default: 3600 s = 1 h). Raises ------ ValueError If *reservation_id* is not a valid UUID or *extension_seconds* ≤ 0. """ _validate_reservation_id(reservation_id) if extension_seconds <= 0: raise ValueError(f"extension_seconds must be > 0: {extension_seconds!r}") _ensure_coord_dirs(root) path = _heartbeats_dir(root) / f"{reservation_id}.json" try: path.resolve().relative_to(_heartbeats_dir(root).resolve()) except ValueError: raise ValueError("reservation_id produces a path outside the heartbeats directory") now = _now_utc() hb = Heartbeat( reservation_id=reservation_id, run_id=run_id, last_beat_at=now, extended_expires_at=now + datetime.timedelta(seconds=extension_seconds), ) write_text_atomic(path, json.dumps(hb.to_dict(), indent=2) + "\n") logger.debug( "💓 Heartbeat %s → expires %s", reservation_id[:8], hb.extended_expires_at.isoformat()[:19], ) return hb def load_heartbeat_map(root: pathlib.Path) -> HeartbeatRecordMap: """Return a ``reservation_id`` → :class:`Heartbeat` mapping for all live heartbeats. Reads ``.muse/coordination/heartbeats/*.json``. Because heartbeat files are mutable (each :func:`create_heartbeat` call atomically replaces the file), only the most recent heartbeat per reservation ID has meaning. Corrupt or unreadable files are skipped with a warning rather than raising, so a single damaged file never blocks the rest. Args: root: Repository root (the directory containing ``.muse/``). Returns: A dict mapping each ``reservation_id`` to its most recent :class:`Heartbeat`. Returns an empty dict when the heartbeats directory does not yet exist. """ hdir = _heartbeats_dir(root) if not hdir.exists(): return {} hb_map: HeartbeatRecordMap = {} for path in hdir.glob("*.json"): try: raw = json.loads(path.read_text()) hb = Heartbeat.from_dict(raw) hb_map[hb.reservation_id] = hb except (json.JSONDecodeError, KeyError) as exc: logger.warning("⚠️ Corrupt heartbeat %s: %s", path.name, exc) return hb_map # --------------------------------------------------------------------------- # Coordination GC # --------------------------------------------------------------------------- @dataclasses.dataclass class CoordGcResult: """Structured result returned by :func:`run_coord_gc`. All ``*_removed`` counters reflect records that were (or would be, in dry-run mode) deleted. ``*_removed_bytes`` is the sum of ``st_size`` values taken *before* deletion. ``removed_ids`` lists reservation UUIDs whose reservation file was removed (or would be). Attributes ---------- reservations_removed: Count of reservation JSON files deleted. reservations_removed_bytes: Total bytes freed from reservation files. releases_removed: Count of release tombstone files deleted (matched and orphaned). releases_removed_bytes: Total bytes freed from release files. heartbeats_removed: Count of heartbeat files deleted (matched and orphaned). heartbeats_removed_bytes: Total bytes freed from heartbeat files. intents_removed: Count of intent files deleted (only when ``include_intents=True``). intents_removed_bytes: Total bytes freed from intent files. total_removed: Sum of all ``*_removed`` counters. total_removed_bytes: Sum of all ``*_removed_bytes`` counters. elapsed_seconds: Wall-clock time for the full GC pass (monotonic). grace_period_seconds: Grace-period value used for this run (echoed for audit). dry_run: ``True`` when no files were actually deleted. include_intents: ``True`` when intent cleanup was enabled. removed_ids: Reservation UUIDs collected this pass. Populated in dry-run mode too so callers can preview what would be removed. """ reservations_removed: int = 0 reservations_removed_bytes: int = 0 releases_removed: int = 0 releases_removed_bytes: int = 0 heartbeats_removed: int = 0 heartbeats_removed_bytes: int = 0 intents_removed: int = 0 intents_removed_bytes: int = 0 total_removed: int = 0 total_removed_bytes: int = 0 elapsed_seconds: float = 0.0 grace_period_seconds: int = 300 dry_run: bool = False include_intents: bool = False removed_ids: list[str] = dataclasses.field(default_factory=list) def run_coord_gc( root: pathlib.Path, *, dry_run: bool = False, grace_period_seconds: int = 300, include_intents: bool = False, max_intent_age_seconds: int = 604800, # 7 days ) -> CoordGcResult: """Remove stale coordination records from ``.muse/coordination/``. What gets collected ------------------- * **Expired reservations** — TTL exhausted (considering heartbeat extensions), past the grace period. * **Released reservations** — release tombstone present and the release is older than the grace period. * **Corresponding release tombstones** — removed alongside their reservation. * **Corresponding heartbeat files** — removed when the reservation is removed. * **Orphaned heartbeats** — heartbeat file exists but the reservation file does not (e.g. a previous partial GC or corrupted write). * **Orphaned releases** — tombstone exists but the reservation file does not. * **Old intents** (opt-in) — ``created_at`` older than *max_intent_age_seconds* when ``include_intents=True``. What is never collected ----------------------- * Active reservations (not released, effective expiry in the future). * Records within the grace period of expiry or release. Parameters ---------- root: Repository root. dry_run: When ``True``, compute what would be removed but delete nothing. grace_period_seconds: Records expired or released within the last N seconds are skipped. Protects against races with agents that are still reading coordination state. Default: 300 s (5 min). include_intents: When ``True``, intents older than *max_intent_age_seconds* are also collected. Default: ``False`` (intents are permanent audit records unless explicitly cleaned). max_intent_age_seconds: Age threshold for intent cleanup when ``include_intents=True``. Default: 604 800 s (7 days). """ t0 = _time_mod.monotonic() result = CoordGcResult( grace_period_seconds=grace_period_seconds, dry_run=dry_run, include_intents=include_intents, ) now = _now_utc() grace = datetime.timedelta(seconds=grace_period_seconds) # ── Load all coordination state ─────────────────────────────────────────── all_reservations = load_all_reservations(root) released_ids = load_released_ids(root) hb_map = load_heartbeat_map(root) # Build a set of reservation IDs that currently exist on disk. existing_res_ids = {r.reservation_id for r in all_reservations} # Map reservation_id → Release for tombstones that match a reservation. release_map: ReleaseMap = {} for rel in load_all_releases(root): release_map[rel.reservation_id] = rel # ── Determine which reservations to collect ─────────────────────────────── collectable_res_ids: set[str] = set() for res in all_reservations: hb = hb_map.get(res.reservation_id) effective_expires = ( max(res.expires_at, hb.extended_expires_at) if hb is not None else res.expires_at ) if res.reservation_id in released_ids: # Released: collect if release is old enough. rel = release_map.get(res.reservation_id) cutoff = rel.released_at if rel is not None else effective_expires if (now - cutoff) >= grace: collectable_res_ids.add(res.reservation_id) else: # Expired: collect if past grace period. if effective_expires < now and (now - effective_expires) >= grace: collectable_res_ids.add(res.reservation_id) # ── Delete reservation files ────────────────────────────────────────────── # Each stat+unlink pair is wrapped in a try/except FileNotFoundError so # that a concurrent GC pass that already deleted the file does not crash # this one. The counters reflect what *this* pass actually removed. res_dir = _reservations_dir(root) if res_dir.exists(): for path in res_dir.glob("*.json"): if path.stem in collectable_res_ids: try: size = path.stat().st_size except FileNotFoundError: continue # already removed by a concurrent GC pass result.reservations_removed += 1 result.reservations_removed_bytes += size result.removed_ids.append(path.stem) if not dry_run: path.unlink(missing_ok=True) # ── Delete corresponding release tombstones ─────────────────────────────── rel_dir = _releases_dir(root) if rel_dir.exists(): for path in rel_dir.glob("*.json"): stem = path.stem if stem in collectable_res_ids or stem not in existing_res_ids: # Collect if reservation is being removed, or orphaned. try: size = path.stat().st_size except FileNotFoundError: continue result.releases_removed += 1 result.releases_removed_bytes += size if not dry_run: path.unlink(missing_ok=True) # ── Delete corresponding heartbeat files ────────────────────────────────── hb_dir = _heartbeats_dir(root) if hb_dir.exists(): for path in hb_dir.glob("*.json"): stem = path.stem if stem in collectable_res_ids or stem not in existing_res_ids: try: size = path.stat().st_size except FileNotFoundError: continue result.heartbeats_removed += 1 result.heartbeats_removed_bytes += size if not dry_run: path.unlink(missing_ok=True) # ── Optionally collect old intents ──────────────────────────────────────── if include_intents: intent_dir = _intents_dir(root) max_age = datetime.timedelta(seconds=max_intent_age_seconds) if intent_dir.exists(): for path in intent_dir.glob("*.json"): try: raw = json.loads(path.read_text()) created_raw = raw.get("created_at") if not created_raw: continue created_at = _parse_dt(str(created_raw)) if (now - created_at) >= max_age: size = path.stat().st_size result.intents_removed += 1 result.intents_removed_bytes += size if not dry_run: path.unlink(missing_ok=True) except FileNotFoundError: continue # concurrent GC removed it first except (json.JSONDecodeError, KeyError, ValueError): pass # Corrupt intent — skip silently. result.total_removed = ( result.reservations_removed + result.releases_removed + result.heartbeats_removed + result.intents_removed ) result.total_removed_bytes = ( result.reservations_removed_bytes + result.releases_removed_bytes + result.heartbeats_removed_bytes + result.intents_removed_bytes ) result.elapsed_seconds = round(_time_mod.monotonic() - t0, 4) return result # --------------------------------------------------------------------------- # filter_intents (placed after all classes are defined) # --------------------------------------------------------------------------- def filter_intents( intents: list[Intent], *, run_id: str | None = None, branch: str | None = None, address_glob: str | None = None, operation: str | None = None, ) -> list[Intent]: """Return a filtered subset of *intents*. All filters are applied with AND semantics. Parameters ---------- intents: Source list, typically from :func:`load_all_intents`. run_id: Exact-match filter on :attr:`~Intent.run_id`. branch: Exact-match filter on :attr:`~Intent.branch`. address_glob: :mod:`fnmatch` glob applied to each address. An intent is included when *any* of its addresses match. String-only — no filesystem I/O. operation: Exact-match filter on :attr:`~Intent.operation` (e.g. ``"rename"``, ``"delete"``). ``None`` returns all intents regardless of operation. Returns ------- list[Intent] Filtered list, preserving the order of *intents*. """ result: list[Intent] = [] for intent in intents: if run_id is not None and intent.run_id != run_id: continue if branch is not None and intent.branch != branch: continue if address_glob is not None: if not any(fnmatch.fnmatch(addr, address_glob) for addr in intent.addresses): continue if operation is not None and intent.operation != operation: continue result.append(intent) return result