"""★★★재판정 줄이 **구매 장부의 회계에 섞이던 것** (Codex BLOCK 2026-09-02). 재판정을 처음에 구매 장부에 적었다. 원 줄을 지우지 않고 재판정 장부로 **복사**했지만, 복사만으로는 회계가 안 돌아온다 — `bought()` 가 같은 epoch 의 `replay:` 4줄까지 세어 구매 lane 이 12 가 아니라 **16** 으로 읽혔다. 지우는 것은 금지다(역사가 사라진다). **덧붙여 정정**한다. """ from __future__ import annotations import json import pytest from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, TransferBroken) def _lanes(tmp_path, buys=12, replays=4, copied=True): buy = ChunkJournal(tmp_path / "journal.json") rep = ChunkJournal(tmp_path / "journal_replay.json") for i in range(buys): buy.put(f"ident_{i}", {"bought": i}) for i in range(replays): rid = f"{ca.REPLAY_IDENTITY_PREFIX}{i:024d}" buy.put(rid, {"replayed": i}) # ★잘못 적힌 자리 if copied: rep.put(rid, {"replayed": i}) # ★복사만 해 뒀다 return buy, rep class TestTheJournalDoesNotKnowWhatAReplayIs: """★★`ChunkJournal` 전역에 `replay:` 뜻을 **안 박는다** (Codex).""" def test_no_lane_word_in_the_general_journal(self): import inspect from app.modules.pipeline import grounding_chunk_journal as gj # ★★★**글자로 막으면 그 금지를 적은 주석이 걸린다** — 실제로 걸렸다. # AST 로 보고 docstring 을 벗긴 뒤, **실행되는 문자열**만 본다. import ast tree = ast.parse(inspect.getsource(gj)) docs = set() for n in ast.walk(tree): if isinstance(n, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): d = ast.get_docstring(n, clean=False) if d: docs.add(d) live = [n.value for n in ast.walk(tree) if isinstance(n, ast.Constant) and isinstance(n.value, str) and n.value not in docs] bad = [x for x in live if "replay" in x.lower()] assert bad == [], ( f"★일반 장부가 특정 lane 을 안다: {bad} — 다음 lane 이 생기면 " f"또 고친다") def test_it_only_records_that_a_row_is_not_its_own(self, tmp_path): j = ChunkJournal(tmp_path / "j.json") j.put("a", {"x": 1}) rec = j.transfer_out("a", moved_to="다른 장부", why="까닭") assert rec["identity"] == "a" and rec["moved_to"] == "다른 장부" assert j.bought() == 0 assert j.get("a") is None, "★넘긴 줄을 이 lane 이 되쓰면 안 된다" assert "a" in j.entries, "★원 줄을 지웠다 — 역사가 사라진다" def test_a_transfer_of_a_row_it_does_not_have_stops(self, tmp_path): j = ChunkJournal(tmp_path / "j.json") with pytest.raises(TransferBroken): j.transfer_out("없는줄", moved_to="x", why="y") def test_a_broken_transfer_record_stops_on_load(self, tmp_path): p = tmp_path / "j.json" j = ChunkJournal(p) j.put("a", {"x": 1}) j.transfer_out("a", moved_to="다른 장부", why="까닭") d = json.loads(p.read_text(encoding="utf-8")) d["transfers"][0].pop("moved_to") p.write_text(json.dumps(d, ensure_ascii=False), encoding="utf-8") with pytest.raises(TransferBroken): ChunkJournal(p) def test_a_transfer_pointing_at_a_missing_row_stops_on_load(self, tmp_path): p = tmp_path / "j.json" j = ChunkJournal(p) j.put("a", {"x": 1}) j.transfer_out("a", moved_to="다른 장부", why="까닭") d = json.loads(p.read_text(encoding="utf-8")) d["calls"] = [] p.write_text(json.dumps(d, ensure_ascii=False), encoding="utf-8") with pytest.raises(TransferBroken): ChunkJournal(p) class TestTheBoundaryOwnsTheLaneMeaning: def test_it_gives_twelve_and_ten(self, tmp_path): """★★★끝점 — 실물 회계가 **구매 12 · 재판정 10** 이다.""" buy, rep = _lanes(tmp_path, buys=12, replays=4) for i in range(4, 10): # ★뒤에 제대로 적힌 6건 rep.put(f"{ca.REPLAY_IDENTITY_PREFIX}{i:024d}", {"replayed": i}) assert buy.bought() == 16, "★고치기 전에는 16 으로 읽힌다" got = ca.reconcile_lanes(buy, rep) assert got["acquisition_effective_bought"] == 12 assert got["replay_effective_bought"] == 10 assert got["double_counted"] == 0 assert len(got["moved_to_replay"]) == 4 def test_it_survives_a_restart(self, tmp_path): """★정정이 **파일에 남아야** 다음 판도 12 로 읽는다.""" buy, rep = _lanes(tmp_path) ca.reconcile_lanes(buy, rep) again = ChunkJournal(tmp_path / "journal.json") assert again.bought() == 12 assert again.transferred_out() == 4 def test_running_it_twice_changes_nothing(self, tmp_path): buy, rep = _lanes(tmp_path) a = ca.reconcile_lanes(buy, rep) b = ca.reconcile_lanes(buy, rep) assert a["acquisition_effective_bought"] == 12 assert b["acquisition_effective_bought"] == 12 assert b["moved_to_replay"] == [], "★두 번째 판이 또 옮겼다" def test_it_refuses_when_the_row_is_not_in_the_other_lane(self, tmp_path): """★★옮긴 곳에 없는데 회계에서 빼면 **그 판의 답이 어디에도 없다**.""" buy, rep = _lanes(tmp_path, copied=False) with pytest.raises(ca.LaneAccountingBroken, match="재판정 장부에"): ca.reconcile_lanes(buy, rep) assert buy.bought() == 16, "★막았는데 회계를 건드렸다" def test_a_row_counted_in_both_lanes_stops(self, tmp_path): buy, rep = _lanes(tmp_path, buys=1, replays=0) rep.put("ident_0", {"replayed": 0}) with pytest.raises(ca.LaneAccountingBroken, match="두 lane"): ca.lane_accounting(buy, rep) class TestTheStepReconcilesBeforeItJudges: def test_the_central_rejudge_calls_it_first(self): import ast import inspect import textwrap from app.core.steps.reference_acquisition_step import ( ReferenceAcquisitionStep as S) src = textwrap.dedent(inspect.getsource(S.central_rejudge)) tree = ast.parse(src) order = [n.lineno for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "attr", "") in ("reconcile_lanes", "rejudge_rows")] names = [n.func.attr for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "attr", "") in ("reconcile_lanes", "rejudge_rows")] assert names, "★둘 다 안 부른다" pairs = sorted(zip(order, names)) assert pairs[0][1] == "reconcile_lanes", ( "★판정부터 하면 상한이 틀린 채로 산다")