"""사이드카가 **CP 에 없는 칸**을 읽어 실패하던 것 (2026-09-02 · attempt aec5ce986c4c · 유료 30). production `detail_steps` 는 `chosen.content_sha256` 을 읽었지만 `reference_acquisition` CP 의 chosen 에는 그 칸이 없다(caption·index·path·source_website_url·url). 사람 판정과 canary probe 는 파일 bytes 로 세어 통과 → 도구가 production 과 다른 입력으로 쟀다. 이제 셋이 `row_content_sha256` 하나를 쓴다. 시험은 **실제 CP 줄 모양**으로 돈다. """ from __future__ import annotations import ast import hashlib import inspect import textwrap from pathlib import Path import pytest from app.core.steps import detail_steps as ds from app.modules.pipeline import grounding_fidelity_review as fr from app.modules.pipeline import grounding_sidecar_writer as sw from app.modules.pipeline import reference_acquisition as ra from tools.grounding_audit import canary_payload_probe as pp CANARY_EP = Path("/Users/manta/Documents/Projects/TheRoad-I1/artifact/canary_69e821758f3d/projects/8e2e65b7-e910-4b12-b081-c23de0affab5/checkpoints/episodes/9e64c302-82e6-4407-b560-8c70c2ab7192") def _real_shaped_row(fid: str, rel: str) -> dict: """★실제 CP 줄과 같은 chosen 키 — sha 칸이 **없다**.""" return {"research_subject_id": f"{fid}#detail", "identity": f"acq-{fid}", "disposition": "acquired", "status": ra.STATUS_SELECTED, "outcome": ra.STATUS_SELECTED, "why": "", "why_unbought": None, "grounding_fidelity": {"state": ra.FIDELITY_VERIFIED}, "acquisition": {"chosen": {"caption": "", "index": 1, "path": rel, "source_website_url": "", "url": "https://x.invalid/a"}, "chosen_path": rel}, "ledger_row": {"research_subject_id": f"{fid}#detail", "owner_type": "location_part", "final_id": fid, "parent_final_id": None, "purpose": "detail", "covers": [fid]}} class TestTheHelperReadsTheBytesNotAMissingField: def test_sha_is_of_the_file_bytes(self, tmp_path): rel = "projects/p/references/x.png" (tmp_path / rel).parent.mkdir(parents=True) (tmp_path / rel).write_bytes(b"\x89PNG\r\n\x1a\nBYTES") row = _real_shaped_row("LP01", rel) assert "content_sha256" not in row["acquisition"]["chosen"] assert sw.row_content_sha256(row, root=tmp_path) == hashlib.sha256(b"\x89PNG\r\n\x1a\nBYTES").hexdigest() assert sw.row_file_coordinate(row, root=tmp_path) == {"source": "file", "path": str(tmp_path / rel)} def test_a_missing_file_gives_no_hash(self, tmp_path): row = _real_shaped_row("LP01", "projects/p/references/nope.png") assert sw.row_content_sha256(row, root=tmp_path) == "" def test_the_real_shaped_row_attaches_through_production_write_for_shot(self, tmp_path): """★production 사이드카 경로 — 실제 CP 줄 모양으로 붙는다.""" rel = "projects/p/references/LP01.png" (tmp_path / rel).parent.mkdir(parents=True) (tmp_path / rel).write_bytes(b"\x89PNG\r\n\x1a\nLP01") cp = {"data": {"rows": [_real_shaped_row("LP01", rel)]}} rpc = {} n = sw.write_for_shot(rpc, cp, ["LP01"], content_sha_of=lambda r: sw.row_content_sha256(r, root=tmp_path), coordinate_of=lambda r: sw.row_file_coordinate(r, root=tmp_path)) assert n == 1 from app.modules.pipeline import grounding_reference_bundle as rb m = rpc[rb.RPC_MEMBERS_KEY][0] assert m["content_sha256"] == hashlib.sha256(b"\x89PNG\r\n\x1a\nLP01").hexdigest() def test_the_old_field_read_fails_on_the_real_shape(self, tmp_path): """★양성 대조 — 옛 방식(`chosen.content_sha256` 칸)은 실측 그대로 선다.""" cp = {"data": {"rows": [_real_shaped_row("LP01", "projects/p/references/LP01.png")]}} with pytest.raises(Exception, match="내용 해시가 없다"): sw.write_for_shot({}, cp, ["LP01"], content_sha_of=lambda r: str(((r.get("acquisition") or {}).get("chosen") or {}).get("content_sha256") or ""), coordinate_of=lambda r: {"source": "file", "path": ""}) class TestAllThreeConsumersShareTheHelper: @pytest.mark.parametrize("fn,name", [ (ds._write_grounding_sidecar, "detail_steps"), (fr.central_cp_with_reviews, "fidelity_review"), (pp.measure_shot, "probe"), ]) def test_it_calls_row_content_sha256(self, fn, name): src = textwrap.dedent(inspect.getsource(fn)) assert "row_content_sha256(" in src, f"★{name} 가 helper 를 안 쓴다" tree = ast.parse(src) literals = {n.value for n in ast.walk(tree) if isinstance(n, ast.Constant) and isinstance(n.value, str)} assert "content_sha256" not in literals or name == "probe", f"★{name} 가 CP 칸을 직접 읽는다" class TestTheRealCanaryRowsAttach: def test_lp01_and_lp03_attach_from_the_real_checkpoint(self): """★실제 canary CP + 실제 파일 — production helper 로 두 참조가 붙는다.""" import json p = CANARY_EP / "reference_acquisition" / "manifest.json" if not p.is_file(): pytest.skip("canary 산출이 이 기계에 없다") cp = json.loads(p.read_text(encoding="utf-8")) root = Path("/Users/manta/Documents/Projects/TheRoad-I1/artifact/canary_69e821758f3d") # 사람 판정은 DB 에 있어 여기서는 줄의 fidelity 를 verified 로 두고 잰다(결속 경로만) for r in cp["data"]["rows"]: if r.get("research_subject_id") in ("LP01#detail", "LP03#detail"): r["grounding_fidelity"] = {"state": ra.FIDELITY_VERIFIED} rpc = {} n = sw.write_for_shot(rpc, cp, ["LP01", "LP03"], content_sha_of=lambda r: sw.row_content_sha256(r, root=root), coordinate_of=lambda r: sw.row_file_coordinate(r, root=root)) assert n == 2 class TestTheTwoCoordinatesMustAgree: """★Codex BLOCK (2026-09-02): selected 줄엔 `chosen.path` 와 `acquisition.chosen_path` 가 둘 다 있다(실측 · 같다). 한쪽만 정본으로 택하지 않는다 — 다르면 붙이기 전에 선다.""" def _row(self, a, b): row = _real_shaped_row("LP01", a) row["acquisition"]["chosen_path"] = b return row def test_equal_coordinates_give_sha_and_coordinate_of_the_same_file(self, tmp_path): rel = "projects/p/references/LP01.png" (tmp_path / rel).parent.mkdir(parents=True) (tmp_path / rel).write_bytes(b"\x89PNG\r\n\x1a\nSAME") row = self._row(rel, rel) assert sw.resolved_reference_path(row) == rel assert sw.row_content_sha256(row, root=tmp_path) == hashlib.sha256(b"\x89PNG\r\n\x1a\nSAME").hexdigest() assert sw.row_file_coordinate(row, root=tmp_path)["path"] == str(tmp_path / rel) def test_conflicting_coordinates_stop_before_attachment(self, tmp_path): """★양성 대조 — 두 좌표가 갈리면 sha 도 좌표도 안 낸다 · write_for_shot 도 선다.""" row = self._row("projects/p/references/A.png", "projects/p/references/B.png") with pytest.raises(sw.ReferenceCoordinateConflict): sw.resolved_reference_path(row) with pytest.raises(sw.ReferenceCoordinateConflict): sw.row_content_sha256(row, root=tmp_path) cp = {"data": {"rows": [row]}} with pytest.raises(sw.ReferenceCoordinateConflict): sw.write_for_shot({}, cp, ["LP01"], content_sha_of=lambda r: sw.row_content_sha256(r, root=tmp_path), coordinate_of=lambda r: sw.row_file_coordinate(r, root=tmp_path)) def test_one_side_only_is_accepted(self, tmp_path): rel = "projects/p/references/LP01.png" assert sw.resolved_reference_path(self._row(rel, "")) == rel assert sw.resolved_reference_path(self._row("", rel)) == rel def test_a_selected_row_with_no_coordinate_stops(self, tmp_path): cp = {"data": {"rows": [self._row("", "")]}} with pytest.raises(Exception, match="내용 해시가 없다"): sw.write_for_shot({}, cp, ["LP01"], content_sha_of=lambda r: sw.row_content_sha256(r, root=tmp_path), coordinate_of=lambda r: sw.row_file_coordinate(r, root=tmp_path)) def test_the_fidelity_api_uses_the_same_helper(self): from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api._candidates) assert "row_content_sha256(" in src class TestTheProductionWrapperItselfAttaches: """★Codex 조건 (2026-09-02): 유료 결함이 난 자리는 `write_for_shot` 이 아니라 그 위의 `detail_steps._write_grounding_sidecar` 람다였다. **그 함수 자체**를 실제 canary CP· 파일로 돌린다 — 읽기 경계(`central_cp_with_reviews`)와 root 만 결정적으로 준다.""" CANARY = Path("/Users/manta/Documents/Projects/TheRoad-I1/artifact/canary_69e821758f3d") def test_write_grounding_sidecar_records_two_members_with_hash_and_coordinate(self, monkeypatch): import json from app.core.config import settings from app.modules.pipeline import grounding_reference_bundle as rb p = CANARY_EP / "reference_acquisition" / "manifest.json" if not p.is_file(): pytest.skip("canary 산출이 이 기계에 없다") cp = json.loads(p.read_text(encoding="utf-8")) for r in cp["data"]["rows"]: if r.get("research_subject_id") in ("LP01#detail", "LP03#detail"): r["grounding_fidelity"] = {"state": ra.FIDELITY_VERIFIED} # ★읽기 경계만 결정적으로 — DB 의 사람 판정 대신 위 verified 를 얹은 CP # ★HITL 0: production wrapper 는 날것 CP 를 runner 에서 읽는다 — 아래에서 runner 에 준다 monkeypatch.setattr(settings, "projects_dir", str(self.CANARY / "projects")) runner = ds.SceneDetailStep.__new__(ds.SceneDetailStep) runner.project_config = {"grounding_mode": "v2_chunk"} runner._load_prev_checkpoint = lambda sid: cp if sid == "reference_acquisition" else None card = {} n = ds._write_grounding_sidecar(runner, card, ["LP01", "LP03"]) assert n == 2, f"★production wrapper 가 {n} 만 붙였다" members = card[rb.RPC_MEMBERS_KEY] # ★멤버는 **subject 단위**다 — 같은 final_id 의 `#context`(못 구함 · 논리 의무로만 # 남음)와 `#detail`(붙음)이 나란히 적힌다. final_id 로 dict 를 만들면 덮인다 # (오늘 아침 canary ② probe 가 낸 바로 그 결함). attached = {(m["subject_final_id"], m["purpose"]): m for m in members if m.get("member_identity")} assert set(attached) == {("LP01", "detail"), ("LP03", "detail")}, sorted(attached) unattached = [m for m in members if not m.get("member_identity")] assert all(m["content_sha256"] is None and "path" not in m for m in unattached) for (fid, _purpose), m in attached.items(): path = Path(m["path"]) assert path.is_file() and str(path).startswith(str(self.CANARY)), m assert m["content_sha256"] == hashlib.sha256(path.read_bytes()).hexdigest() assert m["source"] == "file" def test_the_wrapper_stops_on_a_coordinate_conflict(self, monkeypatch): """★양성 대조 — 같은 입구, 좌표가 갈린 줄이면 부착 전에 선다.""" import json from app.core.config import settings p = CANARY_EP / "reference_acquisition" / "manifest.json" if not p.is_file(): pytest.skip("canary 산출이 이 기계에 없다") cp = json.loads(p.read_text(encoding="utf-8")) for r in cp["data"]["rows"]: if r.get("research_subject_id") == "LP01#detail": r["grounding_fidelity"] = {"state": ra.FIDELITY_VERIFIED} r["acquisition"]["chosen_path"] = "projects/other/path.png" # ★HITL 0: production wrapper 는 날것 CP 를 runner 에서 읽는다 — 아래에서 runner 에 준다 monkeypatch.setattr(settings, "projects_dir", str(self.CANARY / "projects")) runner = ds.SceneDetailStep.__new__(ds.SceneDetailStep) runner.project_config = {"grounding_mode": "v2_chunk"} runner._load_prev_checkpoint = lambda sid: cp if sid == "reference_acquisition" else None with pytest.raises(sw.ReferenceCoordinateConflict): ds._write_grounding_sidecar(runner, {}, ["LP01"])