"""Codex BLOCK 둘 (2026-09-02 밤): ① 검색 저작 입력 = 뼈대 하나 — 나가는 것과 신원이 같은 객체. 세부 묘사·근거 문장·원문은 둘 다에서 뺀다. ② 아웃룩은 final_id 당 의무 하나 — 조각은 details 로만. + 옛 신원의 구매는 alias 로 잇는다(재구매 0) · provider 거절은 빈 자리 · 재판정 lane 은 provider 실패에 안 죽는다. """ from __future__ import annotations import pytest from app.modules.pipeline import grounding_search_brief as sb from app.modules.pipeline import grounding_chunk_journal as cj def _target(**over): t = {"subject_id": "rs_x", "owner_type": "outlook", "coarse_type_label": "학생 교복", "terms_native": ["1960년대 교복", "학생 교복 사진"], "language_lock_native": "ko", "directive_native": "검정 교복 윗도리와 바지, 깃 안쪽에 흰 천", "surface_form": "검은 교복 윗도리", "visual_brief": "깃 안쪽에 누렇게 때가 탄 흰 천이 덧대어진 검정 교복"} t.update(over); return t class TestSkeletonIsWhatGoesOutAndWhatIsIdentified: def test_outbound_carries_the_skeleton_only(self): o = sb.outbound(_target(), world_facts="세계 사실 전문", source_text="원고 전문 표본", narrow=False, era="1960년대", region="대한민국") assert "세계 사실 전문" not in o["user"] and "원고 전문" not in o["user"] assert "누렇게 때가 탄" not in o["user"], "★세부 묘사가 나갔다" assert "학생 교복" in o["user"] and "1960년대 교복" in o["user"] and "1960년대" in o["user"] and "대한민국" in o["user"] assert o["skeleton"] == sb.skeleton_search_input(_target(), era="1960년대", region="대한민국") def test_identity_ignores_prose_but_moves_with_the_skeleton(self): base = dict(world_facts="w", source_text="s", era="1960년대", region="대한민국") a = sb.target_identity(_target(), **base) assert sb.target_identity(_target(visual_brief="다른 묘사", directive_native="다른 지시", surface_form="다른 표기"), **base) == a assert sb.target_identity(_target(), **{**base, "world_facts": "다른 세계 사실", "source_text": "다른 원고"}) == a assert sb.target_identity(_target(terms_native=["다른 검색어"]), **base) != a assert sb.target_identity(_target(coarse_type_label="두루마기"), **base) != a assert sb.target_identity(_target(), **{**base, "era": "1970년대"}) != a assert "world_facts" not in sb.identity_inputs(world_facts="w", era="e", region="r") def test_legacy_identity_still_moves_with_prose(self): base = dict(world_facts="w", source_text="s", era="1960년대", region="대한민국") assert sb.legacy_target_identity(_target(), **base) != sb.legacy_target_identity(_target(visual_brief="다른 묘사"), **base) class TestOldPurchasesAreAliasedNotRebought: def test_alias_joins_the_old_line_and_takes_no_new_slot(self, tmp_path): j = cj.ChunkJournal(tmp_path / "j.json", contract={"acquisition": {"x": 1}}) cj.buy_or_reuse(j, "old-ident", cap=2, send=lambda: {"subject_id": "S", "rounds": [{"round_no": 1}]}, slot="S:p1") j.alias("new-ident", "old-ident", why="뼈대 신원") assert j.get("new-ident") == {"subject_id": "S", "rounds": [{"round_no": 1}]} assert j.bought_slots() == 1 again = cj.ChunkJournal(tmp_path / "j.json", contract={"acquisition": {"x": 1}}) assert again.get("new-ident") is not None and again.settlements[0]["kind"] == "alias" assert again.entries["new-ident"]["aliased_from"] == "old-ident" def test_the_central_pass_aliases_when_the_writer_knows_the_legacy_identity(self, monkeypatch, tmp_path): from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import reference_acquisition_rounds as rr calls = [] def fake_acquire_one(target, *, rounds, resume_from=None, **kw): calls.append(target["subject_id"]); return {"subject_id": target["subject_id"], "status": "selected", "chosen": {"path": "p"}, "rounds": [{"round_no": 1, "queries": [], "downloaded_candidates": []}], "candidates": []} monkeypatch.setattr(rr, "acquire_one", fake_acquire_one) monkeypatch.setattr(ca, "resolve_rounds", lambda r: 1) monkeypatch.setattr(ca._aa, "inputs_from_ledger", lambda ledger: { "targets": [{"target": {"subject_id": "A", "terms_native": ["t"]}, "source_evidence": {}, "ledger_row": {"final_id": "A"}}], "skipped": [], "refused": [], "auto_completed_rows": [], "not_applicable_rows": [], "ledger_rows": []}) class _W: identity_inputs = {"brief_pack": "5", "era": "e", "region": "r"} legacy_identity_inputs = {"brief_pack": "5", "era": "e", "region": "r", "world_facts": "w"} @staticmethod def target_identity(t): return "skeleton-hash" @staticmethod def legacy_target_identity(t): return "legacy-hash" def __call__(self, *a, **k): return {} journal = cj.ChunkJournal(tmp_path / "j.json", contract={"wiring": ca.CONTRACT_VERSION}) sha = ca.ra.acquisition_contract_sha(rounds=1) if hasattr(ca, "ra") else None # 옛 신원으로 먼저 산 장부를 만든다 from app.modules.pipeline import reference_acquisition as ra sha = ra.acquisition_contract_sha(rounds=1) one = {"target": {"subject_id": "A", "terms_native": ["t"]}, "ledger_row": {"final_id": "A"}} legacy = ca.identity_of(one, contract_sha=sha, brief_inputs={**_W.legacy_identity_inputs, "target_brief": "legacy-hash"}) cj.buy_or_reuse(journal, legacy, cap=2, send=lambda: {"subject_id": "A", "status": "selected", "chosen": {"path": "p"}, "rounds": [{"round_no": 1, "queries": [], "downloaded_candidates": []}], "candidates": []}) got = ca.run({"rows": []}, journal=journal, cap=2, workdir=tmp_path, search=lambda *a, **k: {}, download=lambda *a, **k: True, judge=lambda c: {}, write_brief=_W(), rounds=1) assert calls == [], "★옛 구매를 alias 로 이었으니 다시 사지 않는다" assert got["rows"][0]["disposition"] == "acquired" and journal.settlements[0]["kind"] == "alias" class TestAliasOnlyJoinsSelectedAnswersAndMatchesPassTwo: def _run_with_old(self, monkeypatch, tmp_path, *, old_status, rounds=1, old_pass=1): from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import reference_acquisition_rounds as rr from app.modules.pipeline import reference_acquisition as ra calls = [] def fake_acquire_one(target, *, rounds, resume_from=None, **kw): prev = list((resume_from or {}).get("rounds") or []) for r in range(len(prev) + 1, int(rounds) + 1): calls.append(f"{target['subject_id']}{r}") prev.append({"round_no": r, "queries": [{"text": f"q{r}"}], "downloaded_candidates": [{"index": 1}], "decision": {"next": "narrow_retry"}}) return {"subject_id": target["subject_id"], "status": "retryable", "chosen": None, "rounds": prev, "candidates": []} monkeypatch.setattr(rr, "acquire_one", fake_acquire_one) monkeypatch.setattr(ca, "resolve_rounds", lambda r: rounds) monkeypatch.setattr(ca._aa, "inputs_from_ledger", lambda ledger: { "targets": [{"target": {"subject_id": "A", "terms_native": ["t"]}, "source_evidence": {}, "ledger_row": {"final_id": "A"}}], "skipped": [], "refused": [], "auto_completed_rows": [], "not_applicable_rows": [], "ledger_rows": []}) class _W: identity_inputs = {"brief_pack": "5", "era": "e", "region": "r"} legacy_identity_inputs = {"brief_pack": "5", "era": "e", "region": "r", "world_facts": "w"} @staticmethod def target_identity(t): return "skeleton-hash" @staticmethod def legacy_target_identity(t): return "legacy-hash" def __call__(self, *a, **k): return {} journal = cj.ChunkJournal(tmp_path / "j.json", contract={"wiring": ca.CONTRACT_VERSION}) sha = ra.acquisition_contract_sha(rounds=rounds) one = {"target": {"subject_id": "A", "terms_native": ["t"]}, "ledger_row": {"final_id": "A"}} r1 = {"round_no": 1, "queries": [{"text": "q1"}], "downloaded_candidates": [{"index": 1}], "decision": {"next": "narrow_retry"}} resume = {"rounds": [r1]} if old_pass == 2 else None legacy = ca.identity_of(one, contract_sha=sha, brief_inputs={**_W.legacy_identity_inputs, "target_brief": "legacy-hash", **ca.pass_suffix(old_pass, resume)}) old_resp = {"subject_id": "A", "status": old_status, "chosen": ({"path": "p"} if old_status == "selected" else None), "rounds": [r1] + ([{"round_no": 2, "queries": [{"text": "q2"}], "downloaded_candidates": [], "decision": {"next": "no_match"}}] if old_pass == 2 else []), "candidates": []} cj.buy_or_reuse(journal, legacy, cap=4, send=lambda: old_resp, slot=f"A:p{old_pass}") if old_pass == 2: # 1차도 있어야 2차로 간다 first = ca.identity_of(one, contract_sha=sha, brief_inputs={**_W.legacy_identity_inputs, "target_brief": "legacy-hash"}) cj.buy_or_reuse(journal, first, cap=4, send=lambda: {"subject_id": "A", "status": "selected" if old_status == "selected" else "retryable", "chosen": ({"path": "p"} if old_status == "selected" else None), "rounds": [r1], "candidates": []}, slot="A:p1") got = ca.run({"rows": []}, journal=journal, cap=4, workdir=tmp_path, search=lambda *a, **k: {}, download=lambda *a, **k: True, judge=lambda c: {}, write_brief=_W(), rounds=rounds) return calls, journal, got def test_a_selected_pass_one_answer_is_aliased_and_not_rebought(self, monkeypatch, tmp_path): calls, journal, got = self._run_with_old(monkeypatch, tmp_path, old_status="selected") assert calls == [] and any(s["kind"] == "alias" for s in journal.settlements) def test_an_old_no_match_is_not_aliased_and_the_broad_search_runs(self, monkeypatch, tmp_path): """★Codex BLOCK 2026-09-03: 옛 세부 검색의 못 찾음을 새 넓은 검색의 완료 답으로 접으면 새 검색이 안 나간다.""" calls, journal, got = self._run_with_old(monkeypatch, tmp_path, old_status="no_match_after_retry") assert calls == ["A1"] and not any(s["kind"] == "alias" for s in journal.settlements) def test_the_pass_two_suffix_is_one_function_for_new_and_legacy(self): """★pass 접미(pass+done_rounds)는 한 함수 — 새 신원과 legacy 신원이 같은 접미를 받는다(소스 래칫). 옛 1차가 못 찾았으면 alias 안 하고 1차·2차를 새로 사므로, 2차 alias 는 옛 1차가 골랐을 때만 뜻이 있다.""" import inspect from app.modules.pipeline import grounding_central_acquisition as ca r1 = {"round_no": 1, "queries": [{"text": "q1"}], "downloaded_candidates": [{"index": 1}], "decision": {"next": "narrow_retry"}} bad = {"round_no": 2, "queries": [{"text": "q2"}], "downloaded_candidates": [], "decision": {"next": "retryable"}} got = ca.pass_suffix(2, {"rounds": [r1, bad]}) assert got == {"pass": 2, "done_rounds": [(1, [{"text": "q1"}])]}, "★끝나지 못한 라운드는 접미에 없다" assert ca.pass_suffix(1, {"rounds": [r1]}) == {} src = inspect.getsource(ca) assert src.count("suffix = pass_suffix(pass_no, resume_from)") == 1 and src.count("**suffix}") >= 2 def test_an_old_selected_pass_two_is_aliased_when_the_fresh_pass_one_matches(self, monkeypatch, tmp_path): """★Codex NON-BLOCK (2026-09-03): 옛 1차 retryable + 옛 2차 selected. 새 1차가 같은 질의로 끝나면 2차 접미 (done_rounds)가 같아 옛 2차 selected 를 alias 로 잇고 acquire_one 은 1차 한 번뿐이다.""" from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import reference_acquisition_rounds as rr from app.modules.pipeline import reference_acquisition as ra calls = [] r1 = {"round_no": 1, "queries": [{"text": "q1"}], "downloaded_candidates": [{"index": 1}], "decision": {"next": "narrow_retry"}} def fake_acquire_one(target, *, rounds, resume_from=None, **kw): prev = list((resume_from or {}).get("rounds") or []) for r in range(len(prev) + 1, int(rounds) + 1): calls.append(f"{target['subject_id']}{r}"); prev.append(dict(r1, round_no=r)) return {"subject_id": target["subject_id"], "status": "retryable", "chosen": None, "rounds": prev, "candidates": []} monkeypatch.setattr(rr, "acquire_one", fake_acquire_one) monkeypatch.setattr(ca, "resolve_rounds", lambda r: 2) monkeypatch.setattr(ca._aa, "inputs_from_ledger", lambda ledger: { "targets": [{"target": {"subject_id": "A", "terms_native": ["t"]}, "source_evidence": {}, "ledger_row": {"final_id": "A"}}], "skipped": [], "refused": [], "auto_completed_rows": [], "not_applicable_rows": [], "ledger_rows": []}) class _W: identity_inputs = {"brief_pack": "5", "era": "e", "region": "r"} legacy_identity_inputs = {"brief_pack": "5", "era": "e", "region": "r", "world_facts": "w"} @staticmethod def target_identity(t): return "skeleton-hash" @staticmethod def legacy_target_identity(t): return "legacy-hash" def __call__(self, *a, **k): return {} journal = cj.ChunkJournal(tmp_path / "j.json", contract={"wiring": ca.CONTRACT_VERSION}) sha = ra.acquisition_contract_sha(rounds=2) one = {"target": {"subject_id": "A", "terms_native": ["t"]}, "ledger_row": {"final_id": "A"}} legacy_p2 = ca.identity_of(one, contract_sha=sha, brief_inputs={**_W.legacy_identity_inputs, "target_brief": "legacy-hash", **ca.pass_suffix(2, {"rounds": [r1]})}) cj.buy_or_reuse(journal, legacy_p2, cap=4, slot="A:p2", send=lambda: { "subject_id": "A", "status": "selected", "chosen": {"path": "p"}, "rounds": [r1, {"round_no": 2, "queries": [{"text": "q2"}], "downloaded_candidates": [{"index": 1}], "decision": {"next": "select"}}], "candidates": []}) got = ca.run({"rows": []}, journal=journal, cap=4, workdir=tmp_path, search=lambda *a, **k: {}, download=lambda *a, **k: True, judge=lambda c: {}, write_brief=_W(), rounds=2) assert calls == ["A1"], calls assert sum(1 for s in journal.settlements if s["kind"] == "alias") == 1 assert got["rows"][0]["outcome"] == ra.STATUS_SELECTED def test_an_old_retryable_pass_one_is_rebought_broad_and_pass_two_follows(self, monkeypatch, tmp_path): calls, journal, got = self._run_with_old(monkeypatch, tmp_path, old_status="retryable", rounds=2, old_pass=2) assert calls == ["A1", "A2"], calls assert not any(s["kind"] == "alias" for s in journal.settlements) class TestOneObligationPerOutfit: def _ledger(self): from app.modules.pipeline.grounding_entity_contract import PRODUCER_PAYLOAD from app.modules.pipeline import grounding_acquisition_ledger as gl from app.modules.pipeline.grounding_screen import SCREEN_OBLIGATION def row(rsid, fid, label, terms, brief): return {"research_subject_id": rsid, "owner_type": "outlook", "final_id": fid, "parent_final_id": "C01", "status": gl.RESOLVED, "screen": SCREEN_OBLIGATION, "source_evidence": {"surface_form": label}, PRODUCER_PAYLOAD: {"coarse_type_label": label, "surface_form": label, "visual_brief": brief, "search_terms_native": terms, "language_lock_native": "ko", "hard_to_generate": True, "viewers_would_notice": True}} return {"rows": [row("rs_1", "O01", "교복 윗도리", ["교복 윗도리"], "검정 윗도리"), row("rs_2", "O01", "깃 흰 천", ["교복 깃"], "깃 안쪽 흰 천"), row("rs_3", "O01", "고무신", ["검정 고무신"], "코가 올라간 고무신"), row("rs_4", "O02", "흰 겉옷", ["흰 무명 겉옷"], "이발사 가운")], "outfits": {"O01": {"name": "낡은검정교복", "description": "검정 교복 윗도리와 바지"}, "O02": {"name": "흰무명작업복", "description": "흰 무명 긴 겉옷"}}} def test_three_piece_rows_become_one_outfit_obligation(self): from app.modules.pipeline import grounding_reference_obligations as ro from app.modules.pipeline import grounding_acquisition_ledger as gl from app.modules.pipeline.grounding_entity_contract import PRODUCER_PAYLOAD got = ro.plan(self._ledger()) outs = [r for r in got["rows"] if r.get("owner_type") == "outlook"] assert sorted(r["final_id"] for r in outs) == ["O01", "O02"], "★final_id 당 하나" o1 = next(r for r in outs if r["final_id"] == "O01") assert o1["research_subject_id"] == "O01#outfit" and o1["obligation_kind"] == ro.PURPOSE_OUTFIT and o1["purpose"] is None assert o1[PRODUCER_PAYLOAD]["coarse_type_label"] == "낡은검정교복", "★뼈대 = 아웃룩 엔티티 이름" # ★검색어는 옷 전체 뼈대에서만 — 조각 검색어는 details 에만 (Codex BLOCK 2026-09-03) assert o1[PRODUCER_PAYLOAD]["search_terms_native"] == ["낡은검정교복"] assert [d["search_terms_native"] for d in o1["details"]] == [["교복 윗도리"], ["교복 깃"], ["검정 고무신"]] assert [d["surface_form"] for d in o1["details"]] == ["교복 윗도리", "깃 흰 천", "고무신"] assert sorted(o1["origin_subject_ids"]) == ["rs_1", "rs_2", "rs_3"] # 조사 대상으로 세어도 둘이다 assert len([r for r in gl.acquisition_targets(got) if r["owner_type"] == "outlook"]) == 2 def test_the_outbound_query_for_an_outfit_is_the_whole_outfit_skeleton(self): from app.modules.pipeline import grounding_reference_obligations as ro from app.modules.pipeline import grounding_acquisition_adapter as aa got = ro.plan(self._ledger()) o1 = next(r for r in got["rows"] if r.get("final_id") == "O01") t = aa.search_target(o1) o = sb.outbound(t, world_facts="w", source_text="s", narrow=False, era="1960년대", region="대한민국") assert "낡은검정교복" in o["user"] and "코가 올라간" not in o["user"] and "깃 안쪽" not in o["user"] for piece_term in ("교복 윗도리", "교복 깃", "검정 고무신"): assert piece_term not in o["user"], f"★조각 검색어가 나갔다: {piece_term}" assert o["skeleton"]["terms_native"] == ["낡은검정교복"] class TestProviderFailuresDoNotKillTheLane: def test_a_refusal_frees_the_slot_for_the_next_attempt(self, tmp_path): class ServiceUnavailableError(Exception): ... j = cj.ChunkJournal(tmp_path / "j.json", contract={"a": 1}) with pytest.raises(ServiceUnavailableError): cj.buy_or_reuse(j, "id1", cap=1, send=lambda: (_ for _ in ()).throw(ServiceUnavailableError("503"))) assert j.entries["id1"]["status"] == cj.STATUS_REFUSED and j.bought() == 0 assert cj.buy_or_reuse(j, "id1", cap=1, send=lambda: {"r": 1}) == {"r": 1}, "★사람 없이 다시 산다" def test_a_connection_error_is_uncertain_not_refused(self, tmp_path): """★Codex BLOCK 2026-09-03: 연결 예외는 bytes 가 나갔는지 모른다 — 사람/운반층 증거 전엔 다시 안 산다.""" class APIConnectionError(Exception): ... j = cj.ChunkJournal(tmp_path / "j.json", contract={"a": 1}) with pytest.raises(APIConnectionError): cj.buy_or_reuse(j, "id1", cap=1, send=lambda: (_ for _ in ()).throw(APIConnectionError("conn"))) assert j.entries["id1"]["status"] == cj.STATUS_UNCERTAIN with pytest.raises(cj.NeedsHumanDecision): cj.buy_or_reuse(j, "id1", cap=1, send=lambda: {"r": 1}) def test_a_timeout_is_still_uncertain(self, tmp_path): class Timeout(Exception): ... j = cj.ChunkJournal(tmp_path / "j.json", contract={"a": 1}) with pytest.raises(Timeout): cj.buy_or_reuse(j, "id1", cap=1, send=lambda: (_ for _ in ()).throw(Timeout("t"))) assert j.entries["id1"]["status"] == cj.STATUS_UNCERTAIN def test_rejudge_rows_skips_a_row_whose_provider_failed(self, monkeypatch, tmp_path): from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import reference_acquisition_rounds as rr class ServiceUnavailableError(Exception): ... j = cj.ChunkJournal(tmp_path / "j.json", contract={"a": 1}) monkeypatch.setattr(rr, "resolve_cached_candidates", lambda c, root: [{"index": 1, "path": "p", "rel": "p", "sha256": "0"}]) monkeypatch.setattr(rr, "replay_identity_inputs", lambda acq, resolved: {"k": 1}) monkeypatch.setattr(rr, "cached_round", lambda acq: {"downloaded_candidates": [{"index": 1}]}) monkeypatch.setattr(rr, "rejudge_cached", lambda acq, root, judge: (_ for _ in ()).throw(ServiceUnavailableError("503"))) row = {"identity": "base", "disposition": ca.DISP_ACQUIRED, "status": "retryable", "acquisition": {"chosen": None, "rounds": [{"round_no": 1, "downloaded_candidates": [{"index": 1}]}]}} got = ca.rejudge_rows([row], journal=j, root=tmp_path, judge=lambda c: {}, cap=3) r = got["rows"][0] assert r.get("replay_why_code") == ca.WHY_CURRENT_UNCONFIRMED and "답을 못 받았다" in r["replay_skipped"] class TestOrphanSlotsDoNotStarveCurrentTargets: def test_slots_of_subjects_that_are_no_longer_targets_widen_the_cap(self, monkeypatch, tmp_path): """★실측 bb85e798: 옛 조각별 아웃룩 자리 7 이 상한을 먹어 지금 대상의 2차가 굶었다.""" from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import reference_acquisition_rounds as rr calls = [] def fake_acquire_one(target, *, rounds, resume_from=None, **kw): prev = list((resume_from or {}).get("rounds") or []) for r in range(len(prev) + 1, int(rounds) + 1): calls.append(f"{target['subject_id']}{r}"); prev.append({"round_no": r, "queries": [{"text": "q"}], "downloaded_candidates": [{"index": 1}], "decision": {"next": "narrow_retry"}}) return {"subject_id": target["subject_id"], "status": "retryable", "chosen": None, "rounds": prev, "candidates": []} monkeypatch.setattr(rr, "acquire_one", fake_acquire_one) monkeypatch.setattr(ca, "resolve_rounds", lambda r: 2) monkeypatch.setattr(ca._aa, "inputs_from_ledger", lambda ledger: { "targets": [{"target": {"subject_id": "A", "terms_native": ["t"]}, "source_evidence": {}, "ledger_row": {"final_id": "A"}}], "skipped": [], "refused": [], "auto_completed_rows": [], "not_applicable_rows": [], "ledger_rows": []}) journal = cj.ChunkJournal(tmp_path / "j.json", contract={"wiring": ca.CONTRACT_VERSION}) # 옛 대상 구조의 자리 둘(지금 대상 아님)이 상한 2 를 다 먹은 상태 cj.buy_or_reuse(journal, "old-x1", cap=9, send=lambda: {"subject_id": "X"}, slot="X:p1") cj.buy_or_reuse(journal, "old-y1", cap=9, send=lambda: {"subject_id": "Y"}, slot="Y:p1") got = ca.run({"rows": []}, journal=journal, cap=2, workdir=tmp_path, search=lambda *a, **k: {}, download=lambda *a, **k: True, judge=lambda c: {}, write_brief=None, rounds=2) assert calls == ["A1", "A2"], calls assert got["purchases"]["orphan_slots"] == 2 and got["purchases"]["cap_reached"] is False