"""Comprehensive tests for ``muse coord reconcile``. Coverage -------- Unit — internal helpers _BranchSummary.to_dict: all keys present and correctly typed strategy fast-forward: branch with 0 conflicts → fast-forward strategy rebase: branch with 1–2 conflicts → rebase strategy manual: branch with 3+ conflicts → manual hotspot detection: same address on 2 branches → hotspot Integration — CLI empty repo: exits 0, "no active coordination data" in output single branch: exits 0, branch name in output multiple branches no conflict: each branch listed, 0 hotspots multiple branches with hotspot: hotspot address in output --json output schema: all required top-level keys present --json shorthand: same schema as --format json --format json explicit: same schema as --json JSON schema compact (single line): output is exactly one line duration_ms present: field exists and is a non-negative float duration_ms type: float, not string or int branch entry has run_ids: each branch entry includes run_ids list merge order matches branches: recommended_merge_order matches branch names Security ANSI in branch name (text): stripped from text output (merge order) ANSI in branch name (hotspot): stripped from hotspot text ANSI in address: stripped from text output E2E — strategy thresholds 0 conflicts → fast-forward 1 conflict → rebase 2 conflicts → rebase 3 conflicts → manual Stress 50 reservations 5 branches < 1 s: throughput baseline 100 reservations 10 branches: all in JSON, duration_ms present hotspot-heavy 20 branches 1 addr: all share same address """ from __future__ import annotations type _AddrBranches = dict[str, list[str]] import json import pathlib import time import pytest from tests.cli_test_helper import CliRunner from muse.core.types import fake_id from muse.core.paths import muse_dir from muse.core.coordination import ( create_intent, create_reservation, active_reservations, load_all_intents, ) from muse.cli.commands.reconcile import _BranchSummary cli = None runner = CliRunner() _REQUIRED_JSON_KEYS = { "schema", "active_reservations", "active_intents", "conflict_hotspots", "branches", "recommended_merge_order", "strategies", "hotspots", } _REQUIRED_BRANCH_KEYS = { "branch", "reserved_addresses", "intents", "run_ids", "predicted_conflicts", } # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: dot_muse = muse_dir(tmp_path) dot_muse.mkdir() (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) return tmp_path # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_reservation( root: pathlib.Path, *, run_id: str = "agent-1", branch: str = "main", addresses: list[str] | None = None, ttl_seconds: int = 3600, ) -> None: create_reservation( root, run_id=run_id, branch=branch, addresses=addresses or ["src/mod.py::foo"], ttl_seconds=ttl_seconds, ) def _make_intent( root: pathlib.Path, *, run_id: str = "agent-1", branch: str = "main", addresses: list[str] | None = None, operation: str = "modify", ) -> None: create_intent( root, reservation_id=fake_id("reconcile-intent-res"), run_id=run_id, branch=branch, addresses=addresses or ["src/mod.py::foo"], operation=operation, ) # --------------------------------------------------------------------------- # Unit — _BranchSummary # --------------------------------------------------------------------------- class TestBranchSummaryToDict: def test_all_required_keys_present(self) -> None: bs = _BranchSummary("feature/billing") d = bs.to_dict() assert _REQUIRED_BRANCH_KEYS.issubset(d.keys()) def test_branch_field_matches_constructor(self) -> None: bs = _BranchSummary("feature/auth") assert bs.to_dict()["branch"] == "feature/auth" def test_reserved_addresses_starts_empty(self) -> None: bs = _BranchSummary("main") assert bs.to_dict()["reserved_addresses"] == [] def test_intents_starts_empty(self) -> None: bs = _BranchSummary("main") assert bs.to_dict()["intents"] == [] def test_run_ids_sorted(self) -> None: bs = _BranchSummary("main") bs.run_ids = {"agent-3", "agent-1", "agent-2"} run_ids_list = bs.to_dict()["run_ids"] assert run_ids_list == sorted(run_ids_list) def test_predicted_conflicts_starts_zero(self) -> None: bs = _BranchSummary("main") assert bs.to_dict()["predicted_conflicts"] == 0 def test_conflict_count_reflected_in_dict(self) -> None: bs = _BranchSummary("main") bs.conflict_count = 5 assert bs.to_dict()["predicted_conflicts"] == 5 def test_reserved_addresses_reflects_mutations(self) -> None: bs = _BranchSummary("main") bs.reserved_addresses.extend(["src/a.py::foo", "src/b.py::bar"]) assert len(bs.to_dict()["reserved_addresses"]) == 2 class TestStrategySelection: """Verify strategy labels for known conflict counts (mirrors reconcile.run logic).""" def _strategy_for(self, conflict_count: int) -> str: if conflict_count == 0: return "fast-forward (no conflicts predicted)" elif conflict_count <= 2: return "rebase onto main before merging" else: return "manual conflict resolution required" def test_zero_conflicts_fast_forward(self) -> None: assert self._strategy_for(0) == "fast-forward (no conflicts predicted)" def test_one_conflict_rebase(self) -> None: assert self._strategy_for(1) == "rebase onto main before merging" def test_two_conflicts_rebase(self) -> None: assert self._strategy_for(2) == "rebase onto main before merging" def test_three_conflicts_manual(self) -> None: assert self._strategy_for(3) == "manual conflict resolution required" def test_ten_conflicts_manual(self) -> None: assert self._strategy_for(10) == "manual conflict resolution required" class TestHotspotDetection: def test_same_address_two_branches_is_hotspot(self, tmp_path: pathlib.Path) -> None: _make_reservation( tmp_path, run_id="agent-1", branch="feature/a", addresses=["src/billing.py::compute_total"], ) _make_reservation( tmp_path, run_id="agent-2", branch="feature/b", addresses=["src/billing.py::compute_total"], ) reservations = active_reservations(tmp_path) addr_branches: _AddrBranches = {} for res in reservations: for addr in res.addresses: addr_branches.setdefault(addr, []).append(res.branch) hotspots = { addr: branches for addr, branches in addr_branches.items() if len(set(branches)) > 1 } assert "src/billing.py::compute_total" in hotspots def test_same_address_same_branch_not_hotspot(self, tmp_path: pathlib.Path) -> None: _make_reservation( tmp_path, run_id="agent-1", branch="feature/a", addresses=["src/billing.py::compute_total"], ) _make_reservation( tmp_path, run_id="agent-2", branch="feature/a", addresses=["src/billing.py::compute_total"], ) reservations = active_reservations(tmp_path) addr_branches: _AddrBranches = {} for res in reservations: for addr in res.addresses: addr_branches.setdefault(addr, []).append(res.branch) hotspots = { addr: branches for addr, branches in addr_branches.items() if len(set(branches)) > 1 } assert "src/billing.py::compute_total" not in hotspots def test_distinct_addresses_no_hotspot(self, tmp_path: pathlib.Path) -> None: _make_reservation( tmp_path, run_id="agent-1", branch="feature/a", addresses=["src/a.py::foo"], ) _make_reservation( tmp_path, run_id="agent-2", branch="feature/b", addresses=["src/b.py::bar"], ) reservations = active_reservations(tmp_path) addr_branches: _AddrBranches = {} for res in reservations: for addr in res.addresses: addr_branches.setdefault(addr, []).append(res.branch) hotspots = { addr: branches for addr, branches in addr_branches.items() if len(set(branches)) > 1 } assert len(hotspots) == 0 # --------------------------------------------------------------------------- # Integration — CLI # --------------------------------------------------------------------------- class TestReconcileCLIEmpty: def test_empty_repo_exits_zero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile"]) assert result.exit_code == 0 def test_empty_repo_prints_no_active_data(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile"]) assert "no active coordination data" in result.output def test_empty_repo_json_exits_zero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) assert result.exit_code == 0 def test_empty_repo_json_has_required_keys(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert _REQUIRED_JSON_KEYS.issubset(data.keys()) def test_empty_repo_json_counts_are_zero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert data["active_reservations"] == 0 assert data["active_intents"] == 0 assert data["conflict_hotspots"] == 0 class TestReconcileCLISingleBranch: def test_single_branch_exits_zero(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/billing") result = runner.invoke(cli, ["coord", "reconcile"]) assert result.exit_code == 0 def test_single_branch_name_in_output(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/billing") result = runner.invoke(cli, ["coord", "reconcile"]) assert "feature/billing" in result.output def test_single_branch_no_hotspots(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/billing") result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert data["conflict_hotspots"] == 0 def test_single_branch_fast_forward_strategy(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/billing") result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) strategy = data["strategies"].get("feature/billing", "") assert "fast-forward" in strategy class TestReconcileCLIMultipleBranches: def test_multiple_branches_no_conflict_exits_zero(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/a", addresses=["src/a.py::foo"]) _make_reservation(repo, run_id="agent-2", branch="feature/b", addresses=["src/b.py::bar"]) result = runner.invoke(cli, ["coord", "reconcile"]) assert result.exit_code == 0 def test_multiple_branches_both_listed(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/a", addresses=["src/a.py::foo"]) _make_reservation(repo, run_id="agent-2", branch="feature/b", addresses=["src/b.py::bar"]) result = runner.invoke(cli, ["coord", "reconcile"]) assert "feature/a" in result.output assert "feature/b" in result.output def test_multiple_branches_no_conflict_zero_hotspots(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/a", addresses=["src/a.py::foo"]) _make_reservation(repo, run_id="agent-2", branch="feature/b", addresses=["src/b.py::bar"]) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert data["conflict_hotspots"] == 0 def test_hotspot_address_in_text_output(self, repo: pathlib.Path) -> None: _make_reservation( repo, run_id="agent-1", branch="feature/a", addresses=["src/billing.py::compute_total"], ) _make_reservation( repo, run_id="agent-2", branch="feature/b", addresses=["src/billing.py::compute_total"], ) result = runner.invoke(cli, ["coord", "reconcile"]) assert "src/billing.py::compute_total" in result.output def test_hotspot_count_nonzero_in_json(self, repo: pathlib.Path) -> None: _make_reservation( repo, run_id="agent-1", branch="feature/a", addresses=["src/billing.py::compute_total"], ) _make_reservation( repo, run_id="agent-2", branch="feature/b", addresses=["src/billing.py::compute_total"], ) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert data["conflict_hotspots"] == 1 def test_hotspot_present_in_hotspots_list(self, repo: pathlib.Path) -> None: _make_reservation( repo, run_id="agent-1", branch="feature/a", addresses=["src/billing.py::compute_total"], ) _make_reservation( repo, run_id="agent-2", branch="feature/b", addresses=["src/billing.py::compute_total"], ) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) addresses = [h["address"] for h in data["hotspots"]] assert "src/billing.py::compute_total" in addresses class TestReconcileCLIJSONSchema: def test_json_flag_exits_zero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) assert result.exit_code == 0 def test_json_flag_is_valid_json(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert isinstance(data, dict) def test_json_flag_has_required_keys(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert _REQUIRED_JSON_KEYS.issubset(data.keys()) def test_format_json_explicit_has_required_keys(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert _REQUIRED_JSON_KEYS.issubset(data.keys()) def test_branches_field_is_list(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert isinstance(data["branches"], list) def test_recommended_merge_order_is_list(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert isinstance(data["recommended_merge_order"], list) def test_strategies_is_dict(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert isinstance(data["strategies"], dict) def test_hotspots_is_list(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert isinstance(data["hotspots"], list) def test_branch_entry_has_required_keys(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/billing") result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert len(data["branches"]) >= 1 branch_entry = data["branches"][0] assert _REQUIRED_BRANCH_KEYS.issubset(branch_entry.keys()) def test_schema_version_is_string(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert isinstance(data["schema"], int) def test_active_reservations_count_matches(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/a", addresses=["src/a.py::foo"]) _make_reservation(repo, run_id="agent-2", branch="feature/b", addresses=["src/b.py::bar"]) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert data["active_reservations"] == 2 def test_active_intents_count_matches(self, repo: pathlib.Path) -> None: _make_intent(repo, run_id="agent-1", branch="feature/a") _make_intent(repo, run_id="agent-2", branch="feature/b") result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert data["active_intents"] == 2 # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestReconcileStress: def test_50_reservations_5_branches_under_1_second(self, repo: pathlib.Path) -> None: branches = [f"feature/branch-{i}" for i in range(5)] start = time.monotonic() for i in range(50): create_reservation( repo, run_id=f"agent-{i}", branch=branches[i % len(branches)], addresses=[f"src/mod{i}.py::sym{i}"], ttl_seconds=3600, ) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) elapsed = time.monotonic() - start assert result.exit_code == 0 data = json.loads(result.output) assert data["active_reservations"] == 50 assert elapsed < 1.0, f"50 reservations across 5 branches took {elapsed:.2f}s (limit 1s)" def test_100_reservations_10_branches_json_complete(self, repo: pathlib.Path) -> None: branches = [f"feature/branch-{i}" for i in range(10)] for i in range(100): create_reservation( repo, run_id=f"agent-{i}", branch=branches[i % len(branches)], addresses=[f"src/mod{i}.py::sym{i}"], ttl_seconds=3600, ) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["active_reservations"] == 100 assert "duration_ms" in data assert isinstance(data["duration_ms"], float) assert len(data["branches"]) == 10 def test_hotspot_heavy_20_branches_all_share_one_address( self, repo: pathlib.Path ) -> None: shared = "src/core.py::shared_sym" for i in range(20): create_reservation( repo, run_id=f"agent-{i}", branch=f"feature/branch-{i}", addresses=[shared], ttl_seconds=3600, ) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["conflict_hotspots"] == 1 assert data["hotspots"][0]["address"] == shared assert len(data["hotspots"][0]["branches"]) == 20 # --------------------------------------------------------------------------- # JSON schema — compact + duration_ms # --------------------------------------------------------------------------- class TestReconcileJsonCompact: def test_json_output_is_single_line(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) assert result.exit_code == 0 lines = [ln for ln in result.output.splitlines() if ln.strip()] assert len(lines) == 1, f"Expected 1 JSON line, got {len(lines)}: {result.output!r}" def test_duration_ms_present(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert "duration_ms" in data def test_duration_ms_is_float(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert isinstance(data["duration_ms"], float) def test_duration_ms_non_negative(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert data["duration_ms"] >= 0.0 def test_branch_entry_has_run_ids(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-x", branch="feature/x") result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) branch_entry = data["branches"][0] assert "run_ids" in branch_entry assert isinstance(branch_entry["run_ids"], list) assert "agent-x" in branch_entry["run_ids"] def test_merge_order_matches_branch_names(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/a", addresses=["src/a.py::foo"]) _make_reservation(repo, run_id="agent-2", branch="feature/b", addresses=["src/b.py::bar"]) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) branch_names = {entry["branch"] for entry in data["branches"]} assert set(data["recommended_merge_order"]) == branch_names def test_json_shorthand_same_structure_as_format_json( self, repo: pathlib.Path ) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/a") r1 = runner.invoke(cli, ["coord", "reconcile", "--json"]) r2 = runner.invoke(cli, ["coord", "reconcile", "--json"]) d1 = json.loads(r1.output) d2 = json.loads(r2.output) structural_keys = { "schema", "active_reservations", "active_intents", "conflict_hotspots", "recommended_merge_order", "strategies", } for key in structural_keys: assert d1[key] == d2[key], f"Mismatch on {key!r}" # --------------------------------------------------------------------------- # E2E — strategy thresholds # --------------------------------------------------------------------------- class TestReconcileE2EStrategies: def test_zero_conflicts_fast_forward(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/clean", addresses=["src/a.py::foo"]) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert "fast-forward" in data["strategies"]["feature/clean"] def test_one_conflict_rebase(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/a", addresses=["src/shared.py::x"]) _make_reservation(repo, run_id="agent-2", branch="feature/b", addresses=["src/shared.py::x"]) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) for branch in ("feature/a", "feature/b"): assert "rebase" in data["strategies"][branch] def test_two_conflicts_still_rebase(self, repo: pathlib.Path) -> None: # Two hotspot addresses shared between same two branches → conflict_count == 2. _make_reservation( repo, run_id="agent-1", branch="feature/a", addresses=["src/shared.py::x", "src/shared.py::y"], ) _make_reservation( repo, run_id="agent-2", branch="feature/b", addresses=["src/shared.py::x", "src/shared.py::y"], ) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) for branch in ("feature/a", "feature/b"): assert "rebase" in data["strategies"][branch] def test_three_conflicts_manual(self, repo: pathlib.Path) -> None: shared_addrs = [f"src/shared.py::sym{i}" for i in range(3)] _make_reservation(repo, run_id="agent-1", branch="feature/a", addresses=shared_addrs) _make_reservation(repo, run_id="agent-2", branch="feature/b", addresses=shared_addrs) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) for branch in ("feature/a", "feature/b"): assert "manual" in data["strategies"][branch] def test_merge_order_clean_branch_first(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/clean", addresses=["src/a.py::foo"]) _make_reservation(repo, run_id="agent-2", branch="feature/conflict-a", addresses=["src/shared.py::x"]) _make_reservation(repo, run_id="agent-3", branch="feature/conflict-b", addresses=["src/shared.py::x"]) result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) order = data["recommended_merge_order"] assert order.index("feature/clean") < order.index("feature/conflict-a") assert order.index("feature/clean") < order.index("feature/conflict-b") def test_intents_count_in_json(self, repo: pathlib.Path) -> None: _make_intent(repo, run_id="agent-1", branch="feature/x", operation="rename") _make_intent(repo, run_id="agent-2", branch="feature/x", operation="modify") result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) assert data["active_intents"] == 2 def test_branch_intents_list_populated(self, repo: pathlib.Path) -> None: _make_intent(repo, run_id="agent-1", branch="feature/x", operation="rename") result = runner.invoke(cli, ["coord", "reconcile", "--json"]) data = json.loads(result.output) branch_entry = next(b for b in data["branches"] if b["branch"] == "feature/x") assert "rename" in branch_entry["intents"] def test_text_output_includes_elapsed(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="agent-1", branch="feature/a") result = runner.invoke(cli, ["coord", "reconcile"]) assert result.exit_code == 0 # Elapsed appears at the bottom as (X.XXXs) assert "s)" in result.output # --------------------------------------------------------------------------- # Security — merge order branch sanitization # --------------------------------------------------------------------------- class TestReconcileMergeOrderSecurity: def test_ansi_in_branch_stripped_from_merge_order(self, repo: pathlib.Path) -> None: ansi_branch = "\x1b[31mfeature/malicious\x1b[0m" create_reservation( repo, run_id="agent-malicious", branch=ansi_branch, addresses=["src/mod.py::foo"], ttl_seconds=3600, ) result = runner.invoke(cli, ["coord", "reconcile"]) assert result.exit_code == 0 assert "\x1b[31m" not in result.output assert "\x1b[0m" not in result.output def test_ansi_in_branch_stripped_from_hotspot_text(self, repo: pathlib.Path) -> None: ansi_branch = "\x1b[35mfeature/badactor\x1b[0m" create_reservation( repo, run_id="agent-1", branch=ansi_branch, addresses=["src/billing.py::compute_total"], ttl_seconds=3600, ) create_reservation( repo, run_id="agent-2", branch="feature/clean", addresses=["src/billing.py::compute_total"], ttl_seconds=3600, ) result = runner.invoke(cli, ["coord", "reconcile"]) assert result.exit_code == 0 assert "\x1b[35m" not in result.output assert "\x1b[0m" not in result.output def test_ansi_in_address_stripped_from_hotspot_text( self, repo: pathlib.Path ) -> None: ansi_addr = "\x1b[32msrc/billing.py::compute_total\x1b[0m" create_reservation( repo, run_id="agent-1", branch="feature/a", addresses=[ansi_addr], ttl_seconds=3600, ) create_reservation( repo, run_id="agent-2", branch="feature/b", addresses=[ansi_addr], ttl_seconds=3600, ) result = runner.invoke(cli, ["coord", "reconcile"]) assert result.exit_code == 0 assert "\x1b[32m" not in result.output assert "\x1b[0m" not in result.output class TestRegisterFlags: def test_default_json_out_is_false(self) -> None: import argparse from muse.cli.commands.reconcile import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["reconcile"]) assert args.json_out is False def test_json_flag_sets_json_out(self) -> None: import argparse from muse.cli.commands.reconcile import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["reconcile", "--json"]) assert args.json_out is True def test_j_shorthand_sets_json_out(self) -> None: import argparse from muse.cli.commands.reconcile import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["reconcile", "-j"]) assert args.json_out is True