"""``muse coord reserve`` — advisory symbol reservation for parallel agents. Places an advisory lock on one or more symbol addresses. This does NOT block other agents from editing those symbols — it is a coordination signal, not an enforcement mechanism. Other agents can check existing reservations via ``muse forecast`` or ``muse reconcile`` before starting work. Why reservations? ----------------- When millions of agents operate on a codebase simultaneously, merge conflicts are inevitable *if* agents don't communicate intent. Reservations give agents a low-cost way to say "I'm about to touch this function" before they do it, so that: 1. Other agents can check with ``muse forecast`` and re-route if needed. 2. ``muse plan-merge`` can predict conflicts with higher accuracy. 3. ``muse reconcile`` can recommend merge ordering. A reservation expires after ``--ttl`` seconds (default: 1 hour) and is never enforced — the VCS engine ignores them for correctness. They are purely advisory. Dependency ordering ------------------- Pass ``--depends-on RESID`` (repeatable) to declare that this reservation must wait for another reservation to complete (be released or expire) before it starts work. The declared dependencies form a DAG that agents can inspect with ``muse coord dag``. Examples:: # B must wait for A to finish muse coord reserve "billing.py::process" --run-id agent-B \\ --depends-on # C must wait for both A and B muse coord reserve "auth.py::validate" --run-id agent-C \\ --depends-on --depends-on Usage:: muse reserve "src/billing.py::compute_total" --run-id agent-42 muse reserve "src/auth.py::validate_token" "src/auth.py::refresh_token" \\ --run-id pipeline-7 --ttl 7200 muse reserve "src/core.py::hash_content" --op rename --run-id refactor-bot Output (text):: ✅ Reserved 1 address(es) for run-id 'agent-42' Reservation ID: Expires: 2026-03-18T13:00:00 ⚠️ src/billing.py::compute_total already reserved by run-id 'agent-41' (expires 2026-03-18T12:30:00) Output (``--json``):: { "schema_version": "...", "reservation_id": "", "run_id": "agent-42", "branch": "main", "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, "conflicts": [], "depends_on": [], "dependency_error": null } Exit codes:: 0 — reservation created (conflicts are warnings, not errors) 1 — validation error (bad --ttl, bad --op, bad --depends-on UUID, cycle) 3 — internal error (repository not found) Flags: ``--run-id ID`` Agent/pipeline identifier for conflict detection (default: ``"unknown"``). Maximum length: 256 characters. ``--ttl N`` Reservation duration in seconds (default: 3600; range: 1–31 536 000). ``--op OPERATION`` Declared operation type. Must be one of: ``rename``, ``move``, ``modify``, ``extract``, ``delete``. Omit to leave unspecified. ``--depends-on RESID`` UUID of a reservation that must complete before this one starts. May be repeated to declare multiple dependencies. Each ID is validated as a UUID before any file I/O. ``--json`` / ``--format json`` Emit reservation details as JSON on stdout. """ from __future__ import annotations import argparse import json import logging import sys from muse.core.coordination import active_reservations, create_reservation from muse.core.dag import add_dependencies from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.store import read_current_branch from muse.core.validation import clamp_int type _ReserveOut = dict[str, str | int | list[str] | None] type _ConflictMap = dict[str, list[str]] logger = logging.getLogger(__name__) # ── Input constraints ───────────────────────────────────────────────────────── # These limits prevent unbounded file growth and keep reservation records # human-readable. They are generous enough for any real agent workflow. #: Maximum number of symbol addresses allowed in a single reservation call. _MAX_ADDRESSES: int = 1000 #: Maximum byte-length of the ``--run-id`` value. _MAX_RUN_ID_LEN: int = 256 #: Allowed values for ``--op``. Stored verbatim; validated here so the #: coordination layer never receives an arbitrary string from the CLI. _VALID_OPS: frozenset[str] = frozenset( {"rename", "move", "modify", "extract", "delete"} ) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse coord reserve`` subcommand and its flags. Adds the ``reserve`` parser to *subparsers* (the ``coord`` subgroup) and wires its ``func`` default to :func:`run`. All flag defaults and choices are defined here so ``--help`` output is accurate and complete. """ parser = subparsers.add_parser( "reserve", help="Place advisory reservations on symbol addresses.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "addresses", nargs="+", metavar="ADDRESS", help=( "Symbol address(es) to reserve, e.g. " '"src/billing.py::compute_total". ' f"Maximum {_MAX_ADDRESSES} per call." ), ) parser.add_argument( "--run-id", dest="run_id", default="unknown", metavar="ID", help=( "Agent/pipeline identifier used for conflict detection " f"(default: 'unknown'; max {_MAX_RUN_ID_LEN} chars)." ), ) parser.add_argument( "--ttl", type=int, default=3600, metavar="SECONDS", help="Reservation duration in seconds (default: 3600; range: 1–31 536 000).", ) parser.add_argument( "--op", dest="operation", default=None, choices=sorted(_VALID_OPS), metavar="OPERATION", help=( "Declared operation type — one of: " + ", ".join(sorted(_VALID_OPS)) + ". Omit to leave unspecified." ), ) parser.add_argument( "--depends-on", dest="depends_on", action="append", default=[], metavar="RESID", help=( "UUID of a reservation that must complete before this one starts. " "May be repeated: --depends-on A --depends-on B. " "Each value is validated as a UUID before any file I/O." ), ) 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) def run(args: argparse.Namespace) -> None: """Place advisory reservations on one or more symbol addresses. Reservations are write-once, expiry-based advisory signals. They do not block other agents or affect VCS correctness — they enable conflict *prediction* via ``muse forecast`` and ``muse reconcile``. Execution order --------------- 1. **Validate inputs** — ``--ttl`` range, ``--run-id`` length, address count, ``--op`` membership (enforced by argparse ``choices``), and ``--depends-on`` UUID syntax. Any validation failure exits 1 with a message to *stderr* before any file I/O. 2. **Conflict check** — calls :func:`~muse.core.coordination.active_reservations` to identify existing reservations by other agents on the same addresses. Conflicts are *warnings*, not errors — the reservation is always created. 3. **Create reservation** — writes ``.muse/coordination/reservations/.json`` atomically via :func:`~muse.core.store.write_text_atomic`. 4. **Record dependency edges** (optional) — if ``--depends-on`` IDs are supplied, calls :func:`~muse.core.dag.add_dependencies` to write the DAG record. If the new edges would create a cycle the command exits 1, but **the reservation itself is not rolled back** (reservations are advisory and immutable; the DAG is best-effort). 5. **Emit output** — text to *stdout* (or JSON when ``--format json``). Security -------- * ``--ttl`` is clamped to ``[1, 31_536_000]`` by :func:`~muse.core.validation.clamp_int`. * ``--run-id`` is truncated to :data:`_MAX_RUN_ID_LEN` bytes to prevent arbitrarily large reservation files. * ``addresses`` count is capped at :data:`_MAX_ADDRESSES`. * ``--op`` is restricted to :data:`_VALID_OPS` by argparse ``choices``. * Every ``--depends-on`` UUID is validated by :func:`~muse.core.dag.add_dependencies` before any path is constructed, preventing directory traversal. Args: args: Parsed ``argparse.Namespace`` with attributes ``addresses``, ``run_id``, ``ttl``, ``operation``, ``depends_on``, and ``fmt``. Exit codes: 0 — reservation created successfully (conflicts are warnings). 1 — validation error or dependency cycle detected. 3 — repository not found (``require_repo`` raises ``SystemExit``). """ addresses: list[str] = args.addresses run_id: str = args.run_id operation: str | None = args.operation as_json: bool = args.fmt == "json" depends_on: list[str] = args.depends_on or [] # ── 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 len(addresses) > _MAX_ADDRESSES: print( f"❌ Too many addresses: {len(addresses)} (max {_MAX_ADDRESSES}).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) try: ttl: int = clamp_int(args.ttl, 1, 31_536_000, "ttl") except ValueError as exc: print(f"❌ Invalid --ttl: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from exc root = require_repo() branch = read_current_branch(root) # ── Conflict check ──────────────────────────────────────────────────────── # Build an address → conflicting-reservation map in a single pass over the # active reservations list, instead of an O(addresses × reservations) nested # loop. This keeps conflict detection O(active_reservations + addresses) # even when both are large. existing = active_reservations(root) addr_to_conflicts: _ConflictMap = {} for res in existing: if res.run_id == run_id: continue for addr in res.addresses: if addr in set(addresses): addr_to_conflicts.setdefault(addr, []).append( f" ⚠️ {addr}\n" f" already reserved by run-id {res.run_id!r}" f" (expires {res.expires_at.isoformat()[:19]})" ) conflicts: list[str] = [msg for msgs in addr_to_conflicts.values() for msg in msgs] # ── Create reservation ──────────────────────────────────────────────────── res = create_reservation(root, run_id, branch, addresses, ttl, operation) # ── Record dependency edges ─────────────────────────────────────────────── dep_record = None dep_error: str | None = None if depends_on: try: dep_record = add_dependencies(root, res.reservation_id, depends_on) except (ValueError, FileExistsError) as exc: dep_error = str(exc) logger.warning( "Failed to record dependencies for %s: %s", res.reservation_id[:8], exc, ) # ── Output ──────────────────────────────────────────────────────────────── if as_json: out: _ReserveOut = { **res.to_dict(), "conflicts": conflicts, "depends_on": dep_record.depends_on if dep_record else [], "dependency_error": dep_error, } print(json.dumps(out)) if dep_error: raise SystemExit(ExitCode.USER_ERROR) return if conflicts: for c in conflicts: print(c) print( f"\n✅ Reserved {len(addresses)} address(es) for run-id {run_id!r}\n" f" Reservation ID: {res.reservation_id}\n" f" Expires: {res.expires_at.isoformat()[:19]}" ) if operation: print(f" Operation: {operation}") if dep_record and dep_record.depends_on: print(f" Depends on ({len(dep_record.depends_on)}):") for dep_id in dep_record.depends_on: print(f" {dep_id[:8]}…") if conflicts: print( f"\n ⚠️ {len(conflicts)} conflict(s) detected. " "Run 'muse forecast' for details." ) if dep_error: print(f"\n ⚠️ Dependency error: {dep_error}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR)