"""``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, "duration_ms": 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 """ import argparse import json import logging from typing import TypedDict import sys from muse.core.envelope import EnvelopeJson, make_envelope 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.refs import read_current_branch from muse.core.commits import resolve_commit_ref from muse.core.snapshots import get_commit_snapshot_manifest from muse.core.validation import sanitize_display from muse.core.timing import start_timer 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]] class _ForecastJson(EnvelopeJson): """JSON output for ``muse coord forecast --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ current_branch Branch from which the forecast was computed. branch_filter ``--branch`` filter value used, or ``None`` when all branches were considered. run_id_filter ``--run-id`` filter value used, or ``None`` when not set. min_confidence Minimum confidence threshold applied to predictions (range 0.0–1.0; predictions below this were suppressed). active_reservations Number of active agent reservations visible at forecast time. intents_count Number of declared intents that fed the conflict model. call_graph_available True when a full call graph was available — False means blast-radius predictions are approximate. partial_forecast True when the forecast is based on incomplete data (missing call graph or sparse intent coverage). conflicts List of predicted conflict dicts; each has type, paths, agents, and risk_level fields. high_risk Count of predicted conflicts classified as high risk. medium_risk Count of predicted conflicts classified as medium risk. low_risk Count of predicted conflicts classified as low risk. """ current_branch: str branch_filter: str | None run_id_filter: str | None min_confidence: float active_reservations: int intents_count: int call_graph_available: bool partial_forecast: bool conflicts: list[_ConflictDict] high_risk: int medium_risk: int low_risk: int 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( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON.", ) parser.set_defaults(func=run) # ── Command implementation ──────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Predict merge conflicts from active reservations and intents. Runs three analysis passes against active coordination state: direct address overlap (confidence 1.0), blast-radius overlap via the reverse call graph (confidence 0.75), and operation conflicts between delete/modify intents (confidence 0.9). ``partial_forecast`` is ``true`` when the call graph was unavailable and Pass 2 was skipped — always check this before treating an empty ``conflicts`` list as authoritative. Agent quickstart ---------------- :: muse coord forecast --format json muse coord forecast --branch feat/billing --format json muse coord forecast --run-id agent-42 --format json muse coord forecast --min-confidence 0.8 --format json JSON fields ----------- current_branch Branch being analysed. branch_filter ``--branch`` value, or ``null``. run_id_filter ``--run-id`` value, or ``null``. active_reservations Number of reservations loaded. intents_count Number of intents loaded. call_graph_available ``false`` if the blast-radius pass was skipped. partial_forecast ``true`` when any analysis pass was skipped. conflicts List of conflict objects: ``addresses``, ``agents``, ``confidence``, ``reason``. high_risk Conflicts with confidence ≥ 0.9. medium_risk Conflicts with 0.5 ≤ confidence < 0.9. Exit codes ---------- 0 Success (zero conflicts is still success). 1 Invalid arguments. 2 Not inside a Muse repository. """ elapsed = start_timer() branch_filter: str | None = args.branch_filter run_id_filter: str | None = args.run_id_filter min_confidence: float = args.min_confidence json_out: bool = args.json_out # ── 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 json_out: 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 json_out: 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. commit = resolve_commit_ref(root, 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) # ── JSON output ─────────────────────────────────────────────────────────── if json_out: print(json.dumps(_ForecastJson( **make_envelope(elapsed, warnings=warnings), 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, 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), ))) 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)")