"""C(c) producer 스텝 — ★**아직 안 켠다**. 유료 0. 여기서 재는 것은 「돈다」가 아니라 **「켜지 않았는데 사지 않는가」**와 **「규칙을 다시 적지 않았는가」**다. 실제 주행은 사람 GO 뒤 D 에서 본다. """ from __future__ import annotations import ast import inspect import json import textwrap import pytest from app.core.steps import grounding_chunk_step as gs def _send_call(fn): """실제 전송 호출 노드. ★**참조로 넘긴 것**도 잡는다. `call_structured` 를 `call_with_deadline(call_structured, ...)` 로 넘기면 그것은 `ast.Call` 이 아니라 **인자**다 — 이름만 찾으면 「보내는 자리 0곳」이 된다. 오늘 같은 부류를 감사 도구에서도 겪었다. """ import ast import inspect as _i import textwrap tree = ast.parse(textwrap.dedent(_i.getsource(fn))) for n in ast.walk(tree): if not isinstance(n, ast.Call): continue name = getattr(n.func, "id", "") if name == "call_structured": return n if name == "call_with_deadline" and n.args and \ getattr(n.args[0], "id", "") == "call_structured": return n return None class TestItIsWiredNow: """★★**뒤집은 시험** (2026-09-01 D 활성화). 앞에는 「아무 데도 안 붙는다」를 잠갔는데 그것이 이 커밋이 바꾸는 것이다. 지우지 않고 **어떻게** 붙었나를 잠근다 — 붙은 자리가 하나여야 한다. """ def test_the_step_registry_knows_it_under_one_id(self): from app.core.steps import STEP_CLASSES got = [k for k, v in STEP_CLASSES.items() if v is gs.GroundingChunkStep] assert got == ["grounding_chunk"] def test_the_manifest_knows_it_under_the_same_id(self): from app.core.step_manifest import STEP_MANIFEST assert [k for k in STEP_MANIFEST if "chunk" in k] == ["grounding_chunk"] assert STEP_MANIFEST["grounding_chunk"]["applicability"] == \ "if_chunk_producer" def test_only_the_registry_imports_this_step(self): """★붙은 자리가 **하나**여야 한다 — 여럿이면 경계가 흐려진다.""" from pathlib import Path root = Path(gs.__file__).resolve().parents[3] / "app" hits = set() for f in root.rglob("*.py"): if f.name == "grounding_chunk_step.py": continue tree = ast.parse(f.read_text(encoding="utf-8")) for n in ast.walk(tree): mod = getattr(n, "module", None) if isinstance(n, ast.ImportFrom) and mod and \ mod.endswith("grounding_chunk_step"): hits.add(f.name) if isinstance(n, (ast.Name, ast.Attribute)) and \ getattr(n, "id", getattr(n, "attr", "")) == \ "GroundingChunkStep": hits.add(f.name) assert hits == {"__init__.py"}, f"★붙은 자리가 여럿이다: {sorted(hits)}" class TestItBuysNothingWhenTheModeIsOff: """★★「켜지 않았는데 샀다」가 있으면 안 된다.""" class _Stub(gs.GroundingChunkStep): def __init__(self, mode): self.project_config = {"grounding_mode": mode} self.step_id = "grounding_chunk" def _plan(self): # ★여기 오면 안 된다 raise AssertionError("★모드가 꺼졌는데 계획을 세웠다") @pytest.mark.parametrize("mode", ["legacy", "shadow_plan", "v2"]) def test_every_other_mode_returns_without_planning(self, mode): out = self._Stub(mode)._execute() assert out["data"]["skipped"] is True assert out["data"]["provider_calls"] == 0 assert out["completed_count"] == 0 def test_it_asks_the_single_activation_predicate(self): """★모드 판정을 제 손으로 하면 경계가 두 곳이 된다.""" src = inspect.getsource(gs.GroundingChunkStep._execute) assert "uses_chunk_producer" in src for banned in ('== "v2_chunk"', "== GROUNDING_MODE_V2_CHUNK"): assert banned not in src, f"★경계를 제 손으로 비교한다: {banned}" class TestItDoesNotReimplementAnything: """★★규칙을 다시 적으면 도구가 재는 것과 주행이 굽는 것이 갈린다.""" def test_it_calls_the_shared_helpers(self): src = inspect.getsource(gs) for want in ("bundle_scenes", "build_chunk_payload", "resolve_rows", "reduce_episode", "build_catalog", "to_entity_rows", "assert_bundle_target_agrees"): assert want in src, f"★공용 함수를 안 쓴다: {want}" def test_it_does_not_slice_any_text(self): """★원문도 샷 전문도 **자르지 않는다**. ★앞 판은 **모든** 자르기를 막았다가 `_config_hash` 의 `hexdigest()[:16]` 을 잡았다 — 해시를 줄이는 것은 LLM 에 보내는 글을 자르는 것이 아니다. 막아야 하는 자리는 **payload 를 짓는 곳**이다. """ for name in ("_plan", "_read_chunks", "_one", "_merge", "_wrap"): fn = getattr(gs.GroundingChunkStep, name) tree = ast.parse(textwrap.dedent(inspect.getsource(fn))) for n in ast.walk(tree): if isinstance(n, ast.Subscript) and \ isinstance(n.slice, ast.Slice): raise AssertionError(f"★{name} 안 {n.lineno}줄에서 잘랐다") def test_it_does_not_define_a_second_bundle_target(self): tree = ast.parse(inspect.getsource(gs)) names = {t.id for n in ast.walk(tree) if isinstance(n, ast.Assign) for t in n.targets if isinstance(t, ast.Name)} assert "BUNDLE_TARGET" not in names def test_results_are_gathered_in_plan_order_not_completion_order(self): """★★완료 순서로 쌓으면 같은 입력에도 행 순서가 달라져 지문이 안 선다. ★앞 판 시험은 **문자열 위치**로 봤는데 `for c in plan` 이 futures 를 만드는 자리에도 있어 헛짚었다. 「**`resolve_rows` 를 부르는 loop 가 무엇을 도는가**」를 AST 로 본다. """ import textwrap tree = ast.parse(textwrap.dedent( inspect.getsource(gs.GroundingChunkStep._read_chunks))) loops = [n for n in ast.walk(tree) if isinstance(n, ast.For) and any(isinstance(c, ast.Call) and getattr(c.func, "attr", "") == "resolve_rows" for c in ast.walk(n))] assert len(loops) == 1, f"★해석하는 loop 가 {len(loops)}개다" it = loops[0].iter assert isinstance(it, ast.Name) and it.id == "plan", \ "★해석을 **완료 순서**로 돈다 — 같은 입력에도 행 순서가 달라진다" class TestTheHalfOnModeIsStructurallyImpossible: # noqa: D101 """★★`v2_chunk` 를 받는 집합에 미리 넣었더니 **옛 유료 사슬만 켜졌다**. Codex 가 재현했다 (2026-08-31): `buys_v2_research("v2_chunk")` 가 True 라 `if_grounding_v2` 가 통과하고, 그 술어가 여는 것은 **옛 스텝 넷** (`grounding_a0`·`grounding_plan`·`grounding_research`· `reference_acquisition`)이다. 새 producer 는 manifest 에 없으니 결과는 「새 것은 안 돌고 옛 유료만 켜진 판」이다. ★「아무도 지금 안 고른다」는 **운영 스냅샷**이지 구조가 아니다. """ @staticmethod def _old_v2_steps(): from app.core.step_manifest import STEP_MANIFEST return sorted(k for k, v in STEP_MANIFEST.items() if v.get("applicability") == "if_grounding_v2") def test_the_old_paid_chain_is_what_that_predicate_opens(self): """★positive control — 무엇이 걸려 있는지 **먼저** 보인다.""" assert self._old_v2_steps(), "★if_grounding_v2 스텝이 하나도 없다" def test_project_config_selects_it_now(self): """★★**뒤집었다** (D 활성화) — 이제 고를 수 있다. 앞에는 「고를 수 없다」를 잠갔다. 그때는 새 producer 가 manifest 에 없어서 고르는 순간 옛 유료 사슬만 켜졌기 때문이다. 이번에는 넷을 같은 커밋에 넣었고, 하나라도 빠지면 **조립이 선다**. """ from app.core.grounding_mode import resolve_grounding_mode assert resolve_grounding_mode( {"grounding_mode": "v2_chunk"}) == "v2_chunk" def test_env_selects_it_now(self, monkeypatch): from app.core.grounding_mode import resolve_grounding_mode monkeypatch.setenv("GROUNDING_MODE", "v2_chunk") assert resolve_grounding_mode() == "v2_chunk" def test_an_unknown_value_is_still_refused(self): """★음성 대조 — 열었다고 아무 값이나 받으면 안 된다.""" from app.core.errors import AppError from app.core.grounding_mode import resolve_grounding_mode with pytest.raises(AppError, match="없는 값"): resolve_grounding_mode({"grounding_mode": "v2_chunkk"}) def test_the_two_sets_still_do_not_overlap(self): from app.core.grounding_mode import GROUNDING_MODES, PLANNED_MODES assert not (PLANNED_MODES & GROUNDING_MODES), \ "★계획된 값이 받는 값에 들어갔다 — 반만 켠 판이 생긴다" assert "v2_chunk" in GROUNDING_MODES and not PLANNED_MODES def test_exactly_one_accepted_mode_reaches_the_chunk_producer(self): """★★**하나만** 연다 — 둘이면 경계가 흐려진다.""" from app.core.grounding_mode import (GROUNDING_MODES, uses_chunk_producer) assert [m for m in GROUNDING_MODES if uses_chunk_producer(m)] == ["v2_chunk"] def test_the_old_chain_is_off_in_that_very_mode(self): """★★같은 판에서 둘 다 켜지면 **같은 것을 두 번 산다**.""" from app.core.grounding_mode import (buys_v2_research, uses_chunk_producer) assert buys_v2_research("v2_chunk") is False assert uses_chunk_producer("v2") is False def test_the_predicate_that_opens_the_old_chain_stays_v2_only(self): """★★이 술어가 True 가 되는 순간 **옛 유료 넷**이 열린다.""" from app.core.grounding_mode import GROUNDING_MODES, buys_v2_research assert {m for m in GROUNDING_MODES if buys_v2_research(m)} == {"v2"} class TestCancellationIsNotASingleChunkFailure: """★★멈추라는 말을 **구간 하나의 실패로 접으면** 나머지 worker 가 계속 돌아 돈이 나가고, rows 가 남아 **merge 유료 호출까지** 이어진다. Codex 가 실제로 태워 재현했다 (2026-08-31): `CANCELLATION_SWALLOWED`. """ class _Step(gs.GroundingChunkStep): def __init__(self, code): self.step_id = "grounding_chunk" self.project_config = {} self._code = code self.merge_calls = 0 def _one(self, chunk, jr=None, cap=0): from app.core.errors import AppError if chunk["chunk_id"] == "c0": raise AppError(code=self._code, message="멈춰", status_code=409) return {"rows": []} def _merge(self, rows, segs, jr=None, cap=0): self.merge_calls += 1 return super()._merge(rows, segs, jr, cap) PLAN = [{"chunk_id": "c0", "segment_ids": ["scene-1"], "payload": {}}, {"chunk_id": "c1", "segment_ids": ["scene-2"], "payload": {}}] SEGS = {"scene-1": "가", "scene-2": "나"} @pytest.mark.parametrize("code", ["step.cancelled", "step.owner_lost", "step.gate_unreadable"]) def test_it_is_re_raised_immediately(self, code): from app.core.errors import AppError st = self._Step(code) with pytest.raises(AppError) as got: st._read_chunks(self.PLAN, self.SEGS, {}) assert got.value.code == code, "★다른 오류로 바뀌었다" assert st.merge_calls == 0, "★취소 뒤에 merge 를 샀다" def test_an_ordinary_error_stops_differently_and_later(self): """★★일반 provider 오류도 이제 **선다** — 다만 **다른 이유·다른 때**다. 계약이 바뀌었다 (Codex §10 partial): 구간 하나가 빠지면 등록 횟수가 **구간을 걸쳐** 세어지므로 남은 것만으로 내면 **수가 틀린다**. 그래서 산출을 안 낸다. ★가르는 자리는 남아 있다 — 취소 계열 **원래 예외 그대로** 즉시 올린다 (다른 worker 를 안 기다림) 일반 오류 전부 모은 뒤 `grounding_chunk.incomplete_read` 로 선다 시험을 지우지 않고 **뒤집는다** — 지키던 것(취소와 일반 오류를 가른다)은 그대로다. """ from app.core.errors import AppError st = self._Step("provider.timeout") with pytest.raises(AppError) as got: st._read_chunks(self.PLAN, self.SEGS, {}) assert got.value.code == "grounding_chunk.incomplete_read", \ "★일반 오류가 취소처럼 원래 코드로 올라갔다" assert "c0" in got.value.message assert st.merge_calls == 0 def test_the_two_kinds_are_told_apart_by_the_code_that_surfaces(self): """★★둘 다 서지만 **무엇으로 섰는지**가 달라야 감사에서 갈린다.""" from app.core.errors import AppError codes = {} for kind in ("step.cancelled", "provider.timeout"): with pytest.raises(AppError) as got: self._Step(kind)._read_chunks(self.PLAN, self.SEGS, {}) codes[kind] = got.value.code assert codes["step.cancelled"] == "step.cancelled", \ "★취소가 provider 장애로 기록된다" assert codes["provider.timeout"] == "grounding_chunk.incomplete_read" def test_the_abort_list_lives_in_one_place(self): """★`detail_steps` 가 같은 목록을 **두 번 인라인**으로 적어 뒀다.""" import ast import app.core.run_control as rc src = ast.parse(inspect.getsource(gs)) assert "ABORT_CODES" in {n.id for n in ast.walk(src) if isinstance(n, ast.Name)} or \ "is_abort" in {n.id for n in ast.walk(src) if isinstance(n, ast.Name)}, \ "★목록을 제 손으로 다시 적었다" assert "step.cancelled" in rc.ABORT_CODES def test_the_merge_checks_for_a_stop_before_paying(self): """★구간을 다 읽는 사이에 멈추라는 말이 올 수 있다.""" src = inspect.getsource(gs.GroundingChunkStep._merge) i = src.index("checkpoint_gate") j = src.index("build_merge_payload") assert i < j, "★유료 payload 를 만든 **뒤에** 검사한다" class TestMyStubMustMatchTheRealSignature: """★★내 대역이 실제 시그니처와 갈리면 **진짜 동작을 가린다**. 실제로 그랬다 (2026-08-31): `_one` 에 인자가 늘었는데 시험 대역은 옛 모양이라 `TypeError` 가 났고, 그것이 일반 오류로 접혀 **취소 재전파 축이 통과처럼 보였다**. 대역은 구현을 흉내 내는 것이지 제 모양을 짓는 것이 아니다. """ @pytest.mark.parametrize("name", ["_one", "_merge", "_read_chunks", "_wrap", "_call"]) def test_every_overridden_method_keeps_the_real_signature(self, name): import inspect as _i stub = TestCancellationIsNotASingleChunkFailure._Step if name not in vars(stub): return # 안 덮은 것은 볼 것이 없다 want = list(_i.signature(getattr(gs.GroundingChunkStep, name)) .parameters) got = list(_i.signature(getattr(stub, name)).parameters) assert got == want, f"★{name} 대역이 {got} · 실제는 {want}" class TestTheJournalDoesNotBuyTwice: """★★crash 뒤 재개하면 **이미 끝난 구간을 다시 사면 안 된다**. 구간 하나가 원문 전문 + 샷 전문이라 재구매는 그대로 돈이다. """ def _jr(self, tmp_path, **kw): from app.modules.pipeline.grounding_chunk_journal import ChunkJournal return ChunkJournal(tmp_path / "j.json", contract={"v": 1}, **kw) def test_a_saved_answer_is_reused_without_sending(self, tmp_path): from app.modules.pipeline.grounding_chunk_journal import buy_or_reuse jr = self._jr(tmp_path) calls = [] got = buy_or_reuse(jr, "abc", cap=5, send=lambda: calls.append(1) or {"r": 1}) again = buy_or_reuse(jr, "abc", cap=5, send=lambda: calls.append(1) or {"r": 2}) assert got == again == {"r": 1} assert len(calls) == 1, "★이미 산 것을 다시 샀다" def test_it_survives_a_restart(self, tmp_path): from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) buy_or_reuse(self._jr(tmp_path), "abc", cap=5, send=lambda: {"r": 1}) fresh = ChunkJournal(tmp_path / "j.json", contract={"v": 1}) calls = [] got = buy_or_reuse(fresh, "abc", cap=5, send=lambda: calls.append(1) or {"r": 2}) assert got == {"r": 1} and calls == [], "★재개했는데 다시 샀다" def test_a_changed_contract_is_visible(self, tmp_path): from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) buy_or_reuse(self._jr(tmp_path), "abc", cap=5, send=lambda: {"r": 1}) other = ChunkJournal(tmp_path / "j.json", contract={"v": 2}) assert other.contract_drifted() is True, \ "★팩·계약을 바꿨는데 옛 장부를 그대로 쓴다" def test_a_send_that_never_answered_is_not_re_bought(self, tmp_path): """★★「보냈는데 답을 못 받은 것」은 **샀을 수도 있다**.""" from app.modules.pipeline.grounding_chunk_journal import ( ChunkJournal, NeedsHumanDecision, buy_or_reuse) jr = self._jr(tmp_path) with pytest.raises(RuntimeError, match="끊김"): buy_or_reuse(jr, "abc", cap=5, send=lambda: (_ for _ in ()).throw( RuntimeError("끊김"))) again = ChunkJournal(tmp_path / "j.json", contract={"v": 1}) assert again.bought() == 1, "★안 산 것으로 세면 다음 판이 다시 산다" with pytest.raises(NeedsHumanDecision, match="샀는지 아닌지"): again.assert_no_uncertain() class TestTheBudgetIsCheckedBeforeSending: def test_it_stops_instead_of_quietly_buying_less(self, tmp_path): from app.modules.pipeline.grounding_chunk_journal import ( BudgetExceeded, ChunkJournal, buy_or_reuse) jr = ChunkJournal(tmp_path / "j.json", contract={}) for i in range(2): buy_or_reuse(jr, f"id{i}", cap=2, send=lambda: {"r": 1}) calls = [] with pytest.raises(BudgetExceeded, match="상한"): buy_or_reuse(jr, "id9", cap=2, send=lambda: calls.append(1) or {"r": 1}) assert calls == [], "★상한을 넘고 나서 알았다 — 이미 샀다" def test_the_stop_check_runs_before_the_budget_and_the_send(self, tmp_path): """★순서가 계약이다 — 멈추라는 말이 제일 먼저.""" from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) seen = [] jr = ChunkJournal(tmp_path / "j.json", contract={}) with pytest.raises(RuntimeError, match="멈춰"): buy_or_reuse(jr, "a", cap=0, send=lambda: seen.append("send") or {}, stop_check=lambda: (_ for _ in ()).throw( RuntimeError("멈춰"))) assert seen == [], "★멈추라는 말이 왔는데 보냈다" assert jr.bought() == 0, "★안 보냈는데 샀다고 적혔다" def test_the_step_caps_at_chunks_plus_merge(self): class _S(gs.GroundingChunkStep): def __init__(self): self.project_config = {} self.step_id = "grounding_chunk" assert _S()._cap([1, 2, 3]) == 4 class TestTheCallIdentityRidesToOpik: """★신원을 안 실으면 장부의 어느 줄이 어느 trace 인지 못 잇는다.""" def test_the_identity_rides_on_the_parent_trace_not_the_call_metadata( self): """★★`opik_metadata` 에 실으면 **버려진다**. litellm 이 `metadata["opik"]` 에서 읽는 키는 넷뿐이고 (`project_name`·`current_span_data`·`tags`·`thread_id`), 태그로 실어도 축 whitelist 가 거른다. 신원은 **부모 trace 의 metadata** 로 간다. ★앞 판 시험은 `opik_metadata` 에 있는지를 봤다 — **틀린 자리를 정답으로 못박고 있었다**. 뒤집는다. """ import ast import inspect as _i import textwrap tree = ast.parse(textwrap.dedent( _i.getsource(gs.GroundingChunkStep._call))) opens = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "open_trace"] assert len(opens) == 1, "★부모 trace 를 안 연다" kw = {k.arg for k in opens[0].keywords} assert {"name", "tags", "metadata", "thread_id"} <= kw # ★호출 metadata 에는 **태그만** 간다 — 다른 키는 어차피 버려진다 send = _send_call(gs.GroundingChunkStep._call) assert send is not None, "★보내는 자리를 못 찾았다" meta = [k.value for k in send.keywords if k.arg == "opik_metadata"] assert meta and isinstance(meta[0], ast.Dict) keys = {k.value for k in meta[0].keys if isinstance(k, ast.Constant)} assert keys == {"tags"}, f"★버려질 키를 실었다: {keys - {'tags'}}" def test_the_tag_survives_the_axis_whitelist(self): """★축 접두사가 없으면 태그가 **통째로 걸러진다**.""" from app.modules.llm.opik_trace import is_axis_tag assert is_axis_tag(gs.CHUNK_TAG), \ f"★{gs.CHUNK_TAG!r} 는 축 태그가 아니라 걸러진다" def test_the_join_key_matches_the_audit_tool(self): """★도구와 이름이 다르면 장부↔trace 대조가 통째로 안 된다.""" from tools.grounding_audit import cc_runner as rr assert gs.ID_META_KEY == rr.ID_META_KEY def test_there_is_only_one_place_that_sends(self): """★보내는 자리가 여럿이면 장부·예산·정지를 한 곳이 못 건다. ★**참조로 넘긴 것**까지 센다 — 이름만 찾으면 0곳으로 보인다. """ import ast import inspect as _i tree = ast.parse(_i.getsource(gs)) # ★`from … import call_structured` 는 `ImportFrom` 이라 `Name` 이 # 아니다 — 이름 등장 수는 **쓰는 자리**만 센다. used = [x for x in ast.walk(tree) if isinstance(x, ast.Name) and x.id == "call_structured"] assert len(used) == 1, f"★쓰는 자리가 {len(used)}곳이다" brought = [n for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and any(a.name == "call_structured" for a in n.names)] assert len(brought) == 1, f"★들여오는 자리가 {len(brought)}곳이다" def test_the_send_is_wrapped_in_a_per_call_deadline(self): """★★주행 마감만으로는 **매달린 provider 를 못 끊는다** — 정지 표는 *새* 전송 앞에서만 보인다 (Codex 2026-08-31).""" send = _send_call(gs.GroundingChunkStep._call) assert getattr(send.func, "id", "") == "call_with_deadline", \ "★전송이 한 호출 마감으로 안 감싸였다" assert any(k.arg == "deadline_seconds" for k in send.keywords) assert gs.CHUNK_CALL_DEADLINE_SECONDS > 0 def test_the_send_arms_the_research_budget(self): """★팔을 안 들면 `reserve` 가 no-op 이라 물리 전송이 안 세어진다.""" import ast import inspect as _i import textwrap tree = ast.parse(textwrap.dedent( _i.getsource(gs.GroundingChunkStep._call))) armed = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "research_calls_armed"] assert len(armed) == 1, "★조사 예산 팔을 안 든다" class TestTheJournalSurvivesParallelWriters: """★★구간을 **병렬로** 읽는다 — 여러 worker 가 같은 장부에 적는다. Codex 가 잡았다 (2026-08-31): 같은 이름의 `.tmp` 에 동시에 쓰면 `os.replace` 가 **반쪽을 확정**하거나 나중 쓰기가 앞 쓰기를 통째로 덮는다. 그러면 **산 것이 장부에서 사라지고**, 다음 판이 다시 사서 상한을 넘긴다. """ def test_every_parallel_buy_is_in_the_journal(self, tmp_path): from concurrent.futures import ThreadPoolExecutor from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) n = 40 jr = ChunkJournal(tmp_path / "j.json", contract={}) with ThreadPoolExecutor(max_workers=12) as ex: list(ex.map(lambda i: buy_or_reuse( jr, f"id{i:03d}", cap=n, send=lambda: {"r": 1}), range(n))) assert jr.bought() == n, f"★{n}개를 샀는데 장부에 {jr.bought()}개" # ★끝점 — **파일에서 되읽는다**. 메모리만 맞으면 재개가 안 된다 again = ChunkJournal(tmp_path / "j.json", contract={}) assert again.bought() == n, \ f"★파일에는 {again.bought()}개뿐 — 다음 판이 다시 산다" assert {f"id{i:03d}" for i in range(n)} == set(again.entries) def test_no_stray_temp_file_is_left_behind(self, tmp_path): from concurrent.futures import ThreadPoolExecutor from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) jr = ChunkJournal(tmp_path / "j.json", contract={}) with ThreadPoolExecutor(max_workers=8) as ex: list(ex.map(lambda i: buy_or_reuse( jr, f"x{i}", cap=20, send=lambda: {"r": 1}), range(20))) leftovers = sorted(p.name for p in tmp_path.glob("*.tmp")) assert leftovers == [], f"★임시 파일이 남았다: {leftovers}" def test_the_file_is_always_valid_json_mid_flight(self, tmp_path): """★반쪽 파일을 다음 판이 읽으면 **장부를 통째로 버린다**.""" import json as _j import threading from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) path = tmp_path / "j.json" jr = ChunkJournal(path, contract={}) bad = [] stop = threading.Event() def _watch(): while not stop.is_set(): if path.exists(): try: _j.loads(path.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 bad.append(1) w = threading.Thread(target=_watch, daemon=True) w.start() try: from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers=8) as ex: list(ex.map(lambda i: buy_or_reuse( jr, f"y{i}", cap=60, send=lambda: {"r": 1}), range(60))) finally: stop.set() w.join(timeout=2) assert bad == [], f"★쓰는 도중에 반쪽 파일이 {len(bad)}번 보였다" class TestTheDeadlineReachesTheWorkers: """★정지 표는 **스레드 지역**이다 — 안 실어 보내면 팬아웃에서 no-op 이다.""" def test_the_step_opens_a_run_scope_with_a_deadline(self): import ast import inspect as _i import textwrap tree = ast.parse(textwrap.dedent( _i.getsource(gs.GroundingChunkStep._execute))) scopes = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "research_run_scope"] assert len(scopes) == 1, "★주행 상한·마감을 안 연다" kw = {k.arg for k in scopes[0].keywords} assert {"cap", "deadline_seconds"} <= kw def test_the_workers_get_the_stop_check_bound(self): import inspect as _i src = _i.getsource(gs.GroundingChunkStep._read_chunks) assert "bind_current_research_budget" in src, \ "★worker 가 정지 표를 못 본다 — 마감이 그 자리에서 no-op 이다" def test_a_deadline_actually_stops_a_worker(self): """★★끝점 — 실제 스레드에서 정지 표가 **걸리는지** 본다.""" from concurrent.futures import ThreadPoolExecutor from app.core.image_call_budget import get_current_stop_check from app.core.research_call_budget import ( bind_current_research_budget, research_run_scope) seen = [] with research_run_scope(cap=5, deadline_seconds=0.0): def _work(_): chk = get_current_stop_check() seen.append(chk is not None) if chk is not None: try: chk() except Exception as exc: # noqa: BLE001 return type(exc).__name__ return "no-stop" with ThreadPoolExecutor(max_workers=3) as ex: got = list(ex.map(bind_current_research_budget(_work), range(3))) assert all(seen), "★worker 가 정지 표를 아예 못 봤다" assert set(got) == {"RunDeadlineExceeded"}, f"★막지 못했다: {got}" class TestProcessingOnlyChangesDoNotRebuy: """★★**후처리만 바꿨는데 유료 raw 를 다시 사면 안 된다** (Codex BLOCK). 앞 판은 `_config_hash` 가 후처리·adapter 계약까지 접고, 계약이 하나라도 달라지면 `jr.entries.clear()` 를 했다. 그러면 **같은 `acquisition_identity` 의 저장 응답도 지워져** provider 가 다시 나간다. 갈라야 하는 두 가지 — 획득(acquisition) 팩·모델·실제 요청 계약 → 바뀌면 **다시 산다** 해석(processing) 후처리·adapter·구간 나누기 → 바뀌면 **무료 재해석** """ class _S(gs.GroundingChunkStep): def __init__(self): self.project_config = {} self.step_id = "grounding_chunk" def test_the_identity_ignores_the_processing_contract(self, monkeypatch): from app.modules.pipeline import grounding_chunk as gc st = self._S() payload = {"system": "s", "parts": [{"type": "text", "text": "t"}], "schema": {}} before = st._identity(payload) monkeypatch.setattr(gc, "PROCESSING_CONTRACT_VERSION", "9.9") assert st._identity(payload) == before, \ "★후처리 계약이 신원에 섞였다 — 해석만 고쳐도 다시 산다" def test_the_identity_moves_when_the_request_contract_moves(self, monkeypatch): """★positive control — 재시도를 켠 판과 끈 판은 **같은 것이 아니다**.""" st = self._S() payload = {"system": "s", "parts": [{"type": "text", "text": "t"}], "schema": {}} before = st._identity(payload) monkeypatch.setitem(gs.REQUEST_CONTRACT, "num_retries", 3) assert st._identity(payload) != before, \ "★재시도를 바꿔도 신원이 같다 — 다른 계약으로 산 것을 재사용한다" def test_the_journal_contract_holds_only_acquisition(self): st = self._S() got = st._acquisition() assert {"alias", "physical", "pack", "request"} <= set(got) blob = json.dumps(got, ensure_ascii=False, sort_keys=True) for banned in ("adapter", "chunk_plan", "bundle_target"): assert banned not in blob, f"★획득 계약에 해석 것이 섞였다: {banned}" def test_a_processing_change_keeps_the_saved_answers(self, tmp_path, monkeypatch): """★★끝점 — 후처리를 바꾸고 다시 열어도 **되쓴다**.""" from app.modules.pipeline import grounding_chunk as gc from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) st = self._S() path = tmp_path / "j.json" jr = ChunkJournal(path, contract={"acquisition": st._acquisition()}) buy_or_reuse(jr, "ident-1", cap=3, send=lambda: {"r": 1}) monkeypatch.setattr(gc, "PROCESSING_CONTRACT_VERSION", "9.9") again = ChunkJournal(path, contract={"acquisition": st._acquisition()}) assert again.contract_drifted() is False, \ "★후처리를 바꿨는데 획득 계약이 흔들렸다" calls = [] got = buy_or_reuse(again, "ident-1", cap=3, send=lambda: calls.append(1) or {"r": 2}) assert got == {"r": 1} and calls == [], "★후처리만 바꿨는데 다시 샀다" def test_the_config_hash_separates_the_two(self): import ast import inspect as _i import textwrap tree = ast.parse(textwrap.dedent( _i.getsource(gs.GroundingChunkStep._config_hash))) keys = {n.value for n in ast.walk(tree) if isinstance(n, ast.Constant) and isinstance(n.value, str)} assert {"acquisition", "processing"} <= keys, \ "★지문이 사는 것과 해석하는 것을 안 가른다" class TestTheRequestContractIsActuallySent: """★★기본값에 맡기면 `enable_fallback=True`·라우터 기본 재시도라 **몇 번 나가는지 모른 채** 산다.""" def test_the_send_passes_the_pinned_contract(self): import ast import inspect as _i import textwrap send = _send_call(gs.GroundingChunkStep._call) assert send is not None, "★보내는 자리를 못 찾았다" kw = {k.arg for k in send.keywords} assert {"enable_fallback", "num_retries", "temperature"} <= kw, \ f"★못박은 계약을 안 넘긴다 (넘긴 것: {sorted(kw)})" def test_the_pinned_values_are_the_quiet_ones(self): assert gs.REQUEST_CONTRACT["enable_fallback"] is False assert gs.REQUEST_CONTRACT["num_retries"] == 0 def test_the_contract_rides_in_the_identity(self): st = TestProcessingOnlyChangesDoNotRebuy._S() blob = json.dumps(st._acquisition(), sort_keys=True) assert "enable_fallback" in blob and "num_retries" in blob class TestABrokenJournalIsNotAnEmptyOne: """★★깨진 장부는 「안 샀다」가 **아니라** 「무엇을 샀는지 모른다」다.""" def test_it_stops_instead_of_re_buying_everything(self, tmp_path): from app.modules.pipeline.grounding_chunk_journal import ( ChunkJournal, JournalUnreadable) p = tmp_path / "j.json" p.write_text('{"calls": [반쪽', encoding="utf-8") with pytest.raises(JournalUnreadable, match="무엇을 샀는지 모른다"): ChunkJournal(p, contract={}) def test_a_missing_journal_is_still_a_fresh_start(self, tmp_path): """★positive control — **없는 것**과 **깨진 것**은 다르다.""" from app.modules.pipeline.grounding_chunk_journal import ChunkJournal assert ChunkJournal(tmp_path / "none.json", contract={}).bought() == 0 class TestThePhysicalBoundIsDecidedNotGuessed: """★★글 호출 경로(`_completion`)에는 이미지 쪽 `reserve_current_call` 같은 **세는 자리가 없다.** 그래서 물리 수를 그 자리에서 거절할 수 없다. 대신 **구조적으로 상한을 정한다** — 재시도와 fallback 을 끄면 남는 것은 키 슬롯 loop 뿐이고 그 위는 `slot_count()` 가 정한다. 그 수가 **손으로 적은 승인 수**를 넘으면 아무것도 안 사고 선다. ★이것은 **상한이지 실측이 아니다.** 실제 물리 수는 Opik+provider 로그로 본다. """ class _S(gs.GroundingChunkStep): def __init__(self): self.project_config = {} self.step_id = "grounding_chunk" def test_the_formula_follows_the_request_contract(self): f = gs.physical_per_logical assert f({"enable_fallback": False, "num_retries": 0}, 2) == 2 # ★fallback 3 tier × (1 + 재시도 2) × 슬롯 3 assert f({"enable_fallback": True, "num_retries": 2}, 3) == 27 assert f({"enable_fallback": False, "num_retries": 0}, 0) == 1 def test_the_pinned_contract_keeps_it_to_slots_only(self): from app.core import openai_keys assert gs.physical_per_logical( gs.REQUEST_CONTRACT, openai_keys.slot_count() ) == max(1, openai_keys.slot_count()), \ "★재시도·fallback 이 켜져 있어 상한이 슬롯 수보다 크다" def test_a_plan_over_the_approved_cap_stops_before_buying(self): """★상한은 **설정 SOT** 에서 온다 — 내가 지어낸 수가 아니다. 앞 판은 `APPROVED_PHYSICAL_MAX = 24` 를 코드에 박았다. 사용자 승인도 설정 근거도 없는 수였다 (Codex 2026-08-31, 하드코딩 금지). """ from app.core.errors import AppError st = self._S() st.project_config = {"research_transmission_cap": 4} with pytest.raises(AppError, match="research_transmission_cap"): st._assert_physical_bound(list(range(20))) def test_the_cap_comes_from_the_same_key_the_research_step_uses(self): """★새 설정 키를 만들지 않는다 — `grounding_research` 와 같은 것.""" import inspect as _i from app.core.steps import grounding_steps as gsteps mine = _i.getsource(gs.GroundingChunkStep._transmission_cap) assert "research_transmission_cap" in mine assert "research_transmission_cap" in _i.getsource(gsteps) def test_a_plan_inside_the_bound_passes(self): """★막는 조건을 넓혔으면 **걸릴 정상 계획**을 대 본다.""" assert self._S()._assert_physical_bound([1, 2]) > 0 def test_it_is_checked_before_the_run_scope_opens(self): """★★**문자열 위치로 보지 않는다.** `research_run_scope` 는 import 줄에도 나와서 그것이 먼저 걸린다 — 오늘 세 번째로 같은 부류다. `with` 문과 호출의 **줄 번호**를 AST 로 견준다. """ import ast import inspect as _i import textwrap tree = ast.parse(textwrap.dedent( _i.getsource(gs.GroundingChunkStep._execute))) scope = [n for n in ast.walk(tree) if isinstance(n, ast.With) and any(isinstance(i.context_expr, ast.Call) and getattr(i.context_expr.func, "id", "") == "research_run_scope" for i in n.items)] checks = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "attr", "") == "_assert_physical_bound"] assert len(scope) == 1 and len(checks) == 1 assert checks[0].lineno < scope[0].lineno, \ "★예산을 연 뒤에 상한을 본다 — 순서가 뒤집혔다" def test_the_output_says_it_is_a_bound_not_a_count(self): import inspect as _i src = _i.getsource(gs.GroundingChunkStep._wrap) assert "physical_upper_bound" in src assert "transmission_cap" in src # ★이름이 「몇 번 샀다」로 읽히면 안 된다 assert "physical_calls" not in src class TestThePhysicalReserveIsAtTheProviderBoundary: """★★물리 전송을 **그 자리에서** 막는다 — 키 슬롯 loop 안, `router.completion` **직전**. 논리 호출당 한 번 세면 primary→secondary 두 전송을 못 본다 (Codex 2026-08-31). ★`research_calls_armed()` **밖에서는 no-op** 이라 다른 글 호출부 (`scene_detail` 팬아웃 등)는 한 글자도 안 바뀐다. """ class _Binding: slot = 0 class router: n = 0 @staticmethod def completion(**kw): TestThePhysicalReserveIsAtTheProviderBoundary._Binding \ .router.n += 1 if TestThePhysicalReserveIsAtTheProviderBoundary._Binding \ .router.n == 1: raise RuntimeError("primary 죽음") return {"ok": True} @pytest.fixture() def wired(self, monkeypatch): from app.core import openai_keys from app.modules.llm import llm_client as lc self._Binding.router.n = 0 monkeypatch.setattr(openai_keys, "slot_count", lambda: 2) monkeypatch.setattr(openai_keys, "failover_on", lambda *a, **k: True) monkeypatch.setattr(lc, "_get_router_binding", lambda: self._Binding) return lc def test_unarmed_calls_are_unchanged(self, wired): """★positive control — 팔을 안 든 호출부는 **그대로 돈다**.""" wired._completion(self._Binding, "m", {}) assert self._Binding.router.n == 2 def test_both_transmissions_are_reserved(self, wired): from app.core.research_call_budget import (research_calls_armed, research_run_scope) with research_run_scope(cap=2) as b: with research_calls_armed(): wired._completion(self._Binding, "m", {}) assert self._Binding.router.n == 2 assert b.snapshot()["used"] == 2, \ "★두 번 나갔는데 예약이 두 번이 아니다" def test_a_cap_of_one_blocks_the_secondary(self, wired): from app.core.research_call_budget import (ResearchCallBudgetExceeded, research_calls_armed, research_run_scope) with research_run_scope(cap=1): with pytest.raises(ResearchCallBudgetExceeded): with research_calls_armed(): wired._completion(self._Binding, "m", {}) assert self._Binding.router.n == 1, \ "★상한 1 인데 secondary 가 나갔다" def test_parallel_workers_share_one_cap(self, wired): """★worker 마다 따로 세면 상한이 worker 수만큼 늘어난다.""" from concurrent.futures import ThreadPoolExecutor from app.core.research_call_budget import ( bind_current_research_budget, research_calls_armed, research_run_scope) self._Binding.router.n = 0 monkeyed = self._Binding class _Ok(monkeyed): class router: n = 0 @staticmethod def completion(**kw): _Ok.router.n += 1 return {"ok": True} import app.modules.llm.llm_client as lc lc._get_router_binding = lambda: _Ok ok, denied = [], [] def _go(_): from app.core.research_call_budget import ( ResearchCallBudgetExceeded) try: with research_calls_armed(): lc._completion(_Ok, "m", {}) ok.append(1) except ResearchCallBudgetExceeded: denied.append(1) with research_run_scope(cap=3): with ThreadPoolExecutor(max_workers=8) as ex: list(ex.map(bind_current_research_budget(_go), range(8))) assert _Ok.router.n == 3, f"★상한 3 인데 {_Ok.router.n}번 나갔다" assert len(denied) == 5 def test_the_reserve_is_inside_the_slot_loop(self): """★loop **밖**에 두면 primary→secondary 를 한 번으로 센다.""" import ast import inspect as _i import textwrap from app.modules.llm import llm_client as lc tree = ast.parse(textwrap.dedent(_i.getsource(lc._completion))) loops = [n for n in ast.walk(tree) if isinstance(n, ast.For)] assert len(loops) == 1 inside = [n for n in ast.walk(loops[0]) if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "reserve_current_research_call"] assert len(inside) == 1, "★예약이 슬롯 loop 안에 없다" class TestTheJournalCountsOnlyThisEpoch: """★★계약을 바꾼 뒤 **첫 구매가 영구히 막히던 것** (Codex 재현).""" def _j(self, tmp_path, acq): from app.modules.pipeline.grounding_chunk_journal import ChunkJournal return ChunkJournal(tmp_path / "j.json", contract={"acq": acq}) def test_a_new_acquisition_contract_starts_a_fresh_budget(self, tmp_path): from app.modules.pipeline.grounding_chunk_journal import buy_or_reuse buy_or_reuse(self._j(tmp_path, "old"), "id-old", cap=1, send=lambda: {"r": 1}) new = self._j(tmp_path, "new") assert new.bought() == 0, "★옛 계약 줄을 이 판으로 센다" sent = [] buy_or_reuse(new, "id-new", cap=1, send=lambda: sent.append(1) or {"r": 2}) assert sent == [1], "★계약을 바꾼 뒤 첫 구매가 막혔다" def test_the_old_rows_are_kept_for_audit(self, tmp_path): from app.modules.pipeline.grounding_chunk_journal import buy_or_reuse buy_or_reuse(self._j(tmp_path, "old"), "id-old", cap=1, send=lambda: {"r": 1}) new = self._j(tmp_path, "new") assert new.kept_from_other_epochs() == 1, "★옛 줄을 버렸다" def test_an_old_epoch_answer_is_not_reused(self, tmp_path): """★다른 팩·모델로 산 답은 **같은 대상의 답이 아니다**.""" from app.modules.pipeline.grounding_chunk_journal import buy_or_reuse buy_or_reuse(self._j(tmp_path, "old"), "same-id", cap=2, send=lambda: {"r": "old"}) new = self._j(tmp_path, "new") got = buy_or_reuse(new, "same-id", cap=2, send=lambda: {"r": "new"}) assert got == {"r": "new"}, "★옛 계약의 답을 되썼다" def test_processing_only_change_keeps_the_same_epoch(self, tmp_path): """★positive control — 해석만 바꾸면 **되써야** 한다.""" from app.modules.pipeline.grounding_chunk_journal import buy_or_reuse buy_or_reuse(self._j(tmp_path, "same"), "x", cap=2, send=lambda: {"r": 1}) again = self._j(tmp_path, "same") calls = [] got = buy_or_reuse(again, "x", cap=2, send=lambda: calls.append(1) or {"r": 2}) assert got == {"r": 1} and calls == [] class TestTheLogicalCapIsAtomic: """★★`get → bought → send` 가 잠금 밖이라 **상한 1 에 12개가 나갔다**.""" def test_only_one_buy_gets_through_a_cap_of_one(self, tmp_path): from concurrent.futures import ThreadPoolExecutor from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) jr = ChunkJournal(tmp_path / "j.json", contract={}) sent, errs = [], [] def _go(i): try: buy_or_reuse(jr, f"id{i}", cap=1, send=lambda: sent.append(i) or {"r": i}) except Exception as exc: # noqa: BLE001 errs.append(type(exc).__name__) with ThreadPoolExecutor(max_workers=12) as ex: list(ex.map(_go, range(12))) assert len(sent) == 1, f"★상한 1 인데 {len(sent)}번 보냈다" assert len(errs) == 11 def test_the_same_identity_is_bought_once(self, tmp_path): from concurrent.futures import ThreadPoolExecutor from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal, buy_or_reuse) jr = ChunkJournal(tmp_path / "j.json", contract={}) sent = [] got = [] with ThreadPoolExecutor(max_workers=8) as ex: got = list(ex.map(lambda _: buy_or_reuse( jr, "same", cap=5, send=lambda: (sent.append(1), {"r": 9})[1]), range(8))) assert len(sent) == 1, f"★같은 신원을 {len(sent)}번 샀다" assert all(g == {"r": 9} for g in got) assert jr.reused() == 7