"""★대상 병렬 (2026-09-03 · Codex 계약 7): bounded pool · 1차 전원 join 뒤 2차 · 결과 순서 결정적 · 동시 폭 정확히 N · 예산/정지 표 워커 전달 · 재개 중복 구매 0 · 겹친 subject_id fail-closed · 문(OutboundBudget) 원자적 · 조사 저작기는 legacy alias 를 안 낸다. 무료 · 바깥 호출 0.""" from __future__ import annotations import threading import time from pathlib import Path import pytest from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import grounding_chunk_journal as cj from app.modules.pipeline import grounding_outlook_binding as ob from app.modules.pipeline import grounding_target_research as gtr from app.modules.pipeline import reference_acquisition as ra from tests.grounding.test_central_acquisition import P3, _Spy, _row # noqa: E402 — 진짜 fixture 재사용 ERA, REGION = "가나다 무렵", "라마바 지방" def _research(client, *, skeleton, evidence, opik_metadata=None, **_kw): kind = skeleton["coarse_type_label"] return {"what_it_is": f"{kind}", "appearance_criteria": f"- {kind} 생김새", "narrow_queries": [f"{REGION} {ERA} {kind} 사진"], "rough_queries": [f"{REGION} {ERA} {kind}"], "search_directive_native": f"{REGION} {ERA} 의 {kind}", "language_lock_native": "한국어", "sources": [], "provenance": {"provider": "fake"}} def _writer(calls=None): calls = calls if calls is not None else [] def _rc(client, **kw): calls.append(kw); return _research(client, **kw) w = gtr.make_writer(world_facts="w", source_text="원문", era=ERA, region=REGION, research_call=_rc, client=object()) w.calls = calls return w def _ledger(n): return ob.bind([_row(f"rs{i}") for i in range(n)], P3) class _Gauge: """동시에 몇 워커가 안에 있나 — 최대치를 센다.""" def __init__(self): self.lock = threading.Lock(); self.now = 0; self.peak = 0; self.order = [] def __enter__(self): with self.lock: self.now += 1; self.peak = max(self.peak, self.now) def __exit__(self, *a): with self.lock: self.now -= 1 def _run(led, tmp_path, *, workers, judge_match=True, gauge=None, delay=None, write_brief=None, journal=None, cap=99): spy = _Spy(match=judge_match) g = gauge or _Gauge() def _search(**kw): with g: if delay: delay(kw) g.order.append(kw.get("directive_native")) return spy.search(**kw) out = ca.run(led, journal=journal or cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}), cap=cap, workdir=tmp_path, rel_root=tmp_path, search=_search, download=spy.download, judge=spy.judge, write_brief=write_brief or _writer(), workers=workers) return out, g, spy class TestTheBoundedPool: def test_the_output_order_is_the_target_order_even_when_the_last_finishes_first(self, tmp_path): led = _ledger(6) n = {"k": 0} lock = threading.Lock() def _delay(kw): # ★늦게 시작한 워커일수록 빨리 끝난다 — 완료 순서가 뒤집혀도 출력은 대상 순서 with lock: n["k"] += 1; k = n["k"] time.sleep(0.03 * max(0, 6 - k)) out, g, _ = _run(led, tmp_path, workers=4, delay=_delay) sids = [r["research_subject_id"] for r in out["rows"] if r.get("disposition") == ca.DISP_ACQUIRED] assert sids == [f"rs{i}" for i in range(6)], sids assert g.peak >= 2, "★병렬이 아니었다" def test_at_most_n_workers_are_inside_at_once(self, tmp_path): led = _ledger(8) out, g, _ = _run(led, tmp_path, workers=3, delay=lambda kw: time.sleep(0.05)) assert g.peak == 3, f"★동시 폭이 {g.peak} 다 — 정확히 3 이어야 한다" assert len([r for r in out["rows"] if r.get("disposition") == ca.DISP_ACQUIRED]) == 8 def test_no_second_pass_starts_before_every_first_pass_ended(self, tmp_path): led = _ledger(4) seen = [] lock = threading.Lock() def _delay(kw): with lock: seen.append(("p2" if "좁" in str(kw.get("directive_native")) or len(seen) >= 4 else "p1")) out, g, spy = _run(led, tmp_path, workers=2, judge_match=False, delay=_delay) # 1차는 4번 · 2차는 그 뒤 4번 — 검색 호출 8번, 앞 넷은 전부 1차 assert spy.searched == 8, spy.searched assert seen[:4] == ["p1"] * 4, seen def test_workers_see_the_research_budget_and_the_stop_check(self, tmp_path): from app.core import research_call_budget as rb seen = {"budget": [], "thread": set()} def _delay(kw): seen["budget"].append(rb.get_current_budget() is not None) seen["thread"].add(threading.get_ident()) budget = rb.ResearchCallBudget(cap=50) if hasattr(rb, "ResearchCallBudget") else None rb.install_budget(budget) try: _run(_ledger(4), tmp_path, workers=4, delay=_delay) finally: rb.uninstall_budget() assert all(seen["budget"]), "★워커가 바깥 예산을 못 본다 (thread-local)" assert threading.get_ident() not in seen["thread"] or len(seen["thread"]) > 1 def test_a_duplicate_subject_id_is_refused_before_the_pool(self, tmp_path): led = _ledger(2) rows = led["rows"] rows[1]["research_subject_id"] = rows[0]["research_subject_id"] with pytest.raises(ValueError): _run(led, tmp_path, workers=2) def test_a_resume_buys_nothing_twice(self, tmp_path): led = _ledger(5) j = cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}) out1, _g, spy1 = _run(led, tmp_path, workers=4, journal=j) j2 = cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}) out2, _g2, spy2 = _run(led, tmp_path, workers=4, journal=j2) assert spy2.searched == 0, "★재개가 다시 샀다" assert [r["identity"] for r in out1["rows"]] == [r["identity"] for r in out2["rows"]] def test_the_cap_is_the_journal_gate_not_a_shared_flag(self, tmp_path): led = _ledger(6) out, _g, spy = _run(led, tmp_path, workers=4, cap=3) got = [r for r in out["rows"] if r.get("disposition") == ca.DISP_ACQUIRED] capped = [r for r in out["rows"] if r.get("disposition") == ca.DISP_CAP_REACHED] assert len(got) == 3 and len(capped) == 3, (len(got), len(capped)) assert spy.searched == 3, "★상한 뒤에도 네트워크를 썼다" assert out["purchases"]["cap_reached"] is True def test_workers_are_bounded_by_the_module_maximum(self): assert ca.resolve_workers(999) == ca.MAX_ACQUIRE_WORKERS with pytest.raises(ca.BadPurchaseCap): ca.resolve_workers(0) with pytest.raises(ca.BadPurchaseCap): ca.resolve_workers(True) class TestTheResearchWriterDoesNotAlias: def test_an_old_selected_purchase_under_the_old_identity_is_not_aliased(self, tmp_path): """★Codex 긴급 BLOCK (2026-09-03): 옛 뼈대 신원으로 산 selected 줄이 있어도 새 조사 신원은 그것을 잇지 않는다 — 조사가 실제로 1회 나가고 alias 정산 0.""" from app.modules.pipeline import grounding_search_brief as gsb led = _ledger(1) j = cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}) # 옛 저작기(뼈대 신원)로 한 번 산다 → selected 줄 old_writer = gsb.make_writer(world_facts="w", source_text="원문", era=ERA, region=REGION, call=lambda *a, **k: {"search_directive_native": f"{REGION} {ERA} x", "search_terms_native": [f"{REGION} {ERA} x"], "language_lock_native": "한국어"}) spy = _Spy(match=True) ca.run(led, journal=j, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge, write_brief=old_writer, workers=1) assert j.bought() == 1 # 새 조사 저작기로 같은 대상 → alias 0 · 조사 1회 · 다시 산다 w = _writer() j2 = cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}) spy2 = _Spy(match=True) ca.run(led, journal=j2, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy2.search, download=spy2.download, judge=spy2.judge, write_brief=w, workers=1) assert len(w.calls) == 1, "★새 웹 조사가 안 나갔다" assert sum(1 for s_ in j2.settlements if s_.get("kind") == "alias") == 0, "★옛 구매에 alias 했다" assert spy2.searched >= 1 assert w.legacy_target_identity is None and w.legacy_identity_inputs is None class TestTheOutboundGateIsAtomic: def test_many_threads_cannot_exceed_the_cap(self): from tools.grounding_audit import canary_outbound_gates as og b = og.OutboundBudget("검색", 5) hits = [] def _go(): try: b.reserve(where="t"); hits.append(1) except og.OutboundDenied: hits.append(0) ts = [threading.Thread(target=_go) for _ in range(40)] [t.start() for t in ts]; [t.join() for t in ts] assert b.used == 5 and b.denied == 35 and sum(hits) == 5 raw = og.RawFetchObserver() ts = [threading.Thread(target=raw.bump) for _ in range(50)] [t.start() for t in ts]; [t.join() for t in ts] assert raw.count == 50 class TestAFatalWorkerStopsThePending: def test_pending_jobs_never_reach_the_provider_after_a_fatal(self, tmp_path): """★Codex BLOCK (2026-09-03): job 0 은 오래 걸리고 job 1 은 즉시 fatal · workers=2 → job 2..N 은 provider 경계에 0 회. 도는 job 0 은 join 되고 원예외가 올라온다.""" class _Fatal(BaseException): """검색 안의 Exception 은 라운드에 접힌다(설계) — 멈추라는 말은 BaseException 으로 온다.""" led = _ledger(6) started = [] lock = threading.Lock() def _delay(kw): with lock: started.append(kw.get("directive_native")) k = len(started) if k == 1: time.sleep(0.4) elif k == 2: raise _Fatal("멈춰라") with pytest.raises(_Fatal): _run(led, tmp_path, workers=2, delay=_delay) assert len(started) == 2, f"★fatal 뒤에도 {len(started) - 2}개가 더 나갔다"