"""``muse coord intent`` — declare a specific operation before executing it. Records a structured intent extending an existing reservation. Whereas ``muse reserve`` says "I will touch these symbols", ``muse coord intent`` says "I will rename src/billing.py::compute_total to compute_invoice_total". This additional detail enables: * ``muse forecast`` to compute more precise conflict predictions (a rename conflicts differently with a delete than a modify does). * ``muse plan-merge`` to classify conflicts by a semantic taxonomy. * Audit trail of what each agent intended before committing. Usage:: muse coord intent "src/billing.py::compute_total" \\ --op rename --detail "rename to compute_invoice_total" \\ --reservation-id muse coord intent "src/auth.py::validate_token" \\ --op extract --detail "extract into src/auth/validators.py" \\ --run-id agent-42 muse coord intent "src/core.py::hash_content" --op delete --run-id refactor-bot JSON output schema:: { "schema_version": str, "intent_id": str, // UUID "reservation_id": str, // UUID or "" if standalone "run_id": str, "branch": str, "addresses": [str, ...], "operation": str, "created_at": str, // ISO 8601 "detail": str } Exit codes:: 0 — intent recorded successfully 1 — bad arguments (unknown --op, --run-id too long, --detail too long, too many addresses, --reservation-id not a valid UUID) Flags: ``--op OPERATION`` Required. The operation being declared: delete | extract | inline | merge | modify | move | rename | split. ``--detail TEXT`` Human-readable description of the intended change. Maximum length: 4096 characters. ``--reservation-id UUID`` Link to an existing reservation (optional; creates a standalone intent if omitted). When provided, must be a valid UUID. ``--run-id ID`` Agent identifier (used when --reservation-id is not provided). Maximum length: 256 characters. ``--json`` / ``--format json`` Emit intent details as compact JSON on stdout. """ from __future__ import annotations import argparse import json import logging import sys from muse.core.coordination import _validate_reservation_id, create_intent 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 sanitize_display logger = logging.getLogger(__name__) # ── Input constraints ───────────────────────────────────────────────────────── #: Allowed values for ``--op``. Intent supports more granular operations #: than ``reserve`` (which only tracks which symbols will change). _VALID_OPS: frozenset[str] = frozenset({ "rename", "move", "modify", "extract", "delete", "inline", "split", "merge", }) #: Maximum byte-length of the ``--run-id`` value. Mirrors all other #: coordination commands so audit records stay bounded in size. _MAX_RUN_ID_LEN: int = 256 #: Maximum character-length of the ``--detail`` value. Bounds intent file #: sizes and prevents pathological memory use when loading large intent sets. _MAX_DETAIL_LEN: int = 4096 #: Maximum number of symbol addresses per intent call. Mirrors reserve.py. _MAX_ADDRESSES: int = 1000 # ── CLI registration ────────────────────────────────────────────────────────── def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``intent`` 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( "intent", help="Declare a specific operation before executing it.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "addresses", nargs="+", metavar="ADDRESS", help=( "Symbol address(es) this intent applies to. " f"Maximum {_MAX_ADDRESSES} per call." ), ) parser.add_argument( "--op", dest="operation", required=True, choices=sorted(_VALID_OPS), metavar="OPERATION", help=( "Operation to declare — one of: " + ", ".join(sorted(_VALID_OPS)) + "." ), ) parser.add_argument( "--detail", default="", metavar="TEXT", help=( "Human-readable description of the intended change. " f"Maximum {_MAX_DETAIL_LEN} characters." ), ) parser.add_argument( "--reservation-id", dest="reservation_id", default="", metavar="UUID", help=( "Link to an existing reservation. When provided, must be a valid " "UUID. Omit to create a standalone intent." ), ) parser.add_argument( "--run-id", dest="run_id", default="unknown", metavar="ID", help=( "Agent identifier (used when --reservation-id is not provided). " f"Maximum {_MAX_RUN_ID_LEN} characters." ), ) 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: """Declare a specific operation before executing it. Execution order --------------- 1. **Validate inputs** — ``--run-id`` length, ``--detail`` length, address count, and ``--reservation-id`` UUID syntax (when non-empty). All validation fires before any file I/O; any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message to *stderr*. ``--op`` is validated by argparse ``choices`` before this function is called. 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo` and reads the current branch from the ref store. 3. **Write intent** — calls :func:`~muse.core.coordination.create_intent`, which atomically writes ``.muse/coordination/intents/.json``. 4. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. What intents enable ------------------- ``muse coord intent`` extends a reservation with operational detail. The operation type enables ``muse forecast`` to compute more precise conflict predictions — a rename conflicts differently from a delete. Intents are write-once audit records and never affect VCS correctness. Security -------- * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` to bound audit record size. * ``--detail`` is capped at :data:`_MAX_DETAIL_LEN` to bound intent file sizes and memory use when loading large intent sets. * Address count is capped at :data:`_MAX_ADDRESSES`. * ``--reservation-id``, when non-empty, is validated as a UUID before any path construction (even though intent files reference, not construct, the reservation path). * All display strings in text output are passed through :func:`~muse.core.validation.sanitize_display`. Args: args: Parsed ``argparse.Namespace`` with attributes ``addresses``, ``operation``, ``detail``, ``reservation_id``, ``run_id``, and ``fmt``. Exit codes: 0 — intent recorded successfully. 1 — bad arguments. """ addresses: list[str] = args.addresses operation: str = args.operation detail: str = args.detail reservation_id: str = args.reservation_id run_id: str = args.run_id as_json: bool = args.fmt == "json" # ── Input validation (before any file I/O) ──────────────────────────────── if len(run_id) > _MAX_RUN_ID_LEN: msg = f"--run-id is too long ({len(run_id)} chars; max {_MAX_RUN_ID_LEN})" if as_json: print(json.dumps({"error": msg, "status": "bad_args"})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(detail) > _MAX_DETAIL_LEN: msg = f"--detail is too long ({len(detail)} chars; max {_MAX_DETAIL_LEN})" if as_json: print(json.dumps({"error": msg, "status": "bad_args"})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(addresses) > _MAX_ADDRESSES: msg = f"Too many addresses: {len(addresses)} (max {_MAX_ADDRESSES})" if as_json: print(json.dumps({"error": msg, "status": "bad_args"})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if reservation_id: try: _validate_reservation_id(reservation_id) except ValueError as exc: msg = str(exc) if as_json: print(json.dumps({"error": msg, "status": "bad_reservation_id"})) else: print(f"❌ --reservation-id: {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() branch = read_current_branch(root) intent_record = create_intent( root=root, reservation_id=reservation_id, run_id=run_id, branch=branch, addresses=addresses, operation=operation, detail=detail, ) if as_json: print(json.dumps(intent_record.to_dict())) return safe_intent_id = sanitize_display(intent_record.intent_id) safe_run_id = sanitize_display(intent_record.run_id) print( f"\n✅ Intent recorded\n" f" Intent ID: {safe_intent_id}\n" f" Operation: {operation}\n" f" Addresses: {len(addresses)}\n" f" Run ID: {safe_run_id}" ) if detail: print(f" Detail: {sanitize_display(detail)}") if reservation_id: print(f" Reservation: {sanitize_display(reservation_id)}") print("\nRun 'muse forecast' to check for predicted conflicts.")