"""★★★판정이 **못 여는 경로**를 받던 것 (2026-09-02 유료 canary ① 실측).

검색 10회·받기 38장을 **다 사고 나서** 12대상 전부가
`판정 실패: [Errno 2] No such file or directory: 'projects/…/rs_…_r1_01.png'`
로 떨어졌다 (`selected_count: 0`). 파일은 `PROJECTS_DIR` 아래 멀쩡히 있었고,
기록용 **상대** 경로를 그대로 판정에 넘겨 판정이 **프로세스 cwd 기준**으로
열었을 뿐이다.

★기록은 옮겨 다녀야 하니 상대로 두고, **여는 쪽**에는 절대 경로를 준다.
그러니 두 가지를 같이 잠근다 — 판정이 받은 것은 열리고, 남는 기록은 상대다.
"""
from __future__ import annotations

from pathlib import Path

from app.modules.pipeline import reference_acquisition_rounds as rr

TARGET = {
    "subject_id": "rs_test01",
    "directive_native": "무엇을 찾을지 적힌 지시문",
    "terms_native": ["질의 하나", "질의 둘"],
    "language_lock_native": "그 나라 말로만",
}


def _search(**_kw):
    # ★칸 이름은 **production `_rows_from` 이 읽는 그대로**다 — 지어내면
    #  후보가 0 이 되어 이 시험이 아무것도 안 잠근다 (실제로 그랬다).
    return {"queries": [["질의 하나"]],
            "images": [{"image_url": f"https://example.invalid/{i}.jpg",
                        "thumbnail_url": "",
                        "source_website_url": f"https://example.invalid/{i}",
                        "caption": f"사진 {i}"} for i in (1, 2, 3)]}


def _download(_url, dest: Path, _thumb=""):
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 32)
    return True


class TestWhatTheJudgeGetsCanBeOpened:
    def test_every_path_the_judge_receives_is_a_real_file(self, tmp_path):
        seen: list = []

        def _judge(cands):
            seen.extend(cands)
            for c in cands:
                p = Path(c["path"])
                assert p.is_absolute(), f"★상대 경로를 받았다: {c['path']}"
                assert p.is_file(), f"★못 여는 경로다: {c['path']}"
            return {}

        work = tmp_path / "refs"
        work.mkdir()
        rr.acquire_one(TARGET, workdir=work, rel_root=tmp_path,
                       search=_search, download=_download, judge=_judge,
                       rounds=1, per_round_cap=3)
        assert seen, "★판정이 아예 안 불렸다 — 이 시험이 아무것도 안 잠근다"

    def test_the_record_keeps_the_relative_path(self, tmp_path):
        """★기록은 **상대**여야 한다 — 옮기면 절대 경로는 죽는다."""
        work = tmp_path / "refs"
        work.mkdir()
        got = rr.acquire_one(TARGET, workdir=work, rel_root=tmp_path,
                             search=_search, download=_download,
                             judge=lambda _c: {}, rounds=1, per_round_cap=3)
        cands = got["rounds"][0]["downloaded_candidates"]
        assert cands, "★받은 것이 하나도 안 적혔다"
        for c in cands:
            assert not Path(c["path"]).is_absolute(), (
                f"★기계 경로가 기록에 샜다: {c['path']}")
            assert (tmp_path / c["path"]).is_file()

    def test_the_judge_input_is_not_what_gets_recorded(self, tmp_path):
        """★★둘이 **같은 dict 가 아니어야** 한다 — 하나를 고치면 다른 쪽이
        조용히 따라간다."""
        judged: list = []
        work = tmp_path / "refs"
        work.mkdir()
        got = rr.acquire_one(TARGET, workdir=work, rel_root=tmp_path,
                             search=_search, download=_download,
                             judge=lambda c: judged.extend(c) or {},
                             rounds=1, per_round_cap=3)
        rec = {c["path"] for c in got["rounds"][0]["downloaded_candidates"]}
        assert {c["path"] for c in judged} & rec == set(), (
            "★판정에 준 경로가 기록에 그대로 들어갔다")


def _record(tmp_path, n=3, sha=True):
    """받아 둔 판 하나 — 사진은 **실제로** 있다."""
    import hashlib

    work = tmp_path / "refs"
    work.mkdir(exist_ok=True)
    cands = []
    for i in range(1, n + 1):
        f = work / f"rs_x_r1_{i:02d}.png"
        f.write_bytes(b"\x89PNG\r\n\x1a\n" + bytes([i]) * 32)
        c = {"index": i, "path": f"refs/{f.name}",
             "url": f"https://example.invalid/{i}.jpg",
             "source_website_url": "", "caption": ""}
        if sha:
            c["sha256"] = hashlib.sha256(f.read_bytes()).hexdigest()
        cands.append(c)
    return {"subject_id": "rs_x", "status": "retryable", "chosen": None,
            "contract": rr.ROUNDS_CONTRACT_VERSION,
            "rounds": [{"round_no": 1, "downloaded_candidates": cands,
                        "decision": {"next": "retryable",
                                     "why": "판정 실패: …"}}]}


def _pick(index):
    """심판 하나가 `index` 를 고른 모양. ★production 파서가 읽는 그대로."""
    from app.modules.pipeline.era_research import PICK_MODEL

    # ★칸 이름은 production `_parse_one` 이 읽는 그대로 — `verdicts` 다.
    #  지어내면 심판이 통째로 제외되고, 「안 골랐다」가 「못 골랐다」로
    #  섞여 시험이 아무것도 안 잠근다 (실제로 그랬다).
    return lambda cands: {PICK_MODEL: {"verdicts": [
        {"index": c["index"],
         "object_type_match": "yes" if c["index"] == index else "no",
         "visible": c["index"] == index} for c in cands]}}


class TestRejudgeUsesOnlyWhatWeAlreadyBought:
    def test_it_cannot_even_reach_search_or_download(self):
        """★★계약: replay 경로에서 검색·받기 **callable 이 안 닿는다**."""
        import inspect

        sig = inspect.signature(rr.rejudge_cached)
        assert "search" not in sig.parameters
        assert "download" not in sig.parameters
        src = inspect.getsource(rr.rejudge_cached)
        for word in ("search(", "download(", "_rows_from", "dedupe_candidates"):
            assert word not in src, f"★replay 가 {word} 에 닿는다"

    def test_the_judge_gets_openable_paths_and_the_record_stays_relative(
            self, tmp_path):
        seen: list = []

        def _judge(cands):
            seen.extend(cands)
            for c in cands:
                assert Path(c["path"]).is_file()
                assert Path(c["path"]).is_absolute()
            return _pick(2)(cands)

        got = rr.rejudge_cached(_record(tmp_path), root=tmp_path,
                                judge=_judge)
        assert len(seen) == 3
        assert got["status"] == "selected"
        assert got["chosen"]["index"] == 2
        assert got["chosen_path"] == "refs/rs_x_r1_02.png"
        assert got["downstream_blocked"] is False
        blob = __import__("json").dumps(got, ensure_ascii=False)
        assert str(tmp_path) not in blob, "★기계 경로가 산출에 샜다"

    def test_nothing_eligible_is_recorded_as_no_match(self, tmp_path):
        got = rr.rejudge_cached(_record(tmp_path), root=tmp_path,
                                judge=_pick(0))
        assert got["chosen"] is None
        from app.modules.pipeline import reference_acquisition as ra

        assert got["status"] in (ra.STATUS_NO_MATCH, ra.STATUS_RETRYABLE)
        assert got["replay"]["chosen_index"] == 0


class TestItStopsBeforeSpendingWhenTheFilesAreNotWhatWeSaved:
    """★★★VLM 을 부르기 **전에** 선다 — 돈을 쓰고 나서 못 여는 것이
    바로 앞 판에 난 일이다."""

    @staticmethod
    def _never(_c):
        raise AssertionError("★VLM 이 불렸다 — 문 앞에서 서야 한다")

    def test_a_missing_file_stops(self, tmp_path):
        import pytest

        rec = _record(tmp_path)
        (tmp_path / "refs" / "rs_x_r1_02.png").unlink()
        with pytest.raises(rr.CandidateMissing):
            rr.rejudge_cached(rec, root=tmp_path, judge=self._never)

    def test_a_changed_file_stops(self, tmp_path):
        import pytest

        rec = _record(tmp_path)
        (tmp_path / "refs" / "rs_x_r1_03.png").write_bytes(b"different")
        with pytest.raises(rr.CandidateChanged):
            rr.rejudge_cached(rec, root=tmp_path, judge=self._never)

    def test_a_path_outside_the_root_stops(self, tmp_path):
        import pytest

        rec = _record(tmp_path)
        rec["rounds"][0]["downloaded_candidates"][0]["path"] = \
            "../outside/x.png"
        with pytest.raises(rr.CandidateOutsideRoot):
            rr.rejudge_cached(rec, root=tmp_path, judge=self._never)

    def test_an_absolute_path_in_the_record_stops(self, tmp_path):
        import pytest

        rec = _record(tmp_path)
        rec["rounds"][0]["downloaded_candidates"][0]["path"] = \
            str(tmp_path / "refs" / "rs_x_r1_01.png")
        with pytest.raises(rr.CandidateOutsideRoot):
            rr.rejudge_cached(rec, root=tmp_path, judge=self._never)

    def test_a_round_with_no_candidates_stops(self, tmp_path):
        import pytest

        rec = _record(tmp_path)
        rec["rounds"][0]["downloaded_candidates"] = []
        with pytest.raises(rr.NothingToRejudge):
            rr.rejudge_cached(rec, root=tmp_path, judge=self._never)


class TestTheReplayHasItsOwnIdentity:
    """★원 구매 신원과 **따로** — 같은 replay 재개는 VLM 0회여야 한다."""

    def _ident(self, tmp_path, **ch):
        rec = _record(tmp_path)
        cands = rec["rounds"][0]["downloaded_candidates"]
        if ch.get("swap"):
            cands[0], cands[1] = cands[1], cands[0]
        if ch.get("url"):
            cands[0]["url"] = ch["url"]
        if ch.get("bytes"):
            (tmp_path / "refs" / "rs_x_r1_01.png").write_bytes(ch["bytes"])
            cands[0].pop("sha256", None)
        got = rr.resolve_cached_candidates(cands, root=tmp_path)
        import json
        return json.dumps(rr.replay_identity_inputs(rec, got),
                          ensure_ascii=False, sort_keys=True)

    def test_the_same_pictures_in_the_same_order_are_the_same_replay(
            self, tmp_path):
        assert self._ident(tmp_path) == self._ident(tmp_path)

    def test_order_url_and_bytes_each_change_it(self, tmp_path):
        base = self._ident(tmp_path)
        assert self._ident(tmp_path, swap=True) != base, "★순서가 안 접혔다"
        assert self._ident(tmp_path, url="https://other.invalid/z.jpg") \
            != base, "★출처가 안 접혔다"
        assert self._ident(tmp_path, bytes=b"\x89PNG\r\n\x1a\nZZZZ") != base, \
            "★사진 내용이 안 접혔다"


class TestTheCentralAssemblyOnlyAssembles:
    """★★조립부는 **고르고 부르기만** 한다 — 되보는 법은 `rr` 한 곳이 안다."""

    @staticmethod
    def _journal(tmp_path):
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal

        return ChunkJournal(tmp_path / "j.json")

    @staticmethod
    def _rows(tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        return [{"research_subject_id": "rs_x", "identity": "ident_x",
                 "disposition": ca.DISP_ACQUIRED, "status": "retryable",
                 "acquisition": _record(tmp_path)},
                {"research_subject_id": "rs_y", "identity": "ident_y",
                 "disposition": ca.DISP_SKIPPED, "status": "skipped",
                 "acquisition": {}}]

    def test_it_never_touches_search_or_download(self):
        import inspect

        from app.modules.pipeline import grounding_central_acquisition as ca

        sig = inspect.signature(ca.rejudge_rows)
        assert "search" not in sig.parameters
        assert "download" not in sig.parameters
        src = inspect.getsource(ca.rejudge_rows)
        assert "acquire_one" not in src, "★replay 가 구매 경로에 닿는다"

    def test_it_judges_once_and_a_second_run_judges_zero(self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        calls = {"n": 0}

        def _judge(cands):
            calls["n"] += 1
            return _pick(1)(cands)

        j = self._journal(tmp_path)
        got = ca.rejudge_rows(self._rows(tmp_path), journal=j, root=tmp_path,
                              judge=_judge, cap=5)
        assert calls["n"] == 1
        assert got["replayed"] == 1 and got["skipped"] == 1
        assert len(got["rows"]) == 2, "★안 되본 줄이 목록에서 사라졌다"
        assert got["rows"][0]["status"] == "selected"
        assert got["rows"][0]["replay_identity"].startswith(
            ca.REPLAY_IDENTITY_PREFIX)

        # ★★같은 replay 를 다시 — **VLM 0회**
        again = ca.rejudge_rows(got["rows"], journal=j, root=tmp_path,
                                judge=_judge, cap=5)
        assert calls["n"] == 1, "★재개가 또 판정했다"
        assert again["replayed"] == 0

    def test_the_replay_identity_never_collides_with_the_purchase_one(
            self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        j = self._journal(tmp_path)
        ca.rejudge_rows(self._rows(tmp_path), journal=j, root=tmp_path,
                        judge=_pick(1), cap=5)
        keys = list(j.entries)
        # ★★빈손이면 아래 두 축을 **그냥 지나간다** — 먼저 있는지 본다
        assert len(keys) == 1, f"★장부에 적힌 것이 {len(keys)}개다: {keys}"
        assert keys[0].startswith(ca.REPLAY_IDENTITY_PREFIX), keys
        assert "ident_x" not in keys

    def test_the_cap_stops_without_widening_and_keeps_what_is_done(
            self, tmp_path):
        """★상한에 닿으면 **자동으로 안 넓히고**, 이미 끝난 것은 남는다."""
        from app.modules.pipeline import grounding_central_acquisition as ca

        rows = self._rows(tmp_path)
        second = tmp_path / "refs2"
        second.mkdir()
        rec2 = _record(tmp_path)
        rows.insert(1, {"research_subject_id": "rs_z", "identity": "ident_z",
                        "disposition": ca.DISP_ACQUIRED, "status": "retryable",
                        "acquisition": rec2})
        rows[1]["acquisition"]["subject_id"] = "rs_z"

        calls = {"n": 0}

        def _judge(c):
            calls["n"] += 1
            return _pick(1)(c)

        got = ca.rejudge_rows(rows, journal=self._journal(tmp_path),
                              root=tmp_path, judge=_judge, cap=1)
        assert calls["n"] == 1, "★상한 1 인데 더 판정했다"
        assert got["cap_reached"] is True
        assert len(got["rows"]) == 3
        assert got["rows"][0]["status"] == "selected", "★끝난 것이 사라졌다"
        assert "replay_skipped" in got["rows"][1]


class TestTheStepPicksTheReplayOnlyWhenThereIsSomethingToRedo:
    """★되볼 것이 없으면 **아무 일도 안 한다** — VLM 0."""

    @staticmethod
    def _S():
        from app.core.steps.reference_acquisition_step import (
            ReferenceAcquisitionStep as S)
        return S

    def test_a_bought_row_with_pictures_and_no_choice_counts(self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        res = {"rows": [{"disposition": ca.DISP_ACQUIRED,
                         "acquisition": _record(tmp_path)}]}
        assert self._S().rows_to_rejudge(res) == 1

    def test_an_already_chosen_row_does_not_count(self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        rec = _record(tmp_path)
        rec["chosen"] = {"index": 1, "path": "refs/rs_x_r1_01.png"}
        res = {"rows": [{"disposition": ca.DISP_ACQUIRED, "acquisition": rec}]}
        assert self._S().rows_to_rejudge(res) == 0

    def test_a_row_that_was_never_bought_does_not_count(self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        res = {"rows": [{"disposition": ca.DISP_SKIPPED,
                         "acquisition": _record(tmp_path)},
                        {"disposition": ca.DISP_NOT_APPLICABLE,
                         "acquisition": {}}]}
        assert self._S().rows_to_rejudge(res) == 0

    def test_a_row_with_no_pictures_does_not_count(self):
        from app.modules.pipeline import grounding_central_acquisition as ca

        res = {"rows": [{"disposition": ca.DISP_ACQUIRED,
                         "acquisition": {"rounds": [
                             {"round_no": 1, "downloaded_candidates": []}]}}]}
        assert self._S().rows_to_rejudge(res) == 0

    def test_the_central_branch_only_calls_it_when_there_is_work(self):
        """★★조립부가 **무조건** 부르면 되볼 것이 없어도 판정이 돈다."""
        import inspect

        src = inspect.getsource(self._S()._central)
        assert "rows_to_rejudge(res)" in src
        assert "if self.rows_to_rejudge" in src


class TestABoughtButUnjudgedPassIsNotFinished:
    """★★★**산 것을 못 본 판은 끝난 판이 아니다** (2026-09-02 실측).

    유료 판이 검색 10회·받기 38장을 다 사고 12대상 전부 판정에서 죽었는데
    `failed_count: 0` 으로 **completed** 가 닫혔다. 그래서 재개가 `SKIP` 으로
    지나가고 — 받아 둔 사진을 다시 볼 길이 없었다.

    ★`step_runner` 의 접는 법: `failed==0 → completed` ·
    `completed>0 → partial` · 아니면 `failed`. 우리가 원하는 것은
    **`partial`** 이다 — 재개가 `RERUN_SELF` 로 다시 돌되 **cleanup 을 안
    한다**(`force` 는 `cleanup_artifacts`+`clear_checkpoint` 를 부른다).
    """

    @staticmethod
    def _wrap(rows):
        from app.core.steps.reference_acquisition_step import (
            ReferenceAcquisitionStep as S)

        # ★의무 장부의 줄과 산출의 줄은 **같아야** 한다 — 덮개 검사가 그것을
        #  본다. 빈 장부로 부르면 「의무가 산출에서 사라졌다」로 선다.
        ob = {"rows": [{"research_subject_id": r["research_subject_id"]}
                       for r in rows]}
        return S.central_wrap(ob, {"rows": rows, "purchases": {},
                                   "dispositions": {}})

    @staticmethod
    def _status(got):
        c = got["completed_count"]
        f = got["failed_count"]
        return "completed" if f == 0 else ("partial" if c > 0 else "failed")

    def test_bought_pictures_nobody_looked_at_make_it_partial(self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        got = self._wrap([{"research_subject_id": "rs_x",
                           "disposition": ca.DISP_ACQUIRED,
                           "outcome": "reference_unavailable",
                           "acquisition": _record(tmp_path)}])
        assert got["failed_count"] == 1
        assert got["data"]["rejudge_pending"] == 1
        assert self._status(got) == "partial", "★재개가 그냥 지나간다"

    def test_a_pass_that_looked_and_found_nothing_is_completed(self,
                                                               tmp_path):
        """★「다 보고 없었다」는 **끝난 것**이다 — 다시 안 본다."""
        from app.modules.pipeline import grounding_central_acquisition as ca

        rec = _record(tmp_path)
        rec["chosen"] = None
        rec["rounds"][0]["downloaded_candidates"] = []
        # ★뒤집음 (Codex 2026-09-03 05:30): 「다 보고 없었다」는 production 이 `no_match_after_retry`(terminal) 로 적는다
        #  (rounds.py:595). `retryable` 은 「다 못 봤다」(정지·검색 실패)라 끝난 것이 아니다.
        rec["status"] = "no_match_after_retry"
        got = self._wrap([{"research_subject_id": "rs_x",
                           "disposition": ca.DISP_ACQUIRED,
                           "outcome": "reference_unavailable",
                           "acquisition": rec}])
        assert got["failed_count"] == 0
        assert self._status(got) == "completed"

    def test_a_pass_that_could_not_look_is_a_retry_debt(self, tmp_path):
        """★후보 없이 `retryable` 로 끝난 줄(정지·검색 실패)은 **자동 재시도 빚** — 재개가 그 대상만 다시 산다."""
        from app.modules.pipeline import grounding_central_acquisition as ca

        rec = _record(tmp_path)
        rec["chosen"] = None
        rec["rounds"][0]["downloaded_candidates"] = []
        assert rec["status"] == "retryable"
        got = self._wrap([{"research_subject_id": "rs_x",
                           "disposition": ca.DISP_ACQUIRED,
                           "outcome": "reference_unavailable",
                           "acquisition": rec}])
        assert got["failed_count"] == 1 and got["data"]["research_retry_pending"] == 1
        assert self._status(got) == "partial"

    def test_a_pass_that_chose_is_completed(self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        rec = _record(tmp_path)
        rec["chosen"] = {"index": 1, "path": "refs/rs_x_r1_01.png"}
        got = self._wrap([{"research_subject_id": "rs_x",
                           "disposition": ca.DISP_ACQUIRED,
                           "outcome": "selected", "acquisition": rec}])
        assert got["failed_count"] == 0
        assert self._status(got) == "completed"

    def test_it_never_becomes_failed(self, tmp_path):
        """★`failed` 면 하류가 통째로 막힌다 — 우리가 원하는 것이 아니다."""
        from app.modules.pipeline import grounding_central_acquisition as ca

        rows = [{"research_subject_id": f"rs_{i}",
                 "disposition": ca.DISP_ACQUIRED,
                 "outcome": "reference_unavailable",
                 "acquisition": _record(tmp_path)} for i in range(5)]
        got = self._wrap(rows)
        assert got["completed_count"] > 0
        assert self._status(got) == "partial"


class TestTheStepTellsResumeItIsNotDone:
    """★★★재개가 `SKIP` 으로 지나가면 받아 둔 사진을 다시 볼 길이 없다.

    `status='completed'` + cp clean + **verify pass** → `SKIP`
    (`step_runner.py:683`). 그러니 verify 가 「아직 못 본 것이 있다」를
    말해야 `RERUN_SELF` 로 다시 돈다.
    ★`origin='artifact_missing'` 이라 **cleanup 이 안 일어난다** — `force`
    로 가면 받아 둔 사진과 장부를 지운다.
    """

    @staticmethod
    def _step(cp_data):
        from app.core.steps.reference_acquisition_step import (
            ReferenceAcquisitionStep as S)

        s = S.__new__(S)
        s._load_prev_checkpoint = lambda _sid: {"data": cp_data}
        return s

    def test_pending_rejudge_makes_verify_incomplete(self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        got = self._step({"rows": [{"disposition": ca.DISP_ACQUIRED,
                                    "acquisition": _record(tmp_path)}]}
                         ).verify_completion()
        assert got.is_complete is False
        assert got.origin == "artifact_missing", (
            "★`contract_drift`/`invariant_drift` 로 가면 cleanup 이 붙는다")
        assert got.metadata["rejudge_pending"] == 1

    def test_a_finished_pass_verifies_clean(self, tmp_path):
        from app.modules.pipeline import grounding_central_acquisition as ca

        rec = _record(tmp_path)
        rec["chosen"] = {"index": 1, "path": "refs/rs_x_r1_01.png"}
        got = self._step({"rows": [{"disposition": ca.DISP_ACQUIRED,
                                    "acquisition": rec}]}).verify_completion()
        assert got.is_complete is True and got.origin == "clean"

    def test_no_checkpoint_verifies_clean(self):
        """★아직 안 돈 스텝을 「못 봤다」로 만들지 않는다."""
        s = self._step({})
        assert s.verify_completion().is_complete is True


class TestTheReplayHasItsOwnLedger:
    """★★★재판정을 구매 장부에 적었더니 `reserve` 가 **이미 산 12건**을 제
    상한에서 빼서, 12대상 중 **4개만** 되보고 멈췄다 (2026-09-02 실측).

    상한은 「이 판이 몇 번 부르나」인데 앞 판의 구매가 그것을 먹었다.
    문이 다르면 장부도 다르다. 그리고 원 구매 줄은 한 글자도 안 건드린다.
    """

    def test_the_two_journals_are_different_files(self, tmp_path,
                                                  monkeypatch):
        from app.core.config import settings
        from app.core.steps.reference_acquisition_step import (
            ReferenceAcquisitionStep as S)

        monkeypatch.setattr(settings, "projects_dir", str(tmp_path))
        s = S.__new__(S)
        s.project_id, s.episode_id = "p", "e"
        buy = s._journal()
        rep = s._journal(S.REPLAY_JOURNAL)
        assert buy.path != rep.path
        assert rep.path.name == "journal_replay.json"

    def test_the_central_branch_uses_the_replay_ledger(self):
        import inspect

        from app.core.steps.reference_acquisition_step import (
            ReferenceAcquisitionStep as S)

        src = inspect.getsource(S._central)
        assert "self._journal(self.REPLAY_JOURNAL)" in src, (
            "★재판정이 구매 장부를 쓰면 앞 판의 구매가 상한을 먹는다")

    def test_a_full_run_rejudges_every_pending_row(self, tmp_path):
        """★★12개를 넣으면 12개를 다 되본다 — 앞 판은 4개에서 멈췄다."""
        from app.modules.pipeline import grounding_central_acquisition as ca
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal

        buy = ChunkJournal(tmp_path / "journal.json")
        rows = []
        for i in range(12):
            rec = _record(tmp_path)
            rec["subject_id"] = f"rs_{i}"
            buy.put(f"ident_{i}", {"already": "bought"})
            rows.append({"research_subject_id": f"rs_{i}",
                         "identity": f"ident_{i}",
                         "disposition": ca.DISP_ACQUIRED,
                         "acquisition": rec})
        calls = {"n": 0}

        def _judge(c):
            calls["n"] += 1
            return _pick(1)(c)

        got = ca.rejudge_rows(
            rows, journal=ChunkJournal(tmp_path / "journal_replay.json"),
            root=tmp_path, judge=_judge, cap=12)
        assert calls["n"] == 12, f"★{calls['n']}개에서 멈췄다"
        assert got["replayed"] == 12 and got["cap_reached"] is False
        # ★원 구매 장부는 **한 글자도** 안 바뀐다
        assert len(buy.entries) == 12
        assert all(not k.startswith(ca.REPLAY_IDENTITY_PREFIX)
                   for k in buy.entries)


class TestAlreadyJudgedRowsAreNotJudgedAgain:
    """★★★「다 보고 없었다」를 **끝난 것**으로 안 세면, 재개마다 그 줄을
    다시 판정한다 (실측 2026-09-02: `rejudge_pending` 이 5 에서 안 줄었다).

    같은 것을 계속 사는 것이고, 스텝이 영원히 `partial` 이다.
    """

    def test_no_match_after_retry_is_terminal(self, tmp_path):
        from app.core.steps.reference_acquisition_step import (
            ReferenceAcquisitionStep as S)
        from app.modules.pipeline import grounding_central_acquisition as ca
        from app.modules.pipeline import reference_acquisition as ra

        rec = _record(tmp_path)
        rec["status"] = ra.STATUS_NO_MATCH
        res = {"rows": [{"disposition": ca.DISP_ACQUIRED,
                         "status": ra.STATUS_NO_MATCH, "acquisition": rec}]}
        assert S.rows_to_rejudge(res) == 0
        assert ra.is_terminal(ra.STATUS_NO_MATCH) is True
        assert ra.is_terminal(ra.STATUS_RETRYABLE) is False

    def test_the_central_assembly_skips_it_too(self, tmp_path):
        """★★고르는 규칙이 **두 곳**이면 한쪽만 고쳐진다."""
        from app.modules.pipeline import grounding_central_acquisition as ca
        from app.modules.pipeline import reference_acquisition as ra
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal

        rec = _record(tmp_path)
        rec["status"] = ra.STATUS_NO_MATCH
        calls = {"n": 0}
        got = ca.rejudge_rows(
            [{"research_subject_id": "rs_x", "identity": "i",
              "disposition": ca.DISP_ACQUIRED, "status": ra.STATUS_NO_MATCH,
              "acquisition": rec}],
            journal=ChunkJournal(tmp_path / "jr.json"), root=tmp_path,
            judge=lambda c: calls.__setitem__("n", calls["n"] + 1) or {},
            cap=5)
        assert calls["n"] == 0, "★다 보고 없었던 줄을 또 판정했다"
        assert got["replayed"] == 0 and got["skipped"] == 1

    def test_a_retryable_row_is_still_picked_up(self, tmp_path):
        """★음성 대조 — 「못 봤다」는 여전히 되본다."""
        from app.core.steps.reference_acquisition_step import (
            ReferenceAcquisitionStep as S)
        from app.modules.pipeline import grounding_central_acquisition as ca
        from app.modules.pipeline import reference_acquisition as ra

        rec = _record(tmp_path)
        rec["status"] = ra.STATUS_RETRYABLE
        res = {"rows": [{"disposition": ca.DISP_ACQUIRED,
                         "status": ra.STATUS_RETRYABLE, "acquisition": rec}]}
        assert S.rows_to_rejudge(res) == 1