"""``muse coord release`` — mark a reservation as done and free its addresses. Writing a release tombstone tells the swarm that the agent has finished (or abandoned) its work on the reserved addresses. Once released, the reservation is excluded from :func:`~muse.core.coordination.active_reservations` immediately — no need to wait for the TTL to expire. This is the cooperative handshake that prevents the coordination layer from becoming stale. Usage:: muse coord release --run-id AGENT-42 muse coord release --run-id AGENT-42 --reason cancelled muse coord release --run-id AGENT-42 --reason superseded muse coord release --all-for-run AGENT-42 --run-id AGENT-42 muse coord release --run-id AGENT-42 --json Reasons:: completed Work finished successfully (default). cancelled Work was abandoned before completion. superseded A newer reservation replaced this one. JSON output schema (single-release mode):: { "status": "released" | "already_released" | "not_found", "reservation_id": str, "run_id": str, "released_at": str, // ISO 8601; null when already_released "reason": str, "elapsed_seconds": float } JSON output schema (``--all-for-run`` mode):: { "status": "ok", "released": [{"reservation_id": str, "run_id": str, "released_at": str, "reason": str}, ...], "skipped_already_released": int, "elapsed_seconds": float } Exit codes:: 0 — released (or already released — idempotent) 1 — bad arguments (mutual exclusion, missing required flag, run-id too long) 4 — reservation not found Flags: ``--run-id ID`` Agent/pipeline identifier for the audit trail (required). Maximum length: 256 characters. ``--reason REASON`` Why the reservation is being released. One of ``completed`` (default), ``cancelled``, or ``superseded``. ``--all-for-run RUN_ID`` Release every active reservation whose ``run_id`` matches *RUN_ID*. Cannot be combined with a positional ``RESERVATION_ID``. ``--json`` / ``--format json`` Emit result as compact JSON on stdout. """ from __future__ import annotations import argparse import json import pathlib import sys import time from muse.core.coordination import ( _validate_reservation_id, create_release, load_all_reservations, load_released_ids, ) from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.validation import sanitize_display # ── Input constraints ───────────────────────────────────────────────────────── #: Maximum byte-length of the ``--run-id`` value. Mirrors the limit in #: ``reserve.py`` so audit records stay bounded in size. _MAX_RUN_ID_LEN: int = 256 # ── CLI registration ────────────────────────────────────────────────────────── def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``release`` subcommand on *subparsers* (under ``muse coord``). Wires all flags with their defaults, choices, and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run`. """ parser = subparsers.add_parser( "release", help="Mark a reservation as done and free its addresses.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "reservation_id", nargs="?", default=None, metavar="RESERVATION_ID", help="UUID of the reservation to release.", ) parser.add_argument( "--run-id", required=True, dest="run_id", metavar="ID", help=( "Agent run-id performing the release (for audit trail). " f"Max {_MAX_RUN_ID_LEN} characters." ), ) parser.add_argument( "--reason", default="completed", choices=("completed", "cancelled", "superseded"), help="Why the reservation is being released (default: completed).", ) parser.add_argument( "--all-for-run", default=None, dest="all_for_run", metavar="RUN_ID", help=( "Release all active reservations for this run-id. Cannot be " "combined with a positional RESERVATION_ID." ), ) parser.add_argument( "--format", "-f", default="text", dest="fmt", choices=("text", "json"), help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.set_defaults(func=run) # ── Command implementation ──────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Release one or all reservations for an agent run. Execution order --------------- 1. **Validate inputs** — ``--run-id`` length, mutual exclusion of ``RESERVATION_ID`` and ``--all-for-run``, UUID syntax of ``RESERVATION_ID`` (single mode only). Any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message to *stderr* before any file I/O. 2. **Dispatch** — single mode calls :func:`_run_single`; batch mode calls :func:`_run_batch`. Single-reservation mode ----------------------- ``reservation_id`` (positional) is required. If the reservation is already released, a warning is printed and exit code 0 is returned (idempotent). If the reservation does not exist, exit code :attr:`~muse.core.errors.ExitCode.NOT_FOUND` (4) is returned. Batch mode (``--all-for-run``) -------------------------------- Releases every reservation whose ``run_id`` matches *all_for_run*. Already- released reservations are counted and reported but do not cause a non-zero exit. Security -------- * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` bytes to bound audit record size. * ``reservation_id`` is validated as a UUID before any path is constructed, preventing directory traversal attacks. * All display strings are passed through :func:`~muse.core.validation.sanitize_display`. Args: args: Parsed ``argparse.Namespace`` with attributes ``reservation_id``, ``run_id``, ``reason``, ``all_for_run``, and ``fmt``. Exit codes: 0 — released successfully (or already released — idempotent). 1 — bad arguments or ``--run-id`` too long. 4 — reservation not found (single mode only). """ t0 = time.monotonic() reservation_id: str | None = args.reservation_id run_id: str = args.run_id reason: str = args.reason all_for_run: str | None = args.all_for_run fmt: str = args.fmt as_json = fmt == "json" # ── Input validation (before any file I/O) ──────────────────────────────── if len(run_id) > _MAX_RUN_ID_LEN: print( f"❌ --run-id is too long ({len(run_id)} chars; max {_MAX_RUN_ID_LEN}).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if all_for_run is not None and reservation_id is not None: print( "❌ cannot specify both RESERVATION_ID and --all-for-run", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if all_for_run is None and reservation_id is None: print( "❌ must specify either RESERVATION_ID or --all-for-run", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # UUID validation for single mode — before require_repo() so we never # hit the filesystem with an untrusted string. if reservation_id is not None: try: _validate_reservation_id(reservation_id) except ValueError as exc: if as_json: print(json.dumps({"error": str(exc), "status": "bad_id"})) else: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() if all_for_run is not None: _run_batch(root, all_for_run, run_id, reason, as_json, t0) else: assert reservation_id is not None _run_single(root, reservation_id, run_id, reason, as_json, t0) def _run_single( root: pathlib.Path, reservation_id: str, run_id: str, reason: str, as_json: bool, t0: float, ) -> None: """Release a single reservation by UUID. Checks the released-IDs set first (stem-only scan, no JSON parsing), then the full reservation list. Writes a release tombstone atomically via :func:`~muse.core.store.write_text_atomic`. Args: root: Repository root (the directory containing ``.muse/``). reservation_id: Pre-validated UUID string. run_id: Agent performing the release (written to the tombstone). reason: One of ``"completed"``, ``"cancelled"``, or ``"superseded"``. as_json: When ``True``, emit compact JSON instead of human text. t0: :func:`time.monotonic` start timestamp used for ``elapsed_seconds``. Exit codes: 0 — released (or already released — idempotent). 4 — reservation not found. """ # Check if already released (fast path — no JSON parsing). released_ids = load_released_ids(root) if reservation_id in released_ids: elapsed = round(time.monotonic() - t0, 4) if as_json: print(json.dumps({ "status": "already_released", "reservation_id": reservation_id, "run_id": run_id, "released_at": None, "reason": reason, "elapsed_seconds": elapsed, })) else: rid = sanitize_display(reservation_id) print(f"reservation {rid} is already released — nothing to do") raise SystemExit(ExitCode.SUCCESS) # Check the reservation exists. all_res = load_all_reservations(root) known_ids = {r.reservation_id for r in all_res} if reservation_id not in known_ids: elapsed = round(time.monotonic() - t0, 4) if as_json: print(json.dumps({ "status": "not_found", "reservation_id": reservation_id, "elapsed_seconds": elapsed, })) else: rid = sanitize_display(reservation_id) print(f"❌ reservation {rid} not found", file=sys.stderr) raise SystemExit(ExitCode.NOT_FOUND) # Create the release tombstone. try: rel = create_release(root, reservation_id, run_id, reason) except FileExistsError: # Race: another agent released between our check and write — idempotent. elapsed = round(time.monotonic() - t0, 4) if as_json: print(json.dumps({ "status": "already_released", "reservation_id": reservation_id, "run_id": run_id, "released_at": None, "reason": reason, "elapsed_seconds": elapsed, })) else: rid = sanitize_display(reservation_id) print(f"reservation {rid} is already released (race) — nothing to do") raise SystemExit(ExitCode.SUCCESS) elapsed = round(time.monotonic() - t0, 4) if as_json: print(json.dumps({ "status": "released", "reservation_id": rel.reservation_id, "run_id": rel.run_id, "released_at": rel.released_at.isoformat(), "reason": rel.reason, "elapsed_seconds": elapsed, })) else: rid = sanitize_display(rel.reservation_id) run_label = sanitize_display(rel.run_id) print( f"✅ released {rid} run={run_label}" f" reason={rel.reason} at={rel.released_at.isoformat()[:19]}Z" ) def _run_batch( root: pathlib.Path, all_for_run: str, run_id: str, reason: str, as_json: bool, t0: float, ) -> None: """Release all active reservations for a given run-id. Loads the full reservation list and released-ID set exactly once, then iterates in memory — O(reservations) with no per-record filesystem I/O beyond the initial directory scan. ``skipped_already_released`` counts reservations that were already released before this call (checked via the released-IDs set). ``FileExistsError`` from concurrent races is folded into the same counter so the caller always gets a complete accounting. Args: root: Repository root (the directory containing ``.muse/``). all_for_run: Only release reservations whose ``run_id`` matches this. run_id: Agent performing the release (written to each tombstone). reason: One of ``"completed"``, ``"cancelled"``, or ``"superseded"``. as_json: When ``True``, emit compact JSON instead of human text. t0: :func:`time.monotonic` start timestamp used for ``elapsed_seconds``. """ all_res = load_all_reservations(root) released_ids = load_released_ids(root) targets = [r for r in all_res if r.run_id == all_for_run] released_entries = [] skipped_already = 0 for res in targets: if res.reservation_id in released_ids: skipped_already += 1 continue try: rel = create_release(root, res.reservation_id, run_id, reason) released_entries.append({ "reservation_id": rel.reservation_id, "run_id": rel.run_id, "released_at": rel.released_at.isoformat(), "reason": rel.reason, }) except (FileExistsError, ValueError): # FileExistsError: concurrent race — another agent beat us. # ValueError: corrupt reservation_id that passed UUID storage # but failed re-validation (defensive, should not happen). skipped_already += 1 elapsed = round(time.monotonic() - t0, 4) if as_json: print(json.dumps({ "status": "ok", "released": released_entries, "skipped_already_released": skipped_already, "elapsed_seconds": elapsed, })) else: n = len(released_entries) run_label = sanitize_display(all_for_run) print(f"✅ released {n} reservation(s) for run={run_label} reason={reason}") if skipped_already: print(f" {skipped_already} already released — skipped") for entry in released_entries: rid = sanitize_display(entry["reservation_id"]) print(f" {rid}") print(f"\n ({elapsed:.3f}s)")