"""참조 획득 구간에 **물리 전송 상한**이 실제로 걸리나. ★유료 0. ★★★2026-08-31 발견 — **여기가 비어 있었다.** `reserve_current_research_call` 은 provider 경계 둘에 이미 박혀 있는데 `research_calls_armed()` **밖에서는 아무 일도 안 한다**. `ref_canary` 는 한 번도 팔을 안 들어서 참조 획득 구간에 **나가는 자리의 상한이 없었다** — 내가 센 논리 수만 있었다. 논리 1이 물리 24까지 간다(tier 3 × 키슬롯 2 × router 4). 대상 5개면 이론상 수백이다. 「실제로는 적게 썼다」는 변명이 못 된다 — 승인은 **나갈 수 있는 최대**에 대한 것이다. ★그래서 이 시험은 **조립부를 안 본다.** `acquire_one` 이 불리는 **그 순간** 팔이 들려 있고 예산이 깔려 있나를 본다. """ from __future__ import annotations import json import sys import pytest from tools.grounding_audit import ref_canary as rc class TestTheBoundIsReadFromRealConstants: """★손으로 안 적는다 — 실제 상수에서 읽는다.""" def test_it_reports_all_three_layers(self): got = rc.physical_upper_bound(10) L = got["layers"] assert L["tiers"] >= 1 and L["key_slots"] >= 1 and L["router_tries"] >= 1 assert L["per_structured_call"] == \ L["tiers"] * L["key_slots"] * L["router_tries"] def test_the_raw_http_bound_is_above_the_counted_one(self): """★★**세는 수와 나가는 HTTP 수는 다르다** (Codex 2026-08-31). 앞 판은 둘을 같다고 적었는데 거짓이었다 — litellm 이 제 client 에 주는 `max_retries=2` 가 `reserve` **아래**라 안 세어진다. """ got = rc.physical_upper_bound(30, targets=5) assert got["raw_http_upper_bound"] > got["counted_attempt_cap"], \ "★안 세어지는 겹을 0 으로 적고 있다" assert got["layers"]["litellm_sdk_max_retries"] >= 1 assert got["layers"]["search_sdk_max_retries"] == 0, \ "★검색 경로는 잠갔어야 한다" def test_the_counted_cap_is_at_least_the_logical_count(self): got = rc.physical_upper_bound(10) assert got["counted_attempt_cap"] >= got["logical"], \ "★재시도 겹을 하나도 안 세고 있다" def test_the_armed_cap_does_not_follow_the_worst_case(self, monkeypatch): """★★코드가 제 승인 범위를 **스스로 넓히면** 안 된다. 겹을 늘려도 **거는 상한은 안 움직인다** — 손으로 적은 배수에서만 나온다. 앞 판은 `armed_cap < worst_case` 만 봤는데, 계약을 잠가 둘이 같아지자 깨졌다. 재는 것은 **크기 비교가 아니라 유래**다. """ base = rc.physical_upper_bound(10) monkeypatch.setattr(rc, "CALL_STRUCTURED_TIERS", 9) monkeypatch.setattr(rc, "REQUEST_LOCK", {"num_retries": 7, "enable_fallback": True}) wide = rc.physical_upper_bound(10) assert wide["counted_attempt_cap"] > base["counted_attempt_cap"], \ "★겹을 안 센다" assert wide["armed_cap"] == base["armed_cap"] == \ 10 * rc.APPROVED_PHYSICAL_FACTOR, "★상한이 최악값을 따라간다" def test_the_bound_reflects_the_locked_contract(self): """★기본값이 아니라 **실제로 넘기는 인자**로 센다.""" got = rc.physical_upper_bound(10) assert got["request_lock"] == rc.REQUEST_LOCK if not rc.REQUEST_LOCK.get("enable_fallback", True): assert got["layers"]["tiers"] == 1, "★안 쓰는 tier 를 센다" assert got["layers"]["router_tries"] == \ 1 + int(rc.REQUEST_LOCK["num_retries"]) def test_it_names_the_layer_below_the_count(self): """★★`reserve` **아래**에 SDK 재시도가 또 있다 — 예산이 못 본다. 「HTTP 요청 수」와 「센 수」가 다르다는 것을 감사 전에 적어 둔다. """ L = rc.physical_upper_bound(1)["layers"] assert "litellm_sdk_max_retries" in L and "counted_below_this" in L class TestTheLockActuallyReachesTheCall: """★★**끝점에서** 잰다 — 「넘긴다」고 적은 것과 나가는 것은 다르다.""" def test_the_writer_call_carries_the_lock(self, monkeypatch): seen = {} def _fake(tag, system, user, schema, **kw): seen.update(kw) return {"source_language": "ko", "search_directive_native": "찾아라" * 20, "search_terms_native": ["가", "나", "다"], "language_lock_native": "원어로만"} import app.modules.llm.llm_client as lc monkeypatch.setattr(lc, "call_structured", _fake) rc._write_brief({"owner_type": "prop", "surface_form": "표기", "coarse_type_label": "부류", "visual_brief": ""}, world_facts="W", source_text="S", narrow=False) for k, v in rc.REQUEST_LOCK.items(): assert seen.get(k) == v, f"★{k} 가 안 나간다: {seen.get(k)!r}" def test_the_lock_moves_the_identity(self, monkeypatch): """★계약이 바뀌면 **다른 방식으로 사는 것**이다.""" ctx = {"world_facts": "W", "source_text": "S", "era_declaration": "E", "region_declaration": "R"} t = {"owner_type": "prop", "surface_form": "표기", "coarse_type_label": "부류", "visual_brief": "", "subject_id": "P01"} a = rc.acquisition_identity(t, pack="1", models={}, context=ctx) monkeypatch.setattr(rc, "REQUEST_LOCK", {"num_retries": 3, "enable_fallback": True}) b = rc.acquisition_identity(t, pack="1", models={}, context=ctx) assert a != b, "★요청 계약이 바뀌어도 같은 신원이다" def five_owner_targets(): """★정본 다섯 갈래 각 1개. 게이트가 이것을 요구한다.""" from app.modules.pipeline.grounding_entity_contract import owners return [{"subject_id": f"S{i}", "owner_type": o, "surface_form": "표기", "coarse_type_label": "부류", "visual_brief": "", "local_id": f"c0#{i}"} for i, o in enumerate(sorted(owners()))] class TestTheLoopRunsInsideTheBudget: """★★**끝점에서** 잰다 — `acquire_one` 이 불리는 그 순간을 본다.""" @staticmethod def _drive(monkeypatch, tmp_path, *, targets, capture, sample="1"): """`main` 을 몰되 바깥 호출은 하나도 안 한다.""" from app.core.research_call_budget import (get_current_budget, is_armed) rows = [{"local_id": f"c0#{i}", "occurrences": [], "shot_appearance_ids": []} for i in range(len(targets))] monkeypatch.setattr(rc.rr, "run", lambda *a, **k: { "reduced": {"rows": rows}, "logical": 3, "bought": 0, "reused": 3, "quarantined": [], "run_id": "test", "processing_stamps": {}, "dispatched": 0, "dispatch_budget": 3}) monkeypatch.setattr(rc, "targets_from", lambda _r: list(targets)) monkeypatch.setattr(rc, "parts_skipped", lambda _r: []) def _fake_acquire(t, **kw): b = get_current_budget() capture.append({ "armed": is_armed(), "cap": (b.snapshot()["cap"] if b else None), }) return {"subject_id": t["subject_id"], "status": "selected", "chosen": None, "rounds": [], "candidates": []} # ★`rar` 는 `main` **안에서** import 된다 — 원 모듈을 갈아야 닿는다 from app.modules.pipeline import reference_acquisition_rounds as rar monkeypatch.setattr(rar, "acquire_one", _fake_acquire) jp = tmp_path / "j.json" monkeypatch.setattr( sys, "argv", ["ref_canary", "--live", str(jp), "fixture=period", # ★`rr.run` 을 갈아 끼운 시험이라 판독 구매가 없다 — 문을 명시로 연다 "allow_chunk_dispatch=1"] + ([f"sample={sample}"] if sample else [])) rc.main() return jp def test_acquire_one_is_called_with_the_arm_up(self, monkeypatch, tmp_path): seen = [] targets = five_owner_targets() self._drive(monkeypatch, tmp_path, targets=targets, capture=seen, sample=None) assert seen, "★`acquire_one` 이 아예 안 불렸다 — 시험이 죽었다" assert all(x["armed"] for x in seen), \ "★팔을 안 들었다 — 나가는 자리 상한이 no-op 이다" def test_the_cap_matches_the_approved_number(self, monkeypatch, tmp_path): seen = [] targets = five_owner_targets() # ★표본을 안 뽑는다 — 다섯 다 산다 self._drive(monkeypatch, tmp_path, targets=targets, capture=seen, sample=None) assert len(seen) == 5, f"★표본이 줄었다: {len(seen)}" # ★★판독 몫은 **이 범위 밖**이다 — 신규 논리에만 건다 want = rc.physical_upper_bound( 5 * rc.PER_TARGET_LOGICAL)["armed_cap"] assert {x["cap"] for x in seen} == {want}, \ f"★건 상한이 승인 수와 다르다: {seen}" def test_the_budget_is_gone_afterwards(self, monkeypatch, tmp_path): """★이 주행이 끝난 뒤 **남의 호출**이 이 상한을 보면 안 된다.""" from app.core.research_call_budget import get_current_budget, is_armed seen = [] self._drive(monkeypatch, tmp_path, targets=five_owner_targets(), capture=seen, sample=None) assert not is_armed() and get_current_budget() is None class TestItStopsBeforeTheNetworkWhenTheCapIsHit: """★상한에 닿으면 **네트워크에 닿기 전에** 오른다.""" def test_the_reserve_raises_at_the_cap(self): from app.core.research_call_budget import (ResearchCallBudgetExceeded, research_calls_armed, research_run_scope, reserve_current_research_call) with research_run_scope(cap=2), research_calls_armed(): reserve_current_research_call(source="t") reserve_current_research_call(source="t") with pytest.raises(ResearchCallBudgetExceeded): reserve_current_research_call(source="t") def test_outside_the_arm_it_does_nothing(self): """★positive control 의 짝 — 안 들면 **아무 일도 안 한다**.""" from app.core.research_call_budget import (research_run_scope, reserve_current_research_call) with research_run_scope(cap=0): reserve_current_research_call(source="t") # ★안 오른다 class TestTheGateStopsBeforeAnythingIsBought: """★★★**사기 직전의 문** — 장부도 안 열고 provider 도 안 부른다. 앞 판은 「참조 획득을 안 지나는 갈래」를 **찍기만 하고 그냥 갔다** (Codex 2026-08-31). 이 판의 목적이 다섯 갈래 각 1개인데 갈래가 빠지면 그 판은 목적을 못 이룬 채 돈만 쓴다. """ def test_a_missing_owner_stops_it(self): t = five_owner_targets()[:-1] with pytest.raises(rc.ApprovedPlanMismatch) as e: rc.assert_plan_matches_approval(t) assert "다섯 갈래" in str(e.value) def test_a_doubled_owner_stops_it(self): t = five_owner_targets() t.append({**t[0], "subject_id": "DUP"}) with pytest.raises(rc.ApprovedPlanMismatch): rc.assert_plan_matches_approval(t) def test_the_full_set_passes(self): rc.assert_plan_matches_approval(five_owner_targets()) def test_a_changed_sample_stops_it(self): """★사기 **전에** 적어 둔 것과 다르면 선다 — 결과를 보고 안 넓힌다.""" t = five_owner_targets() pf = {"sample": [{"subject_id": x["subject_id"]} for x in t]} rc.assert_plan_matches_approval(t, pf) # ★같으면 지난다 moved = [{**x, "subject_id": x["subject_id"] + "X"} for x in t] with pytest.raises(rc.ApprovedPlanMismatch) as e: rc.assert_plan_matches_approval(moved, pf) assert "preflight" in str(e.value) def test_a_different_slot_count_stops_it(self, monkeypatch): """★물리 상한 계산이 승인 장부와 갈리면 안 된다.""" import app.core.openai_keys as ok monkeypatch.setattr(ok, "slot_count", lambda: 3) with pytest.raises(rc.ApprovedPlanMismatch) as e: rc.assert_plan_matches_approval(five_owner_targets()) assert "슬롯" in str(e.value) class TestTheEventsTieTheJournalToOpik: """★참조 장부는 대상마다 **최종 1줄**이라 6번 전송을 못 되짚는다.""" def test_each_stage_and_round_gets_its_own_identity(self): from tools.grounding_audit import ref_events as ev base = dict(target_identity="T", outbound={"q": ["가"]}) a = ev.call_identity(stage=ev.STAGE_WRITE, round_no=1, **base) b = ev.call_identity(stage=ev.STAGE_SEARCH, round_no=1, **base) c = ev.call_identity(stage=ev.STAGE_WRITE, round_no=2, **base) assert len({a, b, c}) == 3, "★단계·라운드가 안 갈린다" def test_the_outbound_moves_it(self): from tools.grounding_audit import ref_events as ev a = ev.call_identity("T", stage=ev.STAGE_SEARCH, round_no=1, outbound={"q": ["가"]}) b = ev.call_identity("T", stage=ev.STAGE_SEARCH, round_no=1, outbound={"q": ["나"]}) assert a != b, "★나간 것이 달라도 같은 신원이다" def test_it_appends_and_survives(self, tmp_path): """★줄마다 fsync — 중간에 죽어도 **산 것까지는** 남는다.""" from tools.grounding_audit import ref_events as ev e = ev.RefEvents(tmp_path / "e.jsonl", run_nonce="N1") e.put(target_identity="T", stage=ev.STAGE_WRITE, round_no=1, outbound={"a": 1}) e.put(target_identity="T", stage=ev.STAGE_SEARCH, round_no=1, outbound={"b": 2}) again = ev.RefEvents(tmp_path / "e.jsonl", run_nonce="N1") assert again.tally()["total"] == 2 assert again.tally()["by_stage"] == {ev.STAGE_WRITE: 1, ev.STAGE_SEARCH: 1} def test_an_unknown_stage_is_refused(self): from tools.grounding_audit import ref_events as ev with pytest.raises(ValueError): ev.call_identity("T", stage="지어낸단계", round_no=1, outbound={}) class TestNoTraceMeansNoPurchase: """★★trace 가 None 이면 **provider 0회**여야 한다 (Codex 2026-08-31). `open_trace` 는 설정이 꺼지거나 client 생성이 실패하면 **예외 없이 None** 을 준다 — production 은 그게 맞다(기록이 본 작업을 막으면 안 된다). 그런데 **이 판은 기록이 목적**이다. 결속 키 없이 사면 나중에 무엇을 샀는지 못 되짚는다. """ def test_it_stops_before_acquire_one(self, monkeypatch, tmp_path): import contextlib import sys as _sys from app.modules.pipeline import reference_acquisition_rounds as rar called = [] monkeypatch.setattr(rar, "acquire_one", lambda *a, **k: called.append(1)) @contextlib.contextmanager def _dead(**_kw): yield None # ★기록이 꺼진 상태 import app.modules.llm.opik_trace as ot monkeypatch.setattr(ot, "open_trace", _dead) targets = five_owner_targets() rows = [{"local_id": t["local_id"], "occurrences": [], "shot_appearance_ids": []} for t in targets] monkeypatch.setattr(rc.rr, "run", lambda *a, **k: { "reduced": {"rows": rows}, "logical": 3, "bought": 0, "reused": 3, "quarantined": [], "run_id": "t", "processing_stamps": {}, "dispatched": 0, "dispatch_budget": 3}) monkeypatch.setattr(rc, "targets_from", lambda _r: list(targets)) monkeypatch.setattr(rc, "parts_skipped", lambda _r: []) monkeypatch.setattr( _sys, "argv", ["ref_canary", "--live", str(tmp_path / "j.json"), "fixture=period", "allow_chunk_dispatch=1"]) with pytest.raises(rc.rr.TraceUnavailable): rc.main() assert not called, "★trace 없이 샀다" class TestTheEventCarriesTheTraceId: """★★★실측 결함 (2026-08-31) — `trace_id` 가 **통째로 비어 있었다**. `TraceHandle` 은 `uid` 를 갖는데 코드가 `id`/`trace_id` 를 찾아 전부 None 이었다. 유료 판 17줄이 다 `trace_id: null` 로 남았다. 「적었다」와 「이어진다」는 다르다. """ def test_the_handle_field_is_uid(self): from app.modules.llm.opik_trace import TraceHandle h = TraceHandle(uid="U1", name="n") assert h.uid == "U1" assert not hasattr(h, "trace_id") and not hasattr(h, "id"), \ "★코드가 찾던 이름이 실제로 있으면 이 시험은 무의미하다" def test_the_runner_reads_uid(self): import inspect src = inspect.getsource(rc.main) assert 'getattr(_tr, "uid"' in src, "★`uid` 를 안 읽는다" def test_a_handle_without_uid_stops_it(self): """★못 이으면 **선다** — 결속 없는 기록을 사고 나서 알면 늦다.""" import inspect src = inspect.getsource(rc.main) i = src.index('getattr(_tr, "uid"') j = src.index("acquire_one", i) assert "TraceUnavailable" in src[i:j], "★uid 가 없어도 그냥 산다"