"""재개 입구의 세 문. ★유료 0. Codex BLOCK (2026-09-02) — ①`record_code_transition` **호출자 0** — 전이 기록 없이 다른 코드로 재개할 수 있었고, `canary_run.json` 을 덮어 첫 판 증거가 사라졌다 ②`cap=0` 거절을 `run_steps_batch` 가 삼키면 **옛 completed manifest** 때문에 되읽기가 통과해 성공처럼 지나갔다 ③중앙 조사에 일반 non-fanout `cap=1` 을 줘서 다섯 갈래를 못 쟀다 """ from __future__ import annotations import ast import inspect import json import textwrap import pytest from tools.grounding_audit import canary_pipeline as cp from tools.grounding_audit import canary_run as cr APPROVED = dict(mode="v2_chunk", fixture="period_episode", target="reference_acquisition") @pytest.fixture def run_root(tmp_path, monkeypatch): monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) d = tmp_path / "canary_abc123abc123" d.mkdir(parents=True, exist_ok=True) return d def _run_json(root, tip): (root / "canary_run.json").write_text( json.dumps({"code": {"tip": tip, "clean": True}}), encoding="utf-8") class TestTheTransitionGateIsActuallyCalled: """★★★만들어 놓고 **아무도 안 부르던** 자리 — 이제 `run()` 이 부른다.""" def test_run_calls_it_before_anything_paid(self): src = inspect.getsource(cr.run) i = src.index("assert_resume_transition(") for later in ("_save()", "ci.create_database", "cbs.bootstrap(", "cp.run_pipeline("): assert i < src.index(later), f"{later} 가 재개 문보다 앞이다" def test_a_new_run_needs_no_transition(self, run_root): run_root.rmdir() got = cr.assert_resume_transition("abc123abc123", cr.scenario(**APPROVED)) assert got["resume"] is False def test_the_same_code_needs_no_transition(self, run_root, monkeypatch): tip = cr.git_tip()["tip"] _run_json(run_root, tip) monkeypatch.setattr(cr, "git_tip", lambda: {"tip": tip, "clean": True, "dirty_files": []}) got = cr.assert_resume_transition("abc123abc123", cr.scenario(**APPROVED)) assert got["same_code"] is True def test_a_new_tip_without_a_transition_stops(self, run_root, monkeypatch): _run_json(run_root, "a" * 40) monkeypatch.setattr(cr, "git_tip", lambda: {"tip": "b" * 40, "clean": True, "dirty_files": []}) with pytest.raises(cr.ScopeMismatch) as e: cr.assert_resume_transition("abc123abc123", cr.scenario(**APPROVED)) assert "유효한 전이 기록이 정확히 하나" in str(e.value) def test_a_dirty_tree_stops(self, run_root, monkeypatch): _run_json(run_root, "a" * 40) monkeypatch.setattr(cr, "git_tip", lambda: {"tip": "b" * 40, "clean": False, "dirty_files": ["x.py"]}) with pytest.raises(cr.ScopeMismatch): cr.assert_resume_transition("abc123abc123", cr.scenario(**APPROVED)) class TestSupersedesIsActuallyResolved: """★★잘못 적은 전이 줄이 「있다」로 세어지면 안 된다.""" def _two(self, run_root, sc): cp.append_event(run_root, { "kind": cp.EVENT_CODE_TRANSITION, "from_tip": "틀린것", "to_tip": "b" * 40, "why": "지어낸 것"}) first = [r for r in cp.read_attempts(run_root) if r.get("kind") == cp.EVENT_CODE_TRANSITION][-1] cp.append_event(run_root, { "kind": cp.EVENT_CODE_TRANSITION, "supersedes": first["event_id"], "from_tip": "a" * 40, "to_tip": "b" * 40, "why": "바로잡음", "scenario": {k: sc[k] for k in cr._SCENARIO_AXES}, "approved": cr.approved_for(sc["mode"]), "dimensions": cr.fixture_dimensions(sc["fixture"]), "locks": dict(cr.LOCKED_CONTRACT)}) def test_two_events_in_the_same_second_are_told_apart(self, run_root): """★★`recorded_kst` 는 초 단위라 열쇠가 못 된다 — 실측으로 겪었다.""" a = cp.append_event(run_root, {"kind": cp.EVENT_CODE_TRANSITION, "to_tip": "x"}) b = cp.append_event(run_root, {"kind": cp.EVENT_CODE_TRANSITION, "to_tip": "x"}) assert a["event_id"] != b["event_id"] def test_only_the_superseding_row_counts(self, run_root, monkeypatch): sc = cr.scenario(**APPROVED) _run_json(run_root, "a" * 40) self._two(run_root, sc) monkeypatch.setattr(cr, "git_tip", lambda: {"tip": "b" * 40, "clean": True, "dirty_files": []}) got = cr.assert_resume_transition("abc123abc123", sc) assert got["from_tip"] == "a" * 40, "★대신된 줄을 읽었다" def test_a_transition_with_other_terms_stops(self, run_root, monkeypatch): sc = cr.scenario(**APPROVED) _run_json(run_root, "a" * 40) cp.append_event(run_root, { "kind": cp.EVENT_CODE_TRANSITION, "from_tip": "a" * 40, "to_tip": "b" * 40, "why": "x", "scenario": {"mode": "legacy", "fixture": "canary_one_scene", "target": "scene_detail"}, "approved": cr.approved_for("legacy"), "dimensions": cr.fixture_dimensions("canary_one_scene"), "locks": dict(cr.LOCKED_CONTRACT)}) monkeypatch.setattr(cr, "git_tip", lambda: {"tip": "b" * 40, "clean": True, "dirty_files": []}) with pytest.raises(cr.ScopeMismatch) as e: cr.assert_resume_transition("abc123abc123", sc) assert "지금과 다르다" in str(e.value) class TestTheFirstRunIsNotOverwritten: def test_a_resume_writes_its_own_file(self): src = inspect.getsource(cr.run) assert "canary_resume_" in src assert 'out_name = "canary_run.json" if not resumed' in src class TestASwallowedDenialIsNotSuccess: """★★★`cap=0` 에서 막혔는데 옛 manifest 때문에 통과하던 것.""" def test_the_pipeline_stops_on_a_denied_reuse(self, tmp_path, monkeypatch): from app.core.research_call_budget import reserve_current_research_call from app.core.config import settings from app.services import analysis_dispatch_service as ads from tools.grounding_audit import canary_cost_table as ct monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) monkeypatch.setattr(settings, "projects_dir", str(tmp_path / "projects")) plan = ct.execution_plan(config=cr.fixture_config("legacy"), target="episode_summary") first = plan["applied_metered"][0] def _mark(step, status="completed"): d = (tmp_path / "projects" / "p" / "checkpoints" / "episodes" / "e" / step) d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text(json.dumps({"status": status}), encoding="utf-8") for s in plan["applied_metered"]: _mark(s) def fake(**kw): """★★앞 판에 끝났는데 **또 사려 든다** — 그리고 예외를 삼킨다.""" try: reserve_current_research_call(source="잘못된 재실행") except Exception: # noqa: BLE001 pass # ★안에서 삼킨다 import unittest.mock as m with m.patch.object(ads, "run_steps_batch", fake): with pytest.raises(cp.CanaryStopped) as e: cp.run_pipeline(run_id="abc123abc123", project_id="p", episode_id="e", plan=plan, caps={s: 9 for s in plan["applied_metered"]}, emergency_counted=70, approved_image_calls=0, live=True) assert "체크포인트 재사용이 안 됐다" in str(e.value) got = json.loads((tmp_path / "canary_abc123abc123" / "pipeline_run.json").read_text()) assert got["budget"]["used"] == 0, "★provider 로 나갔다" assert first not in got["reused_from_earlier_attempt"], ( "★막힌 것을 되쓴 것으로 셌다") class TestTheCentralStepGetsItsOwnCap: """★★일반 non-fanout `1` 로는 다섯 갈래를 못 잰다.""" def test_the_cap_comes_from_the_declared_targets(self): got = cr.central_logical_cap("period_episode") assert got["target_count"] == 5 and got["declared_target_count"] == 5 assert got["one_round_each"] == 10 assert got["logical_cap"] == 25 # 5 × (2 × 2 + 재판정 1) def test_the_production_obligation_count_raises_the_cap_but_never_lowers_it(self, tmp_path, monkeypatch): """★실측(attempt 21c47e1b): 선언 5 · production 의무 21 → 상한 40 에 조사가 죽었다.""" got = cr.central_logical_cap("period_episode", obligations=21) assert got["target_count"] == 21 and got["logical_cap"] == 21 * 5 low = cr.central_logical_cap("period_episode", obligations=2) assert low["target_count"] == 5, "★선언보다 작게는 안 잡는다" import json as _json d = tmp_path / "projects" / "p" / "checkpoints" / "episodes" / "e" / "grounding_screen" d.mkdir(parents=True) (d / "manifest.json").write_text(_json.dumps({"data": {"counts": {"obligation": 21}}}), encoding="utf-8") monkeypatch.setattr(cr.ci, "root_dir", lambda _rid: tmp_path) assert cr.obligation_count_of("r") == 21 assert cr.obligation_count_of(None) is None def test_the_per_round_number_counts_what_production_wires(self): """★★★2026-09-02 뒤집었다 — 「함수에 있다」가 아니라 **production 이 넘기는가**로 센다. 앞 판은 `acquire_one` 소스에 `write_brief` 와 `judge` 가 보인다고 2 로 적었다. 그런데 그때 production 은 `write_brief` 를 **안 넘겼고** `if write_brief is not None` 로 건너뛰었다 — 상한이 실제보다 컸다. (그 뒤 저작기를 production 에 배선해 지금은 정말 2 다.) """ from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import reference_acquisition_rounds as rr in_fn = {n.func.id for n in ast.walk(ast.parse(textwrap.dedent( inspect.getsource(rr.acquire_one)))) if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)} # ★2026-09-03: 심판은 `_call_judge(judge, …, criteria=)` 를 거쳐 불린다 — 같은 한 번이다 assert "write_brief" in in_fn and ({"judge", "_call_judge"} & in_fn), "★함수 모양이 바뀌었다" wired = set() for n in ast.walk(ast.parse(textwrap.dedent( inspect.getsource(ca.run)))): if (isinstance(n, ast.Call) and ast.unparse(n.func).endswith("acquire_one")): wired |= {k.arg for k in n.keywords} want = 1 + (1 if "write_brief" in wired else 0) assert cr.central_calls_per_target_per_round() == want == 2 def test_the_plan_uses_it(self): built = cr.build_plan(cr.scenario(**APPROVED)) assert built["caps"]["reference_acquisition"] == 25 assert built["central_cap"]["target_count"] == 5 def test_the_three_doors_are_different_numbers(self): """★★★글/VLM · 검색 **요청** · 받는 **장수** 는 서로 다른 문이다. 앞서 「40장」으로 뭉뚱그려 드렸는데 40 은 **장수**이고 요청은 10 이다. 글 예산은 검색도 다운로드도 **못 센다**. """ from app.modules.pipeline import coarse_type_pick as ctp g = cr.central_logical_cap("period_episode") n, r = g["target_count"], g["rounds"] assert g["logical_cap"] == n * (r * 2 + 1) # 재판정 1 포함 (2026-09-02 밤) assert g["search_requests_cap"] == n * r assert g["downloads_cap"] == n * r * ctp.PER_ROUND_CAP assert g["search_requests_cap"] < g["downloads_cap"] def test_the_candidate_cap_comes_from_the_shared_constant(self): """★후보 상한을 여기 다시 안 적는다 — `era_research` 것이다.""" from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline.era_research import MAX_CANDIDATES assert ctp.PER_ROUND_CAP == MAX_CANDIDATES assert (cr.central_logical_cap("period_episode") ["candidates_per_round"] == MAX_CANDIDATES) def test_search_and_download_are_one_and_many_per_round(self): """★★계약의 근거 — 라운드마다 검색은 **한 번**, 받기는 **여럿**.""" from app.modules.pipeline import reference_acquisition_rounds as rr tree = ast.parse(textwrap.dedent(inspect.getsource(rr.acquire_one))) loops = [n for n in ast.walk(tree) if isinstance(n, ast.For)] inner = [n for n in loops if "judged" in ast.unparse(n.iter)] assert len(inner) == 1, "★후보 loop 가 하나여야 한다" in_loop = {c.func.id for c in ast.walk(inner[0]) if isinstance(c, ast.Call) and isinstance(c.func, ast.Name)} assert "download" in in_loop, "★받기가 후보 loop 안이 아니다" assert "search" not in in_loop, "★검색이 후보마다 돈다" def test_a_fixture_without_declared_targets_stops(self): with pytest.raises(cr.ScopeMismatch): cr.central_logical_cap("canary_one_scene") class TestTheRunCanMoveMoreThanOnce: """★★★전이를 **한 번밖에** 못 적던 것 (2026-09-02 실측). 앞 tip 을 늘 **첫 판 산출**에서 읽어서, `A→B` 를 적고 B 에서 재개한 뒤 결함을 하나 더 고쳐 C 를 만들면 「앞 tip 이 A 인데 B 로 알고 있다」로 섰다. 그러면 결함 하나마다 run 을 버려야 하는데, 그 run 이 들고 있는 것이 **얼마 썼는지**다. ★고치는 방향을 한 번 틀렸다: 전이 **줄을 따라가게** 만들었더니 머리가 곧 목적지가 되어 `assert_resume_transition` 의 조건 검사(시나리오· 승인선·치수·잠금)가 통째로 건너뛰어졌다 — 잠긴 시험 둘이 잡았다. 맞는 답은 **마지막으로 실제로 돈 판**의 tip 을 읽는 것이다. """ @staticmethod def _at(root, tip, seq=None): import json name = "canary_run.json" if seq is None else f"canary_resume_r_{seq}.json" (root / name).write_text(json.dumps({"code": {"tip": tip}}), encoding="utf-8") @staticmethod def _bind(monkeypatch, root): monkeypatch.setattr(cr.ci, "root_dir", lambda _rid: root) def test_the_tip_moves_when_the_run_actually_runs_again( self, tmp_path, monkeypatch): root = tmp_path / "c" root.mkdir() self._bind(monkeypatch, root) a, b, c = "a" * 40, "b" * 40, "c" * 40 self._at(root, a) assert cr.recorded_tip("x") == a # ★전이를 적기만 해서는 **안 움직인다** — 아직 그 코드로 안 돌았다 cp.append_event(root, {"kind": cp.EVENT_CODE_TRANSITION, "from_tip": a, "to_tip": b}) assert cr.recorded_tip("x") == a, "★적기만 했는데 움직였다" self._at(root, b, seq=1) # ★B 에서 실제로 돌았다 assert cr.recorded_tip("x") == b assert cr.code_lineage("x") == [a, b] self._at(root, c, seq=2) assert cr.recorded_tip("x") == c assert cr.code_lineage("x") == [a, b, c] def test_the_order_is_the_number_not_the_clock(self, tmp_path, monkeypatch): """★10 번째 재개가 2 번째보다 **뒤**다 — 글자로 세면 뒤집힌다.""" root = tmp_path / "c" root.mkdir() self._bind(monkeypatch, root) self._at(root, "a" * 40) self._at(root, "b" * 40, seq=10) self._at(root, "c" * 40, seq=2) assert cr.recorded_tip("x") == "b" * 40 def test_a_hop_that_starts_somewhere_else_is_not_valid(self, tmp_path, monkeypatch): """★★`from_tip` 을 안 보면 **아무 데서나 온 줄**이 통과한다.""" root = tmp_path / "c" root.mkdir() self._bind(monkeypatch, root) a, b, other = "a" * 40, "b" * 40, "7" * 40 self._at(root, a) cp.append_event(root, {"kind": cp.EVENT_CODE_TRANSITION, "from_tip": other, "to_tip": b}) assert cr.valid_transition("x", to_tip=b, from_tip=a) is None assert cr.valid_transition("x", to_tip=b, from_tip=other) is not None def test_the_second_hop_is_checked_the_same_way(self, tmp_path, monkeypatch): """★★두 번째 전이도 **조건 검사를 그대로 받는다** — 건너뛰지 않는다.""" root = tmp_path / "c" root.mkdir() self._bind(monkeypatch, root) sc = cr.scenario(**APPROVED) a, b, c = "a" * 40, "b" * 40, "c" * 40 self._at(root, a) self._at(root, b, seq=1) # ★이미 B 까지 돌았다 terms = {"scenario": {k: sc[k] for k in cr._SCENARIO_AXES}, "approved": cr.approved_for(sc["mode"]), "dimensions": cr.fixture_dimensions(sc["fixture"]), "locks": dict(cr.LOCKED_CONTRACT)} cp.append_event(root, {"kind": cp.EVENT_CODE_TRANSITION, "from_tip": b, "to_tip": c, **terms}) monkeypatch.setattr(cr, "git_tip", lambda: {"tip": c, "clean": True, "dirty_files": []}) got = cr.assert_resume_transition("x", sc) assert got["from_tip"] == b and got["to_tip"] == c assert got["lineage"] == [a, b] # ★조건이 다른 두 번째 전이는 **선다** (root / "attempts.jsonl").unlink(missing_ok=True) (root / "pipeline_attempts.json").unlink(missing_ok=True) bad = dict(terms) bad["locks"] = {"num_retries": 3} cp.append_event(root, {"kind": cp.EVENT_CODE_TRANSITION, "from_tip": b, "to_tip": c, **bad}) with pytest.raises(cr.ScopeMismatch, match="지금과 다르다"): cr.assert_resume_transition("x", sc)