"""★인스턴스 신원은 short_id 다 — entity_detail → entity_t2i (Codex BLOCK 2026-09-03 06:40 · 실측 f7cc45c576c0). 같은 이름·같은 갈래(location_part)인 두 실체(LP05→L03 · LP07→L02)가 (name, type) 큐에서 하나로 접혀 LP07 canon 이 안 생기고 part_of 동기화가 fail-closed 로 섰다. 이름·시나리오 문자열을 코드에 넣지 않는다 — 시험 fixture 는 일반 낱말이다.""" from __future__ import annotations import pytest import app.core.steps.entity_steps as es from app.core.steps.entity_steps import EntityDetailStep, EntityT2iStep, _qkey def _detail_step(monkeypatch, tmp_path, *, mode, parts, results, calls): """fake LLM: 큐 줄의 표식을 entity_id 로 돌려준다(팩 v10 계약).""" from app.core.config import settings monkeypatch.setattr(settings, "projects_dir", str(tmp_path / "projects")) from app.modules.pipeline.grounding_entity_sync_ext import CHUNK_SCHEMA_MARKER, CHUNK_SCHEMA_VERSION st = EntityDetailStep.__new__(EntityDetailStep) st.project_id, st.episode_id = "p", "e" st.project_config = {"grounding_mode": mode} st.db = None st.build_opik_metadata = lambda *a, **k: {} st._load_cleaned_text = lambda: "본문" _marked = {"status": "completed", CHUNK_SCHEMA_MARKER: CHUNK_SCHEMA_VERSION} st._load_prev_checkpoint = lambda sid: ({**_marked, "data": {"filtered_entities": {"location_parts": parts}}} if sid == "entity_filter" else {**_marked, "data": {"removed": []}} if sid == "entity_merge" else None) st._journal = lambda *a, **k: None class _P: has_planning_doc = False is_first_episode = False def inject_if_available(self, section, header=""): return "" monkeypatch.setattr("app.core.planning_doc_context.get_planning_context", lambda pid, eid, db=None: _P()) def fake_call(**kw): calls.append(kw) return results(kw) monkeypatch.setattr(es, "call_structured", fake_call, raising=False) monkeypatch.setattr("app.modules.pipeline.entity_extractor_v3._load_prompt", lambda name, **k: f"{name}:{k.get('entity_list','')}") monkeypatch.setattr("app.modules.pipeline.entity_extractor_v3._load_schema", lambda name: {"type": "object"}) monkeypatch.setattr(st, "_config_hash", lambda: "h", raising=False) return st def _echo(kw): """큐 줄 `- [SID] name (type)` 를 읽어 entity_id 를 그대로 돌려주는 대역.""" import re ents = [] for m in re.finditer(r"- \[(\S+)\] (.+?) \((\S+)\)", str(kw.get("user_prompt") or kw.get("prompt") or "") + str(kw)): ents.append({"entity_id": m.group(1), "name": m.group(2), "entity_type": m.group(3), "description": f"d-{m.group(1)}", "visual_traits": ["v"], "distinctive_visual_traits": []}) return {"entities": ents} class TestTheDetailQueueKeepsBothInstances: def test_same_name_same_type_different_ids_are_both_kept_and_matched_by_id(self, monkeypatch, tmp_path): parts = [{"name": "바닥", "short_id": "LP05"}, {"name": "바닥", "short_id": "LP07"}] calls = [] st = _detail_step(monkeypatch, tmp_path, mode="v2_chunk", parts=parts, results=_echo, calls=calls) out = st._execute() q = out["data"]["entity_queue"] assert [x[2] for x in q] == ["LP05", "LP07"] assert out["completed_count"] == 2 and out["failed_count"] == 0 d = out["data"]["entity_details"] assert d["LP05"]["description"] == "d-LP05" and d["LP07"]["description"] == "d-LP07" assert out["data"]["identity_contract"] == es.ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION def test_chunk_producer_mode_stops_before_the_provider_on_a_missing_or_duplicate_id(self, monkeypatch, tmp_path): from app.core.errors import AppError calls = [] st = _detail_step(monkeypatch, tmp_path, mode="v2_chunk", parts=[{"name": "바닥"}], results=_echo, calls=calls) with pytest.raises(AppError) as e: st._execute() assert e.value.code == "entity_detail.identity_missing" and calls == [] st = _detail_step(monkeypatch, tmp_path, mode="v2_chunk", parts=[{"name": "바닥", "short_id": "LP05"}, {"name": "벽", "short_id": "LP05"}], results=_echo, calls=calls) with pytest.raises(AppError) as e: st._execute() assert e.value.code == "entity_detail.identity_duplicate" and calls == [] def test_legacy_mode_without_ids_still_folds_by_name_and_type(self, monkeypatch, tmp_path): calls = [] st = _detail_step(monkeypatch, tmp_path, mode="legacy", parts=[{"name": "바닥"}, {"name": "바닥"}], results=lambda kw: {"entities": [{"name": "바닥", "entity_type": "location_part", "description": "d", "visual_traits": []}]}, calls=calls) out = st._execute() assert len(out["data"]["entity_queue"]) == 1 and out["completed_count"] == 1 def test_the_accounting_is_per_instance_not_per_name(self, monkeypatch, tmp_path): """한 실체의 답으로 이름이 같은 다른 실체를 채우지 않는다 — 못 받은 쪽은 failed 로 센다.""" parts = [{"name": "바닥", "short_id": "LP05"}, {"name": "바닥", "short_id": "LP07"}] only_first = lambda kw: {"entities": [{"entity_id": "LP05", "name": "바닥", "entity_type": "location_part", "description": "d", "visual_traits": []}]} calls = [] st = _detail_step(monkeypatch, tmp_path, mode="v2_chunk", parts=parts, results=only_first, calls=calls) out = st._execute() assert out["completed_count"] == 1 and out["failed_count"] == 1 assert "LP07" not in out["data"]["entity_details"] assert len(calls) == 2, "누락 재시도가 LP07 만 다시 묻는다" def _t2i_step(monkeypatch, *, queue, details, prev_done, calls): from app.modules.pipeline.grounding_entity_sync_ext import CHUNK_SCHEMA_MARKER, CHUNK_SCHEMA_VERSION st = EntityT2iStep.__new__(EntityT2iStep) st.project_id, st.episode_id = "p", "e" st.project_config = {"grounding_mode": "v2_chunk"} st.db = None st.build_opik_metadata = lambda *a, **k: {} _marked = {"status": "completed", CHUNK_SCHEMA_MARKER: CHUNK_SCHEMA_VERSION} st._load_prev_checkpoint = lambda sid: ({**_marked, "data": {"entity_queue": queue, "entity_details": details}} if sid == "entity_detail" else {**_marked, "data": {"removed": []}} if sid == "entity_merge" else None) st.load_checkpoint = lambda: ({"status": "completed", CHUNK_SCHEMA_MARKER: CHUNK_SCHEMA_VERSION, "data": {"completed": prev_done}} if prev_done is not None else None) st.save_checkpoint = lambda *a, **k: None st.update_progress = lambda *a, **k: None monkeypatch.setattr(st, "_config_hash", lambda: "h", raising=False) def fake_call(**kw): calls.append(kw) return {"t2i_prompt": "p", "metadata_json": {}} monkeypatch.setattr(es, "call_structured", fake_call, raising=False) monkeypatch.setattr(es, "_load_system", lambda: "sys", raising=False) monkeypatch.setattr(es, "validate_entity_metadata_shape", lambda *a, **k: None, raising=False) monkeypatch.setattr(es, "time", type("T", (), {"sleep": staticmethod(lambda s: None)})(), raising=False) return st QUEUE = [("바닥", "location_part", "LP05"), ("바닥", "location_part", "LP07")] DETAILS = {"LP05": {"description": "d5", "visual_traits": []}, "LP07": {"description": "d7", "visual_traits": []}} class TestTheT2iStepKeysByInstance: def test_two_same_named_instances_are_both_generated_and_stored_by_id(self, monkeypatch): calls = [] st = _t2i_step(monkeypatch, queue=QUEUE, details=DETAILS, prev_done=None, calls=calls) out = st._execute() done = out["data"]["completed"] assert set(done) >= {"LP05", "LP07"} and len(calls) == 2 assert done["LP05"]["short_id"] == "LP05" and done["LP07"]["short_id"] == "LP07" assert out["completed_count"] == 2 and out["failed_count"] == 0 assert out["data"]["identity_contract"] == es.ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION def test_a_new_checkpoint_resumes_by_id_with_zero_calls(self, monkeypatch): calls = [] prev = {"LP05": {"name": "바닥", "entity_type": "location_part", "short_id": "LP05", "t2i_prompt": "p"}, "LP07": {"name": "바닥", "entity_type": "location_part", "short_id": "LP07", "t2i_prompt": "p"}} st = _t2i_step(monkeypatch, queue=QUEUE, details=DETAILS, prev_done=prev, calls=calls) out = st._execute() assert calls == [] and out["completed_count"] == 2 def test_one_finished_instance_buys_only_the_other(self, monkeypatch): calls = [] prev = {"LP05": {"name": "바닥", "entity_type": "location_part", "short_id": "LP05", "t2i_prompt": "p"}} st = _t2i_step(monkeypatch, queue=QUEUE, details=DETAILS, prev_done=prev, calls=calls) out = st._execute() assert len(calls) == 1 and set(out["data"]["completed"]) >= {"LP05", "LP07"} def test_a_legacy_name_keyed_record_without_an_id_is_not_reused_for_a_duplicated_pair(self, monkeypatch): """같은 짝이 둘인데 옛 값에 short_id 가 없으면 어느 쪽인지 모른다 — 둘 다 다시 산다.""" calls = [] prev = {_qkey("바닥", "location_part"): {"name": "바닥", "entity_type": "location_part", "t2i_prompt": "p"}} st = _t2i_step(monkeypatch, queue=QUEUE, details=DETAILS, prev_done=prev, calls=calls) out = st._execute() assert len(calls) == 2 and set(out["data"]["completed"]) >= {"LP05", "LP07"} def test_a_legacy_record_for_a_unique_pair_is_still_reused(self, monkeypatch): calls = [] q = [("문", "location_part", "LP01")] prev = {_qkey("문", "location_part"): {"name": "문", "entity_type": "location_part", "t2i_prompt": "p"}} st = _t2i_step(monkeypatch, queue=q, details={"LP01": {"description": "d", "visual_traits": []}}, prev_done=prev, calls=calls) out = st._execute() assert calls == [] and out["completed_count"] == 1 class TestTheContractIsInBothHashes: def test_both_steps_bind_the_identity_contract_and_move_when_it_changes(self, monkeypatch): """★v2_chunk 에서만 — legacy 지문은 옛 값 그대로다(PR #82 리뷰 2026-09-03 · `test_pr82_review_fixes`).""" for cls in (EntityDetailStep, EntityT2iStep): st = cls.__new__(cls); st.project_config = {"grounding_mode": "v2_chunk"}; st.project_id, st.episode_id = "p", "e" monkeypatch.setattr(es._EntityStepMixin, "_config_hash", lambda self: "base", raising=False) monkeypatch.setattr(st, "_a0_candidates", lambda: [], raising=False) before = st._config_hash() assert before != "base" monkeypatch.setattr(es, "ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION", "9.x") assert st._config_hash() != before monkeypatch.undo() class TestTheRealPackCarriesTheIdEcho: """★대역이 아니라 **진짜 loader** 로 본다 — 실측 06:45: 제가 만든 팩이 숫자 버전순에서 최신이 아니라 한 번도 안 실렸고 시험은 _load_prompt/_load_schema 를 대역으로 바꿔 그것을 못 봤다.""" def test_the_resolved_detail_schema_and_prompt_speak_entity_id(self): from app.modules.prompt_loader import resolve_effective sch = resolve_effective("entity_extractor_v2", "turn1_7_detail_batch_schema", kind="schema") items = ((sch["content"].get("properties") or {}).get("entities") or {}).get("items") or {} assert "entity_id" in (items.get("properties") or {}), sch.get("version") assert "entity_id" in (items.get("required") or []), sch.get("version") pr = resolve_effective("entity_extractor_v2", "turn1_7_detail_batch", kind="prompt") assert "entity_id" in str(pr["content"]) and "{entity_list}" in str(pr["content"]), pr.get("version") assert pr["version"] == sch["version"] def test_an_unechoed_answer_still_lands_on_a_unique_pair(self, monkeypatch, tmp_path): """표식 없는 답(entity_id 없음)은 (name,type) 짝이 유일할 때 그 short_id 로 귀속 — 짝이 둘이면 못 받은 것으로 센다.""" parts = [{"name": "바닥", "short_id": "LP05"}, {"name": "바닥", "short_id": "LP07"}, {"name": "문", "short_id": "LP01"}] def _no_ids(kw): return {"entities": [{"name": "바닥", "entity_type": "location_part", "description": "d", "visual_traits": []}, {"name": "문", "entity_type": "location_part", "description": "d", "visual_traits": []}]} calls = [] st = _detail_step(monkeypatch, tmp_path, mode="v2_chunk", parts=parts, results=_no_ids, calls=calls) out = st._execute() d = out["data"]["entity_details"] assert "LP01" in d and "LP05" not in d and "LP07" not in d assert out["completed_count"] == 1 and out["failed_count"] == 2 class TestV15PreservesV14AndOnlyAddsTheIdEcho: """★진짜 팩 비교 (Codex 2026-09-03 06:50): v15 는 v14 의 distinctive 규칙을 그대로 두고 표식 echo 만 더한다.""" def test_only_the_batch_prompt_and_its_schema_differ_and_only_additively(self): import json from pathlib import Path as _P base = _P(__file__).resolve().parents[3] / "prompts" / "_base" / "entity_extractor_v2" v14, v15 = base / "14.202606181515", base / "15.202609030650" assert sorted(p.name for p in v14.iterdir()) == sorted(p.name for p in v15.iterdir()) for p in v14.iterdir(): a, b = p.read_text(encoding="utf-8"), (v15 / p.name).read_text(encoding="utf-8") if p.name == "turn1_7_detail_batch.md": import difflib ops = [t for t, *_ in difflib.SequenceMatcher(a=a.splitlines(), b=b.splitlines()).get_opcodes()] assert set(ops) <= {"equal", "insert"}, ops # ★v14 줄은 하나도 안 바뀌고 안 빠진다 — 더해지기만 assert "entity_id" in b and "entity_id" not in a elif p.name == "turn1_7_detail_batch_schema.json": ja, jb = json.loads(a), json.loads(b) ia, ib = ja["properties"]["entities"]["items"], jb["properties"]["entities"]["items"] assert set(ia["properties"]) | {"entity_id"} == set(ib["properties"]) assert set(ia.get("required", [])) | {"entity_id"} == set(ib.get("required", [])) for k in ia["properties"]: assert ia["properties"][k] == ib["properties"][k], k else: assert a == b, p.name class TestTheDetailFingerprintFollowsTheRealPack: """★Codex 06:50 NON-BLOCK(이번 live 뒤 · 최종 병합 전): entity_detail 지문에 실제로 고른 팩의 버전·bytes 를 싣는다.""" def test_legacy_and_v2_hashes_do_not_move_with_the_pack(self, monkeypatch): """★v2_chunk 만 움직인다 — legacy/v2 는 **옛 값 그대로**(mixin base 와 byte-identical). ★2026-09-03 PR #82 리뷰로 뒤집음: 앞 판은 legacy 도 `_identity_hash` 로 감싸 기존 CP(md5 16자)가 전부 어긋났고 runner 가 그것을 BLOCK(409) 으로 다뤄 기존 에피소드 재개가 막혔다.""" import app.modules.prompt_loader as pl real = pl.resolve_effective for mode in ("legacy", "v2"): st = EntityDetailStep.__new__(EntityDetailStep); st.project_config = {"grounding_mode": mode}; st.project_id, st.episode_id = "p", "e" monkeypatch.setattr(es._EntityStepMixin, "_config_hash", lambda self: "base", raising=False) monkeypatch.setattr(st, "_a0_candidates", lambda: None, raising=False) before = st._config_hash() monkeypatch.setattr(pl, "resolve_effective", lambda m, stem, *, kind, **k: {**real(m, stem, kind=kind), "raw_content_hash": "changed"}) assert st._config_hash() == before == "base" monkeypatch.setattr(pl, "resolve_effective", real) def test_the_hash_moves_when_the_resolved_pack_bytes_change(self, monkeypatch): st = EntityDetailStep.__new__(EntityDetailStep); st.project_config = {"grounding_mode": "v2_chunk"}; st.project_id, st.episode_id = "p", "e" monkeypatch.setattr(es._EntityStepMixin, "_config_hash", lambda self: "base", raising=False) before = st._config_hash() import app.modules.prompt_loader as pl real = pl.resolve_effective monkeypatch.setattr(pl, "resolve_effective", lambda m, stem, *, kind, **k: {**real(m, stem, kind=kind), "raw_content_hash": "changed"}) assert st._config_hash() != before def test_the_hash_reads_the_batch_prompt_and_schema_through_the_real_loader(self, monkeypatch): import app.modules.prompt_loader as pl seen = [] real = pl.resolve_effective monkeypatch.setattr(pl, "resolve_effective", lambda m, stem, *, kind, **k: (seen.append((m, stem, kind)) or real(m, stem, kind=kind))) st = EntityDetailStep.__new__(EntityDetailStep); st.project_config = {"grounding_mode": "v2_chunk"}; st.project_id, st.episode_id = "p", "e" monkeypatch.setattr(es._EntityStepMixin, "_config_hash", lambda self: "base", raising=False) st._config_hash() assert ("entity_extractor_v2", "turn1_7_detail_batch", "prompt") in seen assert ("entity_extractor_v2", "turn1_7_detail_batch_schema", "schema") in seen