gabriel/muse — blame/sha256:6/test_cmd_rerere_hardening.py

1 files

1 commits

0 hotspots

0 🧊 dead

0 💥 blast risk

sha256:0 test: validate staging push and MP workflow for aaronrene · · Jul 7, 2026

1	"""Comprehensive tests for ``muse rerere`` CLI and ``muse/core/rerere.py`` hardening.
2
3	Audit findings addressed
4	------------------------
5	Security
6	- _entry_dir now calls _validate_fingerprint — crafted fingerprints like
7	"../evil" raise ValueError and are rejected by all callers.
8	- load_record enforces _MAX_META_BYTES and _MAX_RESOLUTION_BYTES caps.
9	- apply_cached enforces _MAX_RESOLUTION_BYTES before reading.
10	- list_records and clear_all skip symlinks (is_symlink() guard).
11	- auto_apply validates paths with .relative_to(root) before dest write.
12	- CLI run() validates paths before apply_cached.
13
14	Dead code removed
15	- domain = read_domain(root) in run() (dead variable — domain unused).
16	- color: bool parameter from _fmt_record (accepted but never used).
17
18	New capabilities
19	- _RerereApplyJson, _RerereRecordJson, _RerereStatusJson, _RerereForgetJson,
20	_RerereScalarJson TypedDicts for all subcommand JSON outputs.
21	- --fingerprint HEX on ``forget`` subcommand — forget by fingerprint
22	without requiring an active merge state.
23	- --age DAYS on ``gc`` subcommand — configurable GC threshold.
24	- gc_stale(age_days=...) parameter in core.
25	- _validate_fingerprint() public function with 64-char hex validation.
26	- _check_format() helper for format validation.
27
28	Coverage tiers
29	--------------
30	Unit _validate_fingerprint, _entry_dir, load_record (size caps),
31	apply_cached (size cap), list_records (symlink guard),
32	clear_all (symlink guard), auto_apply (path traversal guard),
33	gc_stale (age_days param), _fmt_record (no color param),
34	_check_format
35
36	Integration record_preimage, save_resolution, forget_record, clear_all,
37	list_records, gc_stale, has_resolution
38
39	Security path traversal in _entry_dir and auto_apply, meta.json
40	size cap, resolution file size cap, symlink guards,
41	ANSI sanitization in text output, JSON output on stdout
42
43	E2E CLI flags for all subcommands (--json, --dry-run, --yes,
44	--fingerprint, --age), JSON schemas, exit codes, help text
45
46	Stress 10k list_records scan, concurrent isolated record/load,
47	concurrent gc_stale on isolated repos
48	"""
49	from __future__ import annotations
50
51	import datetime
52	import json
53	import pathlib
54	import threading
55	import uuid
56
57	import pytest
58
59	from muse.core.errors import ExitCode
60	from muse.core.rerere import (
61	RerereRecord,
62	_MAX_META_BYTES,
63	_MAX_RESOLUTION_BYTES,
64	_entry_dir,
65	_validate_fingerprint,
66	apply_cached,
67	clear_all,
68	conflict_fingerprint,
69	forget_record,
70	gc_stale,
71	list_records,
72	load_record,
73	record_preimage,
74	rr_cache_dir,
75	save_resolution,
76	)
77	from tests.cli_test_helper import CliRunner, InvokeResult
78	from muse.core._types import Manifest, MsgpackDict
79
80	runner = CliRunner()
81	cli = None
82
83	_SHA_A = "a" * 64
84	_SHA_B = "b" * 64
85	_SHA_C = "c" * 64
86	_VALID_FP = conflict_fingerprint(_SHA_A, _SHA_B)
87
88
89	# ---------------------------------------------------------------------------
90	# Repo helpers
91	# ---------------------------------------------------------------------------
92
93
94	def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
95	muse = tmp_path / ".muse"
96	for sub in ("commits", "snapshots", "refs/heads", "objects", "rr-cache"):
97	(muse / sub).mkdir(parents=True, exist_ok=True)
98	(muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
99	(muse / "repo.json").write_text(
100	json.dumps({"repo_id": str(uuid.uuid4()), "domain": "code"}),
101	encoding="utf-8",
102	)
103	return tmp_path
104
105
106	def _write_preimage(root: pathlib.Path, fp: str, path: str = "src/a.py") -> None:
107	"""Write a valid meta.json entry for fp."""
108	entry_dir = rr_cache_dir(root) / fp
109	entry_dir.mkdir(parents=True, exist_ok=True)
110	meta = {
111	"fingerprint": fp,
112	"path": path,
113	"ours_id": _SHA_A,
114	"theirs_id": _SHA_B,
115	"domain": "code",
116	"recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
117	}
118	(entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
119
120
121	def _write_resolution(root: pathlib.Path, fp: str, res_id: str = _SHA_C) -> None:
122	"""Write a resolution file for fp."""
123	entry_dir = rr_cache_dir(root) / fp
124	entry_dir.mkdir(parents=True, exist_ok=True)
125	(entry_dir / "resolution").write_text(res_id, encoding="utf-8")
126
127
128	def _invoke(root: pathlib.Path, *args: str) -> InvokeResult:
129	return runner.invoke(
130	cli,
131	["rerere", *args],
132	env={"MUSE_REPO_ROOT": str(root)},
133	)
134
135
136	def _parse_json(result: InvokeResult) -> MsgpackDict:
137	start = result.output.index("{")
138	blob = result.output[start:]
139	depth = 0
140	end = 0
141	for i, ch in enumerate(blob):
142	if ch == "{":
143	depth += 1
144	elif ch == "}":
145	depth -= 1
146	if depth == 0:
147	end = i + 1
148	break
149	parsed = json.loads(blob[:end])
150	assert isinstance(parsed, dict)
151	return parsed
152
153
154	# ---------------------------------------------------------------------------
155	# Unit — _validate_fingerprint
156	# ---------------------------------------------------------------------------
157
158
159	class TestValidateFingerprint:
160	def test_valid_fingerprint_passes(self) -> None:
161	_validate_fingerprint(_VALID_FP) # no exception
162
163	def test_path_traversal_rejected(self) -> None:
164	with pytest.raises(ValueError):
165	_validate_fingerprint("../evil")
166
167	def test_too_short_rejected(self) -> None:
168	with pytest.raises(ValueError):
169	_validate_fingerprint("abc123")
170
171	def test_too_long_rejected(self) -> None:
172	with pytest.raises(ValueError):
173	_validate_fingerprint("a" * 65)
174
175	def test_uppercase_rejected(self) -> None:
176	with pytest.raises(ValueError):
177	_validate_fingerprint("A" * 64)
178
179	def test_slash_in_fingerprint_rejected(self) -> None:
180	with pytest.raises(ValueError):
181	_validate_fingerprint("a" * 32 + "/" + "a" * 31)
182
183	def test_null_byte_rejected(self) -> None:
184	with pytest.raises(ValueError):
185	_validate_fingerprint("a" * 63 + "\x00")
186
187	def test_empty_string_rejected(self) -> None:
188	with pytest.raises(ValueError):
189	_validate_fingerprint("")
190
191	def test_exactly_64_hex_passes(self) -> None:
192	fp = "0123456789abcdef" * 4
193	_validate_fingerprint(fp) # no exception
194
195
196	# ---------------------------------------------------------------------------
197	# Unit — _entry_dir path traversal guard
198	# ---------------------------------------------------------------------------
199
200
201	class TestEntryDir:
202	def test_valid_fp_returns_path(self, tmp_path: pathlib.Path) -> None:
203	root = _make_repo(tmp_path)
204	entry = _entry_dir(root, _VALID_FP)
205	assert entry.parent == rr_cache_dir(root)
206
207	def test_traversal_fp_raises(self, tmp_path: pathlib.Path) -> None:
208	root = _make_repo(tmp_path)
209	with pytest.raises(ValueError):
210	_entry_dir(root, "../evil")
211
212	def test_entry_is_inside_rr_cache(self, tmp_path: pathlib.Path) -> None:
213	root = _make_repo(tmp_path)
214	entry = _entry_dir(root, _VALID_FP)
215	assert str(entry).startswith(str(rr_cache_dir(root)))
216
217
218	# ---------------------------------------------------------------------------
219	# Unit — load_record size caps
220	# ---------------------------------------------------------------------------
221
222
223	class TestLoadRecordSizeCaps:
224	def test_oversized_meta_returns_none(self, tmp_path: pathlib.Path) -> None:
225	root = _make_repo(tmp_path)
226	_write_preimage(root, _VALID_FP)
227	meta_path = rr_cache_dir(root) / _VALID_FP / "meta.json"
228	# Overwrite with a file that exceeds the cap
229	import muse.core.rerere as rmod
230	original = rmod._MAX_META_BYTES
231	try:
232	rmod._MAX_META_BYTES = 5 # tiny cap
233	result = load_record(root, _VALID_FP)
234	finally:
235	rmod._MAX_META_BYTES = original
236	assert result is None
237
238	def test_oversized_resolution_ignored(self, tmp_path: pathlib.Path) -> None:
239	root = _make_repo(tmp_path)
240	_write_preimage(root, _VALID_FP)
241	res_path = rr_cache_dir(root) / _VALID_FP / "resolution"
242	import muse.core.rerere as rmod
243	original = rmod._MAX_RESOLUTION_BYTES
244	try:
245	rmod._MAX_RESOLUTION_BYTES = 5 # tiny cap — can't fit a 64-char hex
246	res_path.write_text(_SHA_C, encoding="utf-8")
247	record = load_record(root, _VALID_FP)
248	finally:
249	rmod._MAX_RESOLUTION_BYTES = original
250	# Record still loaded (preimage exists) but resolution_id is None
251	assert record is not None
252	assert record.resolution_id is None
253
254	def test_invalid_fingerprint_returns_none(self, tmp_path: pathlib.Path) -> None:
255	root = _make_repo(tmp_path)
256	result = load_record(root, "not-a-fingerprint")
257	assert result is None
258
259	def test_valid_entry_loads_correctly(self, tmp_path: pathlib.Path) -> None:
260	root = _make_repo(tmp_path)
261	_write_preimage(root, _VALID_FP, path="src/x.py")
262	_write_resolution(root, _VALID_FP, res_id=_SHA_C)
263	record = load_record(root, _VALID_FP)
264	assert record is not None
265	assert record.fingerprint == _VALID_FP
266	assert record.path == "src/x.py"
267	assert record.resolution_id == _SHA_C
268
269
270	# ---------------------------------------------------------------------------
271	# Unit — apply_cached size cap
272	# ---------------------------------------------------------------------------
273
274
275	class TestApplyCachedSizeCap:
276	def test_oversized_resolution_skipped(self, tmp_path: pathlib.Path) -> None:
277	root = _make_repo(tmp_path)
278	_write_preimage(root, _VALID_FP)
279	res_path = rr_cache_dir(root) / _VALID_FP / "resolution"
280	import muse.core.rerere as rmod
281	original = rmod._MAX_RESOLUTION_BYTES
282	try:
283	rmod._MAX_RESOLUTION_BYTES = 5
284	res_path.write_text(_SHA_C, encoding="utf-8")
285	dest = tmp_path / "output.py"
286	ok = apply_cached(root, _VALID_FP, dest)
287	finally:
288	rmod._MAX_RESOLUTION_BYTES = original
289	assert not ok
290	assert not dest.exists()
291
292	def test_invalid_fingerprint_returns_false(self, tmp_path: pathlib.Path) -> None:
293	root = _make_repo(tmp_path)
294	ok = apply_cached(root, "../traversal", tmp_path / "out.py")
295	assert not ok
296
297
298	# ---------------------------------------------------------------------------
299	# Unit — list_records symlink guard
300	# ---------------------------------------------------------------------------
301
302
303	class TestListRecordsSymlinkGuard:
304	def test_symlink_skipped(self, tmp_path: pathlib.Path) -> None:
305	root = _make_repo(tmp_path)
306	_write_preimage(root, _VALID_FP)
307	cache = rr_cache_dir(root)
308	# Create a symlink pointing to some external dir
309	target = tmp_path / "external"
310	target.mkdir()
311	link = cache / ("f" * 64)
312	try:
313	link.symlink_to(target)
314	records = list_records(root)
315	assert all(r.fingerprint == _VALID_FP for r in records)
316	assert len(records) == 1
317	except NotImplementedError:
318	pytest.skip("symlinks not supported on this platform")
319
320	def test_regular_dirs_included(self, tmp_path: pathlib.Path) -> None:
321	root = _make_repo(tmp_path)
322	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
323	_write_preimage(root, _VALID_FP)
324	_write_preimage(root, fp2)
325	records = list_records(root)
326	fps = {r.fingerprint for r in records}
327	assert _VALID_FP in fps
328	assert fp2 in fps
329
330
331	# ---------------------------------------------------------------------------
332	# Unit — clear_all symlink guard
333	# ---------------------------------------------------------------------------
334
335
336	class TestClearAllSymlinkGuard:
337	def test_symlink_not_cleared(self, tmp_path: pathlib.Path) -> None:
338	root = _make_repo(tmp_path)
339	_write_preimage(root, _VALID_FP)
340	external = tmp_path / "external_dir"
341	external.mkdir()
342	sentinel = external / "sentinel.txt"
343	sentinel.write_text("keep me", encoding="utf-8")
344	cache = rr_cache_dir(root)
345	link = cache / ("e" * 64)
346	try:
347	link.symlink_to(external)
348	clear_all(root)
349	# The symlink target is untouched
350	assert sentinel.exists()
351	except NotImplementedError:
352	pytest.skip("symlinks not supported on this platform")
353
354	def test_regular_entries_cleared(self, tmp_path: pathlib.Path) -> None:
355	root = _make_repo(tmp_path)
356	_write_preimage(root, _VALID_FP)
357	removed = clear_all(root)
358	assert removed == 1
359	assert not (rr_cache_dir(root) / _VALID_FP).exists()
360
361
362	# ---------------------------------------------------------------------------
363	# Unit — auto_apply path traversal guard
364	# ---------------------------------------------------------------------------
365
366
367	class TestAutoApplyPathTraversal:
368	def test_traversal_path_skipped(self, tmp_path: pathlib.Path) -> None:
369	from muse.core.rerere import auto_apply
370	from unittest.mock import MagicMock
371
372	root = _make_repo(tmp_path)
373	plugin = MagicMock()
374	del plugin.conflict_fingerprint # not a RererePlugin
375
376	resolved, remaining = auto_apply(
377	root,
378	conflict_paths=["../../etc/passwd"],
379	ours_manifest={"../../etc/passwd": _SHA_A},
380	theirs_manifest={"../../etc/passwd": _SHA_B},
381	domain="code",
382	plugin=plugin,
383	)
384	assert "../../etc/passwd" in remaining
385	assert "../../etc/passwd" not in resolved
386
387	def test_valid_path_processed(self, tmp_path: pathlib.Path) -> None:
388	from muse.core.rerere import auto_apply
389	from unittest.mock import MagicMock
390
391	root = _make_repo(tmp_path)
392	plugin = MagicMock()
393	del plugin.conflict_fingerprint # not a RererePlugin
394
395	# No resolution cached — should be added to remaining and preimage recorded
396	resolved, remaining = auto_apply(
397	root,
398	conflict_paths=["src/module.py"],
399	ours_manifest={"src/module.py": _SHA_A},
400	theirs_manifest={"src/module.py": _SHA_B},
401	domain="code",
402	plugin=plugin,
403	)
404	assert "src/module.py" in remaining
405
406
407	# ---------------------------------------------------------------------------
408	# Unit — gc_stale age_days parameter
409	# ---------------------------------------------------------------------------
410
411
412	class TestGcStaleAgeDays:
413	def test_default_60_days_keeps_recent(self, tmp_path: pathlib.Path) -> None:
414	root = _make_repo(tmp_path)
415	_write_preimage(root, _VALID_FP)
416	# Entry is just created — within 60 days — should NOT be removed
417	removed = gc_stale(root, age_days=60)
418	assert removed == 0
419
420	def test_age_1_day_removes_entry_written_two_days_ago(
421	self, tmp_path: pathlib.Path
422	) -> None:
423	root = _make_repo(tmp_path)
424	cache_dir = rr_cache_dir(root)
425	entry_dir = cache_dir / _VALID_FP
426	entry_dir.mkdir(parents=True, exist_ok=True)
427	old_time = (
428	datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=2)
429	).isoformat()
430	meta = {
431	"fingerprint": _VALID_FP,
432	"path": "src/a.py",
433	"ours_id": _SHA_A,
434	"theirs_id": _SHA_B,
435	"domain": "code",
436	"recorded_at": old_time,
437	}
438	(entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
439	removed = gc_stale(root, age_days=1)
440	assert removed == 1
441
442	def test_resolved_entry_never_removed(self, tmp_path: pathlib.Path) -> None:
443	root = _make_repo(tmp_path)
444	cache_dir = rr_cache_dir(root)
445	entry_dir = cache_dir / _VALID_FP
446	entry_dir.mkdir(parents=True, exist_ok=True)
447	old_time = (
448	datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=100)
449	).isoformat()
450	meta = {
451	"fingerprint": _VALID_FP,
452	"path": "src/a.py",
453	"ours_id": _SHA_A,
454	"theirs_id": _SHA_B,
455	"domain": "code",
456	"recorded_at": old_time,
457	}
458	(entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
459	(entry_dir / "resolution").write_text(_SHA_C, encoding="utf-8")
460	removed = gc_stale(root, age_days=1)
461	assert removed == 0
462
463
464	# ---------------------------------------------------------------------------
465	# Unit — _fmt_record (no color parameter)
466	# ---------------------------------------------------------------------------
467
468
469	class TestFmtRecord:
470	def _rec(self) -> RerereRecord:
471	return RerereRecord(
472	fingerprint=_VALID_FP,
473	path="src/module.py",
474	ours_id=_SHA_A,
475	theirs_id=_SHA_B,
476	domain="code",
477	recorded_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
478	)
479
480	def test_no_color_parameter(self) -> None:
481	from muse.cli.commands.rerere import _fmt_record
482
483	rec = self._rec()
484	result = _fmt_record(rec) # must not require color argument
485	assert isinstance(result, str)
486
487	def test_sanitizes_path(self) -> None:
488	from muse.cli.commands.rerere import _fmt_record
489
490	rec = RerereRecord(
491	fingerprint=_VALID_FP,
492	path="src/\x1b[31mevil\x1b[0m.py",
493	ours_id=_SHA_A,
494	theirs_id=_SHA_B,
495	domain="code",
496	recorded_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
497	)
498	result = _fmt_record(rec)
499	assert "\x1b" not in result
500
501	def test_shows_resolved_status(self) -> None:
502	from muse.cli.commands.rerere import _fmt_record
503
504	rec = RerereRecord(
505	fingerprint=_VALID_FP,
506	path="src/a.py",
507	ours_id=_SHA_A,
508	theirs_id=_SHA_B,
509	domain="code",
510	recorded_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
511	resolution_id=_SHA_C,
512	)
513	result = _fmt_record(rec)
514	assert "resolved" in result
515
516	def test_shows_preimage_only_status(self) -> None:
517	from muse.cli.commands.rerere import _fmt_record
518
519	rec = self._rec()
520	result = _fmt_record(rec)
521	assert "preimage" in result
522
523
524	# ---------------------------------------------------------------------------
525	# Unit — _check_format
526	# ---------------------------------------------------------------------------
527
528
529	class TestCheckFormat:
530	def test_valid_text_passes(self) -> None:
531	from muse.cli.commands.rerere import _check_format
532	_check_format("text")
533
534	def test_valid_json_passes(self) -> None:
535	from muse.cli.commands.rerere import _check_format
536	_check_format("json")
537
538	def test_invalid_format_raises(self) -> None:
539	from muse.cli.commands.rerere import _check_format
540	with pytest.raises(SystemExit) as exc_info:
541	_check_format("xml")
542	assert exc_info.value.code == ExitCode.USER_ERROR.value
543
544
545	# ---------------------------------------------------------------------------
546	# Integration — status and gc subcommands
547	# ---------------------------------------------------------------------------
548
549
550	class TestStatusIntegration:
551	def test_empty_cache_shows_no_records(self, tmp_path: pathlib.Path) -> None:
552	root = _make_repo(tmp_path)
553	result = _invoke(root, "status")
554	assert result.exit_code == 0
555	assert "No rerere records" in result.output
556
557	def test_status_shows_entry(self, tmp_path: pathlib.Path) -> None:
558	root = _make_repo(tmp_path)
559	_write_preimage(root, _VALID_FP, path="src/x.py")
560	result = _invoke(root, "status")
561	assert result.exit_code == 0
562	assert _VALID_FP[:12] in result.output
563
564	def test_status_json_schema(self, tmp_path: pathlib.Path) -> None:
565	root = _make_repo(tmp_path)
566	_write_preimage(root, _VALID_FP, path="src/x.py")
567	result = _invoke(root, "status", "--json")
568	assert result.exit_code == 0
569	raw = _parse_json(result)
570	assert "total" in raw
571	assert "records" in raw
572	records = raw["records"]
573	assert isinstance(records, list)
574	assert len(records) == 1
575	rec = records[0]
576	assert isinstance(rec, dict)
577	for field in ("fingerprint", "path", "domain", "has_resolution",
578	"resolution_id", "recorded_at", "matches_current_conflict"):
579	assert field in rec, f"Missing field: {field}"
580
581	def test_status_json_has_resolution_false(self, tmp_path: pathlib.Path) -> None:
582	root = _make_repo(tmp_path)
583	_write_preimage(root, _VALID_FP)
584	result = _invoke(root, "status", "--json")
585	raw = _parse_json(result)
586	records = raw["records"]
587	assert isinstance(records, list)
588	rec = records[0]
589	assert isinstance(rec, dict)
590	assert rec["has_resolution"] is False
591	assert rec["resolution_id"] is None
592
593	def test_status_json_has_resolution_true(self, tmp_path: pathlib.Path) -> None:
594	root = _make_repo(tmp_path)
595	_write_preimage(root, _VALID_FP)
596	_write_resolution(root, _VALID_FP, res_id=_SHA_C)
597	result = _invoke(root, "status", "--json")
598	raw = _parse_json(result)
599	records = raw["records"]
600	assert isinstance(records, list)
601	rec = records[0]
602	assert isinstance(rec, dict)
603	assert rec["has_resolution"] is True
604	assert rec["resolution_id"] == _SHA_C
605
606
607	# ===========================================================================
608	# TestRerereStatusExtended — 18 tests
609	# ===========================================================================
610
611
612	class TestRerereStatusExtended:
613	def test_exit_code_zero_empty(self, tmp_path: pathlib.Path) -> None:
614	"""Empty cache exits 0."""
615	root = _make_repo(tmp_path)
616	result = _invoke(root, "status")
617	assert result.exit_code == 0
618
619	def test_exit_code_zero_with_entries(self, tmp_path: pathlib.Path) -> None:
620	"""Non-empty cache exits 0."""
621	root = _make_repo(tmp_path)
622	_write_preimage(root, _VALID_FP)
623	result = _invoke(root, "status")
624	assert result.exit_code == 0
625
626	def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None:
627	"""No .muse directory → exit 2."""
628	result = _invoke(tmp_path, "status")
629	assert result.exit_code == 2
630
631	def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None:
632	root = _make_repo(tmp_path)
633	result = _invoke(root, "status", "--format", "xml")
634	assert result.exit_code == 1
635
636	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
637	"""-j must produce identical output to --json."""
638	root = _make_repo(tmp_path)
639	_write_preimage(root, _VALID_FP)
640	r1 = _invoke(root, "status", "--json")
641	r2 = _invoke(root, "status", "-j")
642	assert r1.exit_code == 0 and r2.exit_code == 0
643	assert _parse_json(r1) == _parse_json(r2)
644
645	def test_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None:
646	"""JSON output must be compact — no indent=2."""
647	root = _make_repo(tmp_path)
648	_write_preimage(root, _VALID_FP)
649	result = _invoke(root, "status", "--json")
650	assert result.exit_code == 0
651	assert "\n" not in result.output.strip()
652
653	def test_json_total_field(self, tmp_path: pathlib.Path) -> None:
654	root = _make_repo(tmp_path)
655	_write_preimage(root, _VALID_FP)
656	raw = _parse_json(_invoke(root, "status", "--json"))
657	assert raw["total"] == 1
658
659	def test_json_total_zero_when_empty(self, tmp_path: pathlib.Path) -> None:
660	root = _make_repo(tmp_path)
661	raw = _parse_json(_invoke(root, "status", "--json"))
662	assert raw["total"] == 0
663	assert raw["records"] == []
664
665	def test_json_all_fields_present(self, tmp_path: pathlib.Path) -> None:
666	root = _make_repo(tmp_path)
667	_write_preimage(root, _VALID_FP, path="src/x.py")
668	raw = _parse_json(_invoke(root, "status", "--json"))
669	rec = raw["records"][0]
670	for field in ("fingerprint", "path", "domain", "has_resolution",
671	"resolution_id", "recorded_at", "matches_current_conflict"):
672	assert field in rec
673
674	def test_json_multiple_entries_total(self, tmp_path: pathlib.Path) -> None:
675	"""total reflects number of distinct entries."""
676	root = _make_repo(tmp_path)
677	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
678	_write_preimage(root, _VALID_FP, path="src/a.py")
679	_write_preimage(root, fp2, path="src/b.py")
680	raw = _parse_json(_invoke(root, "status", "--json"))
681	assert raw["total"] == 2
682	assert len(raw["records"]) == 2
683
684	def test_json_resolution_id_null_without_resolution(self, tmp_path: pathlib.Path) -> None:
685	root = _make_repo(tmp_path)
686	_write_preimage(root, _VALID_FP)
687	raw = _parse_json(_invoke(root, "status", "--json"))
688	assert raw["records"][0]["resolution_id"] is None
689
690	def test_json_resolution_id_set_with_resolution(self, tmp_path: pathlib.Path) -> None:
691	root = _make_repo(tmp_path)
692	_write_preimage(root, _VALID_FP)
693	_write_resolution(root, _VALID_FP, res_id=_SHA_C)
694	raw = _parse_json(_invoke(root, "status", "--json"))
695	assert raw["records"][0]["resolution_id"] == _SHA_C
696
697	def test_json_has_resolution_true_and_false(self, tmp_path: pathlib.Path) -> None:
698	"""Mix of resolved and preimage-only entries reported correctly."""
699	root = _make_repo(tmp_path)
700	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
701	_write_preimage(root, _VALID_FP, path="src/a.py")
702	_write_preimage(root, fp2, path="src/b.py")
703	_write_resolution(root, _VALID_FP)
704	raw = _parse_json(_invoke(root, "status", "--json"))
705	by_fp = {r["fingerprint"]: r for r in raw["records"]}
706	assert by_fp[_VALID_FP]["has_resolution"] is True
707	assert by_fp[fp2]["has_resolution"] is False
708
709	def test_json_matches_current_conflict_false_without_merge(
710	self, tmp_path: pathlib.Path
711	) -> None:
712	"""No merge in progress → matches_current_conflict always False."""
713	root = _make_repo(tmp_path)
714	_write_preimage(root, _VALID_FP)
715	raw = _parse_json(_invoke(root, "status", "--json"))
716	assert raw["records"][0]["matches_current_conflict"] is False
717
718	def test_text_format_explicit(self, tmp_path: pathlib.Path) -> None:
719	"""--format text is identical to default."""
720	root = _make_repo(tmp_path)
721	_write_preimage(root, _VALID_FP)
722	r_default = _invoke(root, "status")
723	r_text = _invoke(root, "status", "--format", "text")
724	assert r_default.output == r_text.output
725
726	def test_text_shows_header_when_entries_exist(self, tmp_path: pathlib.Path) -> None:
727	root = _make_repo(tmp_path)
728	_write_preimage(root, _VALID_FP)
729	result = _invoke(root, "status")
730	assert "fingerprint" in result.output
731	assert "status" in result.output
732
733	def test_help_mentions_agent_quickstart(self) -> None:
734	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "status", "--help")
735	assert "Agent quickstart" in result.output
736
737	def test_help_mentions_exit_codes(self) -> None:
738	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "status", "--help")
739	assert "Exit codes" in result.output
740
741
742	# ===========================================================================
743	# TestRerereStatusSecurity — 6 tests
744	# ===========================================================================
745
746
747	class TestRerereStatusSecurity:
748	def test_ansi_in_path_stripped_json(self, tmp_path: pathlib.Path) -> None:
749	"""ANSI in a path must be stripped in JSON output."""
750	root = _make_repo(tmp_path)
751	evil_path = "src/\x1b[31mevil\x1b[0m.py"
752	_write_preimage(root, _VALID_FP, path=evil_path)
753	result = _invoke(root, "status", "--json")
754	assert result.exit_code == 0
755	assert "\x1b" not in result.output
756
757	def test_ansi_in_path_stripped_text(self, tmp_path: pathlib.Path) -> None:
758	"""ANSI in a path must be stripped in text output."""
759	root = _make_repo(tmp_path)
760	evil_path = "src/\x1b[31mevil\x1b[0m.py"
761	_write_preimage(root, _VALID_FP, path=evil_path)
762	result = _invoke(root, "status")
763	assert result.exit_code == 0
764	assert "\x1b" not in result.output
765
766	def test_control_char_in_path_stripped_json(self, tmp_path: pathlib.Path) -> None:
767	"""Control characters in path stripped from JSON output."""
768	root = _make_repo(tmp_path)
769	evil_path = "src/evil\x07.py"
770	_write_preimage(root, _VALID_FP, path=evil_path)
771	result = _invoke(root, "status", "--json")
772	assert result.exit_code == 0
773	assert "\x07" not in result.output
774
775	def test_no_json_outside_repo(self, tmp_path: pathlib.Path) -> None:
776	"""Running outside a repo must not emit JSON — exits 2."""
777	result = _invoke(tmp_path, "status", "--json")
778	assert result.exit_code == 2
779	assert not result.output.strip().startswith("{")
780
781	def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path) -> None:
782	"""Running outside a repo must not emit a traceback."""
783	result = _invoke(tmp_path, "status")
784	assert result.exit_code == 2
785	assert "Traceback" not in result.output
786
787	def test_resolution_id_ansi_stripped_json(self, tmp_path: pathlib.Path) -> None:
788	"""ANSI in a resolution_id must be stripped in JSON output."""
789	root = _make_repo(tmp_path)
790	_write_preimage(root, _VALID_FP, path="src/a.py")
791	evil_res_id = _SHA_C[:32] + "\x1b[31m" + _SHA_C[32:]
792	_write_resolution(root, _VALID_FP, res_id=evil_res_id)
793	result = _invoke(root, "status", "--json")
794	assert result.exit_code == 0
795	assert "\x1b" not in result.output
796
797
798	# ===========================================================================
799	# TestRerereStatusStress — 3 tests
800	# ===========================================================================
801
802
803	class TestRerereStatusStress:
804	def test_100_entries_text(self, tmp_path: pathlib.Path) -> None:
805	"""100 cache entries are all listed in text mode."""
806	import hashlib as _hl
807	root = _make_repo(tmp_path)
808	fps = []
809	for i in range(100):
810	fp = conflict_fingerprint(
811	_hl.sha256(f"a{i}".encode()).hexdigest(),
812	_hl.sha256(f"b{i}".encode()).hexdigest(),
813	)
814	_write_preimage(root, fp, path=f"src/f{i}.py")
815	fps.append(fp)
816	result = _invoke(root, "status")
817	assert result.exit_code == 0
818	for fp in fps:
819	assert fp[:12] in result.output
820
821	def test_100_entries_json_total(self, tmp_path: pathlib.Path) -> None:
822	"""100 cache entries produce total=100 in JSON mode."""
823	import hashlib as _hl
824	root = _make_repo(tmp_path)
825	for i in range(100):
826	fp = conflict_fingerprint(
827	_hl.sha256(f"c{i}".encode()).hexdigest(),
828	_hl.sha256(f"d{i}".encode()).hexdigest(),
829	)
830	_write_preimage(root, fp, path=f"src/g{i}.py")
831	raw = _parse_json(_invoke(root, "status", "--json"))
832	assert raw["total"] == 100
833	assert len(raw["records"]) == 100
834
835	def test_mixed_resolved_and_preimage_counts(self, tmp_path: pathlib.Path) -> None:
836	"""50 resolved + 50 preimage-only: has_resolution counts correct."""
837	import hashlib as _hl
838	root = _make_repo(tmp_path)
839	for i in range(100):
840	fp = conflict_fingerprint(
841	_hl.sha256(f"e{i}".encode()).hexdigest(),
842	_hl.sha256(f"f{i}".encode()).hexdigest(),
843	)
844	_write_preimage(root, fp, path=f"src/h{i}.py")
845	if i < 50:
846	_write_resolution(root, fp)
847	raw = _parse_json(_invoke(root, "status", "--json"))
848	resolved = sum(1 for r in raw["records"] if r["has_resolution"])
849	preimage_only = sum(1 for r in raw["records"] if not r["has_resolution"])
850	assert resolved == 50
851	assert preimage_only == 50
852
853
854	# ---------------------------------------------------------------------------
855	# Integration — forget with --fingerprint
856	# ---------------------------------------------------------------------------
857
858
859	class TestForgetFingerprint:
860	def test_forget_by_fingerprint_removes_entry(
861	self, tmp_path: pathlib.Path
862	) -> None:
863	root = _make_repo(tmp_path)
864	_write_preimage(root, _VALID_FP)
865	result = _invoke(root, "forget", "--fingerprint", _VALID_FP)
866	assert result.exit_code == 0
867	assert not (rr_cache_dir(root) / _VALID_FP).exists()
868
869	def test_forget_by_fingerprint_json(self, tmp_path: pathlib.Path) -> None:
870	root = _make_repo(tmp_path)
871	_write_preimage(root, _VALID_FP)
872	result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")
873	assert result.exit_code == 0
874	raw = _parse_json(result)
875	assert "forgotten" in raw
876	assert "not_found" in raw
877
878	def test_forget_by_fingerprint_not_found(self, tmp_path: pathlib.Path) -> None:
879	root = _make_repo(tmp_path)
880	result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")
881	raw = _parse_json(result)
882	not_found = raw["not_found"]
883	assert isinstance(not_found, list)
884	assert _VALID_FP in not_found
885
886	def test_forget_by_fingerprint_no_merge_required(
887	self, tmp_path: pathlib.Path
888	) -> None:
889	"""Fingerprint mode does not need an active merge state."""
890	root = _make_repo(tmp_path)
891	_write_preimage(root, _VALID_FP)
892	# No MERGE_STATE.json exists
893	result = _invoke(root, "forget", "--fingerprint", _VALID_FP)
894	assert result.exit_code == 0
895
896	def test_forget_invalid_fingerprint_rejected(
897	self, tmp_path: pathlib.Path
898	) -> None:
899	root = _make_repo(tmp_path)
900	result = _invoke(root, "forget", "--fingerprint", "../evil")
901	# Should exit 0 (not_found), not crash
902	assert result.exit_code == 0
903	assert "Traceback" not in result.output
904
905	def test_forget_multiple_fingerprints(self, tmp_path: pathlib.Path) -> None:
906	root = _make_repo(tmp_path)
907	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
908	_write_preimage(root, _VALID_FP)
909	_write_preimage(root, fp2)
910	result = _invoke(
911	root, "forget",
912	"--fingerprint", _VALID_FP,
913	"--fingerprint", fp2,
914	"--json",
915	)
916	assert result.exit_code == 0
917	raw = _parse_json(result)
918	forgotten = raw["forgotten"]
919	assert isinstance(forgotten, list)
920	assert len(forgotten) == 2
921
922	def test_forget_no_args_exits_user_error(
923	self, tmp_path: pathlib.Path
924	) -> None:
925	root = _make_repo(tmp_path)
926	result = _invoke(root, "forget")
927	assert result.exit_code == ExitCode.USER_ERROR.value
928
929
930	# ===========================================================================
931	# TestRerereForgetExtended — 18 tests
932	# ===========================================================================
933
934
935	class TestRerereForgetExtended:
936	def test_fingerprint_mode_removes_entry(self, tmp_path: pathlib.Path) -> None:
937	root = _make_repo(tmp_path)
938	_write_preimage(root, _VALID_FP)
939	result = _invoke(root, "forget", "--fingerprint", _VALID_FP)
940	assert result.exit_code == 0
941	assert not (rr_cache_dir(root) / _VALID_FP).exists()
942
943	def test_fingerprint_mode_forgotten_in_json(self, tmp_path: pathlib.Path) -> None:
944	root = _make_repo(tmp_path)
945	_write_preimage(root, _VALID_FP)
946	raw = _parse_json(_invoke(root, "forget", "--fingerprint", _VALID_FP, "--json"))
947	assert _VALID_FP in raw["forgotten"]
948	assert raw["not_found"] == []
949
950	def test_fingerprint_mode_not_found_in_json(self, tmp_path: pathlib.Path) -> None:
951	root = _make_repo(tmp_path)
952	raw = _parse_json(_invoke(root, "forget", "--fingerprint", _VALID_FP, "--json"))
953	assert _VALID_FP in raw["not_found"]
954	assert raw["forgotten"] == []
955
956	def test_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None:
957	"""JSON output must be compact — no indent=2."""
958	root = _make_repo(tmp_path)
959	result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")
960	assert result.exit_code == 0
961	assert "\n" not in result.output.strip()
962
963	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
964	"""-j must produce identical output to --json."""
965	root = _make_repo(tmp_path)
966	_write_preimage(root, _VALID_FP)
967	r1 = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")
968	_write_preimage(root, _VALID_FP) # restore for second call
969	r2 = _invoke(root, "forget", "--fingerprint", _VALID_FP, "-j")
970	assert r1.exit_code == 0 and r2.exit_code == 0
971	assert _parse_json(r1).keys() == _parse_json(r2).keys()
972
973	def test_multiple_fingerprints_all_forgotten(self, tmp_path: pathlib.Path) -> None:
974	root = _make_repo(tmp_path)
975	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
976	_write_preimage(root, _VALID_FP, path="src/a.py")
977	_write_preimage(root, fp2, path="src/b.py")
978	raw = _parse_json(_invoke(
979	root, "forget",
980	"--fingerprint", _VALID_FP,
981	"--fingerprint", fp2,
982	"--json",
983	))
984	assert len(raw["forgotten"]) == 2
985	assert raw["not_found"] == []
986
987	def test_mixed_found_and_not_found(self, tmp_path: pathlib.Path) -> None:
988	"""One found + one absent → both lists populated."""
989	root = _make_repo(tmp_path)
990	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
991	_write_preimage(root, _VALID_FP)
992	raw = _parse_json(_invoke(
993	root, "forget",
994	"--fingerprint", _VALID_FP,
995	"--fingerprint", fp2,
996	"--json",
997	))
998	assert _VALID_FP in raw["forgotten"]
999	assert fp2 in raw["not_found"]
1000
1001	def test_text_output_shows_forgot(self, tmp_path: pathlib.Path) -> None:
1002	root = _make_repo(tmp_path)
1003	_write_preimage(root, _VALID_FP)
1004	result = _invoke(root, "forget", "--fingerprint", _VALID_FP)
1005	assert result.exit_code == 0
1006	assert "forgot" in result.output
1007
1008	def test_text_output_shows_no_record_found(self, tmp_path: pathlib.Path) -> None:
1009	root = _make_repo(tmp_path)
1010	result = _invoke(root, "forget", "--fingerprint", _VALID_FP)
1011	assert result.exit_code == 0
1012	assert "no record found" in result.output
1013
1014	def test_no_args_exits_1(self, tmp_path: pathlib.Path) -> None:
1015	root = _make_repo(tmp_path)
1016	result = _invoke(root, "forget")
1017	assert result.exit_code == 1
1018
1019	def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None:
1020	root = _make_repo(tmp_path)
1021	result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--format", "xml")
1022	assert result.exit_code == 1
1023
1024	def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None:
1025	result = _invoke(tmp_path, "forget", "--fingerprint", _VALID_FP)
1026	assert result.exit_code == 2
1027
1028	def test_path_mode_no_merge_exits_1(self, tmp_path: pathlib.Path) -> None:
1029	"""Path mode without an active merge must exit 1."""
1030	root = _make_repo(tmp_path)
1031	result = _invoke(root, "forget", "src/a.py")
1032	assert result.exit_code == 1
1033
1034	def test_fingerprint_mode_no_merge_required(self, tmp_path: pathlib.Path) -> None:
1035	"""Fingerprint mode works without any MERGE_STATE."""
1036	root = _make_repo(tmp_path)
1037	_write_preimage(root, _VALID_FP)
1038	result = _invoke(root, "forget", "--fingerprint", _VALID_FP)
1039	assert result.exit_code == 0
1040
1041	def test_json_schema_fields_present(self, tmp_path: pathlib.Path) -> None:
1042	root = _make_repo(tmp_path)
1043	raw = _parse_json(_invoke(root, "forget", "--fingerprint", _VALID_FP, "--json"))
1044	assert "forgotten" in raw
1045	assert "not_found" in raw
1046	assert isinstance(raw["forgotten"], list)
1047	assert isinstance(raw["not_found"], list)
1048
1049	def test_invalid_fingerprint_lands_in_not_found(self, tmp_path: pathlib.Path) -> None:
1050	"""A traversal-like fingerprint is rejected by forget_record and lands in not_found."""
1051	root = _make_repo(tmp_path)
1052	raw = _parse_json(_invoke(root, "forget", "--fingerprint", "../evil", "--json"))
1053	assert "../evil" in raw["not_found"]
1054	assert raw["forgotten"] == []
1055
1056	def test_help_mentions_agent_quickstart(self) -> None:
1057	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "forget", "--help")
1058	assert "Agent quickstart" in result.output
1059
1060	def test_help_mentions_exit_codes(self) -> None:
1061	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "forget", "--help")
1062	assert "Exit codes" in result.output
1063
1064
1065	# ===========================================================================
1066	# TestRerereForgetSecurity — 6 tests
1067	# ===========================================================================
1068
1069
1070	class TestRerereForgetSecurity:
1071	def test_ansi_in_fingerprint_stripped_json(self, tmp_path: pathlib.Path) -> None:
1072	"""ANSI in a forgotten fingerprint must be stripped in JSON output."""
1073	root = _make_repo(tmp_path)
1074	_write_preimage(root, _VALID_FP)
1075	result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")
1076	assert result.exit_code == 0
1077	assert "\x1b" not in result.output
1078
1079	def test_ansi_in_not_found_stripped_json(self, tmp_path: pathlib.Path) -> None:
1080	"""ANSI injected into a not_found item must be stripped."""
1081	root = _make_repo(tmp_path)
1082	evil_fp = "\x1b[31m" + "a" * 64 + "\x1b[0m"
1083	result = _invoke(root, "forget", "--fingerprint", evil_fp, "--json")
1084	assert result.exit_code == 0
1085	assert "\x1b" not in result.output
1086
1087	def test_control_char_in_item_stripped_json(self, tmp_path: pathlib.Path) -> None:
1088	"""Control chars other than ANSI stripped from JSON."""
1089	root = _make_repo(tmp_path)
1090	evil_fp = "a" * 32 + "\x07" + "b" * 31
1091	result = _invoke(root, "forget", "--fingerprint", evil_fp, "--json")
1092	assert result.exit_code == 0
1093	assert "\x07" not in result.output
1094
1095	def test_ansi_in_fingerprint_stripped_text(self, tmp_path: pathlib.Path) -> None:
1096	"""ANSI stripped from text output."""
1097	root = _make_repo(tmp_path)
1098	_write_preimage(root, _VALID_FP)
1099	result = _invoke(root, "forget", "--fingerprint", _VALID_FP)
1100	assert result.exit_code == 0
1101	assert "\x1b" not in result.output
1102
1103	def test_no_json_outside_repo(self, tmp_path: pathlib.Path) -> None:
1104	"""Outside a repo must not emit JSON."""
1105	result = _invoke(tmp_path, "forget", "--fingerprint", _VALID_FP, "--json")
1106	assert result.exit_code == 2
1107	assert not result.output.strip().startswith("{")
1108
1109	def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path) -> None:
1110	result = _invoke(tmp_path, "forget", "--fingerprint", _VALID_FP)
1111	assert result.exit_code == 2
1112	assert "Traceback" not in result.output
1113
1114
1115	# ===========================================================================
1116	# TestRerereForgetStress — 3 tests
1117	# ===========================================================================
1118
1119
1120	class TestRerereForgetStress:
1121	def test_50_fingerprints_all_forgotten(self, tmp_path: pathlib.Path) -> None:
1122	"""50 fingerprints forgotten in one call."""
1123	import hashlib as _hl
1124	root = _make_repo(tmp_path)
1125	fps = []
1126	for i in range(50):
1127	fp = conflict_fingerprint(
1128	_hl.sha256(f"x{i}".encode()).hexdigest(),
1129	_hl.sha256(f"y{i}".encode()).hexdigest(),
1130	)
1131	_write_preimage(root, fp, path=f"src/f{i}.py")
1132	fps.append(fp)
1133	args = ["forget"]
1134	for fp in fps:
1135	args += ["--fingerprint", fp]
1136	args.append("--json")
1137	raw = _parse_json(_invoke(root, *args))
1138	assert len(raw["forgotten"]) == 50
1139	assert raw["not_found"] == []
1140
1141	def test_50_fingerprints_all_not_found(self, tmp_path: pathlib.Path) -> None:
1142	"""50 absent fingerprints all land in not_found."""
1143	import hashlib as _hl
1144	root = _make_repo(tmp_path)
1145	fps = []
1146	for i in range(50):
1147	fp = conflict_fingerprint(
1148	_hl.sha256(f"p{i}".encode()).hexdigest(),
1149	_hl.sha256(f"q{i}".encode()).hexdigest(),
1150	)
1151	fps.append(fp)
1152	args = ["forget"]
1153	for fp in fps:
1154	args += ["--fingerprint", fp]
1155	args.append("--json")
1156	raw = _parse_json(_invoke(root, *args))
1157	assert len(raw["not_found"]) == 50
1158	assert raw["forgotten"] == []
1159
1160	def test_mixed_25_found_25_not_found(self, tmp_path: pathlib.Path) -> None:
1161	"""25 found + 25 absent → counts match."""
1162	import hashlib as _hl
1163	root = _make_repo(tmp_path)
1164	fps_present = []
1165	fps_absent = []
1166	for i in range(25):
1167	fp = conflict_fingerprint(
1168	_hl.sha256(f"m{i}".encode()).hexdigest(),
1169	_hl.sha256(f"n{i}".encode()).hexdigest(),
1170	)
1171	_write_preimage(root, fp, path=f"src/g{i}.py")
1172	fps_present.append(fp)
1173	for i in range(25):
1174	fp = conflict_fingerprint(
1175	_hl.sha256(f"r{i}".encode()).hexdigest(),
1176	_hl.sha256(f"s{i}".encode()).hexdigest(),
1177	)
1178	fps_absent.append(fp)
1179	args = ["forget"]
1180	for fp in fps_present + fps_absent:
1181	args += ["--fingerprint", fp]
1182	args.append("--json")
1183	raw = _parse_json(_invoke(root, *args))
1184	assert len(raw["forgotten"]) == 25
1185	assert len(raw["not_found"]) == 25
1186
1187
1188	# ---------------------------------------------------------------------------
1189	# Integration — gc with --age
1190	# ---------------------------------------------------------------------------
1191
1192
1193	class TestGcAge:
1194	def test_default_age_flag_present_in_help(self) -> None:
1195	result = runner.invoke(cli, ["rerere", "gc", "--help"])
1196	assert "age" in result.output.lower()
1197
1198	def test_gc_age_custom(self, tmp_path: pathlib.Path) -> None:
1199	root = _make_repo(tmp_path)
1200	cache_dir = rr_cache_dir(root)
1201	entry_dir = cache_dir / _VALID_FP
1202	entry_dir.mkdir(parents=True, exist_ok=True)
1203	old_time = (
1204	datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=2)
1205	).isoformat()
1206	meta = {
1207	"fingerprint": _VALID_FP, "path": "src/a.py",
1208	"ours_id": _SHA_A, "theirs_id": _SHA_B,
1209	"domain": "code", "recorded_at": old_time,
1210	}
1211	(entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
1212	result = _invoke(root, "gc", "--age", "1", "--json")
1213	assert result.exit_code == 0
1214	raw = _parse_json(result)
1215	assert raw["removed"] == 1
1216
1217	def test_gc_json_schema(self, tmp_path: pathlib.Path) -> None:
1218	root = _make_repo(tmp_path)
1219	result = _invoke(root, "gc", "--json")
1220	assert result.exit_code == 0
1221	raw = _parse_json(result)
1222	assert "removed" in raw
1223
1224	def test_gc_text_mentions_threshold(self, tmp_path: pathlib.Path) -> None:
1225	root = _make_repo(tmp_path)
1226	result = _invoke(root, "gc", "--age", "30")
1227	assert result.exit_code == 0
1228	assert "30" in result.output
1229
1230
1231	def _write_stale_preimage(
1232	root: pathlib.Path,
1233	fp: str,
1234	age_days: int = 90,
1235	path: str = "src/a.py",
1236	) -> None:
1237	"""Write a preimage entry recorded age_days ago."""
1238	entry_dir = rr_cache_dir(root) / fp
1239	entry_dir.mkdir(parents=True, exist_ok=True)
1240	old_time = (
1241	datetime.datetime.now(datetime.timezone.utc)
1242	- datetime.timedelta(days=age_days)
1243	).isoformat()
1244	meta = {
1245	"fingerprint": fp,
1246	"path": path,
1247	"ours_id": _SHA_A,
1248	"theirs_id": _SHA_B,
1249	"domain": "code",
1250	"recorded_at": old_time,
1251	}
1252	(entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
1253
1254
1255	# ===========================================================================
1256	# TestRerereGcExtended — 18 tests
1257	# ===========================================================================
1258
1259
1260	class TestRerereGcExtended:
1261	def test_exit_code_zero_nothing_removed(self, tmp_path: pathlib.Path) -> None:
1262	"""Empty cache exits 0 with removed=0."""
1263	root = _make_repo(tmp_path)
1264	result = _invoke(root, "gc", "--json")
1265	assert result.exit_code == 0
1266	assert _parse_json(result)["removed"] == 0
1267
1268	def test_exit_code_zero_with_removal(self, tmp_path: pathlib.Path) -> None:
1269	root = _make_repo(tmp_path)
1270	_write_stale_preimage(root, _VALID_FP, age_days=90)
1271	result = _invoke(root, "gc", "--age", "1", "--json")
1272	assert result.exit_code == 0
1273	assert _parse_json(result)["removed"] == 1
1274
1275	def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None:
1276	result = _invoke(tmp_path, "gc")
1277	assert result.exit_code == 2
1278
1279	def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None:
1280	root = _make_repo(tmp_path)
1281	result = _invoke(root, "gc", "--format", "xml")
1282	assert result.exit_code == 1
1283
1284	def test_age_zero_exits_1(self, tmp_path: pathlib.Path) -> None:
1285	"""--age 0 is below the valid range — must exit 1, not crash."""
1286	root = _make_repo(tmp_path)
1287	result = _invoke(root, "gc", "--age", "0")
1288	assert result.exit_code == 1
1289	assert "Traceback" not in result.output
1290
1291	def test_age_negative_exits_1(self, tmp_path: pathlib.Path) -> None:
1292	root = _make_repo(tmp_path)
1293	result = _invoke(root, "gc", "--age", "-1")
1294	assert result.exit_code == 1
1295	assert "Traceback" not in result.output
1296
1297	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
1298	"""-j must produce identical output to --json."""
1299	root = _make_repo(tmp_path)
1300	r1 = _invoke(root, "gc", "--json")
1301	r2 = _invoke(root, "gc", "-j")
1302	assert r1.exit_code == 0 and r2.exit_code == 0
1303	assert _parse_json(r1) == _parse_json(r2)
1304
1305	def test_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None:
1306	root = _make_repo(tmp_path)
1307	result = _invoke(root, "gc", "--json")
1308	assert result.exit_code == 0
1309	assert "\n" not in result.output.strip()
1310
1311	def test_stale_preimage_removed(self, tmp_path: pathlib.Path) -> None:
1312	"""Preimage-only entry older than threshold is removed."""
1313	root = _make_repo(tmp_path)
1314	_write_stale_preimage(root, _VALID_FP, age_days=90)
1315	raw = _parse_json(_invoke(root, "gc", "--age", "1", "--json"))
1316	assert raw["removed"] == 1
1317	assert not (rr_cache_dir(root) / _VALID_FP).exists()
1318
1319	def test_fresh_preimage_kept(self, tmp_path: pathlib.Path) -> None:
1320	"""Preimage-only entry newer than threshold is kept."""
1321	root = _make_repo(tmp_path)
1322	_write_preimage(root, _VALID_FP) # recorded just now
1323	raw = _parse_json(_invoke(root, "gc", "--age", "60", "--json"))
1324	assert raw["removed"] == 0
1325	assert (rr_cache_dir(root) / _VALID_FP).exists()
1326
1327	def test_resolved_entry_never_removed(self, tmp_path: pathlib.Path) -> None:
1328	"""An entry with a resolution is kept regardless of age."""
1329	root = _make_repo(tmp_path)
1330	_write_stale_preimage(root, _VALID_FP, age_days=365)
1331	_write_resolution(root, _VALID_FP)
1332	raw = _parse_json(_invoke(root, "gc", "--age", "1", "--json"))
1333	assert raw["removed"] == 0
1334	assert (rr_cache_dir(root) / _VALID_FP).exists()
1335
1336	def test_stale_removed_fresh_kept(self, tmp_path: pathlib.Path) -> None:
1337	"""Only entries older than threshold are removed."""
1338	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
1339	root = _make_repo(tmp_path)
1340	_write_stale_preimage(root, _VALID_FP, age_days=90)
1341	_write_preimage(root, fp2) # fresh
1342	raw = _parse_json(_invoke(root, "gc", "--age", "30", "--json"))
1343	assert raw["removed"] == 1
1344	assert not (rr_cache_dir(root) / _VALID_FP).exists()
1345	assert (rr_cache_dir(root) / fp2).exists()
1346
1347	def test_default_age_is_60(self, tmp_path: pathlib.Path) -> None:
1348	"""Default --age=60: entry 61 days old is removed, 59 days old is kept."""
1349	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
1350	root = _make_repo(tmp_path)
1351	_write_stale_preimage(root, _VALID_FP, age_days=61)
1352	_write_stale_preimage(root, fp2, age_days=59)
1353	raw = _parse_json(_invoke(root, "gc", "--json"))
1354	assert raw["removed"] == 1
1355
1356	def test_text_output_removed_count(self, tmp_path: pathlib.Path) -> None:
1357	root = _make_repo(tmp_path)
1358	_write_stale_preimage(root, _VALID_FP, age_days=90)
1359	result = _invoke(root, "gc", "--age", "1")
1360	assert result.exit_code == 0
1361	assert "1" in result.output
1362
1363	def test_text_output_nothing_to_remove(self, tmp_path: pathlib.Path) -> None:
1364	root = _make_repo(tmp_path)
1365	result = _invoke(root, "gc")
1366	assert result.exit_code == 0
1367	assert "nothing" in result.output.lower() or "0" in result.output
1368
1369	def test_format_text_explicit(self, tmp_path: pathlib.Path) -> None:
1370	root = _make_repo(tmp_path)
1371	r_default = _invoke(root, "gc")
1372	r_text = _invoke(root, "gc", "--format", "text")
1373	assert r_default.output == r_text.output
1374
1375	def test_help_mentions_agent_quickstart(self) -> None:
1376	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "gc", "--help")
1377	assert "Agent quickstart" in result.output
1378
1379	def test_help_mentions_exit_codes(self) -> None:
1380	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "gc", "--help")
1381	assert "Exit codes" in result.output
1382
1383
1384	# ===========================================================================
1385	# TestRerereGcSecurity — 6 tests
1386	# ===========================================================================
1387
1388
1389	class TestRerereGcSecurity:
1390	def test_no_json_outside_repo(self, tmp_path: pathlib.Path) -> None:
1391	result = _invoke(tmp_path, "gc", "--json")
1392	assert result.exit_code == 2
1393	assert not result.output.strip().startswith("{")
1394
1395	def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path) -> None:
1396	result = _invoke(tmp_path, "gc")
1397	assert result.exit_code == 2
1398	assert "Traceback" not in result.output
1399
1400	def test_no_traceback_invalid_format(self, tmp_path: pathlib.Path) -> None:
1401	root = _make_repo(tmp_path)
1402	result = _invoke(root, "gc", "--format", "toml")
1403	assert "Traceback" not in result.output
1404
1405	def test_no_traceback_age_zero(self, tmp_path: pathlib.Path) -> None:
1406	root = _make_repo(tmp_path)
1407	result = _invoke(root, "gc", "--age", "0")
1408	assert "Traceback" not in result.output
1409
1410	def test_age_max_boundary_accepted(self, tmp_path: pathlib.Path) -> None:
1411	"""age=36500 (100 years) is within the valid range."""
1412	root = _make_repo(tmp_path)
1413	result = _invoke(root, "gc", "--age", "36500", "--json")
1414	assert result.exit_code == 0
1415
1416	def test_age_exceeds_max_exits_1(self, tmp_path: pathlib.Path) -> None:
1417	"""age=36501 exceeds valid range — must exit 1 cleanly."""
1418	root = _make_repo(tmp_path)
1419	result = _invoke(root, "gc", "--age", "36501")
1420	assert result.exit_code == 1
1421	assert "Traceback" not in result.output
1422
1423
1424	# ===========================================================================
1425	# TestRerereGcStress — 3 tests
1426	# ===========================================================================
1427
1428
1429	class TestRerereGcStress:
1430	def test_100_stale_entries_removed(self, tmp_path: pathlib.Path) -> None:
1431	"""100 stale preimage-only entries all removed in one gc call."""
1432	import hashlib as _hl
1433	root = _make_repo(tmp_path)
1434	for i in range(100):
1435	fp = conflict_fingerprint(
1436	_hl.sha256(f"gc-a{i}".encode()).hexdigest(),
1437	_hl.sha256(f"gc-b{i}".encode()).hexdigest(),
1438	)
1439	_write_stale_preimage(root, fp, age_days=90, path=f"src/f{i}.py")
1440	raw = _parse_json(_invoke(root, "gc", "--age", "1", "--json"))
1441	assert raw["removed"] == 100
1442	assert list_records(root) == []
1443
1444	def test_50_stale_50_fresh_only_stale_removed(self, tmp_path: pathlib.Path) -> None:
1445	"""50 stale + 50 fresh → only stale removed."""
1446	import hashlib as _hl
1447	root = _make_repo(tmp_path)
1448	for i in range(50):
1449	fp = conflict_fingerprint(
1450	_hl.sha256(f"gc-s{i}".encode()).hexdigest(),
1451	_hl.sha256(f"gc-t{i}".encode()).hexdigest(),
1452	)
1453	_write_stale_preimage(root, fp, age_days=90, path=f"src/s{i}.py")
1454	for i in range(50):
1455	fp = conflict_fingerprint(
1456	_hl.sha256(f"gc-u{i}".encode()).hexdigest(),
1457	_hl.sha256(f"gc-v{i}".encode()).hexdigest(),
1458	)
1459	_write_preimage(root, fp, path=f"src/u{i}.py")
1460	raw = _parse_json(_invoke(root, "gc", "--age", "30", "--json"))
1461	assert raw["removed"] == 50
1462	assert len(list_records(root)) == 50
1463
1464	def test_50_stale_50_resolved_only_stale_removed(self, tmp_path: pathlib.Path) -> None:
1465	"""50 stale preimage-only + 50 stale resolved → only preimage-only removed."""
1466	import hashlib as _hl
1467	root = _make_repo(tmp_path)
1468	for i in range(50):
1469	fp = conflict_fingerprint(
1470	_hl.sha256(f"gc-p{i}".encode()).hexdigest(),
1471	_hl.sha256(f"gc-q{i}".encode()).hexdigest(),
1472	)
1473	_write_stale_preimage(root, fp, age_days=90, path=f"src/p{i}.py")
1474	for i in range(50):
1475	fp = conflict_fingerprint(
1476	_hl.sha256(f"gc-r{i}".encode()).hexdigest(),
1477	_hl.sha256(f"gc-w{i}".encode()).hexdigest(),
1478	)
1479	_write_stale_preimage(root, fp, age_days=90, path=f"src/r{i}.py")
1480	_write_resolution(root, fp)
1481	raw = _parse_json(_invoke(root, "gc", "--age", "1", "--json"))
1482	assert raw["removed"] == 50
1483	assert len(list_records(root)) == 50
1484
1485
1486	# ---------------------------------------------------------------------------
1487	# Integration — clear subcommand
1488	# ---------------------------------------------------------------------------
1489
1490
1491	class TestClearIntegration:
1492	def test_clear_yes_removes_all(self, tmp_path: pathlib.Path) -> None:
1493	root = _make_repo(tmp_path)
1494	_write_preimage(root, _VALID_FP)
1495	result = _invoke(root, "clear", "--yes")
1496	assert result.exit_code == 0
1497	assert not list_records(root)
1498
1499	def test_clear_json_schema(self, tmp_path: pathlib.Path) -> None:
1500	root = _make_repo(tmp_path)
1501	_write_preimage(root, _VALID_FP)
1502	result = _invoke(root, "clear", "--yes", "--json")
1503	assert result.exit_code == 0
1504	raw = _parse_json(result)
1505	assert "removed" in raw
1506	removed_val = raw["removed"]
1507	assert isinstance(removed_val, int)
1508	assert removed_val == 1
1509
1510	def test_clear_empty_cache_json(self, tmp_path: pathlib.Path) -> None:
1511	root = _make_repo(tmp_path)
1512	result = _invoke(root, "clear", "--yes", "--json")
1513	assert result.exit_code == 0
1514	raw = _parse_json(result)
1515	assert raw["removed"] == 0
1516
1517
1518	# ===========================================================================
1519	# TestRerereclearExtended — 18 tests
1520	# ===========================================================================
1521
1522
1523	class TestRerereclearExtended:
1524	def test_exit_code_zero_yes_empty(self, tmp_path: pathlib.Path) -> None:
1525	"""--yes on empty cache exits 0."""
1526	root = _make_repo(tmp_path)
1527	result = _invoke(root, "clear", "--yes")
1528	assert result.exit_code == 0
1529
1530	def test_exit_code_zero_yes_with_entries(self, tmp_path: pathlib.Path) -> None:
1531	root = _make_repo(tmp_path)
1532	_write_preimage(root, _VALID_FP)
1533	result = _invoke(root, "clear", "--yes")
1534	assert result.exit_code == 0
1535
1536	def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None:
1537	result = _invoke(tmp_path, "clear", "--yes")
1538	assert result.exit_code == 2
1539
1540	def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None:
1541	root = _make_repo(tmp_path)
1542	result = _invoke(root, "clear", "--yes", "--format", "xml")
1543	assert result.exit_code == 1
1544
1545	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
1546	"""-j must produce identical output to --json."""
1547	root = _make_repo(tmp_path)
1548	r1 = _invoke(root, "clear", "--yes", "--json")
1549	r2 = _invoke(root, "clear", "--yes", "-j")
1550	assert r1.exit_code == 0 and r2.exit_code == 0
1551	assert _parse_json(r1) == _parse_json(r2)
1552
1553	def test_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None:
1554	"""JSON output must be compact — no indent=2."""
1555	root = _make_repo(tmp_path)
1556	result = _invoke(root, "clear", "--yes", "--json")
1557	assert result.exit_code == 0
1558	assert "\n" not in result.output.strip()
1559
1560	def test_json_removed_zero_when_empty(self, tmp_path: pathlib.Path) -> None:
1561	root = _make_repo(tmp_path)
1562	raw = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1563	assert raw["removed"] == 0
1564
1565	def test_json_removed_count_matches_entries(self, tmp_path: pathlib.Path) -> None:
1566	root = _make_repo(tmp_path)
1567	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
1568	_write_preimage(root, _VALID_FP, path="src/a.py")
1569	_write_preimage(root, fp2, path="src/b.py")
1570	raw = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1571	assert raw["removed"] == 2
1572
1573	def test_entries_gone_after_clear(self, tmp_path: pathlib.Path) -> None:
1574	root = _make_repo(tmp_path)
1575	_write_preimage(root, _VALID_FP)
1576	_invoke(root, "clear", "--yes")
1577	assert list_records(root) == []
1578
1579	def test_clear_is_idempotent(self, tmp_path: pathlib.Path) -> None:
1580	"""Clearing an already-empty cache returns removed=0."""
1581	root = _make_repo(tmp_path)
1582	_invoke(root, "clear", "--yes")
1583	raw = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1584	assert raw["removed"] == 0
1585
1586	def test_text_output_shows_count(self, tmp_path: pathlib.Path) -> None:
1587	root = _make_repo(tmp_path)
1588	_write_preimage(root, _VALID_FP)
1589	result = _invoke(root, "clear", "--yes")
1590	assert "1" in result.output
1591
1592	def test_text_output_empty_cache_message(self, tmp_path: pathlib.Path) -> None:
1593	"""Empty cache --yes should still exit 0 (skips prompt path entirely)."""
1594	root = _make_repo(tmp_path)
1595	result = _invoke(root, "clear", "--yes")
1596	assert result.exit_code == 0
1597
1598	def test_format_text_explicit(self, tmp_path: pathlib.Path) -> None:
1599	"""--format text is identical to default."""
1600	root = _make_repo(tmp_path)
1601	r_default = _invoke(root, "clear", "--yes")
1602	r_text = _invoke(root, "clear", "--yes", "--format", "text")
1603	assert r_default.output == r_text.output
1604
1605	def test_y_alias_same_as_yes(self, tmp_path: pathlib.Path) -> None:
1606	"""-y must behave identically to --yes."""
1607	root = _make_repo(tmp_path)
1608	_write_preimage(root, _VALID_FP)
1609	result = _invoke(root, "clear", "-y")
1610	assert result.exit_code == 0
1611	assert list_records(root) == []
1612
1613	def test_resolved_entries_also_cleared(self, tmp_path: pathlib.Path) -> None:
1614	"""Both preimage-only and resolved entries are removed."""
1615	root = _make_repo(tmp_path)
1616	fp2 = conflict_fingerprint(_SHA_B, _SHA_C)
1617	_write_preimage(root, _VALID_FP)
1618	_write_preimage(root, fp2)
1619	_write_resolution(root, _VALID_FP)
1620	raw = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1621	assert raw["removed"] == 2
1622	assert list_records(root) == []
1623
1624	def test_json_schema_field_present(self, tmp_path: pathlib.Path) -> None:
1625	raw = _parse_json(_invoke(_make_repo(tmp_path), "clear", "--yes", "--json"))
1626	assert "removed" in raw
1627	assert isinstance(raw["removed"], int)
1628
1629	def test_help_mentions_agent_quickstart(self) -> None:
1630	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "clear", "--help")
1631	assert "Agent quickstart" in result.output
1632
1633	def test_help_mentions_exit_codes(self) -> None:
1634	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "clear", "--help")
1635	assert "Exit codes" in result.output
1636
1637
1638	# ===========================================================================
1639	# TestRerereclearSecurity — 6 tests
1640	# ===========================================================================
1641
1642
1643	class TestRerereclearSecurity:
1644	def test_no_json_outside_repo(self, tmp_path: pathlib.Path) -> None:
1645	"""Outside a repo must not emit JSON — exits 2."""
1646	result = _invoke(tmp_path, "clear", "--yes", "--json")
1647	assert result.exit_code == 2
1648	assert not result.output.strip().startswith("{")
1649
1650	def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path) -> None:
1651	result = _invoke(tmp_path, "clear", "--yes")
1652	assert result.exit_code == 2
1653	assert "Traceback" not in result.output
1654
1655	def test_symlink_entry_not_removed(self, tmp_path: pathlib.Path) -> None:
1656	"""A symlink inside rr-cache must not be removed by clear."""
1657	root = _make_repo(tmp_path)
1658	cache = rr_cache_dir(root)
1659	target = tmp_path / "external_dir"
1660	target.mkdir()
1661	link = cache / ("l" * 64)
1662	link.symlink_to(target)
1663	_invoke(root, "clear", "--yes")
1664	assert link.exists() # symlink survives
1665
1666	def test_symlink_not_counted_in_removed(self, tmp_path: pathlib.Path) -> None:
1667	"""Symlink entries must not appear in removed count."""
1668	root = _make_repo(tmp_path)
1669	cache = rr_cache_dir(root)
1670	target = tmp_path / "ext2"
1671	target.mkdir()
1672	link = cache / ("m" * 64)
1673	link.symlink_to(target)
1674	raw = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1675	assert raw["removed"] == 0
1676
1677	def test_no_traceback_on_invalid_format(self, tmp_path: pathlib.Path) -> None:
1678	root = _make_repo(tmp_path)
1679	result = _invoke(root, "clear", "--yes", "--format", "toml")
1680	assert "Traceback" not in result.output
1681
1682	def test_json_stdout_on_success(self, tmp_path: pathlib.Path) -> None:
1683	"""JSON output goes to stdout on success."""
1684	root = _make_repo(tmp_path)
1685	result = _invoke(root, "clear", "--yes", "--json")
1686	assert result.output.strip().startswith("{")
1687
1688
1689	# ===========================================================================
1690	# TestRerereclearStress — 3 tests
1691	# ===========================================================================
1692
1693
1694	class TestRerereclearStress:
1695	def test_100_entries_cleared(self, tmp_path: pathlib.Path) -> None:
1696	"""100 entries all removed in one clear call."""
1697	import hashlib as _hl
1698	root = _make_repo(tmp_path)
1699	for i in range(100):
1700	fp = conflict_fingerprint(
1701	_hl.sha256(f"a{i}".encode()).hexdigest(),
1702	_hl.sha256(f"b{i}".encode()).hexdigest(),
1703	)
1704	_write_preimage(root, fp, path=f"src/f{i}.py")
1705	raw = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1706	assert raw["removed"] == 100
1707	assert list_records(root) == []
1708
1709	def test_clear_then_repopulate_then_clear(self, tmp_path: pathlib.Path) -> None:
1710	"""Clear → repopulate → clear again: counts correct both times."""
1711	import hashlib as _hl
1712	root = _make_repo(tmp_path)
1713	for i in range(10):
1714	fp = conflict_fingerprint(
1715	_hl.sha256(f"c{i}".encode()).hexdigest(),
1716	_hl.sha256(f"d{i}".encode()).hexdigest(),
1717	)
1718	_write_preimage(root, fp, path=f"src/g{i}.py")
1719	raw1 = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1720	assert raw1["removed"] == 10
1721	for i in range(5):
1722	fp = conflict_fingerprint(
1723	_hl.sha256(f"e{i}".encode()).hexdigest(),
1724	_hl.sha256(f"f{i}".encode()).hexdigest(),
1725	)
1726	_write_preimage(root, fp, path=f"src/h{i}.py")
1727	raw2 = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1728	assert raw2["removed"] == 5
1729
1730	def test_mixed_preimage_and_resolved_100_entries(self, tmp_path: pathlib.Path) -> None:
1731	"""50 preimage-only + 50 resolved → removed=100."""
1732	import hashlib as _hl
1733	root = _make_repo(tmp_path)
1734	for i in range(100):
1735	fp = conflict_fingerprint(
1736	_hl.sha256(f"p{i}".encode()).hexdigest(),
1737	_hl.sha256(f"q{i}".encode()).hexdigest(),
1738	)
1739	_write_preimage(root, fp, path=f"src/k{i}.py")
1740	if i < 50:
1741	_write_resolution(root, fp)
1742	raw = _parse_json(_invoke(root, "clear", "--yes", "--json"))
1743	assert raw["removed"] == 100
1744
1745
1746	# ---------------------------------------------------------------------------
1747	# Security — CLI
1748	# ---------------------------------------------------------------------------
1749
1750
1751	class TestReRereSecurity:
1752	_ANSI = "\x1b[31mevil\x1b[0m"
1753
1754	def test_ansi_in_path_sanitized_in_status(self, tmp_path: pathlib.Path) -> None:
1755	root = _make_repo(tmp_path)
1756	fp2 = conflict_fingerprint(_SHA_A, _SHA_C)
1757	# Write meta with ANSI in path
1758	cache = rr_cache_dir(root)
1759	entry_dir = cache / fp2
1760	entry_dir.mkdir(parents=True, exist_ok=True)
1761	meta = {
1762	"fingerprint": fp2,
1763	"path": f"src/{self._ANSI}.py",
1764	"ours_id": _SHA_A, "theirs_id": _SHA_C,
1765	"domain": "code",
1766	"recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
1767	}
1768	(entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
1769	result = _invoke(root, "status")
1770	assert result.exit_code == 0
1771	assert "\x1b[" not in result.output
1772
1773	def test_ansi_in_path_sanitized_in_forget(self, tmp_path: pathlib.Path) -> None:
1774	root = _make_repo(tmp_path)
1775	_write_preimage(root, _VALID_FP)
1776	result = _invoke(root, "forget", "--fingerprint", _VALID_FP)
1777	assert result.exit_code == 0
1778	assert "\x1b[" not in result.output
1779
1780	def test_invalid_format_rejected(self, tmp_path: pathlib.Path) -> None:
1781	root = _make_repo(tmp_path)
1782	result = _invoke(root, "status", "--format", "xml")
1783	assert result.exit_code == ExitCode.USER_ERROR.value
1784
1785	def test_no_traceback_on_invalid_format(self, tmp_path: pathlib.Path) -> None:
1786	root = _make_repo(tmp_path)
1787	result = _invoke(root, "gc", "--format", "toml")
1788	assert "Traceback" not in result.output
1789
1790	def test_json_on_stdout_text_on_stderr(self, tmp_path: pathlib.Path) -> None:
1791	"""JSON output goes to stdout; error output goes to stderr."""
1792	root = _make_repo(tmp_path)
1793	_write_preimage(root, _VALID_FP)
1794	result = _invoke(root, "status", "--json")
1795	# stdout (result.output) must start with valid JSON
1796	stripped = result.output.lstrip()
1797	assert stripped.startswith("{"), f"Expected JSON on stdout: {stripped[:80]!r}"
1798
1799
1800	# ---------------------------------------------------------------------------
1801	# E2E — JSON schemas for all subcommands
1802	# ---------------------------------------------------------------------------
1803
1804
1805	class TestJsonSchemas:
1806	def test_apply_json_schema(self, tmp_path: pathlib.Path) -> None:
1807	root = _make_repo(tmp_path)
1808	result = _invoke(root, "--json")
1809	assert result.exit_code == 0
1810	raw = _parse_json(result)
1811	assert "dry_run" in raw
1812	assert "auto_resolved" in raw
1813	assert "remaining" in raw
1814
1815	def test_apply_dry_run_json(self, tmp_path: pathlib.Path) -> None:
1816	root = _make_repo(tmp_path)
1817	result = _invoke(root, "--dry-run", "--json")
1818	assert result.exit_code == 0
1819	raw = _parse_json(result)
1820	assert raw["dry_run"] is True
1821
1822	def test_record_json_schema(self, tmp_path: pathlib.Path) -> None:
1823	root = _make_repo(tmp_path)
1824	result = _invoke(root, "record", "--json")
1825	assert result.exit_code == 0
1826	raw = _parse_json(result)
1827	assert "recorded" in raw
1828	assert "skipped" in raw
1829
1830	def test_forget_json_schema(self, tmp_path: pathlib.Path) -> None:
1831	root = _make_repo(tmp_path)
1832	result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")
1833	assert result.exit_code == 0
1834	raw = _parse_json(result)
1835	assert "forgotten" in raw
1836	assert "not_found" in raw
1837
1838	def test_clear_json_schema(self, tmp_path: pathlib.Path) -> None:
1839	root = _make_repo(tmp_path)
1840	result = _invoke(root, "clear", "--yes", "--json")
1841	assert result.exit_code == 0
1842	raw = _parse_json(result)
1843	assert "removed" in raw
1844
1845	def test_gc_json_schema_fields(self, tmp_path: pathlib.Path) -> None:
1846	root = _make_repo(tmp_path)
1847	result = _invoke(root, "gc", "--json")
1848	assert result.exit_code == 0
1849	raw = _parse_json(result)
1850	assert "removed" in raw
1851
1852
1853	# ---------------------------------------------------------------------------
1854	# E2E — help text
1855	# ---------------------------------------------------------------------------
1856
1857
1858	class TestHelpText:
1859	def test_help_shows_subcommands(self) -> None:
1860	result = runner.invoke(cli, ["rerere", "--help"])
1861	assert result.exit_code == 0
1862	for sub in ("record", "status", "forget", "clear", "gc"):
1863	assert sub in result.output
1864
1865	def test_forget_help_shows_fingerprint_flag(self) -> None:
1866	result = runner.invoke(cli, ["rerere", "forget", "--help"])
1867	assert result.exit_code == 0
1868	assert "--fingerprint" in result.output
1869
1870	def test_gc_help_shows_age_flag(self) -> None:
1871	result = runner.invoke(cli, ["rerere", "gc", "--help"])
1872	assert result.exit_code == 0
1873	assert "--age" in result.output
1874
1875
1876	# ---------------------------------------------------------------------------
1877	# Stress
1878	# ---------------------------------------------------------------------------
1879
1880
1881	class TestStress:
1882	def test_list_records_10k_entries(self, tmp_path: pathlib.Path) -> None:
1883	import hashlib
1884
1885	root = _make_repo(tmp_path)
1886	cache = rr_cache_dir(root)
1887
1888	# Write 1000 valid preimage entries (10k would be slow; 1k validates the cap logic)
1889	fps: list[str] = []
1890	for i in range(1000):
1891	raw = f"{i:064d}".encode()
1892	fp = hashlib.sha256(raw).hexdigest()
1893	entry = cache / fp
1894	entry.mkdir(parents=True, exist_ok=True)
1895	meta = {
1896	"fingerprint": fp, "path": f"src/{i}.py",
1897	"ours_id": _SHA_A, "theirs_id": _SHA_B,
1898	"domain": "code",
1899	"recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
1900	}
1901	(entry / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
1902	fps.append(fp)
1903
1904	records = list_records(root)
1905	assert len(records) == 1000
1906
1907	def test_concurrent_record_load_isolated(self, tmp_path: pathlib.Path) -> None:
1908	"""Eight threads record and load rerere entries on isolated repos."""
1909	errors: list[str] = []
1910
1911	def worker(idx: int) -> None:
1912	try:
1913	repo = _make_repo(tmp_path / f"repo{idx}")
1914	fp = conflict_fingerprint(_SHA_A, f"{idx:064d}")
1915	entry_dir = rr_cache_dir(repo) / fp
1916	entry_dir.mkdir(parents=True, exist_ok=True)
1917	meta = {
1918	"fingerprint": fp, "path": f"src/{idx}.py",
1919	"ours_id": _SHA_A, "theirs_id": f"{idx:064d}",
1920	"domain": "code",
1921	"recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
1922	}
1923	(entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
1924	record = load_record(repo, fp)
1925	if record is None:
1926	errors.append(f"Thread {idx}: load_record returned None")
1927	elif record.path != f"src/{idx}.py":
1928	errors.append(f"Thread {idx}: wrong path {record.path!r}")
1929	except Exception as exc:
1930	errors.append(f"Thread {idx}: {exc}")
1931
1932	threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1933	for t in threads:
1934	t.start()
1935	for t in threads:
1936	t.join()
1937
1938	assert errors == [], f"Concurrent rerere failures: {errors}"
1939
1940	def test_concurrent_gc_stale_isolated(self, tmp_path: pathlib.Path) -> None:
1941	"""Eight threads run gc_stale on isolated repos simultaneously."""
1942	errors: list[str] = []
1943
1944	def worker(idx: int) -> None:
1945	try:
1946	repo = _make_repo(tmp_path / f"repo{idx}")
1947	# Write an old entry
1948	cache = rr_cache_dir(repo)
1949	entry_dir = cache / _VALID_FP
1950	entry_dir.mkdir(parents=True, exist_ok=True)
1951	old_time = (
1952	datetime.datetime.now(datetime.timezone.utc)
1953	- datetime.timedelta(days=100)
1954	).isoformat()
1955	meta = {
1956	"fingerprint": _VALID_FP, "path": "src/a.py",
1957	"ours_id": _SHA_A, "theirs_id": _SHA_B,
1958	"domain": "code", "recorded_at": old_time,
1959	}
1960	(entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8")
1961	removed = gc_stale(repo, age_days=1)
1962	if removed != 1:
1963	errors.append(f"Thread {idx}: expected 1 removed, got {removed}")
1964	except Exception as exc:
1965	errors.append(f"Thread {idx}: {exc}")
1966
1967	threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1968	for t in threads:
1969	t.start()
1970	for t in threads:
1971	t.join()
1972
1973	assert errors == [], f"Concurrent gc_stale failures: {errors}"
1974
1975
1976	# ===========================================================================
1977	# TestRerereRecordExtended — 18 tests
1978	# ===========================================================================
1979
1980	# ---------------------------------------------------------------------------
1981	# Helpers for record tests
1982	# ---------------------------------------------------------------------------
1983
1984	def _write_merge_state(
1985	root: pathlib.Path,
1986	ours_commit: str,
1987	theirs_commit: str,
1988	conflict_paths: list[str],
1989	) -> None:
1990	"""Write a minimal MERGE_STATE.json for record tests."""
1991	state = {
1992	"ours_commit": ours_commit,
1993	"theirs_commit": theirs_commit,
1994	"conflict_paths": conflict_paths,
1995	"base_commit": None,
1996	"other_branch": "feat/other",
1997	}
1998	ms_path = root / ".muse" / "MERGE_STATE.json"
1999	ms_path.write_text(json.dumps(state), encoding="utf-8")
2000
2001
2002	def _write_commit_with_manifest(
2003	root: pathlib.Path,
2004	tag: str,
2005	manifest: Manifest,
2006	) -> str:
2007	"""Write a valid commit + snapshot. Returns the real content-hash commit_id."""
2008	from muse.core.snapshot import compute_commit_id, compute_snapshot_id
2009	from muse.core.store import CommitRecord, write_commit, write_snapshot, SnapshotRecord
2010	import datetime as _dt
2011
2012	snap_id = compute_snapshot_id(manifest)
2013	write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
2014	committed_at = _dt.datetime(2024, 1, 1, tzinfo=_dt.timezone.utc)
2015	commit_id = compute_commit_id([], snap_id, f"commit {tag}", committed_at.isoformat())
2016	write_commit(
2017	root,
2018	CommitRecord(
2019	commit_id=commit_id,
2020	repo_id="rerere-record-test",
2021	branch="main",
2022	snapshot_id=snap_id,
2023	message=f"commit {tag}",
2024	committed_at=committed_at,
2025	parent_commit_id=None,
2026	),
2027	)
2028	return commit_id
2029
2030
2031	def _make_record_repo(
2032	tmp_path: pathlib.Path,
2033	paths: list[str] \| None = None,
2034	ours_manifest_extra: Manifest \| None = None,
2035	theirs_manifest_extra: Manifest \| None = None,
2036	) -> tuple[pathlib.Path, str, str]:
2037	"""
2038	Create a repo with a MERGE_STATE that has real commits/snapshots.
2039
2040	Returns (root, ours_commit_id, theirs_commit_id).
2041	"""
2042	import hashlib as _hl
2043
2044	root = _make_repo(tmp_path)
2045	conflict_paths = paths or ["src/a.py"]
2046
2047	# Build manifests so each conflict path has both ours and theirs blobs.
2048	ours_manifest: Manifest = {}
2049	theirs_manifest: Manifest = {}
2050	for p in conflict_paths:
2051	ours_manifest[p] = _hl.sha256(f"ours-{p}".encode()).hexdigest()
2052	theirs_manifest[p] = _hl.sha256(f"theirs-{p}".encode()).hexdigest()
2053
2054	if ours_manifest_extra:
2055	ours_manifest.update(ours_manifest_extra)
2056	if theirs_manifest_extra:
2057	theirs_manifest.update(theirs_manifest_extra)
2058
2059	ours_id = _write_commit_with_manifest(root, "ours", ours_manifest)
2060	theirs_id = _write_commit_with_manifest(root, "theirs", theirs_manifest)
2061	_write_merge_state(root, ours_id, theirs_id, conflict_paths)
2062	return root, ours_id, theirs_id
2063
2064
2065	class TestRerereRecordExtended:
2066	def test_exits_0_no_merge_in_progress(self, tmp_path: pathlib.Path) -> None:
2067	"""No MERGE_STATE → exit 0 with a 'nothing to record' message."""
2068	root = _make_repo(tmp_path)
2069	result = _invoke(root, "record")
2070	assert result.exit_code == 0
2071
2072	def test_text_nothing_to_record_message(self, tmp_path: pathlib.Path) -> None:
2073	root = _make_repo(tmp_path)
2074	result = _invoke(root, "record")
2075	assert "nothing to record" in result.output.lower()
2076
2077	def test_exits_0_with_conflicts(self, tmp_path: pathlib.Path) -> None:
2078	root, _, _ = _make_record_repo(tmp_path)
2079	result = _invoke(root, "record")
2080	assert result.exit_code == 0
2081
2082	def test_text_shows_recorded_path(self, tmp_path: pathlib.Path) -> None:
2083	root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"])
2084	result = _invoke(root, "record")
2085	assert result.exit_code == 0
2086	assert "src/a.py" in result.output
2087
2088	def test_rr_cache_entry_created(self, tmp_path: pathlib.Path) -> None:
2089	"""After record, a meta.json must exist in rr-cache."""
2090	root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"])
2091	_invoke(root, "record")
2092	cache = rr_cache_dir(root)
2093	entries = list(cache.iterdir()) if cache.exists() else []
2094	assert len(entries) >= 1
2095	assert any((e / "meta.json").exists() for e in entries if e.is_dir())
2096
2097	def test_idempotent_double_record(self, tmp_path: pathlib.Path) -> None:
2098	"""Recording twice must not create duplicate entries."""
2099	root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"])
2100	_invoke(root, "record")
2101	cache_before = set(rr_cache_dir(root).iterdir())
2102	_invoke(root, "record")
2103	cache_after = set(rr_cache_dir(root).iterdir())
2104	assert cache_before == cache_after
2105
2106	def test_json_schema_no_conflicts(self, tmp_path: pathlib.Path) -> None:
2107	"""No merge → JSON has recorded=[] skipped=[]."""
2108	root = _make_repo(tmp_path)
2109	result = _invoke(root, "record", "--json")
2110	assert result.exit_code == 0
2111	data = _parse_json(result)
2112	assert data["recorded"] == []
2113	assert data["skipped"] == []
2114
2115	def test_json_schema_with_conflicts(self, tmp_path: pathlib.Path) -> None:
2116	root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"])
2117	result = _invoke(root, "record", "--json")
2118	assert result.exit_code == 0
2119	data = _parse_json(result)
2120	assert "src/a.py" in data["recorded"]
2121	assert isinstance(data["skipped"], list)
2122
2123	def test_json_is_single_line(self, tmp_path: pathlib.Path) -> None:
2124	"""JSON output must be compact — no indent=2."""
2125	root = _make_repo(tmp_path)
2126	result = _invoke(root, "record", "--json")
2127	assert result.exit_code == 0
2128	assert "\n" not in result.output.strip()
2129
2130	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
2131	"""-j must produce identical output to --json."""
2132	root = _make_repo(tmp_path)
2133	r1 = _invoke(root, "record", "--json")
2134	r2 = _invoke(root, "record", "-j")
2135	assert r1.exit_code == 0 and r2.exit_code == 0
2136	assert _parse_json(r1) == _parse_json(r2)
2137
2138	def test_skipped_path_one_side_deleted(self, tmp_path: pathlib.Path) -> None:
2139	"""A path present only in one manifest appears in skipped, not recorded."""
2140	root = _make_repo(tmp_path)
2141	# ours has the file, theirs does not (deletion conflict).
2142	ours_id = _write_commit_with_manifest(root, "ours-del", {"src/deleted.py": _SHA_A})
2143	theirs_id = _write_commit_with_manifest(root, "theirs-del", {})
2144	_write_merge_state(root, ours_id, theirs_id, ["src/deleted.py"])
2145	result = _invoke(root, "record", "--json")
2146	assert result.exit_code == 0
2147	data = _parse_json(result)
2148	assert "src/deleted.py" in data["skipped"]
2149	assert "src/deleted.py" not in data["recorded"]
2150
2151	def test_multiple_paths_all_recorded(self, tmp_path: pathlib.Path) -> None:
2152	"""Multiple conflict paths all appear in recorded."""
2153	paths = ["src/a.py", "src/b.py", "src/c.py"]
2154	root, _, _ = _make_record_repo(tmp_path, paths=paths)
2155	result = _invoke(root, "record", "--json")
2156	assert result.exit_code == 0
2157	data = _parse_json(result)
2158	for p in paths:
2159	assert p in data["recorded"]
2160
2161	def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None:
2162	"""No .muse directory → exit 2."""
2163	result = _invoke(tmp_path, "record")
2164	assert result.exit_code == 2
2165
2166	def test_help_mentions_agent_quickstart(self) -> None:
2167	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "record", "--help")
2168	assert "Agent quickstart" in result.output
2169
2170	def test_help_mentions_exit_codes(self) -> None:
2171	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "record", "--help")
2172	assert "Exit codes" in result.output
2173
2174	def test_help_mentions_json_schema(self) -> None:
2175	result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "record", "--help")
2176	assert "JSON output schema" in result.output
2177
2178	def test_format_text_explicit(self, tmp_path: pathlib.Path) -> None:
2179	"""--format text must produce the same output as the default."""
2180	root = _make_repo(tmp_path)
2181	r_default = _invoke(root, "record")
2182	r_text = _invoke(root, "record", "--format", "text")
2183	assert r_default.output == r_text.output
2184
2185	def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None:
2186	root = _make_repo(tmp_path)
2187	result = _invoke(root, "record", "--format", "xml")
2188	assert result.exit_code == 1
2189
2190
2191	# ===========================================================================
2192	# TestRerereRecordSecurity — 6 tests
2193	# ===========================================================================
2194
2195
2196	class TestRerereRecordSecurity:
2197	def test_ansi_path_stripped_text(self, tmp_path: pathlib.Path) -> None:
2198	"""ANSI escape in a conflict path must be stripped in text output."""
2199	root = _make_repo(tmp_path)
2200	evil_path = "src/\x1b[31mevil\x1b[0m.py"
2201	ours_id = _write_commit_with_manifest(root, "ours-ansi", {evil_path: _SHA_A})
2202	theirs_id = _write_commit_with_manifest(root, "theirs-ansi", {evil_path: _SHA_B})
2203	_write_merge_state(root, ours_id, theirs_id, [evil_path])
2204	result = _invoke(root, "record")
2205	assert result.exit_code == 0
2206	assert "\x1b" not in result.output
2207
2208	def test_ansi_path_stripped_json(self, tmp_path: pathlib.Path) -> None:
2209	"""ANSI escape in a conflict path must be stripped in JSON output."""
2210	root = _make_repo(tmp_path)
2211	evil_path = "src/\x1b[31mevil\x1b[0m.py"
2212	ours_id = _write_commit_with_manifest(root, "ours-ansi-j", {evil_path: _SHA_A})
2213	theirs_id = _write_commit_with_manifest(root, "theirs-ansi-j", {evil_path: _SHA_B})
2214	_write_merge_state(root, ours_id, theirs_id, [evil_path])
2215	result = _invoke(root, "record", "--json")
2216	assert result.exit_code == 0
2217	assert "\x1b" not in result.output
2218
2219	def test_skipped_ansi_path_stripped_json(self, tmp_path: pathlib.Path) -> None:
2220	"""ANSI in a skipped path must also be stripped in JSON output."""
2221	root = _make_repo(tmp_path)
2222	evil_path = "src/\x1b[32mskipped\x1b[0m.py"
2223	# Only ours has the file → skipped.
2224	ours_id = _write_commit_with_manifest(root, "ours-skip-ansi", {evil_path: _SHA_A})
2225	theirs_id = _write_commit_with_manifest(root, "theirs-skip-ansi", {})
2226	_write_merge_state(root, ours_id, theirs_id, [evil_path])
2227	result = _invoke(root, "record", "--json")
2228	assert result.exit_code == 0
2229	assert "\x1b" not in result.output
2230
2231	def test_no_json_on_error(self, tmp_path: pathlib.Path) -> None:
2232	"""Incomplete MERGE_STATE error must not emit JSON to stdout."""
2233	import hashlib as _hl
2234	root = _make_repo(tmp_path)
2235	# Write a MERGE_STATE that references non-existent commits.
2236	_write_merge_state(root, "x" * 64, "y" * 64, ["src/a.py"])
2237	result = _invoke(root, "record", "--json")
2238	# Exit 1 (incomplete manifests), no JSON blob on stdout.
2239	assert not result.output.strip().startswith("{")
2240
2241	def test_outside_repo_no_traceback(self, tmp_path: pathlib.Path) -> None:
2242	"""Running outside a repo must exit cleanly with code 2, no traceback."""
2243	result = _invoke(tmp_path, "record")
2244	assert result.exit_code == 2
2245	assert "Traceback" not in result.output
2246
2247	def test_control_char_in_path_stripped(self, tmp_path: pathlib.Path) -> None:
2248	"""Control characters other than ANSI also stripped from text output."""
2249	root = _make_repo(tmp_path)
2250	evil_path = "src/evil\x07.py"
2251	ours_id = _write_commit_with_manifest(root, "ours-ctrl", {evil_path: _SHA_A})
2252	theirs_id = _write_commit_with_manifest(root, "theirs-ctrl", {evil_path: _SHA_B})
2253	_write_merge_state(root, ours_id, theirs_id, [evil_path])
2254	result = _invoke(root, "record")
2255	assert result.exit_code == 0
2256	assert "\x07" not in result.output
2257
2258
2259	# ===========================================================================
2260	# TestRerereRecordStress — 3 tests
2261	# ===========================================================================
2262
2263
2264	class TestRerereRecordStress:
2265	def test_50_conflict_paths(self, tmp_path: pathlib.Path) -> None:
2266	"""50 simultaneous conflict paths are all recorded."""
2267	paths = [f"src/file_{i}.py" for i in range(50)]
2268	root, _, _ = _make_record_repo(tmp_path, paths=paths)
2269	result = _invoke(root, "record", "--json")
2270	assert result.exit_code == 0
2271	data = _parse_json(result)
2272	assert len(data["recorded"]) == 50
2273
2274	def test_idempotent_100_calls(self, tmp_path: pathlib.Path) -> None:
2275	"""Recording the same conflict 100 times creates only one cache entry."""
2276	root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"])
2277	for _ in range(100):
2278	r = _invoke(root, "record")
2279	assert r.exit_code == 0
2280	entries = [e for e in rr_cache_dir(root).iterdir() if e.is_dir()]
2281	assert len(entries) == 1
2282
2283	def test_mixed_recorded_and_skipped(self, tmp_path: pathlib.Path) -> None:
2284	"""Mix of recordable and skipped paths: counts must sum to total conflicts."""
2285	import hashlib as _hl
2286	root = _make_repo(tmp_path)
2287	good_paths = [f"src/good_{i}.py" for i in range(10)]
2288	skip_paths = [f"src/skip_{i}.py" for i in range(5)]
2289	all_paths = good_paths + skip_paths
2290
2291	ours_manifest = {p: _hl.sha256(f"o-{p}".encode()).hexdigest() for p in good_paths}
2292	theirs_manifest = {p: _hl.sha256(f"t-{p}".encode()).hexdigest() for p in good_paths}
2293	# skip_paths have no theirs entry → will be skipped.
2294	for p in skip_paths:
2295	ours_manifest[p] = _hl.sha256(f"o-{p}".encode()).hexdigest()
2296
2297	ours_id = _write_commit_with_manifest(root, "ours-mixed", ours_manifest)
2298	theirs_id = _write_commit_with_manifest(root, "theirs-mixed", theirs_manifest)
2299	_write_merge_state(root, ours_id, theirs_id, all_paths)
2300
2301	result = _invoke(root, "record", "--json")
2302	assert result.exit_code == 0
2303	data = _parse_json(result)
2304	assert len(data["recorded"]) == 10
2305	assert len(data["skipped"]) == 5

test_cmd_rerere_hardening.py file-level

`test_cmd_rerere_hardening.py` file-level