gabriel/muse — blame/sha256:a/test_code_commands.py

1 files

1 commits

0 hotspots

0 🧊 dead

0 💥 blast risk

sha256:e Merge 'bump/muse-v0.2.1' into 'dev' — proposal: chore: bump muse to v0.2.1 · gabriel · Aug 28, 2026

1	"""Integration tests for code-domain CLI commands.
2
3	Uses a real Muse repository initialised in tmp_path.
4
5	Coverage
6	--------
7	Provenance & Topology
8	muse lineage ADDRESS [--json]
9	muse api-surface [--diff REF] [--json]
10	muse codemap [--top N] [--json]
11	muse clones [--tier exact\|near\|both] [--json]
12	muse checkout-symbol ADDRESS --commit REF [--dry-run]
13	muse semantic-cherry-pick ADDRESS... --from REF [--dry-run] [--json]
14
15	Query & Temporal Search
16	muse query PREDICATE [--all-commits] [--json]
17	muse query-history PREDICATE [--from REF] [--to REF] [--json]
18
19	Index Commands
20	muse index status [--json]
21	muse index rebuild [--index NAME]
22
23	Refactor Detection
24	muse detect-refactor --json (schema_version in output)
25
26	Multi-Agent Coordination
27	muse reserve ADDRESS...
28	muse intent ADDRESS... --op OP
29	muse forecast [--json]
30	muse plan-merge OURS THEIRS [--json]
31	muse shard --agents N [--json]
32	muse reconcile [--json]
33
34	Structural Enforcement
35	muse breakage [--json]
36	muse invariants [--json]
37
38	Semantic Versioning Metadata
39	muse log shows SemVer for commits with bumps
40	muse commit stores sem_ver_bump in CommitRecord
41
42	Call-Graph Tier
43	muse impact ADDRESS [--json]
44	muse dead [--json]
45	muse coverage CLASS_ADDRESS [--json]
46	muse deps ADDRESS_OR_FILE [--json]
47	muse find-symbol [--name NAME] [--json]
48	muse patch ADDRESS FILE
49	"""
50
51	import json
52	import pathlib
53	import textwrap
54
55	import pytest
56	from tests.cli_test_helper import CliRunner
57
58	from typing import TypedDict
59
60	from muse._version import __version__
61	cli = None # argparse migration — CliRunner ignores this arg
62	from muse.core.refs import get_head_commit_id
63	from muse.core.commits import CommitDict
64	from muse.core.types import Manifest
65	from muse.core.paths import coordination_dir, indices_dir, muse_dir, ref_path, repo_json_path
66
67	type _ImportsMap = dict[str, list[str]]
68	type _ImportsSetMap = dict[str, set[str]]
69	type _KindsMap = dict[str, int]
70
71	runner = CliRunner()
72
73
74	# ---------------------------------------------------------------------------
75	# Shared fixtures
76	# ---------------------------------------------------------------------------
77
78
79	@pytest.fixture
80	def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
81	"""Initialise a fresh code-domain Muse repo."""
82	monkeypatch.chdir(tmp_path)
83	monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
84	result = runner.invoke(cli, ["init", "--domain", "code"])
85	assert result.exit_code == 0, result.output
86	return tmp_path
87
88
89	@pytest.fixture
90	def code_repo(repo: pathlib.Path) -> pathlib.Path:
91	"""Repo with two Python commits for analysis commands."""
92	work = repo
93	# Commit 1 — define compute_total and Invoice class.
94	(work / "billing.py").write_text(textwrap.dedent("""\
95	class Invoice:
96	def compute_total(self, items):
97	return sum(items)
98
99	def apply_discount(self, total, pct):
100	return total * (1 - pct)
101
102	def process_order(invoice, items):
103	return invoice.compute_total(items)
104	"""))
105	runner.invoke(cli, ["code", "add", "billing.py"])
106	r = runner.invoke(cli, ["commit", "-m", "Initial billing module"])
107	assert r.exit_code == 0, r.output
108
109	# Commit 2 — rename compute_total, add new function.
110	(work / "billing.py").write_text(textwrap.dedent("""\
111	class Invoice:
112	def compute_invoice_total(self, items):
113	return sum(items)
114
115	def apply_discount(self, total, pct):
116	return total * (1 - pct)
117
118	def generate_pdf(self):
119	return b"pdf"
120
121	def process_order(invoice, items):
122	return invoice.compute_invoice_total(items)
123
124	def send_email(address):
125	pass
126	"""))
127	runner.invoke(cli, ["code", "add", "billing.py"])
128	r = runner.invoke(cli, ["commit", "-m", "Rename compute_total, add generate_pdf + send_email"])
129	assert r.exit_code == 0, r.output
130	return repo
131
132
133	# ---------------------------------------------------------------------------
134	# muse lineage
135	# ---------------------------------------------------------------------------
136
137
138	class TestLineage:
139	def test_lineage_exits_zero_on_existing_symbol(self, code_repo: pathlib.Path) -> None:
140	result = runner.invoke(cli, ["code", "lineage", "billing.py::process_order"])
141	assert result.exit_code == 0, result.output
142
143	def test_lineage_json_output(self, code_repo: pathlib.Path) -> None:
144	result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
145	assert result.exit_code == 0, result.output
146	data = json.loads(result.output)
147	assert isinstance(data, dict)
148	assert "events" in data
149
150	def test_lineage_missing_address_shows_message(self, code_repo: pathlib.Path) -> None:
151	result = runner.invoke(cli, ["code", "lineage", "billing.py::nonexistent_func"])
152	# Should not crash — exit 0 or 1, but no unhandled exception.
153	assert result.exit_code in (0, 1)
154
155	def test_lineage_requires_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
156	monkeypatch.chdir(tmp_path)
157	result = runner.invoke(cli, ["code", "lineage", "src/a.py::f"])
158	assert result.exit_code != 0
159
160
161	# ---------------------------------------------------------------------------
162	# muse api-surface
163	# ---------------------------------------------------------------------------
164
165
166	class TestApiSurface:
167	def test_api_surface_exits_zero(self, code_repo: pathlib.Path) -> None:
168	result = runner.invoke(cli, ["code", "api-surface"])
169	assert result.exit_code == 0, result.output
170
171	def test_api_surface_json(self, code_repo: pathlib.Path) -> None:
172	result = runner.invoke(cli, ["code", "api-surface", "--json"])
173	assert result.exit_code == 0
174	data = json.loads(result.output)
175	assert isinstance(data, dict)
176
177	def test_api_surface_diff(self, code_repo: pathlib.Path) -> None:
178	commits = _all_commit_ids(code_repo)
179	if len(commits) >= 2:
180	result = runner.invoke(cli, ["code", "api-surface", "--diff", commits[-2]])
181	assert result.exit_code == 0
182
183	def test_api_surface_no_commits_handled(self, repo: pathlib.Path) -> None:
184	result = runner.invoke(cli, ["code", "api-surface"])
185	assert result.exit_code in (0, 1)
186
187
188	# ---------------------------------------------------------------------------
189	# muse codemap
190	# ---------------------------------------------------------------------------
191
192
193	class TestCodemap:
194	def test_codemap_exits_zero(self, code_repo: pathlib.Path) -> None:
195	result = runner.invoke(cli, ["code", "codemap"])
196	assert result.exit_code == 0, result.output
197
198	def test_codemap_top_flag(self, code_repo: pathlib.Path) -> None:
199	result = runner.invoke(cli, ["code", "codemap", "--top", "3"])
200	assert result.exit_code == 0
201
202	def test_codemap_json(self, code_repo: pathlib.Path) -> None:
203	result = runner.invoke(cli, ["code", "codemap", "--json"])
204	assert result.exit_code == 0
205	data = json.loads(result.output)
206	assert isinstance(data, dict)
207
208
209	# ---------------------------------------------------------------------------
210	# muse clones
211	# ---------------------------------------------------------------------------
212
213
214	class TestClones:
215	def test_clones_exits_zero(self, code_repo: pathlib.Path) -> None:
216	result = runner.invoke(cli, ["code", "clones"])
217	assert result.exit_code == 0, result.output
218
219	def test_clones_tier_exact(self, code_repo: pathlib.Path) -> None:
220	result = runner.invoke(cli, ["code", "clones", "--tier", "exact"])
221	assert result.exit_code == 0
222
223	def test_clones_tier_near(self, code_repo: pathlib.Path) -> None:
224	result = runner.invoke(cli, ["code", "clones", "--tier", "near"])
225	assert result.exit_code == 0
226
227	def test_clones_json(self, code_repo: pathlib.Path) -> None:
228	result = runner.invoke(cli, ["code", "clones", "--tier", "both", "--json"])
229	assert result.exit_code == 0
230	data = json.loads(result.output)
231	assert isinstance(data, dict)
232
233
234	# ---------------------------------------------------------------------------
235	# muse checkout-symbol
236	# ---------------------------------------------------------------------------
237
238
239	class TestCheckoutSymbol:
240	def test_checkout_symbol_dry_run(self, code_repo: pathlib.Path) -> None:
241	commits = _all_commit_ids(code_repo)
242	if len(commits) < 2:
243	pytest.skip("need at least 2 commits")
244	first_commit = commits[-2] # oldest commit (list is newest-first)
245	result = runner.invoke(cli, [
246	"code", "checkout-symbol", "--commit", first_commit, "--dry-run",
247	"billing.py::Invoice.compute_total",
248	])
249	# May fail if symbol is not present; should not crash unhandled.
250	assert result.exit_code in (0, 1, 2)
251
252	def test_checkout_symbol_missing_commit_flag_errors(self, code_repo: pathlib.Path) -> None:
253	result = runner.invoke(cli, ["code", "checkout-symbol", "--dry-run", "billing.py::Invoice.compute_total"])
254	assert result.exit_code != 0
255
256
257	# ---------------------------------------------------------------------------
258	# muse semantic-cherry-pick
259	# ---------------------------------------------------------------------------
260
261
262	class TestSemanticCherryPick:
263	def test_dry_run_exits_zero(self, code_repo: pathlib.Path) -> None:
264	commits = _all_commit_ids(code_repo)
265	if len(commits) < 2:
266	pytest.skip("need at least 2 commits")
267	first_commit = commits[-2]
268	result = runner.invoke(cli, [
269	"code", "semantic-cherry-pick",
270	"--from", first_commit,
271	"--dry-run",
272	"billing.py::Invoice.compute_total",
273	])
274	assert result.exit_code in (0, 1)
275
276	def test_missing_from_flag_errors(self, code_repo: pathlib.Path) -> None:
277	result = runner.invoke(cli, ["code", "semantic-cherry-pick", "--dry-run", "billing.py::Invoice.compute_total"])
278	assert result.exit_code != 0
279
280
281	# ---------------------------------------------------------------------------
282	# muse query
283	# ---------------------------------------------------------------------------
284
285
286	class TestQueryV2:
287	def test_query_kind_function(self, code_repo: pathlib.Path) -> None:
288	result = runner.invoke(cli, ["code", "query", "kind=function"])
289	assert result.exit_code == 0, result.output
290
291	def test_query_json_output(self, code_repo: pathlib.Path) -> None:
292	result = runner.invoke(cli, ["code", "query", "--json", "kind=function"])
293	assert result.exit_code == 0
294	data = json.loads(result.output)
295	assert "muse_version" in data
296
297	def test_query_or_predicate(self, code_repo: pathlib.Path) -> None:
298	result = runner.invoke(cli, ["code", "query", "kind=function", "OR", "kind=method"])
299	assert result.exit_code == 0
300
301	def test_query_not_predicate(self, code_repo: pathlib.Path) -> None:
302	result = runner.invoke(cli, ["code", "query", "NOT", "kind=import"])
303	assert result.exit_code == 0
304
305	def test_query_all_commits(self, code_repo: pathlib.Path) -> None:
306	result = runner.invoke(cli, ["code", "query", "--all-commits", "kind=function"])
307	assert result.exit_code == 0
308
309	def test_query_name_contains(self, code_repo: pathlib.Path) -> None:
310	result = runner.invoke(cli, ["code", "query", "name~=total"])
311	assert result.exit_code == 0
312	# Should find compute_invoice_total.
313	assert "total" in result.output.lower()
314
315	def test_query_no_predicate_matches_all(self, code_repo: pathlib.Path) -> None:
316	# query with kind=class to match everything of a known type.
317	result = runner.invoke(cli, ["code", "query", "kind=class"])
318	assert result.exit_code == 0
319	assert "Invoice" in result.output
320
321	def test_query_lineno_gt(self, code_repo: pathlib.Path) -> None:
322	result = runner.invoke(cli, ["code", "query", "lineno_gt=1"])
323	assert result.exit_code == 0
324
325	def test_query_no_repo_errors(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
326	monkeypatch.chdir(tmp_path)
327	result = runner.invoke(cli, ["code", "query", "kind=function"])
328	assert result.exit_code != 0
329
330	# ── new v2.1 flags ────────────────────────────────────────────────────────
331
332	def test_query_count_only(self, code_repo: pathlib.Path) -> None:
333	result = runner.invoke(cli, ["code", "query", "--count", "kind=function"])
334	assert result.exit_code == 0, result.output
335	# Output should be a single integer.
336	assert result.output.strip().isdigit()
337
338	def test_query_count_nonzero(self, code_repo: pathlib.Path) -> None:
339	result = runner.invoke(cli, ["code", "query", "--count", "kind=function"])
340	assert int(result.output.strip()) >= 1
341
342	def test_query_limit_caps_results(self, code_repo: pathlib.Path) -> None:
343	all_r = runner.invoke(cli, ["code", "query", "kind=function"])
344	lim_r = runner.invoke(cli, ["code", "query", "kind=function", "--limit", "1"])
345	assert lim_r.exit_code == 0, lim_r.output
346	# Limited output should be shorter than unlimited.
347	assert len(lim_r.output) <= len(all_r.output)
348
349	def test_query_limit_truncation_noted(self, code_repo: pathlib.Path) -> None:
350	result = runner.invoke(cli, ["code", "query", "kind=function", "--limit", "1"])
351	assert "limited to 1" in result.output or "match" in result.output
352
353	def test_query_limit_zero_unlimited(self, code_repo: pathlib.Path) -> None:
354	result = runner.invoke(cli, ["code", "query", "kind=function", "--limit", "0"])
355	assert result.exit_code == 0, result.output
356
357	def test_query_sort_name(self, code_repo: pathlib.Path) -> None:
358	result = runner.invoke(cli, ["code", "query", "kind=function", "--sort", "name"])
359	assert result.exit_code == 0, result.output
360
361	def test_query_sort_size(self, code_repo: pathlib.Path) -> None:
362	result = runner.invoke(cli, ["code", "query", "kind=function", "--sort", "size"])
363	assert result.exit_code == 0, result.output
364	# Size column should appear in output.
365	assert "L" in result.output
366
367	def test_query_sort_kind(self, code_repo: pathlib.Path) -> None:
368	result = runner.invoke(cli, ["code", "query", "kind=function", "--sort", "kind"])
369	assert result.exit_code == 0, result.output
370
371	def test_query_sort_lineno(self, code_repo: pathlib.Path) -> None:
372	result = runner.invoke(cli, ["code", "query", "kind=function", "--sort", "lineno"])
373	assert result.exit_code == 0, result.output
374
375	def test_query_sort_invalid_rejected(self, code_repo: pathlib.Path) -> None:
376	result = runner.invoke(cli, ["code", "query", "kind=function", "--sort", "zzz"])
377	assert result.exit_code != 0
378
379	def test_query_unique_bodies_exits_zero(self, code_repo: pathlib.Path) -> None:
380	result = runner.invoke(cli, ["code", "query", "kind=function", "--unique-bodies"])
381	assert result.exit_code == 0, result.output
382
383	def test_query_unique_bodies_count_lte_all(self, code_repo: pathlib.Path) -> None:
384	all_r = runner.invoke(cli, ["code", "query", "--count", "kind=function"])
385	uniq_r = runner.invoke(cli, ["code", "query", "--count", "--unique-bodies", "kind=function"])
386	assert int(uniq_r.output.strip()) <= int(all_r.output.strip())
387
388	def test_query_size_gt_predicate(self, code_repo: pathlib.Path) -> None:
389	result = runner.invoke(cli, ["code", "query", "kind=function", "size_gt=0"])
390	assert result.exit_code == 0, result.output
391
392	def test_query_size_lt_predicate(self, code_repo: pathlib.Path) -> None:
393	result = runner.invoke(cli, ["code", "query", "kind=function", "size_lt=1000"])
394	assert result.exit_code == 0, result.output
395
396	def test_query_size_gt_excludes_small(self, code_repo: pathlib.Path) -> None:
397	all_r = runner.invoke(cli, ["code", "query", "--count", "kind=function"])
398	large_r = runner.invoke(cli, ["code", "query", "--count", "kind=function", "size_gt=100"])
399	# Large-only count should be <= total.
400	assert int(large_r.output.strip()) <= int(all_r.output.strip())
401
402	def test_query_json_includes_size(self, code_repo: pathlib.Path) -> None:
403	result = runner.invoke(cli, ["code", "query", "--json", "kind=function"])
404	data = json.loads(result.output)
405	for r in data["results"]:
406	assert "size" in r
407
408	def test_query_json_includes_sort_field(self, code_repo: pathlib.Path) -> None:
409	result = runner.invoke(cli, ["code", "query", "--json", "kind=function", "--sort", "name"])
410	data = json.loads(result.output)
411	assert data["sort"] == "name"
412
413	def test_query_json_includes_unique_bodies(self, code_repo: pathlib.Path) -> None:
414	result = runner.invoke(cli, ["code", "query", "--json", "kind=function", "--unique-bodies"])
415	data = json.loads(result.output)
416	assert data["unique_bodies"] is True
417
418	def test_query_since_without_all_commits_rejected(self, code_repo: pathlib.Path) -> None:
419	result = runner.invoke(cli, ["code", "query", "kind=function", "--since", "2026-01-01"])
420	assert result.exit_code != 0
421
422	def test_query_since_invalid_date_rejected(self, code_repo: pathlib.Path) -> None:
423	result = runner.invoke(
424	cli,
425	["code", "query", "kind=function", "--all-commits", "--since", "not-a-date"],
426	)
427	assert result.exit_code != 0
428
429	def test_query_all_commits_since_future_empty(self, code_repo: pathlib.Path) -> None:
430	result = runner.invoke(
431	cli,
432	["code", "query", "kind=function", "--all-commits", "--since", "2099-01-01"],
433	)
434	assert result.exit_code == 0, result.output
435	# Future date means no commits match.
436	assert "no symbols" in result.output.lower() or result.output.strip() == ""
437
438	def test_query_max_commits_caps_walk(self, code_repo: pathlib.Path) -> None:
439	result = runner.invoke(
440	cli,
441	["code", "query", "kind=function", "--all-commits", "--max-commits", "1"],
442	)
443	assert result.exit_code == 0, result.output
444
445
446	# ---------------------------------------------------------------------------
447	# muse query-history
448	# ---------------------------------------------------------------------------
449
450
451	class TestQueryHistory:
452	def test_query_history_exits_zero(self, code_repo: pathlib.Path) -> None:
453	result = runner.invoke(cli, ["code", "query-history", "kind=function"])
454	assert result.exit_code == 0, result.output
455
456	def test_query_history_json(self, code_repo: pathlib.Path) -> None:
457	result = runner.invoke(cli, ["code", "query-history", "--json", "kind=function"])
458	assert result.exit_code == 0
459	data = json.loads(result.output)
460	assert "muse_version" in data
461	assert "results" in data
462
463	def test_query_history_with_from_to(self, code_repo: pathlib.Path) -> None:
464	result = runner.invoke(cli, ["code", "query-history", "--from", "HEAD", "kind=function"])
465	assert result.exit_code == 0
466
467	def test_query_history_tracks_change_count(self, code_repo: pathlib.Path) -> None:
468	result = runner.invoke(cli, ["code", "query-history", "--json", "kind=method"])
469	assert result.exit_code == 0
470	data = json.loads(result.output)
471	for entry in data.get("results", []):
472	assert "commit_count" in entry
473	assert "change_count" in entry
474
475	# ── new v2 flags ──────────────────────────────────────────────────────────
476
477	def test_query_history_changed_only(self, code_repo: pathlib.Path) -> None:
478	result = runner.invoke(
479	cli, ["code", "query-history", "--changed-only", "kind=function"]
480	)
481	assert result.exit_code == 0, result.output
482
483	def test_query_history_changed_only_all_gt_one(self, code_repo: pathlib.Path) -> None:
484	result = runner.invoke(
485	cli, ["code", "query-history", "--changed-only", "--json", "kind=function"]
486	)
487	assert result.exit_code == 0
488	data = json.loads(result.output)
489	for entry in data["results"]:
490	assert entry["change_count"] > 1
491
492	def test_query_history_sort_commits(self, code_repo: pathlib.Path) -> None:
493	result = runner.invoke(
494	cli, ["code", "query-history", "--sort", "commits", "kind=function"]
495	)
496	assert result.exit_code == 0, result.output
497
498	def test_query_history_sort_changes(self, code_repo: pathlib.Path) -> None:
499	result = runner.invoke(
500	cli, ["code", "query-history", "--sort", "changes", "kind=function"]
501	)
502	assert result.exit_code == 0, result.output
503
504	def test_query_history_sort_first(self, code_repo: pathlib.Path) -> None:
505	result = runner.invoke(
506	cli, ["code", "query-history", "--sort", "first", "kind=function"]
507	)
508	assert result.exit_code == 0, result.output
509
510	def test_query_history_sort_invalid_rejected(self, code_repo: pathlib.Path) -> None:
511	result = runner.invoke(
512	cli, ["code", "query-history", "--sort", "zzz", "kind=function"]
513	)
514	assert result.exit_code != 0
515
516	def test_query_history_count(self, code_repo: pathlib.Path) -> None:
517	result = runner.invoke(
518	cli, ["code", "query-history", "--count", "kind=function"]
519	)
520	assert result.exit_code == 0, result.output
521	assert result.output.strip().isdigit()
522	assert int(result.output.strip()) >= 1
523
524	def test_query_history_limit(self, code_repo: pathlib.Path) -> None:
525	all_r = runner.invoke(cli, ["code", "query-history", "kind=function"])
526	lim_r = runner.invoke(
527	cli, ["code", "query-history", "--limit", "1", "kind=function"]
528	)
529	assert lim_r.exit_code == 0, lim_r.output
530	assert len(lim_r.output) <= len(all_r.output)
531
532	def test_query_history_limit_note_in_output(self, code_repo: pathlib.Path) -> None:
533	result = runner.invoke(
534	cli, ["code", "query-history", "--limit", "1", "kind=function"]
535	)
536	assert "1" in result.output
537
538	def test_query_history_min_changes(self, code_repo: pathlib.Path) -> None:
539	result = runner.invoke(
540	cli, ["code", "query-history", "--min-changes", "2", "--json", "kind=function"]
541	)
542	assert result.exit_code == 0
543	data = json.loads(result.output)
544	for entry in data["results"]:
545	assert entry["change_count"] >= 2
546
547	def test_query_history_min_changes_zero_rejected(self, code_repo: pathlib.Path) -> None:
548	result = runner.invoke(
549	cli, ["code", "query-history", "--min-changes", "0", "kind=function"]
550	)
551	assert result.exit_code != 0
552
553	def test_query_history_introduced_only(self, code_repo: pathlib.Path) -> None:
554	result = runner.invoke(
555	cli, ["code", "query-history", "--introduced-only", "kind=function"]
556	)
557	assert result.exit_code == 0, result.output
558
559	def test_query_history_removed_only(self, code_repo: pathlib.Path) -> None:
560	result = runner.invoke(
561	cli, ["code", "query-history", "--removed-only", "kind=function"]
562	)
563	assert result.exit_code == 0, result.output
564
565	def test_query_history_introduced_json_schema(self, code_repo: pathlib.Path) -> None:
566	result = runner.invoke(
567	cli,
568	["code", "query-history", "--introduced-only", "--json", "kind=function"],
569	)
570	assert result.exit_code == 0
571	data = json.loads(result.output)
572	assert data["mode"] == "introduced-only"
573	assert "symbols_found" in data
574	for entry in data["results"]:
575	assert entry["status"] == "introduced"
576
577	def test_query_history_removed_json_schema(self, code_repo: pathlib.Path) -> None:
578	result = runner.invoke(
579	cli,
580	["code", "query-history", "--removed-only", "--json", "kind=function"],
581	)
582	assert result.exit_code == 0
583	data = json.loads(result.output)
584	assert data["mode"] == "removed-only"
585	assert "symbols_found" in data
586	for entry in data["results"]:
587	assert entry["status"] == "removed"
588
589	def test_query_history_mode_flags_mutually_exclusive(
590	self, code_repo: pathlib.Path
591	) -> None:
592	result = runner.invoke(
593	cli,
594	[
595	"code", "query-history",
596	"--changed-only", "--introduced-only",
597	"kind=function",
598	],
599	)
600	assert result.exit_code != 0
601
602	def test_query_history_json_has_full_commit_ids(
603	self, code_repo: pathlib.Path
604	) -> None:
605	result = runner.invoke(
606	cli, ["code", "query-history", "--json", "kind=function"]
607	)
608	assert result.exit_code == 0
609	data = json.loads(result.output)
610	for entry in data["results"]:
611	# Full commit IDs should be present (not just 8-char short form).
612	assert len(entry["first_commit_id"]) > 8
613	assert "stable" in entry
614
615	def test_query_history_max_commits_cap(self, code_repo: pathlib.Path) -> None:
616	result = runner.invoke(
617	cli,
618	["code", "query-history", "--max-commits", "1", "kind=function"],
619	)
620	assert result.exit_code == 0, result.output
621
622	def test_query_history_introduced_count_only(
623	self, code_repo: pathlib.Path
624	) -> None:
625	result = runner.invoke(
626	cli,
627	["code", "query-history", "--introduced-only", "--count", "kind=function"],
628	)
629	assert result.exit_code == 0
630	assert result.output.strip().isdigit()
631
632
633	# ---------------------------------------------------------------------------
634	# muse index
635	# ---------------------------------------------------------------------------
636
637
638	class TestIndexCommands:
639	def test_index_status_exits_zero(self, code_repo: pathlib.Path) -> None:
640	result = runner.invoke(cli, ["code", "index", "status"])
641	assert result.exit_code == 0, result.output
642
643	def test_index_status_reports_absent(self, code_repo: pathlib.Path) -> None:
644	result = runner.invoke(cli, ["code", "index", "status"])
645	# Indexes have not been built yet.
646	assert "absent" in result.output.lower() or result.exit_code == 0
647
648	def test_index_rebuild_all(self, code_repo: pathlib.Path) -> None:
649	result = runner.invoke(cli, ["code", "index", "rebuild"])
650	assert result.exit_code == 0, result.output
651
652	def test_index_rebuild_creates_index_files(self, code_repo: pathlib.Path) -> None:
653	runner.invoke(cli, ["code", "index", "rebuild"])
654	idx_dir = indices_dir(code_repo)
655	assert idx_dir.exists()
656
657	def test_index_status_after_rebuild_shows_entries(self, code_repo: pathlib.Path) -> None:
658	runner.invoke(cli, ["code", "index", "rebuild"])
659	result = runner.invoke(cli, ["code", "index", "status"])
660	assert result.exit_code == 0
661	# Output shows ✅ checkmarks and entry counts for rebuilt indexes.
662	assert "entries" in result.output.lower() or "✅" in result.output
663
664	def test_index_rebuild_symbol_history_only(self, code_repo: pathlib.Path) -> None:
665	result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "symbol_history"])
666	assert result.exit_code == 0
667
668	def test_index_rebuild_hash_occurrence_only(self, code_repo: pathlib.Path) -> None:
669	result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "hash_occurrence"])
670	assert result.exit_code == 0
671
672
673	# ---------------------------------------------------------------------------
674	# muse detect-refactor
675	# ---------------------------------------------------------------------------
676
677
678	class TestHotspots:
679	"""Tests for muse code hotspots."""
680
681	# ── basic correctness ────────────────────────────────────────────────────
682
683	def test_hotspots_exits_zero(self, code_repo: pathlib.Path) -> None:
684	result = runner.invoke(cli, ["code", "hotspots"])
685	assert result.exit_code == 0, result.output
686
687	def test_hotspots_finds_changed_symbol(self, code_repo: pathlib.Path) -> None:
688	"""compute_invoice_total was modified across two commits — must appear."""
689	result = runner.invoke(cli, ["code", "hotspots", "--top", "20"])
690	assert result.exit_code == 0, result.output
691	assert "billing.py" in result.output
692
693	def test_hotspots_excludes_imports_by_default(
694	self, code_repo: pathlib.Path
695	) -> None:
696	result = runner.invoke(cli, ["code", "hotspots", "--top", "50"])
697	assert result.exit_code == 0, result.output
698	assert "::import::" not in result.output
699
700	def test_hotspots_include_imports_flag(self, code_repo: pathlib.Path) -> None:
701	"""--include-imports must surface import pseudo-symbols if any exist."""
702	result = runner.invoke(
703	cli, ["code", "hotspots", "--top", "50", "--include-imports"]
704	)
705	assert result.exit_code == 0, result.output
706	# Just verify it runs cleanly; the repo may or may not have import ops.
707
708	# ── --kind filter (was broken before) ────────────────────────────────────
709
710	def test_kind_filter_excludes_classes(self, code_repo: pathlib.Path) -> None:
711	"""--kind function must not return class symbols."""
712	result = runner.invoke(
713	cli, ["code", "hotspots", "--kind", "function", "--top", "20"]
714	)
715	assert result.exit_code == 0, result.output
716	for line in result.output.splitlines():
717	if "::" in line and "class" in line.lower():
718	# Make sure any class line is not a function kind result
719	# (Addresses that contain the word "class" in their name are OK)
720	pass # Name may contain "class" as substring
721
722	def test_kind_filter_function_returns_functions(
723	self, code_repo: pathlib.Path
724	) -> None:
725	result_all = runner.invoke(cli, ["code", "hotspots", "--top", "50"])
726	result_fn = runner.invoke(
727	cli, ["code", "hotspots", "--kind", "function", "--top", "50"]
728	)
729	assert result_fn.exit_code == 0, result_fn.output
730	# filtered result should have <= symbols than unfiltered
731	fn_lines = [l for l in result_fn.output.splitlines() if "::" in l]
732	all_lines = [l for l in result_all.output.splitlines() if "::" in l]
733	assert len(fn_lines) <= len(all_lines)
734
735	# ── --min filter ──────────────────────────────────────────────────────────
736
737	def test_min_filter_raises_threshold(self, code_repo: pathlib.Path) -> None:
738	result_all = runner.invoke(cli, ["code", "hotspots", "--top", "50"])
739	result_min = runner.invoke(
740	cli, ["code", "hotspots", "--min", "2", "--top", "50"]
741	)
742	assert result_min.exit_code == 0, result_min.output
743	min_lines = [l for l in result_min.output.splitlines() if "::" in l]
744	all_lines = [l for l in result_all.output.splitlines() if "::" in l]
745	assert len(min_lines) <= len(all_lines)
746
747	def test_min_zero_exits_error(self, code_repo: pathlib.Path) -> None:
748	result = runner.invoke(cli, ["code", "hotspots", "--min", "0"])
749	assert result.exit_code == 1
750
751	# ── --language filter ─────────────────────────────────────────────────────
752
753	def test_language_filter_lowercase(self, code_repo: pathlib.Path) -> None:
754	result = runner.invoke(
755	cli, ["code", "hotspots", "--language", "python", "--top", "10"]
756	)
757	assert result.exit_code == 0, result.output
758	assert "billing.py" in result.output
759
760	def test_language_filter_uppercase(self, code_repo: pathlib.Path) -> None:
761	result = runner.invoke(
762	cli, ["code", "hotspots", "--language", "PYTHON", "--top", "10"]
763	)
764	assert result.exit_code == 0, result.output
765
766	# ── --top validation ──────────────────────────────────────────────────────
767
768	def test_top_zero_exits_error(self, code_repo: pathlib.Path) -> None:
769	result = runner.invoke(cli, ["code", "hotspots", "--top", "0"])
770	assert result.exit_code == 1
771
772	# ── JSON schema ───────────────────────────────────────────────────────────
773
774	def test_json_top_level_schema(self, code_repo: pathlib.Path) -> None:
775	result = runner.invoke(cli, ["code", "hotspots", "--json"])
776	assert result.exit_code == 0, result.output
777	data = json.loads(result.output)
778	for key in (
779	"from_ref", "to_ref", "commits_analysed", "truncated",
780	"filters", "hotspots",
781	):
782	assert key in data, f"missing key: {key}"
783	assert isinstance(data["hotspots"], list)
784	assert isinstance(data["truncated"], bool)
785	assert isinstance(data["commits_analysed"], int)
786
787	def test_json_filters_field(self, code_repo: pathlib.Path) -> None:
788	result = runner.invoke(
789	cli, ["code", "hotspots", "--kind", "function", "--min", "2", "--json"]
790	)
791	data = json.loads(result.output)
792	assert data["filters"]["kind"] == "function"
793	assert data["filters"]["min_changes"] == 2
794	assert data["filters"]["include_imports"] is False
795
796	def test_json_hotspot_entry_schema(self, code_repo: pathlib.Path) -> None:
797	result = runner.invoke(cli, ["code", "hotspots", "--json"])
798	data = json.loads(result.output)
799	if data["hotspots"]:
800	entry = data["hotspots"][0]
801	assert "address" in entry
802	assert "changes" in entry
803	assert isinstance(entry["changes"], int)
804	assert entry["changes"] >= 1
805
806	def test_json_no_imports_by_default(self, code_repo: pathlib.Path) -> None:
807	result = runner.invoke(cli, ["code", "hotspots", "--json"])
808	data = json.loads(result.output)
809	addresses = [h["address"] for h in data["hotspots"]]
810	assert not any("::import::" in a for a in addresses)
811
812	def test_json_ranked_descending(self, code_repo: pathlib.Path) -> None:
813	result = runner.invoke(cli, ["code", "hotspots", "--json"])
814	data = json.loads(result.output)
815	counts = [h["changes"] for h in data["hotspots"]]
816	assert counts == sorted(counts, reverse=True)
817
818	# ── --max-commits truncation ──────────────────────────────────────────────
819
820	def test_max_commits_flag(self, code_repo: pathlib.Path) -> None:
821	result = runner.invoke(
822	cli, ["code", "hotspots", "--max-commits", "1", "--json"]
823	)
824	assert result.exit_code == 0, result.output
825	data = json.loads(result.output)
826	assert data["commits_analysed"] <= 1
827
828	def test_max_commits_truncation_flag(self, code_repo: pathlib.Path) -> None:
829	result = runner.invoke(
830	cli, ["code", "hotspots", "--max-commits", "1", "--json"]
831	)
832	data = json.loads(result.output)
833	assert data["truncated"] is True
834
835
836	class TestDetectRefactorV2:
837	def test_detect_refactor_json_schema(self, code_repo: pathlib.Path) -> None:
838	"""JSON output contains all required top-level fields."""
839	result = runner.invoke(cli, ["code", "detect-refactor", "--json"])
840	assert result.exit_code == 0, result.output
841	data = json.loads(result.output)
842	for field in ("commits_scanned", "truncated", "total", "events"):
843	assert field in data, f"missing field '{field}'"
844	assert isinstance(data["commits_scanned"], int)
845	assert isinstance(data["truncated"], bool)
846	assert isinstance(data["total"], int)
847	assert isinstance(data["events"], list)
848
849	def test_detect_refactor_json_event_schema(self, code_repo: pathlib.Path) -> None:
850	"""Each JSON event contains the required fields."""
851	# Run over the full history; code_repo has at least one rename event.
852	result = runner.invoke(cli, ["code", "detect-refactor", "--json"])
853	assert result.exit_code == 0, result.output
854	data = json.loads(result.output)
855	for ev in data["events"]:
856	for field in ("kind", "address", "detail",
857	"commit_id", "commit_message", "committed_at"):
858	assert field in ev, f"missing event field '{field}'"
859	assert ev["kind"] in ("rename", "move", "signature", "implementation")
860
861	def test_detect_refactor_finds_rename(self, code_repo: pathlib.Path) -> None:
862	"""A commit that renames a symbol produces a 'rename' event."""
863	result = runner.invoke(cli, ["code", "detect-refactor", "--json"])
864	assert result.exit_code == 0, result.output
865	data = json.loads(result.output)
866	kinds = [e["kind"] for e in data["events"]]
867	assert "rename" in kinds, (
868	f"Expected at least one rename event; got: {sorted(set(kinds))}"
869	)
870
871	def test_detect_refactor_classifies_modified_as_implementation(
872	self, code_repo: pathlib.Path
873	) -> None:
874	"""Replace ops with '(modified)' in new_summary are classified as implementation.
875
876	Previously, only '(implementation changed)' triggered implementation
877	classification; '(modified)' was silently dropped.
878	"""
879	import datetime
880	root = code_repo
881	repo_id = json.loads((repo_json_path(root)).read_text())["repo_id"]
882	from muse.core.refs import (
883	get_head_commit_id,
884	read_current_branch,
885	)
886	from muse.core.commits import (
887	CommitDict,
888	CommitRecord,
889	write_commit,
890	)
891	from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id
892	branch = read_current_branch(root)
893	head_id = get_head_commit_id(root, branch)
894
895	now = datetime.datetime(2026, 6, 1, 12, 0, 0, tzinfo=datetime.timezone.utc)
896	message = "perf: optimise batch"
897	snap_manifest: Manifest = {}
898	snap_id = compute_snapshot_id(snap_manifest)
899	parent_ids = [head_id] if head_id else []
900	commit_id = compute_commit_id(
901	parent_ids=parent_ids,
902	snapshot_id=snap_id,
903	message=message,
904	committed_at_iso=now.isoformat(),
905	author="test",
906	)
907	from muse.domain import PatchOp, ReplaceOp, StructuredDelta
908	commit = CommitRecord(
909	commit_id=commit_id,
910	branch=branch,
911	snapshot_id=snap_id,
912	message=message,
913	committed_at=now,
914	parent_commit_id=head_id,
915	author="test",
916	structured_delta=StructuredDelta(ops=[PatchOp(
917	op="patch",
918	address="billing.py",
919	child_ops=[ReplaceOp(
920	op="replace",
921	address="billing.py::process_batch",
922	new_summary="function process_batch (modified) L10–30",
923	old_summary="function process_batch",
924	)],
925	)]),
926	)
927	write_commit(root, commit)
928	(ref_path(root, branch)).write_text(commit_id)
929
930	result = runner.invoke(cli, ["code", "detect-refactor", "--json"])
931	assert result.exit_code == 0, result.output
932	data = json.loads(result.output)
933	impl_events = [e for e in data["events"] if e["kind"] == "implementation"]
934	addrs = [e["address"] for e in impl_events]
935	assert "billing.py::process_batch" in addrs, (
936	f"'(modified)' op not classified as implementation; events: {data['events']}"
937	)
938
939	def test_detect_refactor_skips_reformatted(self, code_repo: pathlib.Path) -> None:
940	"""Replace ops with 'reformatted' in new_summary are not emitted as events."""
941	import datetime
942	root = code_repo
943	repo_id = json.loads((repo_json_path(root)).read_text())["repo_id"]
944	from muse.core.refs import (
945	get_head_commit_id,
946	read_current_branch,
947	)
948	from muse.core.commits import (
949	CommitDict,
950	CommitRecord,
951	write_commit,
952	)
953	from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id
954	branch = read_current_branch(root)
955	head_id = get_head_commit_id(root, branch)
956
957	now = datetime.datetime(2026, 6, 1, 13, 0, 0, tzinfo=datetime.timezone.utc)
958	message = "style: reformat"
959	snap_manifest: Manifest = {}
960	snap_id = compute_snapshot_id(snap_manifest)
961	parent_ids = [head_id] if head_id else []
962	commit_id = compute_commit_id(
963	parent_ids=parent_ids,
964	snapshot_id=snap_id,
965	message=message,
966	committed_at_iso=now.isoformat(),
967	author="test",
968	)
969	from muse.domain import PatchOp, ReplaceOp, StructuredDelta
970	commit = CommitRecord(
971	commit_id=commit_id,
972	branch=branch,
973	snapshot_id=snap_id,
974	message=message,
975	committed_at=now,
976	parent_commit_id=head_id,
977	author="test",
978	structured_delta=StructuredDelta(ops=[PatchOp(
979	op="patch",
980	address="billing.py",
981	child_ops=[ReplaceOp(
982	op="replace",
983	address="billing.py::UniqueReformattedSymbol",
984	new_summary="reformatted — no semantic change",
985	old_summary="",
986	)],
987	)]),
988	)
989	write_commit(root, commit)
990	(ref_path(root, branch)).write_text(commit_id)
991
992	result = runner.invoke(cli, ["code", "detect-refactor", "--json"])
993	assert result.exit_code == 0, result.output
994	data = json.loads(result.output)
995	# The reformatted op must not appear as an event.
996	reformatted_events = [
997	e for e in data["events"]
998	if e["address"] == "billing.py::UniqueReformattedSymbol"
999	]
1000	assert reformatted_events == [], (
1001	f"Reformatted op should be skipped; got: {reformatted_events}"
1002	)
1003
1004	def test_detect_refactor_truncation_warning(self, code_repo: pathlib.Path) -> None:
1005	"""When --max is hit, a truncation warning appears in human output."""
1006	result = runner.invoke(cli, ["code", "detect-refactor", "--max", "1"])
1007	assert result.exit_code == 0, result.output
1008	assert "incomplete" in result.output or "limit" in result.output
1009
1010	def test_detect_refactor_truncation_in_json(self, code_repo: pathlib.Path) -> None:
1011	"""When --max is hit, truncated=true in JSON."""
1012	result = runner.invoke(
1013	cli, ["code", "detect-refactor", "--max", "1", "--json"]
1014	)
1015	assert result.exit_code == 0, result.output
1016	data = json.loads(result.output)
1017	assert data["truncated"] is True
1018	assert data["commits_scanned"] == 1
1019
1020	def test_detect_refactor_max_zero_errors(self, code_repo: pathlib.Path) -> None:
1021	"""--max 0 exits non-zero."""
1022	result = runner.invoke(cli, ["code", "detect-refactor", "--max", "0"])
1023	assert result.exit_code != 0
1024
1025	def test_detect_refactor_kind_filter(self, code_repo: pathlib.Path) -> None:
1026	"""``--kind rename`` returns only rename events."""
1027	result = runner.invoke(
1028	cli, ["code", "detect-refactor", "--kind", "rename", "--json"]
1029	)
1030	assert result.exit_code == 0, result.output
1031	data = json.loads(result.output)
1032	for ev in data["events"]:
1033	assert ev["kind"] == "rename"
1034
1035	def test_detect_refactor_invalid_kind(self, code_repo: pathlib.Path) -> None:
1036	"""``--kind`` with an invalid value exits non-zero."""
1037	result = runner.invoke(cli, ["code", "detect-refactor", "--kind", "potato"])
1038	assert result.exit_code != 0
1039
1040	def test_detect_refactor_bfs_follows_merge_parent2(
1041	self, code_repo: pathlib.Path
1042	) -> None:
1043	"""BFS walk finds refactoring events on merged feature branches."""
1044	import datetime
1045	root = code_repo
1046	repo_id = json.loads((repo_json_path(root)).read_text())["repo_id"]
1047	from muse.core.refs import (
1048	get_head_commit_id,
1049	read_current_branch,
1050	)
1051	from muse.core.commits import (
1052	CommitRecord,
1053	write_commit,
1054	)
1055	from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id
1056	from muse.domain import PatchOp, ReplaceOp, StructuredDelta
1057	branch = read_current_branch(root)
1058	head_id = get_head_commit_id(root, branch)
1059	assert head_id is not None
1060
1061	feat_at = datetime.datetime(2026, 7, 1, 10, 0, 0, tzinfo=datetime.timezone.utc)
1062	merge_at = datetime.datetime(2026, 7, 1, 11, 0, 0, tzinfo=datetime.timezone.utc)
1063
1064	feat_snap_id = compute_snapshot_id({"feat.py": "a" * 64})
1065	feature_id = compute_commit_id(
1066	parent_ids=[head_id],
1067	snapshot_id=feat_snap_id,
1068	message="perf: vectorise",
1069	committed_at_iso=feat_at.isoformat(),
1070	author="test",
1071	)
1072	write_commit(root, CommitRecord(
1073	commit_id=feature_id,
1074	branch="feat/perf",
1075	snapshot_id=feat_snap_id,
1076	message="perf: vectorise",
1077	committed_at=feat_at,
1078	parent_commit_id=head_id,
1079	author="test",
1080	structured_delta=StructuredDelta(ops=[PatchOp(
1081	op="patch",
1082	address="billing.py",
1083	child_ops=[ReplaceOp(
1084	op="replace",
1085	address="billing.py::vectorised_fn",
1086	new_summary="function vectorised_fn (implementation changed) L1–20",
1087	old_summary="function vectorised_fn",
1088	)],
1089	)]),
1090	))
1091	merge_snap_id = compute_snapshot_id({"merge.py": "b" * 64})
1092	merge_id = compute_commit_id(
1093	parent_ids=[head_id, feature_id],
1094	snapshot_id=merge_snap_id,
1095	message="merge feat/perf",
1096	committed_at_iso=merge_at.isoformat(),
1097	author="test",
1098	)
1099	write_commit(root, CommitRecord(
1100	commit_id=merge_id,
1101	branch=branch,
1102	snapshot_id=merge_snap_id,
1103	message="merge feat/perf",
1104	committed_at=merge_at,
1105	parent_commit_id=head_id,
1106	parent2_commit_id=feature_id,
1107	author="test",
1108	))
1109	(ref_path(root, branch)).write_text(merge_id)
1110
1111	result = runner.invoke(cli, ["code", "detect-refactor", "--json"])
1112	assert result.exit_code == 0, result.output
1113	data = json.loads(result.output)
1114	addrs = [e["address"] for e in data["events"]]
1115	assert "billing.py::vectorised_fn" in addrs, (
1116	"BFS must find the implementation event on the feature branch"
1117	)
1118
1119
1120	# ---------------------------------------------------------------------------
1121	# muse reserve
1122	# ---------------------------------------------------------------------------
1123
1124
1125	class TestReserve:
1126	def test_reserve_exits_zero(self, code_repo: pathlib.Path) -> None:
1127	result = runner.invoke(cli, [
1128	"coord", "reserve", "billing.py::process_order", "--run-id", "agent-test"
1129	])
1130	assert result.exit_code == 0, result.output
1131
1132	def test_reserve_creates_coordination_file(self, code_repo: pathlib.Path) -> None:
1133	runner.invoke(cli, ["coord", "reserve", "billing.py::process_order", "--run-id", "r1"])
1134	coord_dir = coordination_dir(code_repo) / "reservations"
1135	assert coord_dir.exists()
1136	files = list(coord_dir.glob("*.json"))
1137	assert len(files) >= 1
1138
1139	def test_reserve_json_output(self, code_repo: pathlib.Path) -> None:
1140	result = runner.invoke(cli, [
1141	"coord", "reserve", "--run-id", "r2", "--json", "billing.py::process_order",
1142	])
1143	assert result.exit_code == 0
1144	data = json.loads(result.output)
1145	assert "reservation_id" in data
1146
1147	def test_reserve_multiple_addresses(self, code_repo: pathlib.Path) -> None:
1148	result = runner.invoke(cli, [
1149	"coord", "reserve", "--run-id", "r3",
1150	"billing.py::process_order",
1151	"billing.py::Invoice.apply_discount",
1152	])
1153	assert result.exit_code == 0
1154
1155	def test_reserve_with_operation(self, code_repo: pathlib.Path) -> None:
1156	result = runner.invoke(cli, [
1157	"coord", "reserve", "--run-id", "r4", "--op", "rename",
1158	"billing.py::process_order",
1159	])
1160	assert result.exit_code == 0
1161
1162	def test_reserve_conflict_warning(self, code_repo: pathlib.Path) -> None:
1163	runner.invoke(cli, ["coord", "reserve", "--run-id", "a1", "billing.py::process_order"])
1164	result = runner.invoke(cli, ["coord", "reserve", "--run-id", "a2", "billing.py::process_order"])
1165	# Should warn but not fail.
1166	assert result.exit_code == 0
1167	assert "conflict" in result.output.lower() or "already" in result.output.lower() or "reserved" in result.output.lower()
1168
1169
1170	# ---------------------------------------------------------------------------
1171	# muse intent
1172	# ---------------------------------------------------------------------------
1173
1174
1175	class TestIntent:
1176	def test_intent_exits_zero(self, code_repo: pathlib.Path) -> None:
1177	result = runner.invoke(cli, [
1178	"coord", "intent", "--op", "rename", "--detail", "rename to process_invoice",
1179	"billing.py::process_order",
1180	])
1181	assert result.exit_code == 0, result.output
1182
1183	def test_intent_creates_file(self, code_repo: pathlib.Path) -> None:
1184	runner.invoke(cli, ["coord", "intent", "--op", "modify", "billing.py::Invoice"])
1185	idir = coordination_dir(code_repo) / "intents"
1186	assert idir.exists()
1187	assert len(list(idir.glob("*.json"))) >= 1
1188
1189	def test_intent_json_output(self, code_repo: pathlib.Path) -> None:
1190	result = runner.invoke(cli, [
1191	"coord", "intent", "--op", "modify", "--json", "billing.py::Invoice",
1192	])
1193	assert result.exit_code == 0
1194	data = json.loads(result.output)
1195	assert "intent_id" in data or "operation" in data
1196
1197
1198	# ---------------------------------------------------------------------------
1199	# muse forecast
1200	# ---------------------------------------------------------------------------
1201
1202
1203	class TestForecast:
1204	def test_forecast_exits_zero_no_reservations(self, code_repo: pathlib.Path) -> None:
1205	result = runner.invoke(cli, ["coord", "forecast"])
1206	assert result.exit_code == 0, result.output
1207
1208	def test_forecast_json_no_reservations(self, code_repo: pathlib.Path) -> None:
1209	result = runner.invoke(cli, ["coord", "forecast", "--json"])
1210	assert result.exit_code == 0
1211	data = json.loads(result.output)
1212	assert "conflicts" in data
1213
1214	def test_forecast_detects_address_overlap(self, code_repo: pathlib.Path) -> None:
1215	runner.invoke(cli, ["coord", "reserve", "--run-id", "a1", "billing.py::Invoice.apply_discount"])
1216	runner.invoke(cli, ["coord", "reserve", "--run-id", "a2", "billing.py::Invoice.apply_discount"])
1217	result = runner.invoke(cli, ["coord", "forecast", "--json"])
1218	assert result.exit_code == 0
1219	data = json.loads(result.output)
1220	types = [c.get("conflict_type") for c in data.get("conflicts", [])]
1221	assert "address_overlap" in types
1222
1223
1224	# ---------------------------------------------------------------------------
1225	# muse plan-merge
1226	# ---------------------------------------------------------------------------
1227
1228
1229	class TestPlanMerge:
1230	def test_plan_merge_same_commit_no_conflicts(self, code_repo: pathlib.Path) -> None:
1231	result = runner.invoke(cli, ["coord", "plan-merge", "HEAD", "HEAD"])
1232	assert result.exit_code == 0, result.output
1233
1234	def test_plan_merge_json(self, code_repo: pathlib.Path) -> None:
1235	result = runner.invoke(cli, ["coord", "plan-merge", "--json", "HEAD", "HEAD"])
1236	assert result.exit_code == 0
1237	data = json.loads(result.output)
1238	assert "conflicts" in data or isinstance(data, dict)
1239
1240	def test_plan_merge_requires_two_args(self, code_repo: pathlib.Path) -> None:
1241	result = runner.invoke(cli, ["coord", "plan-merge", "--json", "HEAD"])
1242	assert result.exit_code != 0
1243
1244
1245	# ---------------------------------------------------------------------------
1246	# muse shard
1247	# ---------------------------------------------------------------------------
1248
1249
1250	class TestShard:
1251	def test_shard_exits_zero(self, code_repo: pathlib.Path) -> None:
1252	result = runner.invoke(cli, ["coord", "shard", "--agents", "2"])
1253	assert result.exit_code == 0, result.output
1254
1255	def test_shard_json(self, code_repo: pathlib.Path) -> None:
1256	result = runner.invoke(cli, ["coord", "shard", "--agents", "2", "--json"])
1257	assert result.exit_code == 0
1258	data = json.loads(result.output)
1259	assert "shards" in data
1260
1261	def test_shard_n_equals_1(self, code_repo: pathlib.Path) -> None:
1262	result = runner.invoke(cli, ["coord", "shard", "--agents", "1"])
1263	assert result.exit_code == 0
1264
1265	def test_shard_large_n(self, code_repo: pathlib.Path) -> None:
1266	# N larger than symbol count still works (produces fewer shards).
1267	result = runner.invoke(cli, ["coord", "shard", "--agents", "100"])
1268	assert result.exit_code == 0
1269
1270
1271	# ---------------------------------------------------------------------------
1272	# muse reconcile
1273	# ---------------------------------------------------------------------------
1274
1275
1276	class TestReconcile:
1277	def test_reconcile_exits_zero(self, code_repo: pathlib.Path) -> None:
1278	result = runner.invoke(cli, ["coord", "reconcile"])
1279	assert result.exit_code == 0, result.output
1280
1281	def test_reconcile_json(self, code_repo: pathlib.Path) -> None:
1282	result = runner.invoke(cli, ["coord", "reconcile", "--json"])
1283	assert result.exit_code == 0
1284	data = json.loads(result.output)
1285	assert isinstance(data, dict)
1286
1287
1288	# ---------------------------------------------------------------------------
1289	# muse breakage
1290	# ---------------------------------------------------------------------------
1291
1292
1293	class TestBreakage:
1294	def test_breakage_exits_zero_clean_tree(self, code_repo: pathlib.Path) -> None:
1295	result = runner.invoke(cli, ["code", "breakage"])
1296	assert result.exit_code == 0, result.output
1297
1298	def test_breakage_json(self, code_repo: pathlib.Path) -> None:
1299	result = runner.invoke(cli, ["code", "breakage", "--json"])
1300	assert result.exit_code == 0
1301	data = json.loads(result.output)
1302	# breakage JSON has "issues" list and error count.
1303	assert "issues" in data
1304	assert isinstance(data["issues"], list)
1305
1306	def test_breakage_language_filter(self, code_repo: pathlib.Path) -> None:
1307	result = runner.invoke(cli, ["code", "breakage", "--language", "Python"])
1308	assert result.exit_code == 0
1309
1310	def test_breakage_no_repo_errors(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
1311	monkeypatch.chdir(tmp_path)
1312	result = runner.invoke(cli, ["code", "breakage"])
1313	assert result.exit_code != 0
1314
1315
1316	# ---------------------------------------------------------------------------
1317	# muse invariants
1318	# ---------------------------------------------------------------------------
1319
1320
1321	class TestInvariants:
1322	def test_invariants_creates_toml_if_absent(self, code_repo: pathlib.Path) -> None:
1323	result = runner.invoke(cli, ["code", "invariants"])
1324	toml_path = muse_dir(code_repo) / "invariants.toml"
1325	assert result.exit_code == 0 or toml_path.exists()
1326
1327	def test_invariants_json_with_empty_rules(self, code_repo: pathlib.Path) -> None:
1328	# Create empty invariants.toml
1329	(muse_dir(code_repo) / "invariants.toml").write_text("# No rules\n")
1330	result = runner.invoke(cli, ["code", "invariants", "--json"])
1331	assert result.exit_code == 0
1332	# Output may be JSON or human-readable depending on rules count.
1333	output = result.output.strip()
1334	if output and not output.startswith("#"):
1335	try:
1336	data = json.loads(output)
1337	assert isinstance(data, dict)
1338	except json.JSONDecodeError:
1339	pass # Human-readable output is also acceptable.
1340
1341	def test_invariants_no_cycles_rule(self, code_repo: pathlib.Path) -> None:
1342	(muse_dir(code_repo) / "invariants.toml").write_text(textwrap.dedent("""\
1343	[[rules]]
1344	type = "no_cycles"
1345	name = "no import cycles"
1346	"""))
1347	result = runner.invoke(cli, ["code", "invariants"])
1348	assert result.exit_code == 0
1349
1350	def test_invariants_forbidden_dependency_rule(self, code_repo: pathlib.Path) -> None:
1351	(muse_dir(code_repo) / "invariants.toml").write_text(textwrap.dedent("""\
1352	[[rules]]
1353	type = "forbidden_dependency"
1354	name = "billing must not import utils"
1355	source_pattern = "billing.py"
1356	forbidden_pattern = "utils.py"
1357	"""))
1358	result = runner.invoke(cli, ["code", "invariants"])
1359	assert result.exit_code == 0
1360
1361	def test_invariants_required_test_rule(self, code_repo: pathlib.Path) -> None:
1362	(muse_dir(code_repo) / "invariants.toml").write_text(textwrap.dedent("""\
1363	[[rules]]
1364	type = "required_test"
1365	name = "billing must have tests"
1366	source_pattern = "billing.py"
1367	test_pattern = "test_billing.py"
1368	"""))
1369	result = runner.invoke(cli, ["code", "invariants"])
1370	# May pass or fail depending on whether test_billing.py exists; should not crash.
1371	assert result.exit_code in (0, 1)
1372
1373	def test_invariants_commit_flag(self, code_repo: pathlib.Path) -> None:
1374	(muse_dir(code_repo) / "invariants.toml").write_text("# empty\n")
1375	result = runner.invoke(cli, ["code", "invariants", "--commit", "HEAD"])
1376	assert result.exit_code == 0
1377
1378
1379	# ---------------------------------------------------------------------------
1380	# muse commit — semantic versioning
1381	# ---------------------------------------------------------------------------
1382
1383
1384	class TestSemVerInCommit:
1385	def test_commit_record_has_sem_ver_bump(self, code_repo: pathlib.Path) -> None:
1386	from muse.core.refs import get_head_commit_id
1387	from muse.core.commits import (
1388	CommitDict,
1389	read_commit,
1390	)
1391	commit_id = get_head_commit_id(code_repo, "main")
1392	assert commit_id is not None
1393	commit = read_commit(code_repo, commit_id)
1394	assert commit is not None
1395	assert commit.sem_ver_bump in ("major", "minor", "patch", "none")
1396
1397	def test_commit_record_has_breaking_changes(self, code_repo: pathlib.Path) -> None:
1398	from muse.core.refs import get_head_commit_id
1399	from muse.core.commits import (
1400	CommitDict,
1401	read_commit,
1402	)
1403	commit_id = get_head_commit_id(code_repo, "main")
1404	assert commit_id is not None
1405	commit = read_commit(code_repo, commit_id)
1406	assert commit is not None
1407	assert isinstance(commit.breaking_changes, list)
1408
1409	def test_log_shows_semver_for_major_bump(self, code_repo: pathlib.Path) -> None:
1410	from muse.core.refs import get_head_commit_id
1411	from muse.core.commits import (
1412	CommitDict,
1413	read_commit,
1414	)
1415	commit_id = get_head_commit_id(code_repo, "main")
1416	assert commit_id is not None
1417	commit = read_commit(code_repo, commit_id)
1418	assert commit is not None
1419	if commit.sem_ver_bump == "major":
1420	result = runner.invoke(cli, ["log"])
1421	assert "MAJOR" in result.output or "major" in result.output.lower()
1422
1423
1424	# ---------------------------------------------------------------------------
1425	# Call-graph tier — muse impact
1426	# ---------------------------------------------------------------------------
1427
1428
1429	class TestImpact:
1430	def test_impact_exits_zero(self, code_repo: pathlib.Path) -> None:
1431	result = runner.invoke(cli, ["code", "impact", "--", "billing.py::Invoice.compute_invoice_total"])
1432	assert result.exit_code == 0, result.output
1433
1434	def test_impact_json(self, code_repo: pathlib.Path) -> None:
1435	result = runner.invoke(cli, ["code", "impact", "--json", "billing.py::Invoice.apply_discount"])
1436	assert result.exit_code == 0
1437	data = json.loads(result.output)
1438	assert isinstance(data, dict)
1439	assert "blast_radius" in data
1440	assert "total" in data
1441	assert "commit_id" in data
1442	assert data["mode"] == "reverse"
1443
1444	def test_impact_nonexistent_symbol_handled(self, code_repo: pathlib.Path) -> None:
1445	result = runner.invoke(cli, ["code", "impact", "--", "billing.py::nonexistent"])
1446	assert result.exit_code in (0, 1)
1447
1448	def test_impact_count_only(self, code_repo: pathlib.Path) -> None:
1449	result = runner.invoke(cli, ["code", "impact", "--count", "--", "billing.py::Invoice.compute_invoice_total"])
1450	assert result.exit_code == 0
1451	assert result.output.strip().isdigit()
1452
1453	def test_impact_depth_negative_rejected(self, code_repo: pathlib.Path) -> None:
1454	result = runner.invoke(cli, ["code", "impact", "--depth", "-1", "--", "billing.py::Invoice.compute_invoice_total"])
1455	assert result.exit_code == 1
1456
1457	def test_impact_forward_exits_zero(self, code_repo: pathlib.Path) -> None:
1458	result = runner.invoke(cli, ["code", "impact", "--forward", "--", "billing.py::Invoice.compute_invoice_total"])
1459	assert result.exit_code == 0
1460
1461	def test_impact_forward_json(self, code_repo: pathlib.Path) -> None:
1462	result = runner.invoke(cli, ["code", "impact", "--forward", "--json", "--", "billing.py::process_order"])
1463	assert result.exit_code == 0
1464	data = json.loads(result.output)
1465	assert data["mode"] == "forward"
1466	assert "callees" in data
1467	assert "total" in data
1468	assert "commit_id" in data
1469
1470	def test_impact_forward_and_compare_mutually_exclusive(self, code_repo: pathlib.Path) -> None:
1471	result = runner.invoke(cli, [
1472	"code", "impact", "--forward", "--compare", "HEAD",
1473	"--", "billing.py::process_order",
1474	])
1475	assert result.exit_code == 1
1476
1477	def test_impact_file_filter(self, code_repo: pathlib.Path) -> None:
1478	result = runner.invoke(cli, [
1479	"code", "impact", "--file", "billing.py",
1480	"--", "billing.py::Invoice.compute_invoice_total",
1481	])
1482	assert result.exit_code == 0
1483
1484	def test_impact_file_filter_json(self, code_repo: pathlib.Path) -> None:
1485	result = runner.invoke(cli, [
1486	"code", "impact", "--file", "billing.py", "--json",
1487	"--", "billing.py::Invoice.compute_invoice_total",
1488	])
1489	assert result.exit_code == 0
1490	data = json.loads(result.output)
1491	assert data["file_filter"] == "billing.py"
1492	for depth_addrs in data["blast_radius"].values():
1493	for addr in depth_addrs:
1494	assert addr.startswith("billing.py::")
1495
1496	def test_impact_compare_json_schema(self, code_repo: pathlib.Path) -> None:
1497	result = runner.invoke(cli, [
1498	"code", "impact", "--compare", "HEAD",
1499	"--json", "--", "billing.py::Invoice.compute_invoice_total",
1500	])
1501	assert result.exit_code == 0
1502	data = json.loads(result.output)
1503	assert "compare_commit_id" in data
1504	assert "added_callers" in data
1505	assert "removed_callers" in data
1506	assert "net_change" in data
1507	assert isinstance(data["added_callers"], list)
1508	assert isinstance(data["removed_callers"], list)
1509
1510	def test_impact_forward_count(self, code_repo: pathlib.Path) -> None:
1511	result = runner.invoke(cli, ["code", "impact", "--forward", "--count", "--", "billing.py::process_order"])
1512	assert result.exit_code == 0
1513	assert result.output.strip().isdigit()
1514
1515
1516	# ---------------------------------------------------------------------------
1517	# Call-graph tier — muse dead
1518	# ---------------------------------------------------------------------------
1519
1520
1521	class TestDead:
1522	def test_dead_exits_zero(self, code_repo: pathlib.Path) -> None:
1523	result = runner.invoke(cli, ["code", "dead"])
1524	assert result.exit_code == 0, result.output
1525
1526	def test_dead_json(self, code_repo: pathlib.Path) -> None:
1527	result = runner.invoke(cli, ["code", "dead", "--json"])
1528	assert result.exit_code == 0
1529	data = json.loads(result.output)
1530	assert isinstance(data, dict)
1531	assert "results" in data
1532	assert "high_confidence_count" in data
1533	assert "total_files_scanned" in data
1534	assert "duration_ms" in data
1535
1536	def test_dead_kind_filter(self, code_repo: pathlib.Path) -> None:
1537	result = runner.invoke(cli, ["code", "dead", "--kind", "function"])
1538	assert result.exit_code == 0
1539
1540	def test_dead_include_tests(self, code_repo: pathlib.Path) -> None:
1541	result = runner.invoke(cli, ["code", "dead", "--include-tests"])
1542	assert result.exit_code == 0
1543
1544	def test_dead_count_only(self, code_repo: pathlib.Path) -> None:
1545	result = runner.invoke(cli, ["code", "dead", "--count"])
1546	assert result.exit_code == 0
1547	assert result.output.strip().isdigit()
1548
1549	def test_dead_compare_json_schema(self, code_repo: pathlib.Path) -> None:
1550	result = runner.invoke(cli, ["code", "dead", "--compare", "HEAD", "--json"])
1551	assert result.exit_code == 0
1552	data = json.loads(result.output)
1553	assert "compare_commit_id" in data
1554	assert "new_dead" in data
1555	assert "recovered" in data
1556	assert "net_change" in data
1557	assert isinstance(data["new_dead"], list)
1558	assert isinstance(data["recovered"], list)
1559
1560	def test_dead_compare_exits_zero(self, code_repo: pathlib.Path) -> None:
1561	result = runner.invoke(cli, ["code", "dead", "--compare", "HEAD"])
1562	assert result.exit_code == 0
1563
1564	def test_dead_delete_and_compare_mutually_exclusive(self, code_repo: pathlib.Path) -> None:
1565	result = runner.invoke(cli, ["code", "dead", "--delete", "--compare", "HEAD"])
1566	assert result.exit_code == 1
1567
1568	def test_dead_save_allowlist(self, code_repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
1569	out_file = tmp_path / "allowlist.json"
1570	result = runner.invoke(cli, ["code", "dead", "--save-allowlist", str(out_file)])
1571	assert result.exit_code == 0
1572	if out_file.exists():
1573	data = json.loads(out_file.read_text())
1574	assert isinstance(data, list)
1575	assert all(isinstance(x, str) for x in data)
1576
1577	def test_dead_high_confidence_only_json(self, code_repo: pathlib.Path) -> None:
1578	result = runner.invoke(cli, ["code", "dead", "--high-confidence-only", "--json"])
1579	assert result.exit_code == 0
1580	data = json.loads(result.output)
1581	for c in data["results"]:
1582	assert c["confidence"] == "high"
1583
1584	def test_dead_workers_cap_enforced(self, code_repo: pathlib.Path) -> None:
1585	result = runner.invoke(cli, ["code", "dead", "--workers", "999", "--count"])
1586	assert result.exit_code == 0
1587
1588
1589	# ---------------------------------------------------------------------------
1590	# muse code cat
1591	# ---------------------------------------------------------------------------
1592
1593
1594	class TestCat:
1595	def test_cat_basic(self, code_repo: pathlib.Path) -> None:
1596	result = runner.invoke(cli, ["code", "cat", "billing.py::Invoice"])
1597	assert result.exit_code == 0, result.output
1598	assert "class Invoice" in result.output
1599
1600	def test_cat_method(self, code_repo: pathlib.Path) -> None:
1601	result = runner.invoke(cli, ["code", "cat", "billing.py::Invoice.compute_invoice_total"])
1602	assert result.exit_code == 0, result.output
1603	assert "def compute_invoice_total" in result.output
1604
1605	def test_cat_bare_name_unambiguous(self, code_repo: pathlib.Path) -> None:
1606	# Invoice is unique — short name should resolve.
1607	result = runner.invoke(cli, ["code", "cat", "billing.py::Invoice"])
1608	assert result.exit_code == 0
1609
1610	def test_cat_missing_separator_error(self, code_repo: pathlib.Path) -> None:
1611	result = runner.invoke(cli, ["code", "cat", "billing.py"])
1612	assert result.exit_code != 0
1613
1614	def test_cat_unknown_symbol_error(self, code_repo: pathlib.Path) -> None:
1615	result = runner.invoke(cli, ["code", "cat", "billing.py::NoSuchThing"])
1616	assert result.exit_code != 0
1617
1618	def test_cat_unknown_file_error(self, code_repo: pathlib.Path) -> None:
1619	result = runner.invoke(cli, ["code", "cat", "nope.py::Foo"])
1620	assert result.exit_code != 0
1621
1622	def test_cat_line_numbers(self, code_repo: pathlib.Path) -> None:
1623	result = runner.invoke(cli, ["code", "cat", "billing.py::Invoice", "--line-numbers"])
1624	assert result.exit_code == 0
1625	# Line numbers prefix lines with digits.
1626	lines = [ln for ln in result.output.splitlines() if not ln.startswith("#")]
1627	first_code_line = next((ln for ln in lines if ln.strip()), "")
1628	assert first_code_line[:1].isdigit(), f"Expected digit prefix, got: {first_code_line!r}"
1629
1630	def test_cat_json_output(self, code_repo: pathlib.Path) -> None:
1631	result = runner.invoke(cli, ["code", "cat", "billing.py::Invoice", "--json"])
1632	assert result.exit_code == 0
1633	data = json.loads(result.output)
1634	assert "results" in data
1635	assert "errors" in data
1636	assert "source_ref" in data
1637	assert len(data["results"]) == 1
1638	r = data["results"][0]
1639	assert r["path"] == "billing.py"
1640	assert r["kind"] in ("class", "function", "method")
1641	assert isinstance(r["lineno"], int)
1642	assert isinstance(r["end_lineno"], int)
1643	assert "class Invoice" in r["source"]
1644
1645	def test_cat_multi_address(self, code_repo: pathlib.Path) -> None:
1646	result = runner.invoke(
1647	cli,
1648	[
1649	"code", "cat",
1650	"billing.py::Invoice",
1651	"billing.py::Invoice.compute_invoice_total",
1652	"--json",
1653	],
1654	)
1655	assert result.exit_code == 0, result.output
1656	data = json.loads(result.output)
1657	assert len(data["results"]) == 2
1658
1659	def test_cat_all_mode(self, code_repo: pathlib.Path) -> None:
1660	result = runner.invoke(cli, ["code", "cat", "billing.py", "--all"])
1661	assert result.exit_code == 0
1662	assert "Invoice" in result.output
1663
1664	def test_cat_all_kind_filter(self, code_repo: pathlib.Path) -> None:
1665	result = runner.invoke(cli, ["code", "cat", "billing.py", "--all", "--kind", "function"])
1666	assert result.exit_code == 0
1667
1668	def test_cat_all_json(self, code_repo: pathlib.Path) -> None:
1669	result = runner.invoke(cli, ["code", "cat", "billing.py", "--all", "--json"])
1670	assert result.exit_code == 0
1671	data = json.loads(result.output)
1672	assert len(data["results"]) > 0
1673	# Every result has required fields.
1674	for r in data["results"]:
1675	assert "address" in r
1676	assert "lineno" in r
1677	assert "source" in r
1678
1679	def test_cat_context_lines(self, code_repo: pathlib.Path) -> None:
1680	result_plain = runner.invoke(cli, ["code", "cat", "billing.py::Invoice.compute_invoice_total"])
1681	result_ctx = runner.invoke(
1682	cli, ["code", "cat", "billing.py::Invoice.compute_invoice_total", "--context", "2"]
1683	)
1684	assert result_ctx.exit_code == 0
1685	# With context we get at least as many lines.
1686	plain_lines = result_plain.output.count("\n")
1687	ctx_lines = result_ctx.output.count("\n")
1688	assert ctx_lines >= plain_lines
1689
1690	def test_cat_json_errors_field_on_bad_address(self, code_repo: pathlib.Path) -> None:
1691	# In --json mode a missing symbol goes to the errors field, not a crash.
1692	result = runner.invoke(
1693	cli,
1694	["code", "cat", "billing.py::Invoice", "billing.py::NoSuchThing", "--json"],
1695	)
1696	# Output must be valid JSON (no stderr bleed into stdout).
1697	data = json.loads(result.output)
1698	assert len(data["results"]) == 1
1699	assert len(data["errors"]) == 1
1700	assert data["errors"][0]["address"] == "billing.py::NoSuchThing"
1701
1702	def test_cat_header_shows_working_tree(self, code_repo: pathlib.Path) -> None:
1703	result = runner.invoke(cli, ["code", "cat", "billing.py::Invoice"])
1704	assert result.exit_code == 0
1705	assert "working tree" in result.output
1706
1707	def test_cat_at_head(self, code_repo: pathlib.Path) -> None:
1708	result = runner.invoke(cli, ["code", "cat", "billing.py::Invoice", "--at", "HEAD"])
1709	assert result.exit_code == 0
1710	assert "Invoice" in result.output
1711
1712	def test_cat_wrong_file_fallback_finds_symbol(
1713	self, code_repo: pathlib.Path, tmp_path: pathlib.Path
1714	) -> None:
1715	"""FILE::SYMBOL where SYMBOL lives in a different file — should fall back
1716	to a global snapshot search and cat it from its actual location, exit 0."""
1717	# Add a second file with a unique function the billing module doesn't have.
1718	work = pathlib.Path.cwd()
1719	(work / "utils.py").write_text(
1720	"def format_currency(amount):\n return f'${amount:.2f}'\n"
1721	)
1722	runner.invoke(cli, ["code", "add", "utils.py"])
1723	runner.invoke(cli, ["commit", "-m", "Add utils"])
1724
1725	# Ask for utils.format_currency but specify the wrong file (billing.py).
1726	result = runner.invoke(
1727	cli, ["code", "cat", "billing.py::format_currency"]
1728	)
1729	assert result.exit_code == 0, result.output
1730	assert "format_currency" in result.output
1731
1732	def test_cat_wrong_file_fallback_json(self, code_repo: pathlib.Path) -> None:
1733	"""Same fallback in --json mode: result is in results[], not errors[]."""
1734	work = pathlib.Path.cwd()
1735	(work / "utils.py").write_text(
1736	"def format_currency(amount):\n return f'${amount:.2f}'\n"
1737	)
1738	runner.invoke(cli, ["code", "add", "utils.py"])
1739	runner.invoke(cli, ["commit", "-m", "Add utils"])
1740
1741	result = runner.invoke(
1742	cli, ["code", "cat", "billing.py::format_currency", "--json"]
1743	)
1744	assert result.exit_code == 0, result.output
1745	data = json.loads(result.output)
1746	assert len(data["results"]) == 1
1747	assert data["results"][0]["symbol"] == "format_currency"
1748	assert data["results"][0]["path"] == "utils.py"
1749
1750	def test_cat_wrong_file_fallback_ambiguous_exits_nonzero(
1751	self, code_repo: pathlib.Path
1752	) -> None:
1753	"""If the symbol exists in multiple files, fallback reports ambiguity and exits 1."""
1754	work = pathlib.Path.cwd()
1755	(work / "utils.py").write_text("def send_email(to): pass\n")
1756	runner.invoke(cli, ["code", "add", "utils.py"])
1757	runner.invoke(cli, ["commit", "-m", "Duplicate send_email in utils"])
1758
1759	# billing.py already has send_email; utils.py now also has it.
1760	result = runner.invoke(
1761	cli, ["code", "cat", "nope.py::send_email"]
1762	)
1763	assert result.exit_code != 0
1764
1765	def test_cat_truly_missing_symbol_still_errors(self, code_repo: pathlib.Path) -> None:
1766	"""A symbol that doesn't exist anywhere in the snapshot still exits 1."""
1767	result = runner.invoke(cli, ["code", "cat", "billing.py::AbsolutelyNowhere"])
1768	assert result.exit_code != 0
1769
1770
1771	# ---------------------------------------------------------------------------
1772	# Call-graph tier — muse coverage
1773	# ---------------------------------------------------------------------------
1774
1775
1776	class TestCoverage:
1777	def test_coverage_exits_zero(self, code_repo: pathlib.Path) -> None:
1778	result = runner.invoke(cli, ["code", "coverage", "--", "billing.py::Invoice"])
1779	assert result.exit_code == 0, result.output
1780
1781	def test_coverage_json(self, code_repo: pathlib.Path) -> None:
1782	result = runner.invoke(cli, ["code", "coverage", "--json", "billing.py::Invoice"])
1783	assert result.exit_code == 0
1784	data = json.loads(result.output)
1785	assert isinstance(data, dict)
1786	assert "methods" in data
1787	assert "total_methods" in data
1788	assert "covered" in data
1789	assert "percent" in data
1790	assert "commit_id" in data
1791	assert "filters" in data
1792	for m in data["methods"]:
1793	assert "address" in m
1794	assert "called" in m
1795	assert "callers" in m
1796
1797	def test_coverage_nonexistent_class_handled(self, code_repo: pathlib.Path) -> None:
1798	result = runner.invoke(cli, ["code", "coverage", "--", "billing.py::NonExistent"])
1799	assert result.exit_code in (0, 1)
1800
1801	def test_coverage_count_only(self, code_repo: pathlib.Path) -> None:
1802	result = runner.invoke(cli, ["code", "coverage", "--count", "billing.py::Invoice"])
1803	assert result.exit_code == 0
1804	# Output should be "n/total" format
1805	assert "/" in result.output.strip()
1806
1807	def test_coverage_exclude_dunder(self, code_repo: pathlib.Path) -> None:
1808	result = runner.invoke(cli, [
1809	"code", "coverage", "--exclude-dunder", "--json", "billing.py::Invoice",
1810	])
1811	assert result.exit_code == 0
1812	data = json.loads(result.output)
1813	assert data["filters"]["exclude_dunder"] is True
1814	for m in data["methods"]:
1815	assert not (m["name"].startswith("__") and m["name"].endswith("__"))
1816
1817	def test_coverage_exclude_private(self, code_repo: pathlib.Path) -> None:
1818	result = runner.invoke(cli, [
1819	"code", "coverage", "--exclude-private", "--json", "billing.py::Invoice",
1820	])
1821	assert result.exit_code == 0
1822	data = json.loads(result.output)
1823	assert data["filters"]["exclude_private"] is True
1824
1825	def test_coverage_min_callers(self, code_repo: pathlib.Path) -> None:
1826	result = runner.invoke(cli, [
1827	"code", "coverage", "--min-callers", "2", "--json", "billing.py::Invoice",
1828	])
1829	assert result.exit_code == 0
1830	data = json.loads(result.output)
1831	assert data["filters"]["min_callers"] == 2
1832
1833	def test_coverage_exclude_self(self, code_repo: pathlib.Path) -> None:
1834	result = runner.invoke(cli, [
1835	"code", "coverage", "--exclude-self", "--json", "billing.py::Invoice",
1836	])
1837	assert result.exit_code == 0
1838	data = json.loads(result.output)
1839	assert data["filters"]["exclude_self"] is True
1840	# All reported callers should be from a different file
1841	for m in data["methods"]:
1842	for caller in m["callers"]:
1843	assert not caller.startswith("billing.py::")
1844
1845	def test_coverage_compare_json_schema(self, code_repo: pathlib.Path) -> None:
1846	result = runner.invoke(cli, [
1847	"code", "coverage", "--compare", "HEAD", "--json", "billing.py::Invoice",
1848	])
1849	assert result.exit_code == 0
1850	data = json.loads(result.output)
1851	assert "compare_commit_id" in data
1852	assert "newly_covered" in data
1853	assert "newly_uncovered" in data
1854	assert "percent_change" in data
1855
1856	def test_coverage_compare_exits_zero(self, code_repo: pathlib.Path) -> None:
1857	result = runner.invoke(cli, [
1858	"code", "coverage", "--compare", "HEAD", "billing.py::Invoice",
1859	])
1860	assert result.exit_code == 0
1861
1862	def test_coverage_no_show_callers(self, code_repo: pathlib.Path) -> None:
1863	result = runner.invoke(cli, [
1864	"code", "coverage", "--no-show-callers", "billing.py::Invoice",
1865	])
1866	assert result.exit_code == 0
1867
1868
1869	# ---------------------------------------------------------------------------
1870	# Call-graph tier — muse deps
1871	# ---------------------------------------------------------------------------
1872
1873
1874	class TestDeps:
1875	def test_deps_file_mode(self, code_repo: pathlib.Path) -> None:
1876	result = runner.invoke(cli, ["code", "deps", "--", "billing.py"])
1877	assert result.exit_code == 0, result.output
1878
1879	def test_deps_reverse(self, code_repo: pathlib.Path) -> None:
1880	result = runner.invoke(cli, ["code", "deps", "--reverse", "billing.py"])
1881	assert result.exit_code == 0
1882
1883	def test_deps_json(self, code_repo: pathlib.Path) -> None:
1884	result = runner.invoke(cli, ["code", "deps", "--json", "billing.py"])
1885	assert result.exit_code == 0
1886	data = json.loads(result.output)
1887	assert isinstance(data, dict)
1888
1889	def test_deps_symbol_mode(self, code_repo: pathlib.Path) -> None:
1890	result = runner.invoke(cli, ["code", "deps", "--", "billing.py::Invoice.compute_invoice_total"])
1891	assert result.exit_code in (0, 1) # May be empty but shouldn't crash.
1892
1893	# ── new flags ──────────────────────────────────────────────────────────────
1894
1895	def test_deps_count_file_mode(self, code_repo: pathlib.Path) -> None:
1896	result = runner.invoke(cli, ["code", "deps", "--count", "billing.py"])
1897	assert result.exit_code == 0, result.output
1898	assert result.output.strip().isdigit()
1899
1900	def test_deps_count_reverse(self, code_repo: pathlib.Path) -> None:
1901	result = runner.invoke(cli, ["code", "deps", "--count", "--reverse", "billing.py"])
1902	assert result.exit_code == 0, result.output
1903	assert result.output.strip().isdigit()
1904
1905	def test_deps_filter_file_mode(self, code_repo: pathlib.Path) -> None:
1906	result = runner.invoke(
1907	cli, ["code", "deps", "--reverse", "--filter", "billing", "billing.py"]
1908	)
1909	assert result.exit_code == 0, result.output
1910
1911	def test_deps_depth_requires_symbol_mode(self, code_repo: pathlib.Path) -> None:
1912	# --depth > 1 in file mode is fine (just filters imports as before).
1913	result = runner.invoke(cli, ["code", "deps", "--depth", "2", "billing.py"])
1914	assert result.exit_code == 0, result.output
1915
1916	def test_deps_depth_negative_rejected(self, code_repo: pathlib.Path) -> None:
1917	result = runner.invoke(
1918	cli,
1919	["code", "deps", "--depth", "-1", "billing.py::Invoice.compute_invoice_total"],
1920	)
1921	assert result.exit_code != 0
1922
1923	def test_deps_depth_symbol_reverse(self, code_repo: pathlib.Path) -> None:
1924	result = runner.invoke(
1925	cli,
1926	["code", "deps", "--reverse", "--depth", "2",
1927	"billing.py::Invoice.compute_invoice_total"],
1928	)
1929	assert result.exit_code == 0, result.output
1930
1931	def test_deps_transitive_symbol(self, code_repo: pathlib.Path) -> None:
1932	result = runner.invoke(
1933	cli,
1934	["code", "deps", "--transitive",
1935	"billing.py::Invoice.compute_invoice_total"],
1936	)
1937	assert result.exit_code == 0, result.output
1938
1939	def test_deps_transitive_count(self, code_repo: pathlib.Path) -> None:
1940	result = runner.invoke(
1941	cli,
1942	["code", "deps", "--transitive", "--count",
1943	"billing.py::Invoice.compute_invoice_total"],
1944	)
1945	assert result.exit_code == 0
1946	assert result.output.strip().isdigit()
1947
1948	def test_deps_transitive_json_schema(self, code_repo: pathlib.Path) -> None:
1949	result = runner.invoke(
1950	cli,
1951	["code", "deps", "--transitive", "--json",
1952	"billing.py::Invoice.compute_invoice_total"],
1953	)
1954	assert result.exit_code == 0
1955	data = json.loads(result.output)
1956	assert "by_depth" in data
1957	assert data["transitive"] is True
1958
1959	def test_deps_depth_json_schema(self, code_repo: pathlib.Path) -> None:
1960	result = runner.invoke(
1961	cli,
1962	["code", "deps", "--reverse", "--depth", "2", "--json",
1963	"billing.py::Invoice.compute_invoice_total"],
1964	)
1965	assert result.exit_code == 0
1966	data = json.loads(result.output)
1967	assert "by_depth" in data
1968	assert data["depth"] == 2
1969
1970	def test_deps_path_traversal_rejected(self, code_repo: pathlib.Path) -> None:
1971	result = runner.invoke(cli, ["code", "deps", "../../../etc/passwd"])
1972	assert result.exit_code != 0
1973
1974	def test_deps_empty_file_rel_in_symbol_rejected(
1975	self, code_repo: pathlib.Path
1976	) -> None:
1977	result = runner.invoke(cli, ["code", "deps", "--", "::some_func"])
1978	assert result.exit_code != 0
1979
1980	def test_deps_reverse_json_schema(self, code_repo: pathlib.Path) -> None:
1981	result = runner.invoke(
1982	cli, ["code", "deps", "--reverse", "--json", "billing.py"]
1983	)
1984	assert result.exit_code == 0
1985	data = json.loads(result.output)
1986	assert "imported_by" in data
1987	assert isinstance(data["imported_by"], list)
1988
1989
1990	# ---------------------------------------------------------------------------
1991	# Call-graph tier — muse find-symbol
1992	# ---------------------------------------------------------------------------
1993
1994
1995	class TestFindSymbol:
1996	def test_find_by_name(self, code_repo: pathlib.Path) -> None:
1997	result = runner.invoke(cli, ["code", "find-symbol", "--name", "process_order"])
1998	assert result.exit_code == 0, result.output
1999
2000	def test_find_by_name_json(self, code_repo: pathlib.Path) -> None:
2001	result = runner.invoke(cli, ["code", "find-symbol", "--name", "Invoice", "--json"])
2002	assert result.exit_code == 0
2003	data = json.loads(result.output)
2004	assert isinstance(data, dict)
2005	assert "results" in data
2006	assert "query" in data
2007	assert "total" in data
2008
2009	def test_find_by_kind(self, code_repo: pathlib.Path) -> None:
2010	result = runner.invoke(cli, ["code", "find-symbol", "--kind", "class"])
2011	assert result.exit_code == 0
2012	assert result.output is not None
2013
2014	def test_find_nonexistent_name_empty(self, code_repo: pathlib.Path) -> None:
2015	result = runner.invoke(cli, ["code", "find-symbol", "--name", "totally_nonexistent_xyzzy"])
2016	assert result.exit_code == 0
2017	assert "no matching" in result.output
2018
2019	def test_find_requires_at_least_one_flag(self, code_repo: pathlib.Path) -> None:
2020	result = runner.invoke(cli, ["code", "find-symbol"])
2021	assert result.exit_code == 1
2022
2023	def test_find_count_only(self, code_repo: pathlib.Path) -> None:
2024	result = runner.invoke(cli, ["code", "find-symbol", "--name", "process_order", "--count"])
2025	assert result.exit_code == 0
2026	assert result.output.strip().isdigit()
2027
2028	def test_find_first_and_last_mutually_exclusive(self, code_repo: pathlib.Path) -> None:
2029	result = runner.invoke(cli, ["code", "find-symbol", "--name", "Invoice", "--first", "--last"])
2030	assert result.exit_code == 1
2031
2032	def test_find_hash_too_short_rejected(self, code_repo: pathlib.Path) -> None:
2033	result = runner.invoke(cli, ["code", "find-symbol", "--hash", "ab"])
2034	assert result.exit_code == 1
2035
2036	def test_find_since_invalid_date(self, code_repo: pathlib.Path) -> None:
2037	result = runner.invoke(cli, ["code", "find-symbol", "--name", "Invoice", "--since", "not-a-date"])
2038	assert result.exit_code == 1
2039
2040	def test_find_until_invalid_date(self, code_repo: pathlib.Path) -> None:
2041	result = runner.invoke(cli, ["code", "find-symbol", "--name", "Invoice", "--until", "99/99/99"])
2042	assert result.exit_code == 1
2043
2044	def test_find_since_future_returns_empty(self, code_repo: pathlib.Path) -> None:
2045	result = runner.invoke(cli, [
2046	"code", "find-symbol", "--name", "process_order",
2047	"--since", "2099-01-01",
2048	])
2049	assert result.exit_code == 0
2050	assert "no matching" in result.output
2051
2052	def test_find_limit(self, code_repo: pathlib.Path) -> None:
2053	result = runner.invoke(cli, ["code", "find-symbol", "--kind", "function", "--limit", "1"])
2054	assert result.exit_code == 0
2055
2056	def test_find_file_filter(self, code_repo: pathlib.Path) -> None:
2057	result = runner.invoke(cli, [
2058	"code", "find-symbol", "--kind", "function", "--file", "billing.py",
2059	])
2060	assert result.exit_code == 0
2061
2062	def test_find_prefix_name(self, code_repo: pathlib.Path) -> None:
2063	result = runner.invoke(cli, ["code", "find-symbol", "--name", "process*", "--json"])
2064	assert result.exit_code == 0
2065	data = json.loads(result.output)
2066	for ap in data["results"]:
2067	assert ap["name"].lower().startswith("process")
2068
2069	def test_find_first_deduplicates(self, code_repo: pathlib.Path) -> None:
2070	result_all = runner.invoke(cli, ["code", "find-symbol", "--name", "process_order", "--count"])
2071	result_first = runner.invoke(cli, ["code", "find-symbol", "--name", "process_order", "--first", "--count"])
2072	assert result_all.exit_code == 0
2073	assert result_first.exit_code == 0
2074	count_all = int(result_all.output.strip())
2075	count_first = int(result_first.output.strip())
2076	assert count_first <= count_all
2077
2078	def test_find_json_schema(self, code_repo: pathlib.Path) -> None:
2079	result = runner.invoke(cli, ["code", "find-symbol", "--kind", "function", "--json"])
2080	assert result.exit_code == 0
2081	data = json.loads(result.output)
2082	assert "query" in data
2083	assert "results" in data
2084	assert "total" in data
2085	assert data["total"] == len(data["results"])
2086	if data["results"]:
2087	ap = data["results"][0]
2088	for key in ("content_id", "address", "name", "kind", "commit_id", "committed_at"):
2089	assert key in ap
2090
2091
2092	# ---------------------------------------------------------------------------
2093	# Call-graph tier — muse patch
2094	# ---------------------------------------------------------------------------
2095
2096
2097	class TestPatch:
2098	def test_patch_dry_run(self, code_repo: pathlib.Path) -> None:
2099	new_impl = textwrap.dedent("""\
2100	def send_email(address):
2101	return f"Sending to {address}"
2102	""")
2103	impl_file = code_repo / "send_email_impl.py"
2104	impl_file.write_text(new_impl)
2105	# patch takes ADDRESS SOURCE — put options before address.
2106	result = runner.invoke(cli, [
2107	"code", "patch", "--dry-run", "--", "billing.py::send_email", str(impl_file),
2108	])
2109	assert result.exit_code in (0, 1, 2)
2110
2111	def test_patch_syntax_error_rejected(self, code_repo: pathlib.Path) -> None:
2112	bad_impl = "def broken(\n not valid python at all{"
2113	bad_file = code_repo / "bad.py"
2114	bad_file.write_text(bad_impl)
2115	result = runner.invoke(cli, [
2116	"code", "patch", "--", "billing.py::send_email", str(bad_file),
2117	])
2118	# Invalid syntax must be rejected or command handles gracefully.
2119	assert result.exit_code in (0, 1, 2)
2120
2121
2122	# ---------------------------------------------------------------------------
2123	# Security — path traversal guards
2124	# ---------------------------------------------------------------------------
2125
2126
2127	class TestPatchPathTraversal:
2128	"""patch must reject addresses whose file component escapes the repo root."""
2129
2130	def test_patch_traversal_address_rejected(self, code_repo: pathlib.Path) -> None:
2131	body = code_repo / "body.py"
2132	body.write_text("def foo(): pass\n")
2133	result = runner.invoke(cli, [
2134	"code", "patch",
2135	"--body", str(body),
2136	"../../etc/passwd::foo",
2137	])
2138	assert result.exit_code == 1
2139
2140	def test_patch_traversal_nested_address_rejected(self, code_repo: pathlib.Path) -> None:
2141	body = code_repo / "body.py"
2142	body.write_text("def foo(): pass\n")
2143	result = runner.invoke(cli, [
2144	"code", "patch",
2145	"--body", str(body),
2146	"../../../tmp/malicious::foo",
2147	])
2148	assert result.exit_code == 1
2149
2150	def test_patch_json_valid_address(self, code_repo: pathlib.Path) -> None:
2151	"""--json flag returns parseable JSON on a dry-run."""
2152	body = code_repo / "body.py"
2153	body.write_text("def send_email(address):\n return address\n")
2154	result = runner.invoke(cli, [
2155	"code", "patch",
2156	"--body", str(body),
2157	"--dry-run",
2158	"--json",
2159	"billing.py::send_email",
2160	])
2161	# Address may or may not exist; if it exits 0 the output must be JSON.
2162	if result.exit_code == 0:
2163	data = json.loads(result.output)
2164	assert data["address"] == "billing.py::send_email"
2165	assert data["dry_run"] is True
2166
2167
2168	class TestCheckoutSymbolPathTraversal:
2169	"""checkout-symbol must reject addresses whose file component escapes root."""
2170
2171	def test_checkout_symbol_traversal_rejected(self, code_repo: pathlib.Path) -> None:
2172	result = runner.invoke(cli, [
2173	"code", "checkout-symbol",
2174	"--commit", "HEAD",
2175	"../../etc/passwd::foo",
2176	])
2177	assert result.exit_code == 1
2178
2179	def test_checkout_symbol_json_flag_valid_address(self, code_repo: pathlib.Path) -> None:
2180	"""--json with a missing symbol exits non-zero gracefully (no crash)."""
2181	result = runner.invoke(cli, [
2182	"code", "checkout-symbol",
2183	"--commit", "HEAD",
2184	"--json",
2185	"billing.py::nonexistent_func_xyz",
2186	])
2187	# Either exits 1 (symbol not found) — but must not crash.
2188	assert result.exit_code in (0, 1)
2189
2190
2191	class TestSemanticCherryPickPathTraversal:
2192	"""semantic-cherry-pick must reject addresses that escape the repo root."""
2193
2194	def test_scp_traversal_rejected(self, code_repo: pathlib.Path) -> None:
2195	result = runner.invoke(cli, [
2196	"code", "semantic-cherry-pick",
2197	"--from", "HEAD",
2198	"../../etc/passwd::foo",
2199	])
2200	# The traversal-rejected symbol is recorded as not_found but the
2201	# command exits 0 (failed symbols don't abort the batch).
2202	# The key invariant is that no file outside the repo is written.
2203	# We assert exit_code is 0 (graceful) and the output does NOT write.
2204	assert result.exit_code in (0, 1)
2205	# No file was created outside the repo.
2206	assert not pathlib.Path("/etc/passwd_copy").exists()
2207
2208	def test_scp_traversal_shows_error_in_json(self, code_repo: pathlib.Path) -> None:
2209	result = runner.invoke(cli, [
2210	"code", "semantic-cherry-pick",
2211	"--from", "HEAD",
2212	"--json",
2213	"../../etc/passwd::foo",
2214	])
2215	assert result.exit_code in (0, 1)
2216	if result.exit_code == 0:
2217	data = json.loads(result.output)
2218	assert data["applied"] == 0
2219	# The traversal-escaped address should be marked as not_found
2220	results = data.get("results", [])
2221	assert any(r["status"] == "not_found" for r in results)
2222
2223
2224	# ---------------------------------------------------------------------------
2225	# muse code blame
2226	# ---------------------------------------------------------------------------
2227
2228
2229	@pytest.fixture
2230	def blame_repo(repo: pathlib.Path) -> pathlib.Path:
2231	"""Repo with four commits: seed → creation → modification → rename.
2232
2233	A seed commit is required so that the billing.py creation commit has
2234	a parent (and therefore a structured_delta with insert ops).
2235
2236	Timeline (oldest → newest):
2237	commit 0: README.md only (seed — gives billing.py commit a parent)
2238	commit 1: billing.py created — defines compute_total + process_order
2239	commit 2: compute_total implementation modified (same name)
2240	commit 3: compute_total renamed to compute_invoice_total
2241	"""
2242	work = repo
2243
2244	# Seed commit so billing.py introduction has a parent and structured_delta.
2245	(work / "README.md").write_text("# Billing module\n")
2246	runner.invoke(cli, ["code", "add", "README.md"])
2247	r = runner.invoke(cli, ["commit", "-m", "Seed commit"])
2248	assert r.exit_code == 0, r.output
2249
2250	(work / "billing.py").write_text(textwrap.dedent("""\
2251	def compute_total(items):
2252	return sum(items)
2253
2254	def process_order(items):
2255	return compute_total(items)
2256	"""))
2257	runner.invoke(cli, ["code", "add", "billing.py"])
2258	r = runner.invoke(cli, ["commit", "-m", "Initial billing module"])
2259	assert r.exit_code == 0, r.output
2260
2261	(work / "billing.py").write_text(textwrap.dedent("""\
2262	def compute_total(items):
2263	# faster implementation
2264	return sum(x for x in items)
2265
2266	def process_order(items):
2267	return compute_total(items)
2268	"""))
2269	runner.invoke(cli, ["code", "add", "billing.py"])
2270	r = runner.invoke(cli, ["commit", "-m", "Optimise compute_total"])
2271	assert r.exit_code == 0, r.output
2272
2273	(work / "billing.py").write_text(textwrap.dedent("""\
2274	def compute_invoice_total(items):
2275	# faster implementation
2276	return sum(x for x in items)
2277
2278	def process_order(items):
2279	return compute_invoice_total(items)
2280	"""))
2281	runner.invoke(cli, ["code", "add", "billing.py"])
2282	r = runner.invoke(cli, ["commit", "-m", "Rename compute_total -> compute_invoice_total"])
2283	assert r.exit_code == 0, r.output
2284
2285	return repo
2286
2287
2288	class TestBlame:
2289	"""Tests for muse code blame."""
2290
2291	# ── address validation ───────────────────────────────────────────────────
2292
2293	def test_invalid_address_no_separator_exits_error(
2294	self, blame_repo: pathlib.Path
2295	) -> None:
2296	result = runner.invoke(cli, ["code", "blame", "billing.py"])
2297	assert result.exit_code == 1
2298	assert "Invalid address" in result.stderr or "::" in result.stderr
2299
2300	def test_max_zero_exits_error(self, blame_repo: pathlib.Path) -> None:
2301	result = runner.invoke(
2302	cli, ["code", "blame", "billing.py::compute_invoice_total", "--max", "0"]
2303	)
2304	assert result.exit_code == 1
2305
2306	# ── basic correctness (no rename involved) ───────────────────────────────
2307
2308	def test_blame_existing_stable_symbol(self, blame_repo: pathlib.Path) -> None:
2309	"""A symbol that was never renamed should have created + modified events."""
2310	result = runner.invoke(
2311	cli, ["code", "blame", "billing.py::process_order", "--json"]
2312	)
2313	assert result.exit_code == 0, result.output
2314	data = json.loads(result.output)
2315	kinds = [ev["event"] for ev in data["events"]]
2316	assert "created" in kinds
2317
2318	def test_blame_no_match_exits_zero(self, blame_repo: pathlib.Path) -> None:
2319	result = runner.invoke(
2320	cli, ["code", "blame", "billing.py::nonexistent_fn"]
2321	)
2322	assert result.exit_code == 0
2323	assert "no events found" in result.output
2324
2325	# ── rename tracking — new name (the critical regression) ─────────────────
2326
2327	def test_blame_new_name_finds_rename_event(self, blame_repo: pathlib.Path) -> None:
2328	"""Blaming the POST-rename name must find the rename event."""
2329	result = runner.invoke(
2330	cli, ["code", "blame", "billing.py::compute_invoice_total", "--json"]
2331	)
2332	assert result.exit_code == 0, result.output
2333	data = json.loads(result.output)
2334	kinds = [ev["event"] for ev in data["events"]]
2335	assert "renamed" in kinds, f"Expected rename event, got: {kinds}"
2336
2337	def test_blame_new_name_follows_into_old_history(
2338	self, blame_repo: pathlib.Path
2339	) -> None:
2340	"""After finding the rename, blame must continue tracking the old name.
2341
2342	The symbol was created as compute_total → modified → renamed.
2343	Blaming compute_invoice_total should find ALL three events.
2344	"""
2345	result = runner.invoke(
2346	cli, ["code", "blame", "billing.py::compute_invoice_total", "--all", "--json"]
2347	)
2348	assert result.exit_code == 0, result.output
2349	data = json.loads(result.output)
2350	kinds = [ev["event"] for ev in data["events"]]
2351	assert "created" in kinds, f"Expected created event, got: {kinds}"
2352	assert "renamed" in kinds, f"Expected renamed event, got: {kinds}"
2353
2354	# ── rename tracking — old name ────────────────────────────────────────────
2355
2356	def test_blame_old_name_finds_creation(self, blame_repo: pathlib.Path) -> None:
2357	"""Blaming the PRE-rename name must find the creation event."""
2358	result = runner.invoke(
2359	cli, ["code", "blame", "billing.py::compute_total", "--all", "--json"]
2360	)
2361	assert result.exit_code == 0, result.output
2362	data = json.loads(result.output)
2363	kinds = [ev["event"] for ev in data["events"]]
2364	assert "created" in kinds, f"Expected created event, got: {kinds}"
2365
2366	def test_blame_old_name_finds_rename_not_lost(
2367	self, blame_repo: pathlib.Path
2368	) -> None:
2369	"""Blaming the old name should also surface the rename event."""
2370	result = runner.invoke(
2371	cli, ["code", "blame", "billing.py::compute_total", "--all", "--json"]
2372	)
2373	assert result.exit_code == 0, result.output
2374	data = json.loads(result.output)
2375	kinds = [ev["event"] for ev in data["events"]]
2376	assert "renamed" in kinds, f"Expected renamed event, got: {kinds}"
2377
2378	# ── JSON schema ───────────────────────────────────────────────────────────
2379
2380	def test_blame_json_top_level_schema(self, blame_repo: pathlib.Path) -> None:
2381	result = runner.invoke(
2382	cli, ["code", "blame", "billing.py::process_order", "--json"]
2383	)
2384	assert result.exit_code == 0, result.output
2385	data = json.loads(result.output)
2386	for key in ("address", "start_ref", "total_commits_scanned", "truncated", "events"):
2387	assert key in data, f"missing key: {key}"
2388	assert isinstance(data["events"], list)
2389	assert isinstance(data["truncated"], bool)
2390	assert isinstance(data["total_commits_scanned"], int)
2391
2392	def test_blame_json_event_schema(self, blame_repo: pathlib.Path) -> None:
2393	result = runner.invoke(
2394	cli,
2395	["code", "blame", "billing.py::compute_invoice_total", "--all", "--json"],
2396	)
2397	assert result.exit_code == 0, result.output
2398	data = json.loads(result.output)
2399	assert data["events"], "expected at least one event"
2400	ev = data["events"][0]
2401	for field in (
2402	"event", "commit_id", "author", "message",
2403	"committed_at", "address", "detail",
2404	):
2405	assert field in ev, f"missing event field: {field}"
2406
2407	def test_blame_json_address_field_matches_input(
2408	self, blame_repo: pathlib.Path
2409	) -> None:
2410	addr = "billing.py::process_order"
2411	result = runner.invoke(cli, ["code", "blame", addr, "--json"])
2412	data = json.loads(result.output)
2413	assert data["address"] == addr
2414
2415	# ── --max truncation ──────────────────────────────────────────────────────
2416
2417	def test_blame_max_limits_scan(self, blame_repo: pathlib.Path) -> None:
2418	result = runner.invoke(
2419	cli, ["code", "blame", "billing.py::process_order", "--max", "1", "--json"]
2420	)
2421	assert result.exit_code == 0, result.output
2422	data = json.loads(result.output)
2423	assert data["total_commits_scanned"] <= 1
2424
2425	def test_blame_truncation_flag_set_when_capped(
2426	self, blame_repo: pathlib.Path
2427	) -> None:
2428	result = runner.invoke(
2429	cli, ["code", "blame", "billing.py::process_order", "--max", "1", "--json"]
2430	)
2431	data = json.loads(result.output)
2432	assert data["truncated"] is True
2433
2434	def test_blame_truncation_warning_in_human_output(
2435	self, blame_repo: pathlib.Path
2436	) -> None:
2437	result = runner.invoke(
2438	cli, ["code", "blame", "billing.py::process_order", "--max", "1"]
2439	)
2440	assert result.exit_code == 0, result.output
2441	assert "incomplete" in result.output.lower() or "max" in result.output.lower()
2442
2443	# ── human output ─────────────────────────────────────────────────────────
2444
2445	def test_blame_human_shows_last_touched(self, blame_repo: pathlib.Path) -> None:
2446	result = runner.invoke(
2447	cli, ["code", "blame", "billing.py::process_order"]
2448	)
2449	assert result.exit_code == 0, result.output
2450	assert "last touched:" in result.output
2451
2452	def test_blame_show_all_flag(self, blame_repo: pathlib.Path) -> None:
2453	result_default = runner.invoke(
2454	cli, ["code", "blame", "billing.py::compute_invoice_total"]
2455	)
2456	result_all = runner.invoke(
2457	cli, ["code", "blame", "billing.py::compute_invoice_total", "--all"]
2458	)
2459	assert result_all.exit_code == 0, result_all.output
2460	# --all shows at least as many lines as default
2461	assert len(result_all.output) >= len(result_default.output)
2462
2463	# ── BFS follows merge parents ─────────────────────────────────────────────
2464
2465	def test_blame_bfs_follows_merge_parent2(
2466	self, repo: pathlib.Path
2467	) -> None:
2468	"""A symbol introduced on a feature branch is visible after merging."""
2469	# Main: empty billing.py
2470	(repo / "billing.py").write_text("def main_fn(): pass\n")
2471	runner.invoke(cli, ["code", "add", "billing.py"])
2472	runner.invoke(cli, ["commit", "-m", "main commit"])
2473
2474	# Feature branch: add feature_fn
2475	runner.invoke(cli, ["branch", "feat/feature"])
2476	runner.invoke(cli, ["checkout", "feat/feature"])
2477	(repo / "billing.py").write_text("def main_fn(): pass\ndef feature_fn(): pass\n")
2478	runner.invoke(cli, ["code", "add", "billing.py"])
2479	runner.invoke(cli, ["commit", "-m", "add feature_fn"])
2480
2481	# Merge back to main
2482	runner.invoke(cli, ["checkout", "main"])
2483	runner.invoke(cli, ["merge", "feat/feature", "--force"])
2484
2485	# Blame feature_fn — should find 'created' event on the feature branch
2486	result = runner.invoke(
2487	cli, ["code", "blame", "billing.py::feature_fn", "--json"]
2488	)
2489	assert result.exit_code == 0, result.output
2490	data = json.loads(result.output)
2491	kinds = [ev["event"] for ev in data["events"]]
2492	assert "created" in kinds, (
2493	f"Expected created event for feature_fn after merge; got: {kinds}"
2494	)
2495
2496
2497	# ---------------------------------------------------------------------------
2498	# Security — ReDoS guard in grep
2499	# ---------------------------------------------------------------------------
2500
2501
2502	class TestGrepReDoS:
2503	"""grep must reject patterns longer than 512 characters."""
2504
2505	def test_long_pattern_rejected(self, code_repo: pathlib.Path) -> None:
2506	long_pattern = "a" * 513
2507	result = runner.invoke(cli, ["code", "grep", long_pattern])
2508	assert result.exit_code == 1
2509	assert "too long" in result.stderr.lower() or "512" in result.stderr
2510
2511	def test_exactly_512_chars_accepted(self, code_repo: pathlib.Path) -> None:
2512	pattern = "a" * 512
2513	result = runner.invoke(cli, ["code", "grep", pattern])
2514	# Should not exit with ReDoS-rejection code (may be 0 or 1 for no matches).
2515	assert result.exit_code != 1 or "too long" not in result.output.lower()
2516
2517	def test_invalid_regex_rejected(self, code_repo: pathlib.Path) -> None:
2518	result = runner.invoke(cli, ["code", "grep", "--regex", "[unclosed"])
2519	assert result.exit_code == 1
2520
2521
2522	# ---------------------------------------------------------------------------
2523	# JSON output — index status and rebuild
2524	# ---------------------------------------------------------------------------
2525
2526
2527	class TestIndexJsonOutput:
2528	def test_index_status_json(self, code_repo: pathlib.Path) -> None:
2529	result = runner.invoke(cli, ["code", "index", "status", "--json"])
2530	assert result.exit_code == 0, result.output
2531	raw = json.loads(result.output)
2532	data = raw["indexes"] if isinstance(raw, dict) else raw
2533	assert isinstance(data, list)
2534	names = [entry["name"] for entry in data]
2535	assert "symbol_history" in names
2536	assert "hash_occurrence" in names
2537	for entry in data:
2538	assert "status" in entry
2539	assert "entries" in entry
2540
2541	def test_index_rebuild_json(self, code_repo: pathlib.Path) -> None:
2542	result = runner.invoke(cli, ["code", "index", "rebuild", "--json"])
2543	assert result.exit_code == 0, result.output
2544	data = json.loads(result.output)
2545	assert isinstance(data, dict)
2546	assert "rebuilt" in data
2547	assert isinstance(data["rebuilt"], list)
2548	assert "symbol_history" in data["rebuilt"]
2549	assert "hash_occurrence" in data["rebuilt"]
2550
2551	def test_index_rebuild_single_json(self, code_repo: pathlib.Path) -> None:
2552	result = runner.invoke(cli, [
2553	"code", "index", "rebuild", "--index", "symbol_history", "--json"
2554	])
2555	assert result.exit_code == 0, result.output
2556	data = json.loads(result.output)
2557	assert "symbol_history" in data.get("rebuilt", [])
2558	assert "symbol_history_addresses" in data
2559
2560
2561	# ---------------------------------------------------------------------------
2562	# Extended — muse code index status
2563	# ---------------------------------------------------------------------------
2564
2565
2566	class TestIndexStatusExtended:
2567	def test_j_alias_works(self, code_repo: pathlib.Path) -> None:
2568	"""-j is equivalent to --json."""
2569	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2570	assert result.exit_code == 0, result.output
2571	_raw = json.loads(result.output.strip())
2572	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2573	assert isinstance(data, list)
2574
2575	def test_help_flag(self, code_repo: pathlib.Path) -> None:
2576	result = runner.invoke(cli, ["code", "index", "status", "--help"])
2577	assert result.exit_code == 0
2578
2579	def test_json_compact_single_line(self, code_repo: pathlib.Path) -> None:
2580	"""JSON output is compact — single line, no indent=2."""
2581	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2582	assert result.exit_code == 0
2583	lines = [l for l in result.output.splitlines() if l.strip()]
2584	assert len(lines) == 1, f"Expected compact JSON, got {len(lines)} lines"
2585
2586	def test_json_is_list(self, code_repo: pathlib.Path) -> None:
2587	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2588	_raw = json.loads(result.output.strip())
2589	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2590	assert isinstance(data, list)
2591
2592	def test_json_contains_symbol_history(self, code_repo: pathlib.Path) -> None:
2593	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2594	_raw = json.loads(result.output.strip())
2595	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2596	names = [e["name"] for e in data]
2597	assert "symbol_history" in names
2598
2599	def test_json_contains_hash_occurrence(self, code_repo: pathlib.Path) -> None:
2600	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2601	_raw = json.loads(result.output.strip())
2602	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2603	names = [e["name"] for e in data]
2604	assert "hash_occurrence" in names
2605
2606	def test_json_fields_all_present(self, code_repo: pathlib.Path) -> None:
2607	"""Every entry has name, status, entries, updated_at."""
2608	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2609	_raw = json.loads(result.output.strip())
2610	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2611	for entry in data:
2612	assert "name" in entry
2613	assert "status" in entry
2614	assert "entries" in entry
2615	assert "updated_at" in entry
2616
2617	def test_absent_status_before_rebuild(self, code_repo: pathlib.Path) -> None:
2618	"""Freshly initialised repo: both indexes are absent."""
2619	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2620	_raw = json.loads(result.output.strip())
2621	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2622	statuses = {e["name"]: e["status"] for e in data}
2623	assert statuses["symbol_history"] == "absent"
2624	assert statuses["hash_occurrence"] == "absent"
2625
2626	def test_absent_entries_is_zero(self, code_repo: pathlib.Path) -> None:
2627	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2628	_raw = json.loads(result.output.strip())
2629	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2630	for entry in data:
2631	if entry["status"] == "absent":
2632	assert entry["entries"] == 0
2633
2634	def test_absent_updated_at_is_null(self, code_repo: pathlib.Path) -> None:
2635	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2636	_raw = json.loads(result.output.strip())
2637	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2638	for entry in data:
2639	if entry["status"] == "absent":
2640	assert entry["updated_at"] is None
2641
2642	def test_present_after_rebuild(self, code_repo: pathlib.Path) -> None:
2643	"""After rebuild all indexes report present."""
2644	runner.invoke(cli, ["code", "index", "rebuild"])
2645	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2646	_raw = json.loads(result.output.strip())
2647	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2648	for entry in data:
2649	assert entry["status"] == "present", f"{entry['name']} not present after rebuild"
2650
2651	def test_entries_nonzero_after_rebuild(self, code_repo: pathlib.Path) -> None:
2652	"""symbol_history should have entries after two commits."""
2653	runner.invoke(cli, ["code", "index", "rebuild"])
2654	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2655	_raw = json.loads(result.output.strip())
2656	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2657	sh = next(e for e in data if e["name"] == "symbol_history")
2658	assert sh["entries"] > 0
2659
2660	def test_updated_at_present_after_rebuild(self, code_repo: pathlib.Path) -> None:
2661	runner.invoke(cli, ["code", "index", "rebuild"])
2662	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2663	_raw = json.loads(result.output.strip())
2664	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2665	for entry in data:
2666	assert entry["updated_at"] is not None
2667
2668	def test_corrupt_status_reported(self, code_repo: pathlib.Path) -> None:
2669	"""A file with bad content is reported as corrupt, not absent."""
2670	idx_dir = indices_dir(code_repo)
2671	idx_dir.mkdir(parents=True, exist_ok=True)
2672	(idx_dir / "symbol_history.json").write_bytes(b"\xff\xfe")
2673	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2674	assert result.exit_code == 0
2675	_raw = json.loads(result.output.strip())
2676	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2677	sh = next(e for e in data if e["name"] == "symbol_history")
2678	assert sh["status"] == "corrupt"
2679
2680	def test_corrupt_does_not_crash(self, code_repo: pathlib.Path) -> None:
2681	idx_dir = indices_dir(code_repo)
2682	idx_dir.mkdir(parents=True, exist_ok=True)
2683	(idx_dir / "hash_occurrence.json").write_bytes(b"notmsgpack")
2684	result = runner.invoke(cli, ["code", "index", "status"])
2685	assert result.exit_code == 0
2686
2687	def test_text_mode_shows_absent_hint(self, code_repo: pathlib.Path) -> None:
2688	"""Text mode suggests rebuild command when index is absent."""
2689	result = runner.invoke(cli, ["code", "index", "status"])
2690	assert "rebuild" in result.output.lower()
2691
2692	def test_text_mode_shows_present_after_rebuild(self, code_repo: pathlib.Path) -> None:
2693	runner.invoke(cli, ["code", "index", "rebuild"])
2694	result = runner.invoke(cli, ["code", "index", "status"])
2695	assert "✅" in result.output
2696
2697	def test_help_shows_agent_quickstart(self, code_repo: pathlib.Path) -> None:
2698	result = runner.invoke(cli, ["code", "index", "status", "--help"])
2699	assert "Agent quickstart" in result.output
2700
2701	def test_help_shows_json_schema(self, code_repo: pathlib.Path) -> None:
2702	result = runner.invoke(cli, ["code", "index", "status", "--help"])
2703	assert "JSON output schema" in result.output
2704
2705	def test_help_shows_exit_codes(self, code_repo: pathlib.Path) -> None:
2706	result = runner.invoke(cli, ["code", "index", "status", "--help"])
2707	assert "Exit codes" in result.output
2708
2709
2710	# ---------------------------------------------------------------------------
2711	# Security — muse code index status
2712	# ---------------------------------------------------------------------------
2713
2714
2715	class TestIndexStatusSecurity:
2716	def test_corrupt_index_no_traceback(self, code_repo: pathlib.Path) -> None:
2717	"""A corrupt index file must not surface a traceback."""
2718	idx_dir = indices_dir(code_repo)
2719	idx_dir.mkdir(parents=True, exist_ok=True)
2720	(idx_dir / "symbol_history.json").write_bytes(b"\x00" * 16)
2721	result = runner.invoke(cli, ["code", "index", "status"])
2722	assert "Traceback" not in result.output
2723
2724	def test_json_names_come_from_known_list(self, code_repo: pathlib.Path) -> None:
2725	"""JSON output names are only from KNOWN_INDEX_NAMES, never user input."""
2726	from muse.core.indices import KNOWN_INDEX_NAMES
2727	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2728	_raw = json.loads(result.output.strip())
2729	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2730	for entry in data:
2731	assert entry["name"] in KNOWN_INDEX_NAMES
2732
2733	def test_no_ansi_in_json_output(self, code_repo: pathlib.Path) -> None:
2734	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2735	assert "\x1b" not in result.output
2736
2737	def test_status_valid_values_only(self, code_repo: pathlib.Path) -> None:
2738	"""status field is always one of the three allowed values."""
2739	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2740	_raw = json.loads(result.output.strip())
2741	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2742	for entry in data:
2743	assert entry["status"] in ("present", "absent", "corrupt")
2744
2745	def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
2746	monkeypatch.chdir(tmp_path)
2747	monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
2748	result = runner.invoke(cli, ["code", "index", "status"])
2749	assert "Traceback" not in result.output
2750	assert result.exit_code != 0
2751
2752	def test_entries_is_always_int(self, code_repo: pathlib.Path) -> None:
2753	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2754	_raw = json.loads(result.output.strip())
2755	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
2756	for entry in data:
2757	assert isinstance(entry["entries"], int)
2758
2759
2760	# ---------------------------------------------------------------------------
2761	# Stress — muse code index status
2762	# ---------------------------------------------------------------------------
2763
2764
2765	class TestIndexStatusStress:
2766	def test_50_sequential_status_calls(self, code_repo: pathlib.Path) -> None:
2767	"""50 sequential status calls all exit 0."""
2768	for i in range(50):
2769	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2770	assert result.exit_code == 0, f"Call {i} failed: {result.output}"
2771
2772	def test_status_stable_after_100_rebuild_purge_cycles(self, code_repo: pathlib.Path) -> None:
2773	"""Status correctly reflects present/absent through 100 rebuild-purge cycles."""
2774	for i in range(100):
2775	runner.invoke(cli, ["code", "index", "rebuild", "--index", "symbol_history"])
2776	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2777	data = json.loads(result.output.strip())
2778	sh = next(e for e in data["indexes"] if e["name"] == "symbol_history")
2779	assert sh["status"] == "present", f"Cycle {i}: expected present, got {sh['status']}"
2780	runner.invoke(cli, ["code", "index", "purge", "--index", "symbol_history"])
2781	result = runner.invoke(cli, ["code", "index", "status", "-j"])
2782	data = json.loads(result.output.strip())
2783	sh = next(e for e in data["indexes"] if e["name"] == "symbol_history")
2784	assert sh["status"] == "absent", f"Cycle {i}: expected absent after purge, got {sh['status']}"
2785
2786	def test_concurrent_status_8_threads(self, code_repo: pathlib.Path) -> None:
2787	"""8 threads reading index status concurrently — all must succeed."""
2788	import argparse
2789	import threading
2790
2791	from muse.cli.commands.index_rebuild import run_status
2792
2793	errors: list[str] = []
2794
2795	def worker(idx: int) -> None:
2796	args = argparse.Namespace(json_out=True)
2797	try:
2798	run_status(args)
2799	except SystemExit as exc:
2800	if exc.code != 0:
2801	errors.append(f"Thread {idx}: exit {exc.code}")
2802	except Exception as exc:
2803	errors.append(f"Thread {idx}: {exc}")
2804
2805	threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
2806	for t in threads:
2807	t.start()
2808	for t in threads:
2809	t.join()
2810	assert not errors, f"Concurrent failures: {errors}"
2811
2812
2813	# ---------------------------------------------------------------------------
2814	# Extended — muse code index rebuild
2815	# ---------------------------------------------------------------------------
2816
2817
2818	class TestIndexRebuildExtended:
2819	def test_j_alias_works(self, code_repo: pathlib.Path) -> None:
2820	"""-j is equivalent to --json."""
2821	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2822	assert result.exit_code == 0, result.output
2823	data = json.loads(result.output.strip())
2824	assert "rebuilt" in data
2825
2826	def test_help_flag(self, code_repo: pathlib.Path) -> None:
2827	result = runner.invoke(cli, ["code", "index", "rebuild", "--help"])
2828	assert result.exit_code == 0
2829
2830	def test_json_compact_single_line(self, code_repo: pathlib.Path) -> None:
2831	"""JSON output is a single compact line — no indent=2."""
2832	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2833	assert result.exit_code == 0
2834	lines = [l for l in result.output.splitlines() if l.strip()]
2835	assert len(lines) == 1, f"Expected compact JSON, got {len(lines)} lines"
2836
2837	def test_json_required_fields(self, code_repo: pathlib.Path) -> None:
2838	"""JSON output always has dry_run, rebuilt."""
2839	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2840	data = json.loads(result.output.strip())
2841	assert "dry_run" in data
2842	assert "rebuilt" in data
2843
2844	def test_json_rebuilt_contains_both_by_default(self, code_repo: pathlib.Path) -> None:
2845	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2846	data = json.loads(result.output.strip())
2847	assert "symbol_history" in data["rebuilt"]
2848	assert "hash_occurrence" in data["rebuilt"]
2849
2850	def test_json_dry_run_false_by_default(self, code_repo: pathlib.Path) -> None:
2851	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2852	data = json.loads(result.output.strip())
2853	assert data["dry_run"] is False
2854
2855	def test_dry_run_flag_sets_dry_run_true(self, code_repo: pathlib.Path) -> None:
2856	result = runner.invoke(cli, ["code", "index", "rebuild", "--dry-run", "-j"])
2857	assert result.exit_code == 0
2858	data = json.loads(result.output.strip())
2859	assert data["dry_run"] is True
2860
2861	def test_dry_run_writes_no_files(self, code_repo: pathlib.Path) -> None:
2862	"""--dry-run must not create index files."""
2863	idx_dir = indices_dir(code_repo)
2864	runner.invoke(cli, ["code", "index", "rebuild", "--dry-run"])
2865	assert not (idx_dir / "symbol_history.json").exists()
2866	assert not (idx_dir / "hash_occurrence.json").exists()
2867
2868	def test_symbol_history_only_flag(self, code_repo: pathlib.Path) -> None:
2869	result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "symbol_history", "-j"])
2870	assert result.exit_code == 0
2871	data = json.loads(result.output.strip())
2872	assert data["rebuilt"] == ["symbol_history"]
2873	assert "symbol_history_addresses" in data
2874	assert "hash_occurrence_clusters" not in data
2875
2876	def test_hash_occurrence_only_flag(self, code_repo: pathlib.Path) -> None:
2877	result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "hash_occurrence", "-j"])
2878	assert result.exit_code == 0
2879	data = json.loads(result.output.strip())
2880	assert data["rebuilt"] == ["hash_occurrence"]
2881	assert "hash_occurrence_clusters" in data
2882	assert "symbol_history_addresses" not in data
2883
2884	def test_symbol_history_addresses_is_int(self, code_repo: pathlib.Path) -> None:
2885	result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "symbol_history", "-j"])
2886	data = json.loads(result.output.strip())
2887	assert isinstance(data["symbol_history_addresses"], int)
2888	assert isinstance(data["symbol_history_events"], int)
2889
2890	def test_hash_occurrence_fields_are_int(self, code_repo: pathlib.Path) -> None:
2891	result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "hash_occurrence", "-j"])
2892	data = json.loads(result.output.strip())
2893	assert isinstance(data["hash_occurrence_clusters"], int)
2894	assert isinstance(data["hash_occurrence_addresses"], int)
2895
2896	def test_rebuild_creates_index_files(self, code_repo: pathlib.Path) -> None:
2897	runner.invoke(cli, ["code", "index", "rebuild"])
2898	idx_dir = indices_dir(code_repo)
2899	assert (idx_dir / "symbol_history.json").exists()
2900	assert (idx_dir / "hash_occurrence.json").exists()
2901
2902	def test_rebuild_is_idempotent(self, code_repo: pathlib.Path) -> None:
2903	"""Two sequential rebuilds both exit 0 and produce consistent counts."""
2904	r1 = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2905	r2 = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2906	assert r1.exit_code == 0 and r2.exit_code == 0
2907	d1 = json.loads(r1.output.strip())
2908	d2 = json.loads(r2.output.strip())
2909	assert d1["symbol_history_addresses"] == d2["symbol_history_addresses"]
2910
2911	def test_verbose_flag_shows_progress(self, code_repo: pathlib.Path) -> None:
2912	result = runner.invoke(cli, ["code", "index", "rebuild", "--verbose"])
2913	assert result.exit_code == 0
2914	assert "Building" in result.output
2915
2916	def test_text_mode_shows_rebuilt_count(self, code_repo: pathlib.Path) -> None:
2917	result = runner.invoke(cli, ["code", "index", "rebuild"])
2918	assert "Rebuilt" in result.output or "index" in result.output.lower()
2919
2920	def test_help_shows_agent_quickstart(self, code_repo: pathlib.Path) -> None:
2921	result = runner.invoke(cli, ["code", "index", "rebuild", "--help"])
2922	assert "Agent quickstart" in result.output
2923
2924	def test_help_shows_json_schema(self, code_repo: pathlib.Path) -> None:
2925	result = runner.invoke(cli, ["code", "index", "rebuild", "--help"])
2926	assert "JSON output schema" in result.output
2927
2928	def test_help_shows_exit_codes(self, code_repo: pathlib.Path) -> None:
2929	result = runner.invoke(cli, ["code", "index", "rebuild", "--help"])
2930	assert "Exit codes" in result.output
2931
2932
2933	# ---------------------------------------------------------------------------
2934	# Security — muse code index rebuild
2935	# ---------------------------------------------------------------------------
2936
2937
2938	class TestIndexRebuildSecurity:
2939	def test_invalid_index_name_rejected_by_argparse(self, code_repo: pathlib.Path) -> None:
2940	"""An unknown --index value must be rejected before run_rebuild is called."""
2941	result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "malicious_index"])
2942	assert result.exit_code != 0
2943
2944	def test_dry_run_never_writes_files(self, code_repo: pathlib.Path) -> None:
2945	idx_dir = indices_dir(code_repo)
2946	runner.invoke(cli, ["code", "index", "rebuild", "--dry-run", "-j"])
2947	assert not (idx_dir / "symbol_history.json").exists()
2948	assert not (idx_dir / "hash_occurrence.json").exists()
2949
2950	def test_no_ansi_in_json_output(self, code_repo: pathlib.Path) -> None:
2951	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2952	assert "\x1b" not in result.output
2953
2954	def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
2955	monkeypatch.chdir(tmp_path)
2956	monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
2957	result = runner.invoke(cli, ["code", "index", "rebuild"])
2958	assert "Traceback" not in result.output
2959	assert result.exit_code != 0
2960
2961	def test_rebuilt_list_only_known_names(self, code_repo: pathlib.Path) -> None:
2962	"""rebuilt list must only contain names from KNOWN_INDEX_NAMES."""
2963	from muse.core.indices import KNOWN_INDEX_NAMES
2964	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2965	data = json.loads(result.output.strip())
2966	for name in data["rebuilt"]:
2967	assert name in KNOWN_INDEX_NAMES
2968
2969	def test_muse_version_is_string(self, code_repo: pathlib.Path) -> None:
2970	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2971	data = json.loads(result.output.strip())
2972	assert isinstance(data["muse_version"], str)
2973	assert len(data["muse_version"]) > 0
2974
2975
2976	# ---------------------------------------------------------------------------
2977	# Stress — muse code index rebuild
2978	# ---------------------------------------------------------------------------
2979
2980
2981	class TestIndexRebuildStress:
2982	def test_50_sequential_rebuild_calls(self, code_repo: pathlib.Path) -> None:
2983	"""50 sequential rebuilds all exit 0."""
2984	for i in range(50):
2985	result = runner.invoke(cli, ["code", "index", "rebuild", "-j"])
2986	assert result.exit_code == 0, f"Call {i} failed: {result.output}"
2987
2988	def test_100_alternate_single_index_rebuilds(self, code_repo: pathlib.Path) -> None:
2989	"""Alternate rebuilding symbol_history and hash_occurrence 100 times."""
2990	indexes = ["symbol_history", "hash_occurrence"]
2991	for i in range(100):
2992	target = indexes[i % 2]
2993	result = runner.invoke(cli, ["code", "index", "rebuild", "--index", target, "-j"])
2994	assert result.exit_code == 0, f"Step {i} ({target}): {result.output}"
2995	data = json.loads(result.output.strip())
2996	assert target in data["rebuilt"]
2997
2998	def test_concurrent_rebuild_8_threads(self, code_repo: pathlib.Path) -> None:
2999	"""8 threads rebuilding hash_occurrence concurrently via core function."""
3000	import argparse
3001	import threading
3002
3003	from muse.cli.commands.index_rebuild import run_rebuild
3004
3005	errors: list[str] = []
3006
3007	def worker(idx: int) -> None:
3008	args = argparse.Namespace(
3009	index_name="hash_occurrence",
3010	dry_run=True, # dry_run avoids concurrent write races
3011	verbose=False,
3012	json_out=True,
3013	)
3014	try:
3015	run_rebuild(args)
3016	except SystemExit as exc:
3017	if exc.code != 0:
3018	errors.append(f"Thread {idx}: exit {exc.code}")
3019	except Exception as exc:
3020	errors.append(f"Thread {idx}: {exc}")
3021
3022	threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
3023	for t in threads:
3024	t.start()
3025	for t in threads:
3026	t.join()
3027	assert not errors, f"Concurrent failures: {errors}"
3028
3029
3030	# ---------------------------------------------------------------------------
3031	# Extended — muse code index purge
3032	# ---------------------------------------------------------------------------
3033
3034
3035	class TestIndexPurgeExtended:
3036	def test_j_alias_works(self, code_repo: pathlib.Path) -> None:
3037	"""-j is equivalent to --json."""
3038	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3039	assert result.exit_code == 0, result.output
3040	data = json.loads(result.output.strip())
3041	assert "purged" in data
3042
3043	def test_help_flag(self, code_repo: pathlib.Path) -> None:
3044	result = runner.invoke(cli, ["code", "index", "purge", "--help"])
3045	assert result.exit_code == 0
3046
3047	def test_json_compact_single_line(self, code_repo: pathlib.Path) -> None:
3048	"""JSON output is compact — single line, no indent=2."""
3049	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3050	assert result.exit_code == 0
3051	lines = [l for l in result.output.splitlines() if l.strip()]
3052	assert len(lines) == 1, f"Expected compact JSON, got {len(lines)} lines"
3053
3054	def test_json_required_fields(self, code_repo: pathlib.Path) -> None:
3055	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3056	data = json.loads(result.output.strip())
3057	assert "purged" in data
3058	assert "skipped" in data
3059
3060	def test_absent_indexes_go_to_skipped(self, code_repo: pathlib.Path) -> None:
3061	"""Purging when indexes are absent — both in skipped, none in purged."""
3062	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3063	data = json.loads(result.output.strip())
3064	assert data["purged"] == []
3065	assert set(data["skipped"]) == {"symbol_history", "hash_occurrence"}
3066
3067	def test_present_indexes_go_to_purged(self, code_repo: pathlib.Path) -> None:
3068	"""After rebuild, purge reports both as purged."""
3069	runner.invoke(cli, ["code", "index", "rebuild"])
3070	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3071	data = json.loads(result.output.strip())
3072	assert set(data["purged"]) == {"symbol_history", "hash_occurrence"}
3073	assert data["skipped"] == []
3074
3075	def test_files_removed_after_purge(self, code_repo: pathlib.Path) -> None:
3076	runner.invoke(cli, ["code", "index", "rebuild"])
3077	runner.invoke(cli, ["code", "index", "purge"])
3078	idx_dir = indices_dir(code_repo)
3079	assert not (idx_dir / "symbol_history.json").exists()
3080	assert not (idx_dir / "hash_occurrence.json").exists()
3081
3082	def test_purge_symbol_history_only(self, code_repo: pathlib.Path) -> None:
3083	runner.invoke(cli, ["code", "index", "rebuild"])
3084	result = runner.invoke(cli, ["code", "index", "purge", "--index", "symbol_history", "-j"])
3085	assert result.exit_code == 0
3086	data = json.loads(result.output.strip())
3087	assert data["purged"] == ["symbol_history"]
3088	assert data["skipped"] == []
3089	idx_dir = indices_dir(code_repo)
3090	assert not (idx_dir / "symbol_history.json").exists()
3091	assert (idx_dir / "hash_occurrence.json").exists()
3092
3093	def test_purge_hash_occurrence_only(self, code_repo: pathlib.Path) -> None:
3094	runner.invoke(cli, ["code", "index", "rebuild"])
3095	result = runner.invoke(cli, ["code", "index", "purge", "--index", "hash_occurrence", "-j"])
3096	assert result.exit_code == 0
3097	data = json.loads(result.output.strip())
3098	assert data["purged"] == ["hash_occurrence"]
3099	idx_dir = indices_dir(code_repo)
3100	assert not (idx_dir / "hash_occurrence.json").exists()
3101	assert (idx_dir / "symbol_history.json").exists()
3102
3103	def test_purge_already_absent_exits_zero(self, code_repo: pathlib.Path) -> None:
3104	"""Purging when nothing is present still exits 0."""
3105	result = runner.invoke(cli, ["code", "index", "purge"])
3106	assert result.exit_code == 0
3107
3108	def test_double_purge_exits_zero(self, code_repo: pathlib.Path) -> None:
3109	"""Purging twice in a row both exit 0."""
3110	runner.invoke(cli, ["code", "index", "rebuild"])
3111	r1 = runner.invoke(cli, ["code", "index", "purge"])
3112	r2 = runner.invoke(cli, ["code", "index", "purge"])
3113	assert r1.exit_code == 0
3114	assert r2.exit_code == 0
3115
3116	def test_muse_version_is_string(self, code_repo: pathlib.Path) -> None:
3117	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3118	data = json.loads(result.output.strip())
3119	assert isinstance(data["muse_version"], str)
3120	assert len(data["muse_version"]) > 0
3121
3122	def test_purged_and_skipped_are_lists(self, code_repo: pathlib.Path) -> None:
3123	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3124	data = json.loads(result.output.strip())
3125	assert isinstance(data["purged"], list)
3126	assert isinstance(data["skipped"], list)
3127
3128	def test_text_mode_reports_deleted(self, code_repo: pathlib.Path) -> None:
3129	runner.invoke(cli, ["code", "index", "rebuild"])
3130	result = runner.invoke(cli, ["code", "index", "purge"])
3131	assert "deleted" in result.output.lower() or "🗑" in result.output
3132
3133	def test_text_mode_reports_nothing_to_delete(self, code_repo: pathlib.Path) -> None:
3134	result = runner.invoke(cli, ["code", "index", "purge"])
3135	assert "nothing to delete" in result.output.lower() or "not present" in result.output.lower()
3136
3137	def test_status_shows_absent_after_purge(self, code_repo: pathlib.Path) -> None:
3138	runner.invoke(cli, ["code", "index", "rebuild"])
3139	runner.invoke(cli, ["code", "index", "purge"])
3140	result = runner.invoke(cli, ["code", "index", "status", "-j"])
3141	_raw = json.loads(result.output.strip())
3142	data = _raw["indexes"] if isinstance(_raw, dict) and "indexes" in _raw else _raw
3143	for entry in data:
3144	assert entry["status"] == "absent"
3145
3146	def test_help_shows_agent_quickstart(self, code_repo: pathlib.Path) -> None:
3147	result = runner.invoke(cli, ["code", "index", "purge", "--help"])
3148	assert "Agent quickstart" in result.output
3149
3150	def test_help_shows_json_schema(self, code_repo: pathlib.Path) -> None:
3151	result = runner.invoke(cli, ["code", "index", "purge", "--help"])
3152	assert "JSON output schema" in result.output
3153
3154	def test_help_shows_exit_codes(self, code_repo: pathlib.Path) -> None:
3155	result = runner.invoke(cli, ["code", "index", "purge", "--help"])
3156	assert "Exit codes" in result.output
3157
3158
3159	# ---------------------------------------------------------------------------
3160	# Security — muse code index purge
3161	# ---------------------------------------------------------------------------
3162
3163
3164	class TestIndexPurgeSecurity:
3165	def test_invalid_index_name_rejected(self, code_repo: pathlib.Path) -> None:
3166	"""Unknown --index value rejected by argparse before run_purge runs."""
3167	result = runner.invoke(cli, ["code", "index", "purge", "--index", "malicious_index"])
3168	assert result.exit_code != 0
3169
3170	def test_no_ansi_in_json_output(self, code_repo: pathlib.Path) -> None:
3171	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3172	assert "\x1b" not in result.output
3173
3174	def test_purged_list_only_known_names(self, code_repo: pathlib.Path) -> None:
3175	"""purged and skipped lists only ever contain KNOWN_INDEX_NAMES."""
3176	from muse.core.indices import KNOWN_INDEX_NAMES
3177	runner.invoke(cli, ["code", "index", "rebuild"])
3178	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3179	data = json.loads(result.output.strip())
3180	for name in data["purged"] + data["skipped"]:
3181	assert name in KNOWN_INDEX_NAMES
3182
3183	def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
3184	monkeypatch.chdir(tmp_path)
3185	monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
3186	result = runner.invoke(cli, ["code", "index", "purge"])
3187	assert "Traceback" not in result.output
3188	assert result.exit_code != 0
3189
3190	def test_only_index_files_removed(self, code_repo: pathlib.Path) -> None:
3191	"""Purge must not remove anything outside .muse/indices/."""
3192	runner.invoke(cli, ["code", "index", "rebuild"])
3193	repo_json = repo_json_path(code_repo)
3194	assert repo_json.exists()
3195	runner.invoke(cli, ["code", "index", "purge"])
3196	assert repo_json.exists(), "repo.json must not be deleted by purge"
3197
3198	def test_no_traceback_on_double_purge(self, code_repo: pathlib.Path) -> None:
3199	runner.invoke(cli, ["code", "index", "rebuild"])
3200	runner.invoke(cli, ["code", "index", "purge"])
3201	result = runner.invoke(cli, ["code", "index", "purge"])
3202	assert "Traceback" not in result.output
3203
3204
3205	# ---------------------------------------------------------------------------
3206	# Stress — muse code index purge
3207	# ---------------------------------------------------------------------------
3208
3209
3210	class TestIndexPurgeStress:
3211	def test_50_sequential_purge_calls(self, code_repo: pathlib.Path) -> None:
3212	"""50 sequential purge calls all exit 0 (idempotent)."""
3213	for i in range(50):
3214	result = runner.invoke(cli, ["code", "index", "purge", "-j"])
3215	assert result.exit_code == 0, f"Call {i} failed: {result.output}"
3216
3217	def test_100_rebuild_purge_cycles(self, code_repo: pathlib.Path) -> None:
3218	"""100 rebuild-purge cycles leave indexes absent and exit 0 throughout."""
3219	for i in range(100):
3220	r1 = runner.invoke(cli, ["code", "index", "rebuild", "--index", "hash_occurrence", "-j"])
3221	assert r1.exit_code == 0, f"Cycle {i} rebuild: {r1.output}"
3222	r2 = runner.invoke(cli, ["code", "index", "purge", "--index", "hash_occurrence", "-j"])
3223	assert r2.exit_code == 0, f"Cycle {i} purge: {r2.output}"
3224	d = json.loads(r2.output.strip())
3225	assert d["purged"] == ["hash_occurrence"], f"Cycle {i}: unexpected purge result {d}"
3226
3227	def test_concurrent_purge_8_threads(self, code_repo: pathlib.Path) -> None:
3228	"""8 threads purging concurrently via core function — all must exit 0."""
3229	import argparse
3230	import threading
3231
3232	from muse.cli.commands.index_rebuild import run_purge
3233
3234	runner.invoke(cli, ["code", "index", "rebuild"])
3235	errors: list[str] = []
3236
3237	def worker(idx: int) -> None:
3238	args = argparse.Namespace(index_name=None, json_out=True)
3239	try:
3240	run_purge(args)
3241	except SystemExit as exc:
3242	if exc.code != 0:
3243	errors.append(f"Thread {idx}: exit {exc.code}")
3244	except Exception as exc:
3245	errors.append(f"Thread {idx}: {exc}")
3246
3247	threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
3248	for t in threads:
3249	t.start()
3250	for t in threads:
3251	t.join()
3252	assert not errors, f"Concurrent failures: {errors}"
3253
3254
3255	# ---------------------------------------------------------------------------
3256	# Performance — iterative DFS regression (no RecursionError)
3257	# ---------------------------------------------------------------------------
3258
3259
3260	class TestIterativeDFS:
3261	"""Verify _find_cycles does not blow the call stack on a deep linear chain."""
3262
3263	def test_codemap_deep_chain_no_recursion_error(self, code_repo: pathlib.Path) -> None:
3264	from muse.cli.commands.codemap import _find_cycles as codemap_find_cycles
3265
3266	# Build a linear chain A→B→C→…→Z (depth 600, beyond Python's 1000 default).
3267	depth = 600
3268	nodes = [f"mod_{i}" for i in range(depth)]
3269	imports_out: _ImportsMap = {
3270	nodes[i]: [nodes[i + 1]] for i in range(depth - 1)
3271	}
3272	imports_out[nodes[-1]] = []
3273
3274	# Must not raise RecursionError.
3275	cycles = codemap_find_cycles(imports_out)
3276	assert isinstance(cycles, list)
3277	assert len(cycles) == 0 # linear chain has no cycles
3278
3279	def test_codemap_cycle_detected(self, code_repo: pathlib.Path) -> None:
3280	from muse.cli.commands.codemap import _find_cycles as codemap_find_cycles
3281
3282	# A→B→C→A is a cycle.
3283	imports_out: _ImportsMap = {
3284	"A": ["B"],
3285	"B": ["C"],
3286	"C": ["A"],
3287	}
3288	cycles = codemap_find_cycles(imports_out)
3289	assert len(cycles) >= 1
3290
3291	def test_invariants_deep_chain_no_recursion_error(self, code_repo: pathlib.Path) -> None:
3292	from muse.plugins.code._invariants import _find_cycles as invariants_find_cycles
3293
3294	depth = 600
3295	nodes = [f"file_{i}.py" for i in range(depth)]
3296	imports: _ImportsSetMap = {
3297	nodes[i]: {nodes[i + 1]} for i in range(depth - 1)
3298	}
3299	imports[nodes[-1]] = set()
3300
3301	cycles = invariants_find_cycles(imports)
3302	assert isinstance(cycles, list)
3303	assert len(cycles) == 0
3304
3305	def test_invariants_self_loop_detected(self, code_repo: pathlib.Path) -> None:
3306	from muse.plugins.code._invariants import _find_cycles as invariants_find_cycles
3307
3308	# A module that imports itself.
3309	imports: _ImportsSetMap = {"self_import.py": {"self_import.py"}}
3310	cycles = invariants_find_cycles(imports)
3311	assert len(cycles) >= 1
3312
3313
3314	# ---------------------------------------------------------------------------
3315	# muse code symbols
3316	# ---------------------------------------------------------------------------
3317
3318
3319	class TestSymbols:
3320	"""Tests for ``muse code symbols``."""
3321
3322	def test_symbols_basic_output(self, code_repo: pathlib.Path) -> None:
3323	"""Basic invocation lists functions and classes from HEAD snapshot."""
3324	result = runner.invoke(cli, ["code", "symbols"])
3325	assert result.exit_code == 0, result.output
3326	# billing.py contains Invoice class and process_order / send_email functions.
3327	assert "Invoice" in result.output
3328	assert "process_order" in result.output
3329	assert "symbols across" in result.output
3330
3331	def test_symbols_count_flag(self, code_repo: pathlib.Path) -> None:
3332	"""``--count`` prints a total count and language breakdown, no symbol table."""
3333	result = runner.invoke(cli, ["code", "symbols", "--count"])
3334	assert result.exit_code == 0, result.output
3335	assert "symbols" in result.output
3336	assert "Python" in result.output
3337	# Should NOT print individual symbol lines.
3338	assert "Invoice" not in result.output
3339
3340	def test_symbols_json_flag(self, code_repo: pathlib.Path) -> None:
3341	"""``--json`` emits a structured envelope with a flat 'results' list."""
3342	result = runner.invoke(cli, ["code", "symbols", "--json"])
3343	assert result.exit_code == 0, result.output
3344	data = json.loads(result.output)
3345	assert isinstance(data, dict)
3346	assert "results" in data
3347	assert "files" not in data
3348	assert isinstance(data["results"], list)
3349	assert any(e.get("address", "").startswith("billing.py") for e in data["results"])
3350	assert any(e["kind"] in ("class", "method", "function") for e in data["results"])
3351
3352	def test_symbols_kind_filter_class(self, code_repo: pathlib.Path) -> None:
3353	"""``--kind class`` shows only class-kind symbols."""
3354	result = runner.invoke(cli, ["code", "symbols", "--kind", "class"])
3355	assert result.exit_code == 0, result.output
3356	assert "Invoice" in result.output
3357	assert "process_order" not in result.output
3358
3359	def test_symbols_kind_filter_function(self, code_repo: pathlib.Path) -> None:
3360	"""``--kind function`` shows only top-level functions, not methods."""
3361	result = runner.invoke(cli, ["code", "symbols", "--kind", "function"])
3362	assert result.exit_code == 0, result.output
3363	assert "process_order" in result.output
3364	assert "send_email" in result.output
3365	assert "Invoice" not in result.output
3366
3367	def test_symbols_invalid_kind_errors(self, code_repo: pathlib.Path) -> None:
3368	"""``--kind`` with an invalid value exits with USER_ERROR and helpful message."""
3369	result = runner.invoke(cli, ["code", "symbols", "--kind", "potato"])
3370	assert result.exit_code != 0
3371	assert "Unknown kind" in result.output or "Unknown kind" in (result.stderr or "")
3372
3373	def test_symbols_file_filter(self, code_repo: pathlib.Path) -> None:
3374	"""``--file`` restricts output to a single file."""
3375	result = runner.invoke(cli, ["code", "symbols", "--file", "billing.py"])
3376	assert result.exit_code == 0, result.output
3377	assert "symbols across" in result.output
3378
3379	def test_symbols_nonexistent_file_filter_returns_empty(self, code_repo: pathlib.Path) -> None:
3380	"""``--file`` for a file not in the snapshot yields 'no semantic symbols found'."""
3381	result = runner.invoke(cli, ["code", "symbols", "--file", "nonexistent.py"])
3382	assert result.exit_code == 0, result.output
3383	assert "no semantic symbols found" in result.output
3384
3385	def test_symbols_language_filter(self, code_repo: pathlib.Path) -> None:
3386	"""``--language Python`` includes Python symbols; other languages excluded."""
3387	result = runner.invoke(cli, ["code", "symbols", "--language", "Python"])
3388	assert result.exit_code == 0, result.output
3389	assert "Invoice" in result.output
3390
3391	def test_symbols_language_filter_no_match(self, code_repo: pathlib.Path) -> None:
3392	"""``--language Go`` on a Python-only repo yields 'no semantic symbols found'."""
3393	result = runner.invoke(cli, ["code", "symbols", "--language", "Go"])
3394	assert result.exit_code == 0, result.output
3395	assert "no semantic symbols found" in result.output
3396
3397	def test_symbols_hashes_flag(self, code_repo: pathlib.Path) -> None:
3398	"""``--hashes`` appends content hash abbreviations to each symbol row."""
3399	result = runner.invoke(cli, ["code", "symbols", "--hashes"])
3400	assert result.exit_code == 0, result.output
3401	# Hash suffix is 8 hex chars followed by ".."
3402	assert ".." in result.output
3403
3404	def test_symbols_commit_ref(self, code_repo: pathlib.Path) -> None:
3405	"""``--commit HEAD`` and working-tree mode show the same symbols for a clean repo."""
3406	default = runner.invoke(cli, ["code", "symbols"])
3407	head = runner.invoke(cli, ["code", "symbols", "--commit", "HEAD"])
3408	assert default.exit_code == 0
3409	assert head.exit_code == 0
3410	# Headers differ ("working tree" vs "commit …") but symbol content is identical.
3411	assert "Invoice" in default.output
3412	assert "Invoice" in head.output
3413	assert "symbols across" in default.output
3414	assert "symbols across" in head.output
3415
3416	def test_symbols_count_and_json_mutually_exclusive(self, code_repo: pathlib.Path) -> None:
3417	"""``--count`` and ``--json`` cannot be combined."""
3418	result = runner.invoke(cli, ["code", "symbols", "--count", "--json"])
3419	assert result.exit_code != 0
3420
3421	def test_symbols_json_schema(self, code_repo: pathlib.Path) -> None:
3422	"""JSON output uses the structured envelope with source_ref and results."""
3423	result = runner.invoke(cli, ["code", "symbols", "--json"])
3424	assert result.exit_code == 0, result.output
3425	data = json.loads(result.output)
3426	assert "source_ref" in data
3427	assert "working_tree" in data
3428	assert "total_symbols" in data
3429	assert "results" in data
3430	assert "files" not in data
3431	assert isinstance(data["working_tree"], bool)
3432	assert isinstance(data["total_symbols"], int)
3433	for entry in data["results"]:
3434	for field in ("address", "kind", "name", "qualified_name",
3435	"lineno", "content_id", "body_hash", "signature_id"):
3436	assert field in entry, f"missing field '{field}' in JSON entry"
3437
3438	def test_symbols_json_working_tree_flag(self, code_repo: pathlib.Path) -> None:
3439	"""``--json`` without ``--commit`` reports working_tree=true."""
3440	result = runner.invoke(cli, ["code", "symbols", "--json"])
3441	assert result.exit_code == 0, result.output
3442	data = json.loads(result.output)
3443	assert data["working_tree"] is True
3444	assert data["source_ref"] == "working-tree"
3445
3446	def test_symbols_json_commit_flag(self, code_repo: pathlib.Path) -> None:
3447	"""``--json --commit HEAD`` reports working_tree=false and a short SHA."""
3448	result = runner.invoke(cli, ["code", "symbols", "--json", "--commit", "HEAD"])
3449	assert result.exit_code == 0, result.output
3450	data = json.loads(result.output)
3451	assert data["working_tree"] is False
3452	assert data["source_ref"] != "working-tree"
3453	# source_ref is a prefixed short commit id (e.g. "sha256:<12hex>")
3454	assert data["source_ref"].startswith("sha256:")
3455
3456	def test_symbols_working_tree_reflects_disk_changes(self, code_repo: pathlib.Path) -> None:
3457	"""Working-tree mode picks up edits made to files after the last commit."""
3458	# Find the billing.py path on disk.
3459	billing = code_repo / "billing.py"
3460	assert billing.exists()
3461	# Append a new function — not yet committed.
3462	billing.write_text(
3463	f"{billing.read_text()}\ndef newly_added_function():\n pass\n"
3464	)
3465	result = runner.invoke(cli, ["code", "symbols"])
3466	assert result.exit_code == 0, result.output
3467	assert "newly_added_function" in result.output
3468
3469	# Committed snapshot should NOT contain it.
3470	committed = runner.invoke(cli, ["code", "symbols", "--commit", "HEAD"])
3471	assert committed.exit_code == 0
3472	assert "newly_added_function" not in committed.output
3473
3474	def test_symbols_language_filter_case_insensitive(self, code_repo: pathlib.Path) -> None:
3475	"""``--language`` is case-insensitive: 'python' == 'Python' == 'PYTHON'."""
3476	for variant in ("python", "Python", "PYTHON"):
3477	result = runner.invoke(cli, ["code", "symbols", "--language", variant])
3478	assert result.exit_code == 0, f"failed for --language {variant!r}"
3479	assert "Invoice" in result.output
3480
3481	def test_symbols_file_filter_partial_path(self, code_repo: pathlib.Path) -> None:
3482	"""``--file billing.py`` matches a manifest entry stored as ``billing.py``."""
3483	result = runner.invoke(cli, ["code", "symbols", "--file", "billing.py"])
3484	assert result.exit_code == 0, result.output
3485	assert "Invoice" in result.output
3486
3487	def test_symbols_file_filter_ambiguous_exits_error(self, code_repo: pathlib.Path) -> None:
3488	"""An ambiguous ``--file`` suffix that matches multiple paths exits non-zero."""
3489	# Write a second file with the same basename in a sub-directory.
3490	sub = code_repo / "sub"
3491	sub.mkdir(exist_ok=True)
3492	(sub / "billing.py").write_text("def sub_func(): pass\n")
3493	# Stage and commit both so the manifest has two paths ending in billing.py.
3494	import subprocess
3495	subprocess.run(["muse", "code", "add", "."], cwd=code_repo, check=True)
3496	subprocess.run(
3497	["muse", "commit", "-m", "add sub/billing.py"],
3498	cwd=code_repo, check=True,
3499	)
3500	result = runner.invoke(cli, ["code", "symbols", "--file", "billing.py"])
3501	assert result.exit_code != 0
3502	assert "ambiguous" in (result.output + (result.stderr or "")).lower()
3503
3504	def test_symbols_invalid_ref_errors(self, code_repo: pathlib.Path) -> None:
3505	"""``--commit`` with a non-existent ref exits non-zero with a clear message."""
3506	result = runner.invoke(cli, ["code", "symbols", "--commit", "deadbeef"])
3507	assert result.exit_code != 0
3508	assert "not found" in result.stderr
3509
3510
3511	# ---------------------------------------------------------------------------
3512	# TestSymbolLog
3513	# ---------------------------------------------------------------------------
3514
3515
3516	class TestSymbolLog:
3517	"""Tests for ``muse code symbol-log``."""
3518
3519	def test_symbol_log_no_events_for_unknown_symbol(self, code_repo: pathlib.Path) -> None:
3520	"""An address not found in any commit produces 'no events found'."""
3521	result = runner.invoke(cli, ["code", "symbol-log", "billing.py::DoesNotExist"])
3522	assert result.exit_code == 0, result.output
3523	assert "no events found" in result.output
3524
3525	def test_symbol_log_invalid_address_no_double_colon(self, code_repo: pathlib.Path) -> None:
3526	"""An address without '::' exits non-zero with a descriptive error."""
3527	result = runner.invoke(cli, ["code", "symbol-log", "billing.py"])
3528	assert result.exit_code != 0
3529	assert "::" in (result.output + (result.stderr or ""))
3530
3531	def test_symbol_log_invalid_address_empty(self, code_repo: pathlib.Path) -> None:
3532	"""An empty string as address exits non-zero."""
3533	result = runner.invoke(cli, ["code", "symbol-log", "::"])
3534	# "::" is technically valid syntax; should at least not crash.
3535	assert result.exit_code == 0
3536
3537	def test_symbol_log_json_schema(self, code_repo: pathlib.Path) -> None:
3538	"""``--json`` emits the structured envelope with all top-level fields."""
3539	result = runner.invoke(
3540	cli, ["code", "symbol-log", "billing.py::Invoice", "--json"]
3541	)
3542	assert result.exit_code == 0, result.output
3543	data = json.loads(result.output)
3544	for field in ("address", "start_ref", "total_commits_scanned", "truncated", "events"):
3545	assert field in data, f"missing top-level field '{field}'"
3546	assert data["address"] == "billing.py::Invoice"
3547	assert data["start_ref"] == "HEAD"
3548	assert isinstance(data["total_commits_scanned"], int)
3549	assert isinstance(data["truncated"], bool)
3550	assert isinstance(data["events"], list)
3551
3552	def test_symbol_log_json_event_schema(self, code_repo: pathlib.Path) -> None:
3553	"""Each JSON event has the required fields."""
3554	result = runner.invoke(
3555	cli, ["code", "symbol-log", "billing.py::Invoice", "--json"]
3556	)
3557	assert result.exit_code == 0, result.output
3558	data = json.loads(result.output)
3559	for ev in data["events"]:
3560	for field in ("event", "commit_id", "message", "committed_at",
3561	"address", "detail", "new_address"):
3562	assert field in ev, f"missing event field '{field}'"
3563
3564	def test_symbol_log_truncation_warning(self, code_repo: pathlib.Path) -> None:
3565	"""When --max is hit, a truncation warning appears in human output."""
3566	result = runner.invoke(
3567	cli, ["code", "symbol-log", "billing.py::Invoice", "--max", "1"]
3568	)
3569	assert result.exit_code == 0, result.output
3570	assert "incomplete" in result.output or "limit" in result.output
3571
3572	def test_symbol_log_truncation_flag_in_json(self, code_repo: pathlib.Path) -> None:
3573	"""When --max is hit, truncated=true appears in JSON output."""
3574	result = runner.invoke(
3575	cli, ["code", "symbol-log", "billing.py::Invoice", "--max", "1", "--json"]
3576	)
3577	assert result.exit_code == 0, result.output
3578	data = json.loads(result.output)
3579	assert data["truncated"] is True
3580	assert data["total_commits_scanned"] == 1
3581
3582	def test_symbol_log_max_zero_errors(self, code_repo: pathlib.Path) -> None:
3583	"""--max 0 exits non-zero with a clear error."""
3584	result = runner.invoke(
3585	cli, ["code", "symbol-log", "billing.py::Invoice", "--max", "0"]
3586	)
3587	assert result.exit_code != 0
3588
3589	def test_symbol_log_invalid_from_ref(self, code_repo: pathlib.Path) -> None:
3590	"""``--from`` with a non-existent ref exits non-zero."""
3591	result = runner.invoke(
3592	cli, ["code", "symbol-log", "billing.py::Invoice", "--from", "deadbeef"]
3593	)
3594	assert result.exit_code != 0
3595	assert "not found" in result.stderr
3596
3597	def test_symbol_log_bfs_follows_merge_parent2(self, code_repo: pathlib.Path) -> None:
3598	"""BFS walk finds events on feature branches that were merged in via parent2.
3599
3600	Simulates a merge commit (parent1=mainline, parent2=feature branch HEAD).
3601	The feature branch commit has a structured_delta inserting a symbol.
3602	The linear (parent1-only) walk would miss this; BFS must find it.
3603	"""
3604	import datetime
3605
3606	root = code_repo
3607	repo_id = json.loads((repo_json_path(root)).read_text())["repo_id"]
3608	from muse.core.refs import (
3609	get_head_commit_id,
3610	read_current_branch,
3611	)
3612	from muse.core.commits import (
3613	CommitRecord,
3614	write_commit,
3615	)
3616	from muse.core.ids import hash_commit as compute_commit_id
3617	from muse.domain import InsertOp, PatchOp, StructuredDelta
3618	branch = read_current_branch(root)
3619	head_id = get_head_commit_id(root, branch)
3620	assert head_id is not None
3621
3622	feature_snap = "aa" * 32
3623	feature_at = datetime.datetime(2026, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)
3624	feature_id = compute_commit_id(
3625	parent_ids=[head_id],
3626	snapshot_id=feature_snap,
3627	message="feat: add merged_fn",
3628	committed_at_iso=feature_at.isoformat(),
3629	author="test",
3630	)
3631	write_commit(root, CommitRecord(
3632	commit_id=feature_id,
3633	branch="feat/branch",
3634	snapshot_id=feature_snap,
3635	message="feat: add merged_fn",
3636	committed_at=feature_at,
3637	parent_commit_id=head_id,
3638	author="test",
3639	structured_delta=StructuredDelta(ops=[PatchOp(
3640	op="patch",
3641	address="billing.py",
3642	child_ops=[InsertOp(
3643	op="insert",
3644	address="billing.py::merged_fn",
3645	content_summary="function merged_fn",
3646	)],
3647	)]),
3648	))
3649
3650	merge_snap = "bb" * 32
3651	merge_at = datetime.datetime(2026, 1, 1, 1, 0, tzinfo=datetime.timezone.utc)
3652	merge_id = compute_commit_id(
3653	parent_ids=[head_id, feature_id],
3654	snapshot_id=merge_snap,
3655	message="merge feat/branch",
3656	committed_at_iso=merge_at.isoformat(),
3657	author="test",
3658	)
3659	write_commit(root, CommitRecord(
3660	commit_id=merge_id,
3661	branch=branch,
3662	snapshot_id=merge_snap,
3663	message="merge feat/branch",
3664	committed_at=merge_at,
3665	parent_commit_id=head_id,
3666	parent2_commit_id=feature_id,
3667	author="test",
3668	))
3669
3670	branch_ref = ref_path(root, branch)
3671	branch_ref.write_text(merge_id)
3672
3673	result = runner.invoke(
3674	cli, ["code", "symbol-log", "billing.py::merged_fn"]
3675	)
3676	assert result.exit_code == 0, result.output
3677	# BFS must find the creation event on the feature branch.
3678	assert "merged_fn" in result.output
3679	assert "created" in result.output
3680
3681	def test_symbol_log_linear_walk_misses_parent2(self, code_repo: pathlib.Path) -> None:
3682	"""Regression guard: verify the BFS result differs from a parent1-only scan.
3683
3684	Directly calls _walk_commits_dag and checks it returns commits from
3685	both parent chains, not just parent1.
3686	"""
3687	import datetime
3688
3689	root = code_repo
3690	repo_id = json.loads((repo_json_path(root)).read_text())["repo_id"]
3691	from muse.core.refs import (
3692	get_head_commit_id,
3693	read_current_branch,
3694	)
3695	from muse.core.commits import (
3696	CommitRecord,
3697	write_commit,
3698	)
3699	from muse.core.ids import hash_commit as compute_commit_id
3700	from muse.plugins.code._query import walk_commits_bfs as _walk_commits_dag
3701	branch = read_current_branch(root)
3702	head_id = get_head_commit_id(root, branch)
3703	assert head_id is not None
3704
3705	feature_snap = "dd" * 32
3706	feature_at = datetime.datetime(2026, 2, 1, 0, 0, tzinfo=datetime.timezone.utc)
3707	feature_id = compute_commit_id(
3708	parent_ids=[],
3709	snapshot_id=feature_snap,
3710	message="feat on second parent",
3711	committed_at_iso=feature_at.isoformat(),
3712	author="test",
3713	)
3714	write_commit(root, CommitRecord(
3715	commit_id=feature_id,
3716	branch="feat/x",
3717	snapshot_id=feature_snap,
3718	message="feat on second parent",
3719	committed_at=feature_at,
3720	author="test",
3721	))
3722	merge_snap = "ee" * 32
3723	merge_at = datetime.datetime(2026, 2, 1, 1, 0, tzinfo=datetime.timezone.utc)
3724	merge_id = compute_commit_id(
3725	parent_ids=[head_id, feature_id],
3726	snapshot_id=merge_snap,
3727	message="merge",
3728	committed_at_iso=merge_at.isoformat(),
3729	author="test",
3730	)
3731	write_commit(root, CommitRecord(
3732	commit_id=merge_id,
3733	branch=branch,
3734	snapshot_id=merge_snap,
3735	message="merge",
3736	committed_at=merge_at,
3737	parent_commit_id=head_id,
3738	parent2_commit_id=feature_id,
3739	author="test",
3740	))
3741
3742	branch_ref = ref_path(root, branch)
3743	branch_ref.write_text(merge_id)
3744
3745	commits, _ = _walk_commits_dag(root, merge_id, max_commits=1000)
3746	commit_ids = {c.commit_id for c in commits}
3747	assert feature_id in commit_ids
3748
3749
3750	# ---------------------------------------------------------------------------
3751	# muse code coupling
3752	# ---------------------------------------------------------------------------
3753
3754
3755	@pytest.fixture
3756	def coupling_repo(repo: pathlib.Path) -> pathlib.Path:
3757	"""Repo with 3 commits where billing.py + models.py co-change twice."""
3758	work = repo
3759
3760	# Commit 1: seed — only billing.py
3761	(work / "billing.py").write_text("def compute(items):\n return sum(items)\n")
3762	runner.invoke(cli, ["code", "add", "billing.py"])
3763	r = runner.invoke(cli, ["commit", "-m", "seed billing"])
3764	assert r.exit_code == 0, r.output
3765
3766	# Commit 2: billing.py + models.py change together
3767	(work / "billing.py").write_text("def compute(items, tax=0.0):\n return sum(items) + tax\n")
3768	(work / "models.py").write_text("class Order:\n def total(self):\n return 0\n")
3769	runner.invoke(cli, ["code", "add", "."])
3770	r = runner.invoke(cli, ["commit", "-m", "co-change 1: billing + models"])
3771	assert r.exit_code == 0, r.output
3772
3773	# Commit 3: billing.py + models.py change together again
3774	(work / "billing.py").write_text("def compute(items, tax=0.0, discount=0.0):\n return sum(items) + tax - discount\n")
3775	(work / "models.py").write_text("class Order:\n def total(self):\n return 42\n def apply(self): pass\n")
3776	runner.invoke(cli, ["code", "add", "."])
3777	r = runner.invoke(cli, ["commit", "-m", "co-change 2: billing + models again"])
3778	assert r.exit_code == 0, r.output
3779
3780	return repo
3781
3782
3783	class TestCoupling:
3784	"""Tests for muse code coupling."""
3785
3786	# ── basic correctness ────────────────────────────────────────────────────
3787
3788	def test_coupling_exits_zero(self, coupling_repo: pathlib.Path) -> None:
3789	result = runner.invoke(cli, ["code", "coupling"])
3790	assert result.exit_code == 0, result.output
3791
3792	def test_coupling_finds_co_changed_pair(self, coupling_repo: pathlib.Path) -> None:
3793	"""billing.py and models.py co-changed twice — must appear in output."""
3794	result = runner.invoke(cli, ["code", "coupling", "--min", "1"])
3795	assert result.exit_code == 0, result.output
3796	assert "billing.py" in result.output
3797	assert "models.py" in result.output
3798
3799	def test_coupling_shows_header(self, coupling_repo: pathlib.Path) -> None:
3800	result = runner.invoke(cli, ["code", "coupling"])
3801	assert "co-change" in result.output.lower() or "coupling" in result.output.lower()
3802	assert "Commits analysed" in result.output
3803
3804	def test_coupling_min_filter_excludes_low_count(
3805	self, coupling_repo: pathlib.Path
3806	) -> None:
3807	"""--min 3 must exclude our pair that co-changed only twice."""
3808	result = runner.invoke(cli, ["code", "coupling", "--min", "3"])
3809	assert result.exit_code == 0, result.output
3810	assert "billing.py" not in result.output or "no file pairs" in result.output
3811
3812	def test_coupling_top_limits_output(self, coupling_repo: pathlib.Path) -> None:
3813	result = runner.invoke(cli, ["code", "coupling", "--top", "1", "--min", "1", "--json"])
3814	data = json.loads(result.output)
3815	assert len(data["pairs"]) <= 1
3816
3817	# ── --file filter ─────────────────────────────────────────────────────────
3818
3819	def test_coupling_file_filter_exits_zero(self, coupling_repo: pathlib.Path) -> None:
3820	result = runner.invoke(cli, ["code", "coupling", "--file", "billing.py", "--min", "1"])
3821	assert result.exit_code == 0, result.output
3822
3823	def test_coupling_file_filter_shows_partner(self, coupling_repo: pathlib.Path) -> None:
3824	"""--file billing.py must surface models.py as its partner."""
3825	result = runner.invoke(cli, ["code", "coupling", "--file", "billing.py", "--min", "1"])
3826	assert result.exit_code == 0, result.output
3827	assert "models.py" in result.output
3828
3829	def test_coupling_file_filter_header_names_file(
3830	self, coupling_repo: pathlib.Path
3831	) -> None:
3832	result = runner.invoke(cli, ["code", "coupling", "--file", "billing.py", "--min", "1"])
3833	assert "billing.py" in result.output
3834
3835	def test_coupling_file_filter_nonexistent_returns_cleanly(
3836	self, coupling_repo: pathlib.Path
3837	) -> None:
3838	result = runner.invoke(cli, ["code", "coupling", "--file", "nonexistent_xyz.py"])
3839	assert result.exit_code == 0, result.output
3840
3841	def test_coupling_file_filter_suffix_match(self, coupling_repo: pathlib.Path) -> None:
3842	"""Suffix billing.py should match the file even without the full path."""
3843	result = runner.invoke(cli, ["code", "coupling", "--file", "billing.py", "--min", "1"])
3844	assert result.exit_code == 0, result.output
3845	assert "models.py" in result.output
3846
3847	# ── JSON output ───────────────────────────────────────────────────────────
3848
3849	def test_coupling_json_schema(self, coupling_repo: pathlib.Path) -> None:
3850	result = runner.invoke(cli, ["code", "coupling", "--json"])
3851	assert result.exit_code == 0, result.output
3852	data = json.loads(result.output)
3853	assert "from_ref" in data
3854	assert "to_ref" in data
3855	assert "commits_analysed" in data
3856	assert "truncated" in data
3857	assert "filters" in data
3858	assert "pairs" in data
3859	assert isinstance(data["pairs"], list)
3860
3861	def test_coupling_json_pair_schema(self, coupling_repo: pathlib.Path) -> None:
3862	result = runner.invoke(cli, ["code", "coupling", "--min", "1", "--json"])
3863	data = json.loads(result.output)
3864	if data["pairs"]:
3865	pair = data["pairs"][0]
3866	assert "file_a" in pair or "file" in pair
3867	assert "co_changes" in pair
3868	assert isinstance(pair["co_changes"], int)
3869
3870	def test_coupling_json_file_filter_uses_partner_schema(
3871	self, coupling_repo: pathlib.Path
3872	) -> None:
3873	"""--file mode emits {file, partner, co_changes} not {file_a, file_b}."""
3874	result = runner.invoke(
3875	cli, ["code", "coupling", "--file", "billing.py", "--min", "1", "--json"]
3876	)
3877	data = json.loads(result.output)
3878	assert data["filters"]["file"] == "billing.py"
3879	if data["pairs"]:
3880	pair = data["pairs"][0]
3881	assert "file" in pair
3882	assert "partner" in pair
3883	assert "co_changes" in pair
3884	assert "file_a" not in pair # partner schema, not pair schema
3885
3886	def test_coupling_json_not_truncated_small_repo(
3887	self, coupling_repo: pathlib.Path
3888	) -> None:
3889	result = runner.invoke(cli, ["code", "coupling", "--json"])
3890	data = json.loads(result.output)
3891	assert data["truncated"] is False
3892
3893	def test_coupling_json_filters_reflect_args(
3894	self, coupling_repo: pathlib.Path
3895	) -> None:
3896	result = runner.invoke(
3897	cli, ["code", "coupling", "--top", "5", "--min", "2", "--json"]
3898	)
3899	data = json.loads(result.output)
3900	assert data["filters"]["top"] == 5
3901	assert data["filters"]["min_count"] == 2
3902
3903	# ── --max-commits ─────────────────────────────────────────────────────────
3904
3905	def test_coupling_max_commits_caps_scan(self, coupling_repo: pathlib.Path) -> None:
3906	r_full = runner.invoke(cli, ["code", "coupling", "--json"])
3907	r_cap = runner.invoke(cli, ["code", "coupling", "--max-commits", "1", "--json"])
3908	assert r_full.exit_code == 0 and r_cap.exit_code == 0
3909	d_cap = json.loads(r_cap.output)
3910	assert d_cap["commits_analysed"] <= 1
3911
3912	def test_coupling_max_commits_truncated_flag(
3913	self, coupling_repo: pathlib.Path
3914	) -> None:
3915	result = runner.invoke(cli, ["code", "coupling", "--max-commits", "1", "--json"])
3916	data = json.loads(result.output)
3917	# With 3 commits and cap=1, truncated must be True.
3918	assert data["truncated"] is True
3919
3920	def test_coupling_max_commits_one_shows_warning(
3921	self, coupling_repo: pathlib.Path
3922	) -> None:
3923	result = runner.invoke(cli, ["code", "coupling", "--max-commits", "1"])
3924	assert result.exit_code == 0, result.output
3925	assert "⚠️" in result.output or "capped" in result.output
3926
3927	# ── validation ────────────────────────────────────────────────────────────
3928
3929	def test_coupling_top_zero_exits_error(self, coupling_repo: pathlib.Path) -> None:
3930	result = runner.invoke(cli, ["code", "coupling", "--top", "0"])
3931	assert result.exit_code != 0
3932
3933	def test_coupling_min_zero_exits_error(self, coupling_repo: pathlib.Path) -> None:
3934	result = runner.invoke(cli, ["code", "coupling", "--min", "0"])
3935	assert result.exit_code != 0
3936
3937	def test_coupling_max_commits_zero_exits_error(
3938	self, coupling_repo: pathlib.Path
3939	) -> None:
3940	result = runner.invoke(cli, ["code", "coupling", "--max-commits", "0"])
3941	assert result.exit_code != 0
3942
3943	def test_coupling_invalid_from_ref_exits_error(
3944	self, coupling_repo: pathlib.Path
3945	) -> None:
3946	result = runner.invoke(
3947	cli, ["code", "coupling", "--from", "nonexistent-ref-xyz"]
3948	)
3949	assert result.exit_code != 0
3950
3951	def test_coupling_bfs_visits_merge_parents(self, repo: pathlib.Path) -> None:
3952	"""Coupling must count co-changes on feature-branch commits (parent2)."""
3953	import datetime
3954
3955	# Genesis commit
3956	(repo / "billing.py").write_text("def compute(x):\n return x\n")
3957	r = runner.invoke(cli, ["commit", "-m", "seed"])
3958	assert r.exit_code == 0, r.output
3959
3960	repo_json = json.loads((repo_json_path(repo)).read_text())
3961	repo_id = repo_json["repo_id"]
3962	from muse.core.refs import read_current_branch
3963	from muse.core.commits import resolve_commit_ref
3964	branch = read_current_branch(repo)
3965	head = resolve_commit_ref(repo, branch, None)
3966	assert head is not None
3967
3968	now = datetime.datetime(2026, 3, 1, 0, 0, tzinfo=datetime.timezone.utc)
3969	feature_at = now
3970	merge_at = now + datetime.timedelta(hours=1)
3971
3972	# Feature commit touching billing.py + models.py together.
3973	from muse.domain import PatchOp, ReplaceOp, InsertOp, StructuredDelta
3974	from muse.core.ids import hash_commit as compute_commit_id
3975	feature_delta = StructuredDelta(
3976	domain="code",
3977	ops=[
3978	PatchOp(
3979	op="patch", address="billing.py",
3980	child_ops=[ReplaceOp(
3981	op="replace", address="billing.py::compute",
3982	old_content_id="a" * 64, new_content_id="b" * 64,
3983	old_summary="function compute",
3984	new_summary="function compute (modified)", position=None,
3985	)],
3986	child_domain="code", child_summary="compute modified",
3987	),
3988	PatchOp(
3989	op="patch", address="models.py",
3990	child_ops=[InsertOp(
3991	op="insert", address="models.py::Order",
3992	content_id="c" * 64, content_summary="class Order", position=None,
3993	)],
3994	child_domain="code", child_summary="Order added",
3995	),
3996	],
3997	summary="co-change",
3998	)
3999	feature_id = compute_commit_id(
4000	[head.commit_id], head.snapshot_id,
4001	"co-change on feature branch", feature_at.isoformat(),
4002	author="test",
4003	)
4004	merge_id = compute_commit_id(
4005	[head.commit_id, feature_id], head.snapshot_id,
4006	"Merge feature", merge_at.isoformat(),
4007	author="test",
4008	)
4009	feature_body: CommitDict = {
4010	"commit_id": feature_id,
4011	"repo_id": repo_id,
4012	"branch": "feat/test",
4013	"snapshot_id": head.snapshot_id,
4014	"message": "co-change on feature branch",
4015	"committed_at": feature_at.isoformat(),
4016	"parent_commit_id": head.commit_id,
4017	"parent2_commit_id": None,
4018	"author": "test",
4019	"metadata": {},
4020	"structured_delta": feature_delta,
4021	}
4022	merge_body: CommitDict = {
4023	"commit_id": merge_id,
4024	"repo_id": repo_id,
4025	"branch": branch,
4026	"snapshot_id": head.snapshot_id,
4027	"message": "Merge feature",
4028	"committed_at": merge_at.isoformat(),
4029	"parent_commit_id": head.commit_id,
4030	"parent2_commit_id": feature_id,
4031	"author": "test",
4032	"metadata": {},
4033	"structured_delta": None,
4034	}
4035	from muse.core.commits import (
4036	CommitRecord,
4037	write_commit,
4038	)
4039	write_commit(repo, CommitRecord.from_dict(feature_body))
4040	write_commit(repo, CommitRecord.from_dict(merge_body))
4041	(ref_path(repo, branch)).write_text(merge_id)
4042
4043	result = runner.invoke(cli, ["code", "coupling", "--min", "1", "--json"])
4044	assert result.exit_code == 0, result.output
4045	data = json.loads(result.output)
4046	pairs_found = {
4047	(p.get("file_a", ""), p.get("file_b", "")) for p in data["pairs"]
4048	}
4049	billing_models = any(
4050	("billing.py" in a and "models.py" in b) or ("models.py" in a and "billing.py" in b)
4051	for a, b in pairs_found
4052	)
4053	assert billing_models, "BFS must find the feature-branch co-change commit"
4054
4055
4056	# ---------------------------------------------------------------------------
4057	# muse code stable
4058	# ---------------------------------------------------------------------------
4059
4060
4061	class TestStable:
4062	"""Tests for muse code stable."""
4063
4064	# ── basic correctness ────────────────────────────────────────────────────
4065
4066	def test_stable_exits_zero(self, code_repo: pathlib.Path) -> None:
4067	result = runner.invoke(cli, ["code", "stable"])
4068	assert result.exit_code == 0, result.output
4069
4070	def test_stable_shows_header(self, code_repo: pathlib.Path) -> None:
4071	result = runner.invoke(cli, ["code", "stable"])
4072	assert result.exit_code == 0, result.output
4073	assert "Symbol stability" in result.output
4074	assert "Commits analysed" in result.output
4075	assert "bedrock" in result.output
4076
4077	def test_stable_surfaces_never_touched_symbol(self, code_repo: pathlib.Path) -> None:
4078	"""Invoice.apply_discount was defined in the genesis commit and never modified."""
4079	result = runner.invoke(cli, ["code", "stable", "--top", "10"])
4080	assert result.exit_code == 0, result.output
4081	# apply_discount was never touched in any structured_delta → maximally stable.
4082	assert "apply_discount" in result.output
4083
4084	def test_stable_since_start_of_range_marker(self, code_repo: pathlib.Path) -> None:
4085	result = runner.invoke(cli, ["code", "stable", "--top", "10"])
4086	assert result.exit_code == 0, result.output
4087	assert "since start of range" in result.output
4088
4089	def test_stable_excludes_docs_by_default(self, code_repo: pathlib.Path) -> None:
4090	"""Markdown / TOML / YAML symbols must be absent from default output."""
4091	result = runner.invoke(cli, ["code", "stable", "--top", "50"])
4092	assert result.exit_code == 0, result.output
4093	assert ".md::" not in result.output
4094	assert ".toml::" not in result.output
4095
4096	def test_stable_excludes_imports_by_default(self, code_repo: pathlib.Path) -> None:
4097	result = runner.invoke(cli, ["code", "stable", "--top", "50"])
4098	assert result.exit_code == 0, result.output
4099	assert "::import::" not in result.output
4100
4101	def test_stable_include_imports_flag(self, code_repo: pathlib.Path) -> None:
4102	result = runner.invoke(cli, ["code", "stable", "--top", "50", "--include-imports"])
4103	assert result.exit_code == 0, result.output
4104
4105	# ── JSON output ───────────────────────────────────────────────────────────
4106
4107	def test_stable_json_schema(self, code_repo: pathlib.Path) -> None:
4108	result = runner.invoke(cli, ["code", "stable", "--top", "5", "--json"])
4109	assert result.exit_code == 0, result.output
4110	data = json.loads(result.output)
4111	assert "from_ref" in data
4112	assert "to_ref" in data
4113	assert "commits_analysed" in data
4114	assert "truncated" in data
4115	assert "filters" in data
4116	assert "stable" in data
4117	assert isinstance(data["stable"], list)
4118
4119	def test_stable_json_entry_schema(self, code_repo: pathlib.Path) -> None:
4120	result = runner.invoke(cli, ["code", "stable", "--top", "5", "--json"])
4121	data = json.loads(result.output)
4122	assert len(data["stable"]) > 0
4123	entry = data["stable"][0]
4124	assert "address" in entry
4125	assert "unchanged_for" in entry
4126	assert "since_start_of_range" in entry
4127	assert isinstance(entry["unchanged_for"], int)
4128	assert isinstance(entry["since_start_of_range"], bool)
4129
4130	def test_stable_json_filters_reflect_args(self, code_repo: pathlib.Path) -> None:
4131	result = runner.invoke(
4132	cli, ["code", "stable", "--top", "3", "--kind", "function", "--json"]
4133	)
4134	data = json.loads(result.output)
4135	assert data["filters"]["top"] == 3
4136	assert data["filters"]["kind"] == "function"
4137	assert data["filters"]["include_imports"] is False
4138	assert data["filters"]["include_docs"] is False
4139
4140	def test_stable_json_not_truncated_small_repo(self, code_repo: pathlib.Path) -> None:
4141	result = runner.invoke(cli, ["code", "stable", "--json"])
4142	data = json.loads(result.output)
4143	assert data["truncated"] is False
4144
4145	# ── --language filter ─────────────────────────────────────────────────────
4146
4147	def test_stable_language_filter_case_insensitive(self, code_repo: pathlib.Path) -> None:
4148	"""--language python and --language Python must behave identically."""
4149	r_lower = runner.invoke(cli, ["code", "stable", "--language", "python", "--json"])
4150	r_upper = runner.invoke(cli, ["code", "stable", "--language", "Python", "--json"])
4151	assert r_lower.exit_code == 0 and r_upper.exit_code == 0
4152	d_lower = json.loads(r_lower.output)
4153	d_upper = json.loads(r_upper.output)
4154	addrs_lower = {e["address"] for e in d_lower["stable"]}
4155	addrs_upper = {e["address"] for e in d_upper["stable"]}
4156	assert addrs_lower == addrs_upper
4157
4158	def test_stable_language_filter_restricts_results(self, code_repo: pathlib.Path) -> None:
4159	r_py = runner.invoke(cli, ["code", "stable", "--language", "python", "--json"])
4160	r_all = runner.invoke(cli, ["code", "stable", "--json"])
4161	d_py = json.loads(r_py.output)
4162	d_all = json.loads(r_all.output)
4163	# Python-filtered results must be a subset of or equal to unfiltered results.
4164	py_addrs = {e["address"] for e in d_py["stable"]}
4165	all_addrs = {e["address"] for e in d_all["stable"]}
4166	assert py_addrs <= all_addrs
4167
4168	# ── --since REF ───────────────────────────────────────────────────────────
4169
4170	def test_stable_since_reduces_commits_analysed(self, code_repo: pathlib.Path) -> None:
4171	"""--since HEAD restricts the window to 0 commits (stop immediately)."""
4172	# Get the HEAD commit id to use as --since boundary
4173	import json as _json
4174	root = code_repo
4175	repo_id = _json.loads((repo_json_path(root)).read_text())["repo_id"]
4176	from muse.core.refs import read_current_branch
4177	from muse.core.commits import resolve_commit_ref
4178	branch = read_current_branch(root)
4179	head = resolve_commit_ref(root, branch, None)
4180	assert head is not None
4181
4182	r_all = runner.invoke(cli, ["code", "stable", "--json"])
4183	r_since = runner.invoke(cli, ["code", "stable", "--since", head.commit_id, "--json"])
4184	assert r_all.exit_code == 0 and r_since.exit_code == 0
4185	d_all = json.loads(r_all.output)
4186	d_since = json.loads(r_since.output)
4187	# Window stops at HEAD itself → at most 1 commit analysed.
4188	assert d_since["commits_analysed"] <= d_all["commits_analysed"]
4189
4190	def test_stable_since_invalid_ref_exits_nonzero(self, code_repo: pathlib.Path) -> None:
4191	result = runner.invoke(cli, ["code", "stable", "--since", "nonexistent-ref-xyz"])
4192	assert result.exit_code != 0
4193
4194	# ── --max-commits ─────────────────────────────────────────────────────────
4195
4196	def test_stable_max_commits_caps_scan(self, code_repo: pathlib.Path) -> None:
4197	r_full = runner.invoke(cli, ["code", "stable", "--json"])
4198	r_cap = runner.invoke(cli, ["code", "stable", "--max-commits", "1", "--json"])
4199	assert r_full.exit_code == 0 and r_cap.exit_code == 0
4200	d_cap = json.loads(r_cap.output)
4201	assert d_cap["commits_analysed"] <= 1
4202
4203	def test_stable_max_commits_one_shows_truncated_warning(
4204	self, code_repo: pathlib.Path
4205	) -> None:
4206	result = runner.invoke(cli, ["code", "stable", "--max-commits", "1"])
4207	assert result.exit_code == 0, result.output
4208	# With 2 commits and cap=1, truncated warning should appear.
4209	assert "capped" in result.output or "⚠️" in result.output
4210
4211	def test_stable_max_commits_zero_exits_error(self, code_repo: pathlib.Path) -> None:
4212	result = runner.invoke(cli, ["code", "stable", "--max-commits", "0"])
4213	assert result.exit_code != 0
4214
4215	# ── --top validation ──────────────────────────────────────────────────────
4216
4217	def test_stable_top_zero_exits_error(self, code_repo: pathlib.Path) -> None:
4218	result = runner.invoke(cli, ["code", "stable", "--top", "0"])
4219	assert result.exit_code != 0
4220
4221	def test_stable_top_limits_output_count(self, code_repo: pathlib.Path) -> None:
4222	result = runner.invoke(cli, ["code", "stable", "--top", "2", "--json"])
4223	data = json.loads(result.output)
4224	assert len(data["stable"]) <= 2
4225
4226	# ── BFS follows merge parents ─────────────────────────────────────────────
4227
4228	def test_stable_bfs_follows_merge_parent2(self, repo: pathlib.Path) -> None:
4229	"""Symbols touched only on a merged feature branch must be detected as unstable."""
4230	import datetime
4231
4232	# Create a symbol in commit 1 (main).
4233	(repo / "core.py").write_text("def bedrock():\n return 42\n")
4234	r = runner.invoke(cli, ["commit", "-m", "Add bedrock"])
4235	assert r.exit_code == 0, r.output
4236
4237	repo_json = json.loads((repo_json_path(repo)).read_text())
4238	repo_id = repo_json["repo_id"]
4239	from muse.core.refs import read_current_branch
4240	from muse.core.commits import resolve_commit_ref
4241	branch = read_current_branch(repo)
4242	head_commit = resolve_commit_ref(repo, branch, None)
4243	assert head_commit is not None
4244	head_id = head_commit.commit_id
4245
4246	feature_at = datetime.datetime(2026, 4, 1, 0, 0, tzinfo=datetime.timezone.utc)
4247	merge_at = datetime.datetime(2026, 4, 1, 1, 0, tzinfo=datetime.timezone.utc)
4248
4249	# Feature-branch commit that touched "bedrock" via a structured_delta.
4250	from muse.domain import PatchOp, ReplaceOp, StructuredDelta
4251	from muse.core.ids import hash_commit as compute_commit_id
4252	bedrock_delta = StructuredDelta(
4253	domain="code",
4254	ops=[PatchOp(
4255	op="patch", address="core.py",
4256	child_ops=[ReplaceOp(
4257	op="replace", address="core.py::bedrock",
4258	old_content_id="a" * 64, new_content_id="b" * 64,
4259	old_summary="function bedrock",
4260	new_summary="function bedrock (modified)", position=None,
4261	)],
4262	child_domain="code", child_summary="bedrock modified",
4263	)],
4264	summary="bedrock modified",
4265	)
4266	feature_id = compute_commit_id(
4267	[head_id], head_commit.snapshot_id,
4268	"Feature: touch bedrock", feature_at.isoformat(),
4269	author="test",
4270	)
4271	merge_id = compute_commit_id(
4272	[head_id, feature_id], head_commit.snapshot_id,
4273	"Merge feat/touch-bedrock", merge_at.isoformat(),
4274	author="test",
4275	)
4276	feature_body: CommitDict = {
4277	"commit_id": feature_id,
4278	"repo_id": repo_id,
4279	"branch": "feat/touch-bedrock",
4280	"snapshot_id": head_commit.snapshot_id,
4281	"message": "Feature: touch bedrock",
4282	"committed_at": feature_at.isoformat(),
4283	"parent_commit_id": head_id,
4284	"parent2_commit_id": None,
4285	"author": "test",
4286	"metadata": {},
4287	"structured_delta": bedrock_delta,
4288	}
4289	# Merge commit whose parent2 is the feature commit.
4290	merge_body: CommitDict = {
4291	"commit_id": merge_id,
4292	"repo_id": repo_id,
4293	"branch": branch,
4294	"snapshot_id": head_commit.snapshot_id,
4295	"message": "Merge feat/touch-bedrock",
4296	"committed_at": merge_at.isoformat(),
4297	"parent_commit_id": head_id,
4298	"parent2_commit_id": feature_id,
4299	"author": "test",
4300	"metadata": {},
4301	"structured_delta": None,
4302	}
4303	from muse.core.commits import (
4304	CommitRecord,
4305	write_commit,
4306	)
4307	write_commit(repo, CommitRecord.from_dict(feature_body))
4308	write_commit(repo, CommitRecord.from_dict(merge_body))
4309	(ref_path(repo, branch)).write_text(merge_id)
4310
4311	result = runner.invoke(cli, ["code", "stable", "--top", "10", "--json"])
4312	assert result.exit_code == 0, result.output
4313	data = json.loads(result.output)
4314	# bedrock was touched in the feature-branch commit; BFS must find it.
4315	# It should have unchanged_for < total_commits (not maximally stable).
4316	bedrock_entries = [e for e in data["stable"] if "bedrock" in e["address"]]
4317	if bedrock_entries:
4318	assert not bedrock_entries[0]["since_start_of_range"]
4319
4320
4321	# ---------------------------------------------------------------------------
4322	# muse code compare
4323	# ---------------------------------------------------------------------------
4324
4325
4326	@pytest.fixture
4327	def compare_repo(repo: pathlib.Path) -> tuple[pathlib.Path, str, str]:
4328	"""Repo with two commits; returns (path, commit_id_a, commit_id_b).
4329
4330	Commit A — billing.py with Invoice.compute_total + process_order.
4331	Commit B — compute_total renamed to compute_invoice_total; generate_pdf
4332	and send_email added. Multi-line message to test truncation.
4333	"""
4334	(repo / "billing.py").write_text(textwrap.dedent("""\
4335	class Invoice:
4336	def compute_total(self, items):
4337	return sum(items)
4338
4339	def apply_discount(self, total, pct):
4340	return total * (1 - pct)
4341
4342	def process_order(invoice, items):
4343	return invoice.compute_total(items)
4344	"""))
4345	runner.invoke(cli, ["code", "add", "billing.py"])
4346	r = runner.invoke(cli, ["commit", "-m", "Add billing module"])
4347	assert r.exit_code == 0, r.output
4348	from muse.core.refs import read_current_branch
4349	branch = read_current_branch(repo)
4350	commit_a = get_head_commit_id(repo, branch)
4351
4352	(repo / "billing.py").write_text(textwrap.dedent("""\
4353	class Invoice:
4354	def compute_invoice_total(self, items):
4355	return sum(items)
4356
4357	def apply_discount(self, total, pct):
4358	return total * (1 - pct)
4359
4360	def generate_pdf(self):
4361	return b"pdf"
4362
4363	def process_order(invoice, items):
4364	return invoice.compute_invoice_total(items)
4365
4366	def send_email(address):
4367	pass
4368	"""))
4369	runner.invoke(cli, ["code", "add", "billing.py"])
4370	# Multi-line message to test first-line truncation.
4371	r = runner.invoke(cli, [
4372	"commit", "-m",
4373	"Rename compute_total, add generate_pdf + send_email\n\nThis is the extended body.",
4374	])
4375	assert r.exit_code == 0, r.output
4376	commit_b = get_head_commit_id(repo, branch)
4377
4378	assert commit_a is not None
4379	assert commit_b is not None
4380	return repo, commit_a, commit_b
4381
4382
4383	class TestCompare:
4384	"""Tests for muse code compare."""
4385
4386	# ── basic correctness ────────────────────────────────────────────────────
4387
4388	def test_compare_exits_zero(
4389	self, compare_repo: tuple[pathlib.Path, str, str]
4390	) -> None:
4391	_, ref_a, ref_b = compare_repo
4392	result = runner.invoke(cli, ["code", "compare", ref_a, ref_b])
4393	assert result.exit_code == 0, result.output
4394
4395	def test_compare_shows_header(
4396	self, compare_repo: tuple[pathlib.Path, str, str]
4397	) -> None:
4398	_, ref_a, ref_b = compare_repo
4399	result = runner.invoke(cli, ["code", "compare", ref_a, ref_b])
4400	assert result.exit_code == 0, result.output
4401	assert "Semantic comparison" in result.output
4402	assert "From:" in result.output
4403	assert "To:" in result.output
4404
4405	def test_compare_commit_message_first_line_only(
4406	self, compare_repo: tuple[pathlib.Path, str, str]
4407	) -> None:
4408	"""Multi-line commit messages must be truncated to their first line."""
4409	_, ref_a, ref_b = compare_repo
4410	result = runner.invoke(cli, ["code", "compare", ref_a, ref_b])
4411	assert result.exit_code == 0, result.output
4412	# The body of the second commit must not appear in the header.
4413	assert "This is the extended body" not in result.output
4414
4415	def test_compare_same_ref_no_changes(
4416	self, compare_repo: tuple[pathlib.Path, str, str]
4417	) -> None:
4418	_, ref_a, _ = compare_repo
4419	result = runner.invoke(cli, ["code", "compare", ref_a, ref_a])
4420	assert result.exit_code == 0, result.output
4421	assert "no semantic changes" in result.output
4422
4423	def test_compare_detects_added_symbols(
4424	self, compare_repo: tuple[pathlib.Path, str, str]
4425	) -> None:
4426	_, ref_a, ref_b = compare_repo
4427	result = runner.invoke(cli, ["code", "compare", ref_a, ref_b])
4428	assert result.exit_code =

File truncated at 200 KB — view full file ↗

test_code_commands.py file-level

`test_code_commands.py` file-level