"""``muse coord reconcile`` — recommend merge ordering and integration strategy. Reads active reservations, intents, and branch divergence to recommend: 1. **Merge ordering** — which branches should be merged first to minimize downstream conflicts. 2. **Integration strategy** — fast-forward, squash, or rebase for each branch. 3. **Conflict hotspots** — symbols reserved by multiple agents that need special attention. ``muse coord reconcile`` is a *read-only* planning command. It does not write to branches, commit history, or the coordination layer. It provides the plan; agents execute it. Why this exists --------------- In a system with millions of concurrent agents, merges happen constantly. Without coordination, every merge introduces friction. ``muse coord reconcile`` gives an orchestration agent a complete picture of the current coordination state and a recommended action plan. Usage:: muse coord reconcile muse coord reconcile --json Output (text):: Reconciliation report ────────────────────────────────────────────────────────────── Active reservations: 3 Active intents: 2 Conflict hotspots: 1 Recommended merge order: 1. feature/billing (3 addresses, 0 conflict(s)) 2. feature/auth (5 addresses, 1 conflict(s)) Conflict hotspot(s): src/billing.py::compute_total reserved by: agent-41, agent-42 → resolve 'feature/billing' first; feature/auth must rebase Integration strategies: feature/billing → fast-forward (no conflicts predicted) feature/auth → rebase onto main before merging (0.001s) JSON output schema:: { "schema_version": str, "active_reservations": int, "active_intents": int, "conflict_hotspots": int, "branches": [ { "branch": str, "reserved_addresses": [str, ...], "intents": [str, ...], "run_ids": [str, ...], "predicted_conflicts": int }, ... ], "recommended_merge_order": [str, ...], "strategies": {branch: str, ...}, "hotspots": [ {"address": str, "branches": [str, ...]} ], "elapsed_seconds": float } Exit codes:: 0 — success (no active data is also success) 1 — unexpected error loading coordination state Flags: ``--json`` / ``--format json`` Emit the reconciliation report as compact JSON on stdout. """ from __future__ import annotations import argparse import json import logging import sys import time from muse._version import __version__ from muse.core._types import Metadata from muse.core.coordination import active_reservations, load_all_intents from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) type _ReconcileDict = dict[str, str | int | list[str]] type _BranchMap = dict[str, "_BranchSummary"] type _AddrBranchMap = dict[str, list[str]] # ── Error helper ────────────────────────────────────────────────────────────── def _err(msg: str, as_json: bool, status: str = "error") -> None: """Print an error and return. Caller raises SystemExit.""" if as_json: print(json.dumps({"error": msg, "status": status})) else: print(f"❌ {msg}", file=sys.stderr) # ── Internal types ──────────────────────────────────────────────────────────── class _BranchSummary: """Aggregated coordination state for a single branch. Collects reserved addresses, declared intents, participating agent run-IDs, and a predicted conflict count (populated after hotspot detection). Attributes: branch: Branch name this summary covers. reserved_addresses: Symbol addresses reserved on this branch. intents: Operation names declared via ``muse coord intent``. run_ids: Set of agent run-IDs that have reservations or intents here. conflict_count: Number of hotspot addresses this branch participates in. """ def __init__(self, branch: str) -> None: self.branch = branch self.reserved_addresses: list[str] = [] self.intents: list[str] = [] self.run_ids: set[str] = set() self.conflict_count: int = 0 def to_dict(self) -> _ReconcileDict: """Serialise to a plain dict suitable for JSON output.""" return { "branch": self.branch, "reserved_addresses": self.reserved_addresses, "intents": self.intents, "run_ids": sorted(self.run_ids), "predicted_conflicts": self.conflict_count, } # ── CLI registration ────────────────────────────────────────────────────────── def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``reconcile`` 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( "reconcile", help="Recommend merge ordering and integration strategy.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) 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: """Recommend merge ordering and integration strategy. Reads coordination state (reservations + intents) and produces a recommended action plan: which branches to merge first, what strategy to use, and which conflict hotspots need manual attention. Execution order --------------- 1. **Resolve repo** — :func:`~muse.core.repo.require_repo`. 2. **Load state** — :func:`~muse.core.coordination.active_reservations` and :func:`~muse.core.coordination.load_all_intents`. Any unexpected I/O error exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message on *stderr* (or compact JSON on *stdout* when ``--format json``). 3. **Aggregate by branch** — build a :class:`_BranchSummary` per branch. 4. **Detect hotspots** — addresses reserved across >1 branch. 5. **Score branches** — increment ``conflict_count`` for each hotspot participation. 6. **Order branches** — ascending by ``(conflict_count, address_count)`` so cleanest branches merge first. 7. **Assign strategies** — fast-forward (0 conflicts), rebase (1–2), or manual (3+). 8. **Emit output** — compact JSON or human-readable text. This command is *read-only*. It never writes to branches, commit history, or the coordination layer. Security -------- * All branch names, addresses, and run-IDs in text output are passed through :func:`~muse.core.validation.sanitize_display` to prevent ANSI injection. * No user-supplied path strings are used to construct file paths. Performance ----------- * O(R + I) to load reservations and intents (one directory scan each). * O(A) to build the hotspot map where A is the total number of addresses. * O(B log B) to sort branches where B is the number of distinct branches. Args: args: Parsed ``argparse.Namespace`` with attribute ``fmt``. Exit codes: 0 — success (no active data is also success). 1 — unexpected error loading coordination state. """ as_json: bool = args.fmt == "json" t0 = time.monotonic() root = require_repo() try: reservations = active_reservations(root) except Exception as exc: # noqa: BLE001 _err(str(exc), as_json, "load_error") raise SystemExit(ExitCode.USER_ERROR) try: intents = load_all_intents(root) except Exception as exc: # noqa: BLE001 _err(str(exc), as_json, "load_error") raise SystemExit(ExitCode.USER_ERROR) # Aggregate by branch. branch_map: _BranchMap = {} for res in reservations: b = res.branch if b not in branch_map: branch_map[b] = _BranchSummary(b) branch_map[b].reserved_addresses.extend(res.addresses) branch_map[b].run_ids.add(res.run_id) for it in intents: b = it.branch if b not in branch_map: branch_map[b] = _BranchSummary(b) branch_map[b].intents.append(it.operation) branch_map[b].run_ids.add(it.run_id) # Detect conflict hotspots. addr_branches: _AddrBranchMap = {} for res in reservations: for addr in res.addresses: addr_branches.setdefault(addr, []).append(res.branch) hotspots: _AddrBranchMap = { addr: branches for addr, branches in addr_branches.items() if len(set(branches)) > 1 } # Compute conflict counts per branch based on hotspot participation. for addr, branches in hotspots.items(): unique_branches = list(dict.fromkeys(branches)) for b in unique_branches: if b in branch_map: branch_map[b].conflict_count += 1 # Recommend merge order: fewer conflicts → merge first. ordered = sorted( branch_map.values(), key=lambda bs: (bs.conflict_count, len(bs.reserved_addresses)), ) # Recommend integration strategies. strategies: Metadata = {} for bs in ordered: if bs.conflict_count == 0: strategies[bs.branch] = "fast-forward (no conflicts predicted)" elif bs.conflict_count <= 2: strategies[bs.branch] = "rebase onto main before merging" else: strategies[bs.branch] = "manual conflict resolution required" elapsed = round(time.monotonic() - t0, 4) if as_json: print(json.dumps({ "schema_version": __version__, "active_reservations": len(reservations), "active_intents": len(intents), "conflict_hotspots": len(hotspots), "branches": [bs.to_dict() for bs in ordered], "recommended_merge_order": [bs.branch for bs in ordered], "strategies": strategies, "hotspots": [ {"address": addr, "branches": list(dict.fromkeys(brs))} for addr, brs in sorted(hotspots.items()) ], "elapsed_seconds": elapsed, })) return # ── Text output ─────────────────────────────────────────────────────────── print("\nReconciliation report") print("─" * 62) print( f" Active reservations: {len(reservations)} " f"Active intents: {len(intents)} " f"Conflict hotspots: {len(hotspots)}" ) if not reservations and not intents: print( "\n (no active coordination data — run 'muse reserve' or 'muse intent' first)" ) return if ordered: print(f"\n Recommended merge order:") for rank, bs in enumerate(ordered, 1): c = bs.conflict_count print( f" {rank}. {sanitize_display(bs.branch):<30} " f"({len(bs.reserved_addresses)} addresses, {c} conflict(s))" ) if hotspots: print(f"\n Conflict hotspot(s):") for addr, branches in sorted(hotspots.items()): unique = list(dict.fromkeys(branches)) print(f" {sanitize_display(addr)}") print(f" reserved by: {', '.join(sanitize_display(b) for b in unique)}") first = unique[0] rest = ", ".join(sanitize_display(b) for b in unique[1:]) print(f" → resolve {sanitize_display(first)!r} first; {rest} must rebase") print(f"\n Integration strategies:") for bs in ordered: print(f" {sanitize_display(bs.branch):<30} → {strategies[bs.branch]}") print(f"\n ({elapsed:.3f}s)")