"""장부가 **크래시를 completed 로** 적던 것. ★유료 0. Codex BLOCK (2026-09-02) — 실측: artifact/canary_69e821758f3d/pipeline_attempts.json 의 첫 attempt 는 status=completed · stopped_at=null · used=15 인데, 실제로는 `grounding_chunk` 가 AttributeError 로 죽었다. 원인: `analysis_dispatch_service.run_steps_batch` 가 **안에서 실패를 잡아** 기록하고 `None` 을 돌려주므로, 밖에서는 「예외가 안 나왔다」로 보인다. """ from __future__ import annotations import json import pytest from tools.grounding_audit import canary_pipeline as cp def _cp(tmp_path, pid, eid, step, status): d = (tmp_path / pid / "checkpoints" / "episodes" / eid / step) d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text(json.dumps({"status": status}), encoding="utf-8") @pytest.fixture def projects(tmp_path, monkeypatch): from app.core.config import settings monkeypatch.setattr(settings, "projects_dir", str(tmp_path)) return tmp_path class TestItReadsBackWhatActuallyHappened: def test_a_missing_checkpoint_is_not_finished(self, projects): with pytest.raises(cp.StepDidNotFinish) as e: cp.assert_step_finished("p", "e", "grounding_chunk") assert "없다" in str(e.value) @pytest.mark.parametrize("st", ["completed", "partial", "not_applicable"]) def test_the_finished_states_pass(self, projects, st): _cp(projects, "p", "e", "s", st) assert cp.assert_step_finished("p", "e", "s") == st @pytest.mark.parametrize("st", ["failed", "running", "blocked", ""]) def test_anything_else_stops(self, projects, st): _cp(projects, "p", "e", "s", st) with pytest.raises(cp.StepDidNotFinish): cp.assert_step_finished("p", "e", "s") def test_the_real_crashed_run_would_be_caught(self, projects): """★★실제로 죽은 판의 모양 그대로 — 앞은 끝나고 그 자리는 없다.""" for s in ("text_cleanup", "scene_segmentation", "shot_validator"): _cp(projects, "p", "e", s, "completed") for s in ("grounding_a0", "entity_all_character"): _cp(projects, "p", "e", s, "not_applicable") for s in ("text_cleanup", "grounding_a0"): assert cp.assert_step_finished("p", "e", s) with pytest.raises(cp.StepDidNotFinish): cp.assert_step_finished("p", "e", "grounding_chunk") class TestTheLedgerIsAppendOnly: """★★★앞 줄을 **고쳐 쓰지 않는다** — 판정이 바뀐 사실 자체가 증거다.""" def _seed(self, root, status="completed", used=15): cp.append_attempt(root, {"attempt_id": "aaa111", "status": status, "used": used}) def test_a_correction_is_a_new_row(self, tmp_path): self._seed(tmp_path) cp.correct_attempt(tmp_path, "aaa111", was="completed", now="crashed", why="grounding_chunk AttributeError") rows = cp.read_attempts(tmp_path) assert len(rows) == 2 assert rows[0]["status"] == "completed", "★앞 줄을 고쳤다" assert rows[1]["kind"] == cp.EVENT_CORRECTION assert rows[1]["status_now"] == "crashed" def test_the_effective_status_follows_the_correction(self, tmp_path): self._seed(tmp_path) assert cp.effective_status(tmp_path, "aaa111") == "completed" cp.correct_attempt(tmp_path, "aaa111", was="completed", now="crashed", why="x") assert cp.effective_status(tmp_path, "aaa111") == "crashed" def test_it_refuses_to_correct_what_it_misread(self, tmp_path): """★지금 상태가 내가 아는 것과 다르면 **안 고친다**.""" self._seed(tmp_path, status="stopped") with pytest.raises(cp.CanaryStopped): cp.correct_attempt(tmp_path, "aaa111", was="completed", now="crashed", why="x") def test_an_event_row_is_not_counted_as_a_run(self, tmp_path): self._seed(tmp_path, used=15) cp.correct_attempt(tmp_path, "aaa111", was="completed", now="crashed", why="x") assert cp.cumulative_used(tmp_path) == 15, "★정정 줄을 또 셌다" assert cp.open_attempts(tmp_path) == [] def test_a_code_transition_is_recorded_too(self, tmp_path): self._seed(tmp_path) got = cp.append_event(tmp_path, { "kind": cp.EVENT_CODE_TRANSITION, "from_tip": "d992a611", "to_tip": "df7a3446", "why": "producer 상속 수리"}) assert got["kind"] == cp.EVENT_CODE_TRANSITION assert cp.cumulative_used(tmp_path) == 15 def test_an_unknown_event_is_refused(self, tmp_path): with pytest.raises(cp.CanaryStopped): cp.append_event(tmp_path, {"kind": "아무거나"}) class TestAUsageCorrectionActuallyMovesTheTotal: """★★★앞 판은 정정을 「메모」로만 뒀다 — 누계가 **안 움직였다**. 그리고 세 수를 **갈라 적는다** (Codex 2026-09-02) — 구매 수 · 실제 운반 시도(모르면 범위) · 승인선에서 **빼는 수**. 모르는 것을 확정으로 안 쓴다. """ def _seed(self, root, used=0): cp.append_attempt(root, {"attempt_id": "bbb222", "status": "crashed", "used": used}) def _fix(self, root, **kw): base = dict(debit_was=0, debit_now=4, logical_calls=2, physical_lower=2, physical_upper=4, why="운반층이 안 세어졌다", evidence={"opik_spans": 2, "journal_ok": 2}) base.update(kw) return cp.correct_usage(root, "bbb222", **base) def test_the_debit_changes_the_cumulative(self, tmp_path): self._seed(tmp_path) assert cp.cumulative_used(tmp_path) == 0 self._fix(tmp_path) assert cp.cumulative_used(tmp_path) == 4, "★정정이 누계에 안 실렸다" assert cp.effective_used(tmp_path, "bbb222") == 4 def test_the_unknown_stays_unknown(self, tmp_path): """★★실제 수를 **확정으로 안 적는다** — 범위로 남는다.""" self._seed(tmp_path) got = self._fix(tmp_path) assert got["physical_attempts"] == {"exact": None, "lower": 2, "upper": 4} assert got["logical_calls"] == 2 assert got["budget_debit_now"] == 4 def test_the_original_row_is_untouched(self, tmp_path): self._seed(tmp_path) self._fix(tmp_path) rows = cp.read_attempts(tmp_path) assert rows[0]["used"] == 0, "★앞 줄을 고쳤다" def test_it_is_counted_exactly_once(self, tmp_path): self._seed(tmp_path) self._fix(tmp_path) assert cp.cumulative_used(tmp_path) == 4 real = [a for a in cp.read_attempts(tmp_path) if a.get("kind") is None] assert len(real) == 1, "★가짜 판이 생겼다" def test_a_debit_below_the_known_upper_is_refused(self, tmp_path): """★★모르는 만큼은 **크게** 빼야 넘치지 않는다.""" self._seed(tmp_path) with pytest.raises(cp.CanaryStopped) as e: self._fix(tmp_path, debit_now=2) assert "크게" in str(e.value) def test_a_correction_without_any_physical_number_is_refused(self, tmp_path): self._seed(tmp_path) with pytest.raises(cp.CanaryStopped): self._fix(tmp_path, physical_lower=None, physical_upper=None) def test_an_exact_number_is_allowed_too(self, tmp_path): self._seed(tmp_path) got = self._fix(tmp_path, physical_exact=4, physical_lower=None, physical_upper=None) assert got["physical_attempts"]["exact"] == 4 def test_it_refuses_when_it_misread_the_current_value(self, tmp_path): self._seed(tmp_path) with pytest.raises(cp.CanaryStopped): self._fix(tmp_path, debit_was=9) def test_a_second_correction_supersedes(self, tmp_path): self._seed(tmp_path) self._fix(tmp_path) cp.correct_usage(tmp_path, "bbb222", debit_was=4, debit_now=6, physical_exact=6, why="다시 셌다") assert cp.cumulative_used(tmp_path) == 6 def test_a_usage_correction_does_not_touch_the_status(self, tmp_path): """★★실측 2026-09-02 — 사용량 정정이 `crashed` 를 `"None"` 으로 덮어썼다. 두 정정은 **서로 다른 칸**을 고친다. """ self._seed(tmp_path) assert cp.effective_status(tmp_path, "bbb222") == "crashed" self._fix(tmp_path) assert cp.effective_status(tmp_path, "bbb222") == "crashed" def test_a_status_correction_still_works(self, tmp_path): cp.append_attempt(tmp_path, {"attempt_id": "ccc333", "status": "completed", "used": 3}) cp.correct_attempt(tmp_path, "ccc333", was="completed", now="crashed", why="x") assert cp.effective_status(tmp_path, "ccc333") == "crashed" assert cp.effective_used(tmp_path, "ccc333") == 3 class TestTransportIsNotDebitedTwice: """★★★같은 사용량을 **두 번 빼지 않는다** (Codex BLOCK 2026-09-02). 장부의 `effective_used` 가 이미 그 attempt 차감을 누계에 접어 `remaining = 상한 − 누계` 로 여는데, 운반 계수기가 **과거 최대 used** 를 그대로 이어받으면 같은 것이 한 번 더 줄어든다. """ def _log(self, root, rows): (root / cp.TRANSPORT_LOG).write_text( "\n".join(json.dumps(r, ensure_ascii=False) for r in rows), encoding="utf-8") def test_a_settled_attempts_transport_is_not_adopted(self, tmp_path): cp.append_attempt(tmp_path, {"attempt_id": "aa", "status": "crashed", "used": 7}) self._log(tmp_path, [{"scope": "pipeline", "attempt_id": "aa", "used": 7, "by_source": {"x": 7}}]) got = cp.unsettled_transport(tmp_path, "pipeline") assert got["used"] == 0, "★이미 접힌 것을 또 이어받았다" assert cp.cumulative_used(tmp_path) == 7 def test_an_open_attempts_transport_is_adopted(self, tmp_path): """★아직 안 닫힌 판의 몫은 **이어받는다** — 그것이 미정이다.""" cp.append_attempt(tmp_path, {"attempt_id": "bb", "status": "running", "used": 0}) self._log(tmp_path, [{"scope": "pipeline", "attempt_id": "bb", "used": 3, "by_source": {"x": 3}}]) got = cp.unsettled_transport(tmp_path, "pipeline") assert got["used"] == 3 and got["by_source"] == {"x": 3} def test_a_row_with_no_attempt_is_adopted(self, tmp_path): """★★죽어서 attempt 를 못 적은 줄도 **이어받는다** — 안 세면 샌다.""" self._log(tmp_path, [{"scope": "pipeline", "attempt_id": "", "used": 1, "by_source": {"boot": 1}}]) got = cp.unsettled_transport(tmp_path, "pipeline") assert got["used"] == 1 and got["from_attempts"] == ["미정"] def test_scopes_do_not_mix(self, tmp_path): self._log(tmp_path, [ {"scope": "bootstrap", "attempt_id": "", "used": 2, "by_source": {"b": 2}}, {"scope": "pipeline", "attempt_id": "", "used": 5, "by_source": {"p": 5}}]) assert cp.unsettled_transport(tmp_path, "bootstrap")["used"] == 2 assert cp.unsettled_transport(tmp_path, "pipeline")["used"] == 5 def test_a_broken_line_stops(self, tmp_path): (root := tmp_path / cp.TRANSPORT_LOG).write_text("{망가진", encoding="utf-8") assert root.is_file() with pytest.raises(cp.LedgerRefused): cp.unsettled_transport(tmp_path, "pipeline") def test_reopening_a_settled_attempt_does_not_lower_it_twice(self, tmp_path): """★★음성 대조 — terminal 로 반영된 판을 다시 열어도 잔여가 그대로.""" cp.append_attempt(tmp_path, {"attempt_id": "cc", "status": "crashed", "used": 10}) self._log(tmp_path, [{"scope": "pipeline", "attempt_id": "cc", "used": 10, "by_source": {"x": 10}}]) left = cp.remaining_cap(tmp_path, ceiling=93) assert left == 83 assert cp.unsettled_transport(tmp_path, "pipeline")["used"] == 0 assert cp.remaining_cap(tmp_path, ceiling=93) == 83 def test_the_run_uses_the_reconciler(self): import inspect from tools.grounding_audit import canary_run as cr src = inspect.getsource(cr.run) assert "cp.unsettled_transport(" in inspect.getsource(cr.run) or \ "_transport_so_far(" in src assert '"attempt_id": _aid["id"]' in src, "★줄에 신원이 안 붙는다" class TestTheTransportRowCarriesTheRealAttemptId: """★★★운반 줄의 `attempt_id` 가 **진짜 attempt** 와 같아야 한다. Codex BLOCK (2026-09-02): 앞 판은 `run_pipeline` **뒤에** 생기는 자리를 먼저 읽어서 운반 줄의 신원이 늘 **빈 문자열**이었다. 그러면 `unsettled_transport` 계약이 production 에서 성립하지 않는다. ★손으로 id 를 넣는 시험은 부족하다 — 실제 `run_pipeline` 을 태운다. """ def test_the_run_makes_the_id_before_the_pipeline(self): import inspect from tools.grounding_audit import canary_run as cr src = inspect.getsource(cr.run) i = src.index('_aid["id"] = ') assert i < src.index("cp.run_pipeline("), "★신원을 뒤에 만든다" assert "attempt_id=_aid[\"id\"]" in src, "★pipeline 에 안 넘긴다" def test_the_pipeline_takes_the_given_id(self, tmp_path, monkeypatch): """★★실제 `run_pipeline` 이 **받은 신원**으로 장부를 연다.""" 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 from tools.grounding_audit import canary_run as cr 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") def fake(**kw): d = (tmp_path / "projects" / "p" / "checkpoints" / "episodes" / "e" / kw["step_ids"][0]) d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text(json.dumps( {"status": "completed"}), encoding="utf-8") import unittest.mock as m with m.patch.object(ads, "run_steps_batch", fake): got = cp.run_pipeline(run_id="abc123abc123", project_id="p", episode_id="e", plan=plan, caps={s: 1 for s in plan["applied_metered"]}, emergency_counted=70, approved_image_calls=0, attempt_id="deadbeef1234", live=True) assert got["attempt_id"] == "deadbeef1234" rows = [a for a in cp.read_attempts(tmp_path / "canary_abc123abc123") if a.get("kind") is None] assert [a["attempt_id"] for a in rows] == ["deadbeef1234"] def test_a_row_bound_to_that_id_is_settled_after_it_closes(self, tmp_path): """★그 신원으로 적힌 운반 줄은 판이 닫히면 **또 안 빠진다**.""" cp.append_attempt(tmp_path, {"attempt_id": "deadbeef1234", "status": "completed", "used": 5}) (tmp_path / cp.TRANSPORT_LOG).write_text(json.dumps( {"scope": "pipeline", "attempt_id": "deadbeef1234", "used": 5, "by_source": {"x": 5}}) + "\n", encoding="utf-8") assert cp.unsettled_transport(tmp_path, "pipeline")["used"] == 0 assert cp.cumulative_used(tmp_path) == 5 assert cp.remaining_cap(tmp_path, ceiling=93) == 88