"""``muse coord forecast`` — predict merge conflicts before they happen. Reads all active reservations and intents across branches, then uses the reverse call graph to compute *likely* conflicts before any code is written. This turns merge conflict resolution from a reactive ("it broke") problem into a proactive ("we predicted it") workflow — essential when many agents operate on a codebase simultaneously. Conflict types detected ----------------------- ``address_overlap`` Two agents have reserved the same symbol address. Direct collision. Confidence: **1.00** — this will definitely conflict. ``blast_radius_overlap`` Agent A's reserved symbol is in the call chain of Agent B's target, or vice versa. A change to A's symbol will affect B's symbol. Requires the code index to be built (``muse code index``). Confidence: **0.75** — likely to conflict. ``operation_conflict`` Agent A intends to delete/rename a symbol that Agent B intends to modify. Classic use-after-free / use-after-rename semantic conflict. Confidence: **0.90** — will almost certainly conflict. Incomplete forecasts -------------------- When the code index has not been built, or when the index cannot be read (e.g. the repo has no commits yet), blast-radius analysis is skipped. The output includes a ``warnings`` field (JSON) or a visible notice (text) so you know the forecast is partial. **A partial forecast is still reported** — it is never silently presented as complete. Agents should check ``partial_forecast`` (JSON) to gate on whether the full analysis ran. Usage:: muse coord forecast muse coord forecast --branch feature-x muse coord forecast --run-id agent-42 muse coord forecast --min-confidence 0.9 muse coord forecast --format json muse coord forecast --json JSON output schema:: { "schema_version": str, "current_branch": str, "branch_filter": str | null, "run_id_filter": str | null, "min_confidence": float, "active_reservations": int, "intents_count": int, "call_graph_available": bool, "partial_forecast": bool, "warnings": [str, ...], "conflicts": [ { "conflict_type": "address_overlap" | "blast_radius_overlap" | "operation_conflict", "addresses": [str, ...], "agents": [str, ...], "confidence": float, "description": str }, ... ], "high_risk": int, "medium_risk": int, "low_risk": int, "elapsed_seconds": float } Text output:: Conflict forecast — 3 active reservation(s), 1 intent(s) ────────────────────────────────────────────────────────────── ⚠️ blast_radius_overlap (confidence 0.75) src/billing.py::compute_total agent-41 (branch: main) ↔ agent-42 (branch: feature/billing) → compute_total is in the transitive call chain of process_payment 🔴 operation_conflict (confidence 0.90) ... 1 high-risk, 1 medium-risk, 0 low-risk conflict(s) predicted Flags:: --branch BRANCH Restrict to reservations/intents on this branch. --run-id ID Show only conflicts involving a specific agent. Maximum 256 characters. --min-confidence FLOAT Hide conflicts below this confidence threshold (default: 0.0 — show all). Range: [0.0, 1.0]. --format / -f Output format: text (default) or json. --json Shorthand for --format json. Exit codes:: 0 — success (zero conflicts is still success) 1 — bad arguments (--min-confidence out of range, --run-id too long) 2 — unexpected error during analysis """ from __future__ import annotations import argparse import json import logging import sys import time from muse._version import __version__ from muse.core.coordination import active_reservations, load_all_intents from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.core.validation import sanitize_display from muse.plugins.code._callgraph import build_reverse_graph, transitive_callers type _ConflictDict = dict[str, str | float | list[str]] type _AgentListMap = dict[str, list[str]] logger = logging.getLogger(__name__) # ── Input constraints ───────────────────────────────────────────────────────── #: Maximum byte-length of the ``--run-id`` filter value. Matches the cap #: applied to run-id in all other coordination commands. _MAX_RUN_ID_LEN: int = 256 # ── Data model ──────────────────────────────────────────────────────────────── class _ConflictPrediction: """A single predicted merge conflict between two or more agents. Attributes ---------- conflict_type: One of ``"address_overlap"``, ``"blast_radius_overlap"``, or ``"operation_conflict"``. addresses: The symbol address(es) at the centre of the conflict. agents: List of ``"run_id@branch"`` strings identifying the conflicting agents. confidence: Float in ``[0, 1]``. 1.0 = certain conflict; < 0.5 = speculative. description: Human-readable explanation of why this is a conflict. """ def __init__( self, conflict_type: str, addresses: list[str], agents: list[str], confidence: float, description: str, ) -> None: self.conflict_type = conflict_type self.addresses = addresses self.agents = agents self.confidence = confidence self.description = description def to_dict(self) -> _ConflictDict: """Serialise to a JSON-safe dict. Returns ------- dict Keys: ``conflict_type``, ``addresses``, ``agents``, ``confidence`` (rounded to 3 d.p.), ``description``. """ return { "conflict_type": self.conflict_type, "addresses": self.addresses, "agents": self.agents, "confidence": round(self.confidence, 3), "description": self.description, } # ── CLI registration ────────────────────────────────────────────────────────── def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``forecast`` 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( "forecast", help="Predict merge conflicts from active reservations and intents.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--branch", "-b", dest="branch_filter", default=None, metavar="BRANCH", help="Restrict to reservations/intents on this branch.", ) parser.add_argument( "--run-id", dest="run_id_filter", default=None, metavar="ID", help=( "Show only conflicts involving this agent run-id. " f"Maximum {_MAX_RUN_ID_LEN} characters." ), ) parser.add_argument( "--min-confidence", dest="min_confidence", type=float, default=0.0, metavar="FLOAT", help=( "Hide conflicts below this confidence threshold " "(default: 0.0 — show all). Range: [0.0, 1.0]." ), ) 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: """Predict merge conflicts from active reservations and intents. Execution order --------------- 1. **Validate inputs** — ``--run-id`` length and ``--min-confidence`` range. Fires before any file I/O; any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1). 2. **Load state** — calls :func:`~muse.core.coordination.active_reservations` and :func:`~muse.core.coordination.load_all_intents`, then applies ``--branch`` and ``--run-id`` filters in memory. 3. **Run three analysis passes** (see below). 4. **Apply ``--min-confidence`` filter** to the conflict list. 5. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. Analysis passes --------------- **Pass 1 — Direct address overlap** (always runs) Scans all active reservations and finds symbol addresses that more than one agent has reserved simultaneously. Confidence 1.0. **Pass 2 — Blast-radius overlap** (runs when code index is available) Builds the reverse call graph for the current commit and checks whether any two reserved addresses lie in each other's transitive call chains. Confidence 0.75. When the code index is unavailable (no commits yet, or ``muse code index`` has not been run), this pass is skipped. The reason is recorded in ``warnings`` (JSON) or printed as a notice (text). ``partial_forecast`` is ``true`` whenever any pass was skipped. The forecast is **never silently presented as complete** — agents should check ``partial_forecast`` before treating an empty ``conflicts`` list as authoritative. **Pass 3 — Operation conflicts** (always runs) Scans intents and finds addresses where one agent intends to ``delete`` and another intends to ``modify``, ``rename``, or ``extract``. Confidence 0.9. Filtering --------- * ``--branch`` filters reservations and intents to a single branch before any analysis pass runs, reducing the O(n²) cost of Pass 2. * ``--run-id`` keeps only conflicts where the named agent appears in the ``agents`` list. Applied after all passes so the agent sees its full blast radius. * ``--min-confidence`` hides low-signal conflicts. Applied last so ``high_risk`` / ``medium_risk`` / ``low_risk`` counts reflect the *filtered* list. Error handling -------------- * ``(OSError, KeyError, ValueError, AttributeError)`` from the call graph build are treated as "index not available" — a warning is emitted and the blast-radius pass is skipped. * Any other exception propagates so that the caller sees a real error rather than a false-clean forecast. Security -------- All agent-supplied strings (``run_id``, ``branch``, ``addresses``) are passed through :func:`~muse.core.validation.sanitize_display` before printing to strip ANSI escape sequences and control characters. ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` to bound memory use. No user-supplied string is ever used to construct a file path. Performance ----------- Pass 1 and Pass 3 are O(n × m) where n = reservations and m = intents. Pass 2 is O(n² × D) where D is the BFS depth limit (default 5) — on large swarms with many reservations, use ``--branch`` to reduce n. Args: args: Parsed ``argparse.Namespace`` with attributes ``branch_filter``, ``run_id_filter``, ``min_confidence``, and ``fmt``. Exit codes: 0 — success (zero conflicts is still success). 1 — bad arguments. """ t0 = time.monotonic() branch_filter: str | None = args.branch_filter run_id_filter: str | None = args.run_id_filter min_confidence: float = args.min_confidence fmt: str = args.fmt as_json = fmt == "json" # ── Input validation (before any file I/O) ──────────────────────────────── if run_id_filter is not None and len(run_id_filter) > _MAX_RUN_ID_LEN: msg = f"--run-id is too long ({len(run_id_filter)} 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 not (0.0 <= min_confidence <= 1.0): msg = f"--min-confidence must be in [0.0, 1.0], got {min_confidence}" if as_json: print(json.dumps({"error": msg, "status": "bad_args"})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() branch = read_current_branch(root) reservations = active_reservations(root) intents = load_all_intents(root) if branch_filter: reservations = [r for r in reservations if r.branch == branch_filter] intents = [i for i in intents if i.branch == branch_filter] conflicts: list[_ConflictPrediction] = [] warnings: list[str] = [] call_graph_available = False # ── Pass 1: Direct address overlap ──────────────────────────────────────── # Build: address → list of "run_id@branch" labels. addr_agents: _AgentListMap = {} for res in reservations: for addr in res.addresses: addr_agents.setdefault(addr, []).append(f"{res.run_id}@{res.branch}") for addr, agents in sorted(addr_agents.items()): unique_agents = list(dict.fromkeys(agents)) if len(unique_agents) > 1: conflicts.append(_ConflictPrediction( conflict_type="address_overlap", addresses=[addr], agents=unique_agents, confidence=1.0, description=( f"{addr} reserved by {len(unique_agents)} agents simultaneously" ), )) # ── Pass 2: Blast-radius overlap ────────────────────────────────────────── # Use the reverse call graph to detect indirect dependencies between # reserved addresses. Skip silently (but warn) when the index is # unavailable — do NOT catch all exceptions, as unexpected errors # indicate real problems that agents must see. repo_id = read_repo_id(root) commit = resolve_commit_ref(root, repo_id, branch, None) if commit is not None: manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} try: reverse = build_reverse_graph(root, manifest) call_graph_available = True all_addresses = list(addr_agents.keys()) for i, addr_a in enumerate(all_addresses): callers_a = transitive_callers(addr_a, reverse, max_depth=5) callers_set: set[str] = {c for lvl in callers_a.values() for c in lvl} for addr_b in all_addresses[i + 1:]: if addr_b in callers_set: agents_a = addr_agents.get(addr_a, []) agents_b = addr_agents.get(addr_b, []) if set(agents_a) != set(agents_b): conflicts.append(_ConflictPrediction( conflict_type="blast_radius_overlap", addresses=[addr_a, addr_b], agents=list(set(agents_a) | set(agents_b)), confidence=0.75, description=( f"{addr_b} is in the transitive call chain of {addr_a}" ), )) except (OSError, KeyError, ValueError, AttributeError) as exc: # Expected: index not built, object not found, or malformed data. # Record as a warning — the caller must know blast-radius analysis # was skipped rather than receiving a false-clean forecast. warn = f"blast_radius_overlap skipped — call graph unavailable: {exc}" warnings.append(warn) logger.debug("Call graph unavailable for forecast: %s", exc) else: warnings.append( "blast_radius_overlap skipped — no commits on this branch yet" ) # ── Pass 3: Operation conflicts ─────────────────────────────────────────── # Detect intents where one agent intends delete and another modify/rename. intent_ops: _AgentListMap = {} intent_agents: _AgentListMap = {} for it in intents: for addr in it.addresses: intent_ops.setdefault(addr, []).append(it.operation) intent_agents.setdefault(addr, []).append(f"{it.run_id}@{it.branch}") for addr, ops in sorted(intent_ops.items()): if len(set(ops)) <= 1 and len(set(intent_agents.get(addr, []))) <= 1: continue # Same op by same agent — not a conflict. has_delete = "delete" in ops has_modify = any(op in ("modify", "rename", "extract") for op in ops) if has_delete and has_modify: agents = list(dict.fromkeys(intent_agents.get(addr, []))) conflicts.append(_ConflictPrediction( conflict_type="operation_conflict", addresses=[addr], agents=agents, confidence=0.9, description=f"delete vs modify conflict on {addr}", )) # ── Post-pass filters ───────────────────────────────────────────────────── # --run-id: keep only conflicts that mention this agent. if run_id_filter is not None: tag = run_id_filter # agents are stored as "run_id@branch" conflicts = [ c for c in conflicts if any(a.split("@")[0] == tag for a in c.agents) ] # --min-confidence: suppress low-signal predictions. if min_confidence > 0.0: conflicts = [c for c in conflicts if c.confidence >= min_confidence] partial_forecast = bool(warnings) elapsed = round(time.monotonic() - t0, 4) # ── JSON output ─────────────────────────────────────────────────────────── if as_json: print(json.dumps({ "schema_version": __version__, "current_branch": branch, "branch_filter": branch_filter, "run_id_filter": run_id_filter, "min_confidence": min_confidence, "active_reservations": len(reservations), "intents_count": len(intents), "call_graph_available": call_graph_available, "partial_forecast": partial_forecast, "warnings": warnings, "conflicts": [c.to_dict() for c in conflicts], "high_risk": sum(1 for c in conflicts if c.confidence >= 0.9), "medium_risk": sum(1 for c in conflicts if 0.5 <= c.confidence < 0.9), "low_risk": sum(1 for c in conflicts if c.confidence < 0.5), "elapsed_seconds": elapsed, })) return # ── Text output ─────────────────────────────────────────────────────────── print( f"\nConflict forecast — " f"{len(reservations)} active reservation(s), {len(intents)} intent(s)" ) print("─" * 62) if warnings: for w in warnings: print(f"\n ⚠ Note: {w}") if not conflicts: print("\n ✅ No conflicts predicted.") if not reservations: print(" (no active reservations — run 'muse coord reserve' first)") else: for c in conflicts: icon = "🔴" if c.confidence >= 0.9 else "⚠️ " print(f"\n{icon} {c.conflict_type} (confidence {c.confidence:.2f})") for addr in c.addresses: print(f" {sanitize_display(addr)}") for agent in c.agents: print(f" agent: {sanitize_display(agent)}") print(f" → {sanitize_display(c.description)}") high = sum(1 for c in conflicts if c.confidence >= 0.9) med = sum(1 for c in conflicts if 0.5 <= c.confidence < 0.9) print(f"\n {high} high-risk, {med} medium-risk conflict(s) predicted") print(" Run 'muse coord plan-merge' for a detailed merge strategy.") print(f"\n ({elapsed:.3f}s)")