"""★취소에 맞은 대상은 **자동 재시도 빚**이지 completed 봉인이 아니다 (Codex BLOCK 2026-09-03 05:30). 실측 4398a55dc0bb: 정지 요청이 「검색 실패」로 접혀 9줄이 raw retryable · outcome reference_unavailable · disposition acquired · why "" · 후보 0 으로 남았는데, `rows_to_rejudge` 가 후보 있는 줄만 세어 failed_count 0 · completed 로 봉인됐다 — plain resume 이 verify pass → SKIP 으로 지나갈 자리였다.""" from __future__ import annotations import pytest 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 def _row(sid, status, *, candidates=False, chosen=False, disposition=None): rounds = [{"round_no": 1, "downloaded_candidates": [{"index": 1}] if candidates else [], "decision": {"next": "retryable", "why": "다 못 봤다 — 검색 실패: 정지 요청"}}] return {"research_subject_id": sid, "status": status, "outcome": ra.acquisition_outcome(status), "disposition": disposition or ca.DISP_ACQUIRED, "why": "", "acquisition": {"status": status, "chosen": {"index": 1} if chosen else None, "rounds": rounds}} def _real_shape(): """실물 CP 의 모양: selected 7 · retryable(후보 0) 9 · not_applicable 10.""" rows = [_row(f"S{i}", ra.STATUS_SELECTED, candidates=True, chosen=True) for i in range(7)] rows += [_row(f"R{i}", ra.STATUS_RETRYABLE) for i in range(9)] rows += [{"research_subject_id": f"N{i}", "status": None, "disposition": "not_applicable", "acquisition": {}} for i in range(10)] return {"rows": rows} def _step_with(cp): st = S.__new__(S) st.step_id = "reference_acquisition"; st.project_config = {} st._get_step_run = lambda sid: {"status": "completed", "run_id": "r"} st.load_checkpoint = lambda: cp st._check_cp_mismatch = lambda c: None st._load_prev_checkpoint = lambda sid: cp st._last_execute_result = None return st class TestThePendingSplit: def test_candidates_mean_rejudge_and_none_means_research_retry(self): got = S.pending_rows({"rows": [_row("A", ra.STATUS_RETRYABLE, candidates=True), _row("B", ra.STATUS_RETRYABLE)]}) assert got["rejudge"] == 1 and got["research_retry"] == 1 and got["total"] == 2 assert got["rejudge_subjects"] == ["A"] and got["research_retry_subjects"] == ["B"] def test_terminal_and_chosen_and_unbought_rows_are_not_debt(self): rows = [_row("A", ra.STATUS_SELECTED, candidates=True, chosen=True), _row("B", ra.STATUS_NO_MATCH, candidates=True), # 다 보고 없었다 — terminal _row("C", ra.STATUS_RETRYABLE, disposition=ca.DISP_UNCONFIRMED)] # 안 산 줄 assert S.pending_rows({"rows": rows})["total"] == 0 assert S.rows_to_rejudge({"rows": [_row("D", ra.STATUS_RETRYABLE, candidates=True)]}) == 1 class TestTheSealIsRefused: def test_central_wrap_counts_both_debts_in_failed_count(self, monkeypatch): monkeypatch.setattr(ca, "ledger_coverage", lambda ob, res: {"ok": True}) monkeypatch.setattr(ca, "unfinished_rows", lambda res: []) got = S.central_wrap({"rows": []}, _real_shape()) assert got["failed_count"] == 9 d = got["data"] assert d["pending_total"] == 9 and d["research_retry_pending"] == 9 and d["rejudge_pending"] == 0 def test_verify_completion_reports_the_debt_without_asking_a_human(self): st = S.__new__(S) st._last_execute_result = {"data": _real_shape()} rep = st.verify_completion() assert rep.is_complete is False and rep.origin == "artifact_missing" assert rep.metadata["research_retry_pending"] == 9 and rep.metadata["pending_total"] == 9 assert not any("사람" in m for m in rep.missing) class TestTheProductionResumeReopensTheSealedCheckpoint: def test_a_completed_cp_with_nine_candidateless_retryable_rows_is_rerun_self(self): """★끝점 (Codex ④): 실물과 같은 completed CP 를 production `StepRunner._evaluate_resume_decision(resume)` 에 넣으면 RERUN_SELF(origin artifact_missing — cleanup 없음 · 산 것을 지우지 않는다).""" from app.core.step_runner import ResumeAction cp = {"status": "completed", "config_hash": "h", "schema_version": 1, "data": _real_shape()} decision = _step_with(cp)._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF, decision assert decision.origin == "artifact_missing" def test_a_clean_cp_is_still_skipped(self): from app.core.step_runner import ResumeAction rows = [_row(f"S{i}", ra.STATUS_SELECTED, candidates=True, chosen=True) for i in range(3)] cp = {"status": "completed", "config_hash": "h", "schema_version": 1, "data": {"rows": rows}} assert _step_with(cp)._evaluate_resume_decision(mode="resume").action == ResumeAction.SKIP class TestTheJournalReplaysOkAndRebuysIncomplete: def test_ok_entries_replay_and_incomplete_entries_rebuy(self, tmp_path): """★Codex ④ 「7 selected provider 0 · 9 만 다시 산다」의 장부 쪽 근거 — `ok` 는 자리를 되쓰고(reserve False) `incomplete` 는 빈 자리(reserve True).""" import inspect from app.modules.pipeline import grounding_chunk_journal as cj params = inspect.signature(cj.ChunkJournal.__init__).parameters kw = {"contract": {"wiring": "t"}} if "contract" in params else {} j = cj.ChunkJournal(tmp_path / "journal.json", **kw) j.put("ok1", {"status": "selected"}, status=cj.STATUS_OK) j.put("inc1", {"status": "retryable"}, status=cj.STATUS_OK) j.settle_incomplete("inc1", why="마지막 라운드가 빈손", by="test") assert j.reserve("ok1", cap=10) is False # 되쓴다 — provider 0 assert j.reserve("inc1", cap=10) is True # 빈 자리 — 그 대상만 다시 산다 class TestANewStopRequestDoesNotWriteACompletedCheckpoint: def test_a_cancel_inside_the_central_run_propagates_out_of_the_step(self, monkeypatch): """★끝점 (Codex ⑤): cd210f76 뒤 정지 요청은 검색 실패로 접히지 않고 올라온다 → 스텝이 CP 를 쓰지 않는다.""" from app.core.errors import AppError st = S.__new__(S) st.project_config = {} monkeypatch.setattr(S, "central_obligations", lambda self: {"rows": []}) wrapped = [] monkeypatch.setattr(S, "central_wrap", staticmethod(lambda ob, res: wrapped.append(1))) for name in ("_journal", "_cap", "_workdir", "_search", "_download", "_judge", "_write_brief"): monkeypatch.setattr(S, name, lambda self, *a, **k: None) def _run(*a, **k): raise AppError(code="step.cancelled", message="정지 요청", status_code=409) monkeypatch.setattr(S, "central_result", staticmethod(_run)) with pytest.raises(AppError) as e: st._central() assert e.value.code == "step.cancelled" and wrapped == []