"""era_research — 시대·브랜드 인지 대상 사전 이미지 조사 (2026-08-14 사용자 확정). "그냥 이미지 생성으로는 그 시대나 장면을 제대로 묘사하기 어려운 경우 서울역과 같이 사전 이미지 조사가 필요해, 무조건" — 지폐·브랜드 가전· 신발·패션·차량·1980년대 열차 실내 같은 **시청자 인지 대상**은 생성 전에 실물 사진을 조사해 참조로 동봉한다. 서울역 재생성(regen_period_bg)에서 통한 구조(검색→후보→VLM 선택→참조)의 범용 정식 배선이다. 층위 셋(전부 `era_research_enabled` 기본 OFF=byte-identical): B1 엔티티 참조 생성(소품·장소 — 인물 제외) / B2 groupbg 플레이트 / B3 샷 레벨(엔티티로 커버되지 않는 시대 인지 대상). 계약 요점: - 판별+원어 질의 저작 = **1콜**(대상 텍스트가 짧고 이미 원어라 seed 의 장문 언어-격리 브리프 2콜 기계가 불요). 판별 기준은 팩 assess_sys 가 소유 — 코드에는 대상 어휘가 없다(통칭 하드코딩 금지). "당연한 것은 조사하지 않는다·대부분 빈 목록"이 계약 본문에 명문. - 질의는 무조건 원어 + language_lock(원어 잠금문을 지시문 맨 앞에) — `search_grounded_ref.search_reference_images` 재사용(나간 질의 기록 포함). 다운로드·안전 정책·VLM 선택도 같은 모듈 부품 재사용. - 선택 심판 = GPT LVM 단독(모델 분업표 "이미지 비교 선택=GPT"). seed 의 이중 평균은 후보가 구조물일 때의 계약 — 여기서는 단일 대상 사용성 판정이라 1심으로 줄인다. - 산출 meta(질의·나간 질의·선택 URL·sha256)는 호출측이 record/CP 에 영속한다 — 기록 없는 유료 조사는 없다. """ from __future__ import annotations import hashlib import logging from pathlib import Path from typing import Any, Dict, List, Optional from app.modules.prompt_loader import load_prompt, pack_dir_content_hash logger = logging.getLogger(__name__) _MODULE = "era_research" PACK_VERSION_MAP = { "1": "1.202608141330", # v3 (2026-08-25): 저작한 검색어가 여러 대상을 담아 **나간 질의가 # 뭉쳤다** — 완주 판 조사 6건 중 5건. 원인은 검색어 칸이 아니라 **대상 # 칸**이었다: 대상 자체가 「A 및 B」로 나와(「사거리 도로 노면 표시와 # 가로등」) 검색어가 흩어졌다. 그래서 검색어 칸에 「한 대상만」을 적는 # 판(측정용 2.202608251431)은 무효였고, 대상 칸에 **한 대상 = 사진 한 # 장이 혼자 보여줄 수 있는 것 하나**를 박은 이 판이 통했다. # 실측 = backend/tools/prompt_measure/ab_era_terms.py "3": "3.202608251500", # ★★v4 (2026-08-31, Codex BLOCK-3): v3 지문에 **손으로 적은 대상 목록**이 # 있었다 — 「화폐와 그 위조 방지 요소 · 상표 붙은 가전/신발/의류 · 차량 · # 대중교통 실내 · 가전 · 제복 · 점포 설비」와 반대편의 「밋밋한 컵 · 나무 # 의자 · 흔한 나무」. 그건 **하드 프롬프트**다(사용자 계약 1·2): 목록에 # 없는 부류를 못 건지고, 목록에 있는 부류는 시대와 무관해도 건진다. # ★대신 **조건**만 남겼다 — 「그 형태가 부류가 아니라 **시대·지역**이 # 정하는가」 + 「그곳 사람이 알아채는가」. 대상은 world facts 와 원문이 # 준다. 옛 팩은 **그대로 둔다**(소비자가 없어질 때까지 보존). # ★★**아직 pin 하지 않는다.** 이 팩으로 바꾸면 기존 era 캐시가 전부 # 갈리고 뒤쪽 late 호출부(groupbg·confined·plate)의 판별이 그 자리에서 # 달라진다 — 3a 는 inert 여야 한다(Codex). pin 과 정책 버전은 3b # 활성화 커밋에서 **같이** 올린다. "4": "4.202608311200", } ERA_RESEARCH_PACK_VERSION = "3" # 정책 문자열 — ON 시 지문 기여. 판별·선택 계약이 바뀌면 버전을 올린다. # v2 (2026-08-14 #119①): 적용 범위가 계약에 들어간다 — 샷별 bgfirst 판· # confined 합성까지 조사가 붙는 커버리지 확장 + records 사이드카 캐시. # v3 (2026-08-27, 감사 2-C): **캐시 신원 계약이 바뀌었다** — 자유 저작 # (대상 이름·검색어·잠금문) 대신 정본 세 칸(장소 id·실내/실외·정본 내용 # 해시)으로 합친다. 판별 입력도 같은 정본으로 맞춘다. # # ★**올려야 한다.** 이 문자열이 스텝 지문(`image_steps.py:829`)과 샷 # 지문(`still_recipe_service.py:1164`) 둘 다에 접힌다. 안 올리면 # 완주한 판이 옛 신원으로 만든 참조를 **새 계약인 것처럼** 그대로 # 재사용한다 — 고친 것이 한 번도 안 닿는다. # ★적용 범위(plate·confined)를 이름에 **남긴다** — v2 가 그것을 담으려고 # 올린 판이고, 시험이 그 두 낱말을 잠근다. 지우면 v2 의 계약이 사라진다. ERA_RESEARCH_POLICY_VERSION = ( "era_research_v3_canonical_scope_plate_confined") # 한 대상 텍스트에서 조사할 대상 상한 — "보통 하나"(2026-08-03 관례). # 판별이 여럿을 물어도 프레임을 지배하는 것부터 이 수만 조사한다. MAX_SUBJECTS_PER_CALL = 2 # 참조 신원의 **역할** — 코드가 정하는 값이다 (2026-08-27, 감사 2-C). # # ★모델에게 종류를 고르게 하지 않는다. enum 검사는 문자열 범위만 볼 뿐 # 뜻이 맞는지는 여전히 자가신고이고, 그 값을 캐시 신원으로 믿으면 # 캐시 미스보다 나쁜 **다른 장소 참조 재사용**이 난다(Codex 판정). # 값은 `background_classify` 의 `is_indoor` 에서 읽는다. # # ★같은 장소라도 안과 밖은 다른 참조다 — 한 값으로 두면 `L01` 의 # 실내 뷰와 실외 뷰가 한 참조로 합쳐진다. SCOPE_ROLE_INTERIOR = "location_interior" SCOPE_ROLE_EXTERIOR = "location_exterior" SCOPE_ROLES = (SCOPE_ROLE_INTERIOR, SCOPE_ROLE_EXTERIOR) # 후보 다운로드 상한 — search_grounded_ref MAX_PICK_CANDIDATES 와 별도로 # 조사당 비용을 묶는다(후보가 많아도 선택 품질이 늘지 않는 실측 관례). MAX_CANDIDATES = 4 #: ★이 late 경로가 **실제로 도는 라운드 수**. 재검색은 중앙 획득에만 있다. #: 이관이 끝나면 이 경로 자체가 없어진다. LATE_PATH_ROUNDS = 1 #: ★판별 결과의 **명시 상태**. 「비대상」과 「판별 실패」를 둘 다 `None` 으로 #: 접으면 앞쪽 production caller 가 **실패를 정상 빈 목록으로 오독**한다 #: (Codex BLOCK-4). optional side channel(`outcome`)에만 기대지 않는다. PLAN_NO_SUBJECT = "no_subject" # 정당한 era 없음 — 실패 아님 PLAN_FAILED = "failed" # 판별이 못 섰다 — 다시 봐야 한다 PLAN_OK = "plan" # 판별·질의 저작 — 저비용 보조 분업. ★2026-08-25 실측으로 **유지 확정**: # 같은 팩·같은 입력에 모델만 바꿔 재니 gpt 는 17/17 을, gemini-pro 는 6/8 을 # 「비대상」(빈 목록)으로 냈다. 둘 다 파싱 실패가 아니라 실제 판단이라 # 바꾸면 조사가 사실상 꺼진다. 재현 = ab_era_terms.py --arms v3:gpt ASSESS_MODEL = "gemini-flash" # 후보 선택 — 「전부 부적합」을 말할 수 있는 쪽으로 (2026-08-25 사용자 지시). # 근거 = probe_era_pick_model.py 실측: 같은 후보 4장에 Gemini 3.1 Pro 는 # 1·3·4 를 score=0 으로 걸러냈고 GPT 는 이유 한 줄만 남기고 통과시켰다. # ★2번은 Pro 도 놓쳤다 — 선택 모델은 회수된 후보가 나쁠 때의 마지막 방어일 # 뿐이고 근본은 검색이다(그래서 팩 v3 를 같이 올렸다). PICK_MODEL = "gemini-pro" def resolve_model_physical(alias: str) -> str: """alias 뒤의 **물리** 모델 문자열 — 지문·캐시 신원 스탬프용. (2026-08-25 Codex BLOCK) `gemini-pro`·`gemini-flash` 는 모델 이름이 아니라 설정으로 매핑되는 alias 다(`llm_client.py:495-500`). alias 만 신원에 넣으면 **설정의 실제 모델만 바뀌었을 때** 옛 판별·선택이 그대로 재사용된다. 저장소에 같은 관례가 이미 있다 (`multiroll_gemini.resolve_select_judge_model_physical`). 모르는 alias 는 그대로 돌려준다 — 물리 이름을 직접 쓴 경우다. """ from app.core.config import settings return { "gemini-pro": settings.gemini_text_model, "gemini-flash": settings.gemini_flash_model, "gemini-lite": settings.gemini_lite_model, "gpt-mini": settings.gemini_flash_model, "gpt": settings.openai_model, }.get(alias, alias) def resolve_era_pack(selector: str = ERA_RESEARCH_PACK_VERSION) -> str: try: return PACK_VERSION_MAP[selector] except KeyError: raise ValueError( f"unknown era_research pack selector {selector!r} " f"(known: {sorted(PACK_VERSION_MAP)})") def era_pack_content_hash(selector: str = ERA_RESEARCH_PACK_VERSION) -> str: return pack_dir_content_hash(_MODULE, resolve_era_pack(selector)) def build_assess_schema() -> Dict[str, Any]: return { "type": "object", "properties": { # ★maxItems 를 **소비 계약과 맞춘다** (2026-08-27, 감사 2-C). # 캐시 소비자는 `subjects[0]` 하나만 조사한다(아래 :426). # 4를 물으면 두 번째부터는 저작만 되고 **아무도 안 쓴다** — # 「한 장이 혼자 보여줄 수 있는 지배적 대상 하나」로 좁힌다. # 복합 대상을 여러 참조로 쪼개는 것은 새 기능이라 이 판의 # 범위가 아니다(Codex 확정). "subjects": { "type": "array", "maxItems": 1, "items": { "type": "object", "properties": { "subject_native": {"type": "string", "minLength": 2}, "search_terms_native": { "type": "array", "minItems": 2, "maxItems": 4, "items": {"type": "string", "minLength": 2}, }, "language_lock_native": { "type": "string", "minLength": 10}, "reason_ko": {"type": "string", "minLength": 2}, }, "required": ["subject_native", "search_terms_native", "language_lock_native", "reason_ko"], "additionalProperties": False, }, }, }, "required": ["subjects"], "additionalProperties": False, } def assess_subjects( *, step_tag: str, subject_text: str, world_facts_block: str, project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: """조사 필요 대상 판별 + 원어 질의 저작 — 1콜. 빈 목록 = 조사 불요(대부분의 입력). 상한은 호출측이 아니라 여기서 자른다(MAX_SUBJECTS_PER_CALL) — 판별이 길게 물어도 지출이 안 는다. """ from app.modules.llm.llm_client import call_structured assess_sys = load_prompt( _MODULE, "assess_sys", version=resolve_era_pack()).strip() parts = [{ "type": "text", "text": ("SUBJECT TEXT:\n" + subject_text.strip() + "\n\nWORLD FACTS (region and era — authoritative):\n" + (world_facts_block or "").strip()), }] pc = {**(project_config or {}), step_tag: {"model": ASSESS_MODEL}} data = call_structured( step_tag, assess_sys, parts, build_assess_schema(), project_config=pc, schema_name=step_tag, opik_metadata=opik_metadata, ) subjects = [s for s in (data.get("subjects") or []) if isinstance(s, dict)] return subjects[:MAX_SUBJECTS_PER_CALL] def build_ref_role(subject_native: str) -> str: """참조 역할문 — 형태 권위만 갖고 구도·인물 권위는 없다(뼈대/표면 관례).""" return load_prompt( _MODULE, "ref_role", version=resolve_era_pack(), subject=subject_native).strip() def _verdict_rows(verdict: Dict[str, Any]) -> List[Dict[str, Any]]: """심판이 채운 후보별 칸 — 성공·실패 어느 쪽이든 같은 모양으로 남긴다.""" return [ {"index": d.get("index"), "usable": d.get("usable"), "score": d.get("score"), "reason_ko": d.get("reason_ko")} for d in (verdict.get("verdicts") or []) if isinstance(d, dict) ] def research_reference( *, subject: Dict[str, Any], world_facts_block: str, out_path: Path, step_tag: str, project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, openai_client: Any = None, audit: Optional[Dict[str, Any]] = None, ) -> Optional[Dict[str, Any]]: """한 대상을 조사해 참조 1장을 확보한다 — 실패는 None(생성은 계속). 검색→후보 다운로드(안전 정책)→GPT 선택→out_path 저장. meta 반환: {subject, terms, queries, picked_index, picked_url, sha256, file}. 조사 실패가 생성을 막지 않는 것이 계약이다(참조 없이 그리던 기존 동작으로 그대로 진행) — 다만 결과 meta 가 없으면 기록도 없으므로 호출측은 None 도 감사 필드(researched=false)로 남겨야 한다. audit (2026-08-25 Codex BLOCK-2): **실패해도 무엇을 했는지 남긴다.** None 을 돌려주는 갈래에서도 나간 질의·회수 URL·후보별 판정이 여기 채워진다. 특히 「전부 부적합」(chosen_index=0)은 심판 모델을 바꾼 이유 그 자체인데, 종전에는 그 판정이 어디에도 안 남고 사라졌다. """ from app.modules.llm.llm_client import call_structured from app.modules.pipeline.multiroll_gemini import png_part from app.modules.pipeline.search_grounded_ref import ( # ★`build_pick_schema`·`build_pick_user_head` 는 **더 안 쓴다** — # 심판에게 `REGION AND ERA` 를 주고 시대·국적·평범함을 판정시키던 # 것들이다. 야외 legacy 는 그대로 쓰므로 **모듈에서 지우지는 않는다** # (소비자가 없어질 때까지 보존, Codex). download_candidate, resolve_ref_pack_version, search_reference_images, ) if openai_client is None: from app.core.openai_keys import openai_client as _mk openai_client = _mk() aud = audit if audit is not None else {} terms = [str(t) for t in (subject.get("search_terms_native") or [])] lock = str(subject.get("language_lock_native") or "") name = str(subject.get("subject_native") or "").strip() aud.update({"subject": name, "terms": terms, "status": "incomplete"}) if not (name and terms): aud["status"] = "no_subject" return None try: res = search_reference_images( openai_client, directive_native=name, terms_native=terms, language_lock_native=lock, ) except Exception as exc: # noqa: BLE001 — 조사 실패는 생성 비차단 logger.warning("era_research[%s]: 검색 실패 — %r", step_tag, exc) aud["status"] = "search_error" return None aud["queries"] = list(res.get("queries") or []) images = list(res.get("images") or [])[:MAX_CANDIDATES] if not images: logger.info("era_research[%s]: 후보 0 — %s", step_tag, name) aud["status"] = "no_candidates" return None cand_dir = out_path.parent / f".{out_path.stem}_cand" cand_paths: List[Path] = [] cand_urls: List[str] = [] for i, im in enumerate(images, 1): dest = cand_dir / f"cand{i}.png" if download_candidate( str(im.get("image_url") or ""), dest, fallback_url=str(im.get("thumbnail_url") or "")): cand_paths.append(dest) cand_urls.append(str(im.get("image_url") or "")) aud["candidate_urls"] = cand_urls if not cand_paths: logger.info("era_research[%s]: 다운로드 전멸 — %s", step_tag, name) aud["status"] = "download_failed" return None # ★★★**심판에게 시대·지역을 주지 않는다** (사용자 확정 2026-08-30). # 「우리가 찾던 게 맞냐는 **해당 오브젝트 종류만** 보는 것뿐이야. # 상세히는 인간도 몰라 전문가가 아니면. 자동차인지, 화폐인지 등등 만」 # # 옛 `pick_system` + `build_pick_user_head` 는 심판에게 `REGION AND ERA` # 를 주고 「다른 나라나 다른 시대면 탈락」·「관광지·꾸민 것이면 탈락」· # 「방문객용으로 개조됐나」를 판정시켰다. 그건 짐작이고, 사용자 규칙에 # 정면으로 어긋난다. # # ★시대·지역 좌표는 **검색 질의에는 그대로 남는다** — 위 `queries` 가 # 그것을 담는다. VLM 이 **검증하지 않을 뿐**이다. from app.modules.pipeline.coarse_type_pick import ( PROMPT_MODULE as _CT_MODULE, SCHEMA_STEM as _CT_SCHEMA, SYSTEM_STEM as _CT_SYS, combine_coarse_verdicts, load_pack as _ct_pack) _ctp = _ct_pack() pick_sys = _ctp["stems"][_CT_SYS]["content"].strip() parts: List[Dict[str, Any]] = [{ # ★찾는 **종류의 이름만** 준다. 세계 사실 블록은 안 준다. "type": "text", "text": "THE KIND OF THING the photograph must show:\n" + name, }] for i, p in enumerate(cand_paths, 1): parts.append({"type": "text", "text": f"PHOTOGRAPH {i}:"}) parts.append(png_part(p)) pick_tag = f"{step_tag}_pick" pc = {**(project_config or {}), pick_tag: {"model": PICK_MODEL}} try: raw = call_structured( pick_tag, pick_sys, parts, _ctp["stems"][_CT_SCHEMA]["content"], project_config=pc, schema_name=pick_tag, opik_metadata=opik_metadata, ) except Exception as exc: # noqa: BLE001 — 선택 실패도 비차단 logger.warning("era_research[%s]: 선택 실패 — %r", step_tag, exc) aud["status"] = "pick_error" return None # ★**1심이다.** 기존 era 경로는 단일 picker 이므로 새 다중 심판 비용을 # 넣지 않는다 (Codex). coarse 계약은 1심에서도 그대로 돈다 — # 「살아남은 **모든** 심판이 yes+visible」이 심판 하나면 그 하나가 정한다. verdict = combine_coarse_verdicts({PICK_MODEL: raw}, len(cand_paths)) aud["coarse"] = {k: verdict[k] for k in ("eligible", "chosen_index", "reason", "single_judge", "rejected_judges")} aud["verdicts"] = (raw or {}).get("verdicts") chosen = int(verdict.get("chosen_index") or 0) if not (1 <= chosen <= len(cand_paths)): # ★심판 모델을 바꾼 이유가 바로 이 갈래다(「전부 부적합」을 말할 수 # 있는 쪽으로). 그런데 종전에는 이 판정이 어디에도 안 남았다 — # 후보 URL 과 후보별 이유를 남겨야 「심판이 까다로운가 / 검색이 # 나쁜가」를 사후에 가른다. logger.info("era_research[%s]: 사용 가능 후보 없음 — %s", step_tag, name) aud["status"] = "no_usable" aud["chosen_reason_ko"] = str(verdict.get("chosen_reason_ko") or "") return None data = cand_paths[chosen - 1].read_bytes() out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_bytes(data) aud["status"] = "picked" return { "subject": name, "terms": terms, "queries": list(res.get("queries") or []), "candidates": len(cand_paths), "picked_index": chosen, "picked_url": cand_urls[chosen - 1], "picked_reason_ko": str(verdict.get("chosen_reason_ko") or ""), # 고른 것의 이유만 남기면 **왜 나머지를 버렸는지**가 사라진다 — # 실측(2026-08-25): 후보 4장이 전부 대상이 아닌데도 심판이 2번을 # 골랐고, 나머지 3장을 어떻게 봤는지가 어디에도 없어 「심판이 느슨 # 했나 / 후보가 원래 그랬나」를 사후에 가를 수 없었다. 심판이 이미 # 채운 칸이라 추가 지출이 없다. "verdicts": _verdict_rows(verdict), "candidate_urls": cand_urls, "sha256": hashlib.sha256(data).hexdigest(), "file": out_path.name, } def _cache_sha(*parts: str) -> str: basis = "\n".join(p.strip() for p in parts) return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:16] def _identity_of(canonical_scope_id, canonical_scope_role, canonical_scope_sha) -> tuple: """정본 신원 — **셋이 다 있어야** 합친다 (2026-08-27 감사 2-C). 하나라도 비면 옛 자유 키로 떨어진다 — **잘못 합치는 것이 중복 조사보다 나쁘다**(Codex 확정). Returns: ``(scope_ok, identity_note)`` """ scope = [canonical_scope_id, canonical_scope_role, canonical_scope_sha] ok = all(isinstance(x, str) and x.strip() for x in scope) if canonical_scope_role and canonical_scope_role not in SCOPE_ROLES: # 코드가 정하는 값인데 모르는 것이 왔다 — 추측하지 않는다. ok = False note: Dict[str, Any] = ( {"identity": "canonical", "scope_id": canonical_scope_id, "scope_role": canonical_scope_role, "scope_sha": canonical_scope_sha} if ok else {"identity_fallback": True, "reason": ("scope_role_unknown" if canonical_scope_role and canonical_scope_role not in SCOPE_ROLES else "scope_incomplete"), "scope_id": canonical_scope_id, "scope_role": canonical_scope_role, "scope_sha": canonical_scope_sha}) return ok, note def _make_fail_put(cache_get, cache_put): """실패 감사 기록기. ★**한 벌만 둔다** — 두 곳에 복사하면 한쪽만 고쳐진다. (era R1 BLOCK-3) 실패 감사는 시도마다 **단조 변화**해야 한다 — 같은 bytes 로 덮어쓰면 resume 걷기의 반복 유료 실패가 records 변화 0 으로 보여 지출 감지에 안 잡히고 **무상한 재검색이 열린다**. """ def _fail_put(sha: str, payload: Dict[str, Any]) -> None: key = f"era_fail::{sha}" prev = cache_get(key) attempts = (prev.get("attempts", 0) + 1 if isinstance(prev, dict) else 1) cache_put(key, {**payload, "attempts": attempts}) return _fail_put def build_assess_plan(*, subjects: List[Dict[str, Any]], a_key: str, a_sha: str, identity_note: Dict[str, Any], world_facts_block: str, pack: str) -> Dict[str, Any]: """★**판별 결과를 되짚을 수 있게 담는다** — 중앙 획득(19.x)이 읽을 것. ★`subjects[0]` 만 두면 **같은 판별을 되짚을 수 없다** (Codex). 19.x 는 앞쪽이 판별한 뒤 한참 있다가 도는데, 그때 「무엇을 근거로 이 대상을 골랐나」와 「어느 계약으로 판별했나」가 없으면 확인할 길이 없다. 담는 것: - `subjects` — 저작된 대상·질의·잠금문 (그대로) - `a_key`·`a_sha` — 판별 캐시 좌표. 되짚기의 뿌리 - 정본 신원 note — canonical 인지 fallback 인지, 그 사유까지 - **판별 계약 좌표** — 팩·팩 내용 hash·모델 alias·physical model·정책 ★`r_key` 에는 이 `a_sha` 를 **넣지 않는다.** 원본 대상 텍스트가 샷마다 달라 적중이 1.1% 로 떨어진 실측이 있다(2026-08-25 Codex BLOCK-1). 정본이 있으면 **stable subject id + controlled role + canonical payload sha** 를 쓴다. """ return { "contract_version": ERA_RESEARCH_POLICY_VERSION, "subjects": list(subjects), "assess_key": a_key, "assess_sha": a_sha, **dict(identity_note), "assess_contract": { "pack": pack, "pack_content_hash": era_pack_content_hash(), "model_alias": ASSESS_MODEL, "model_physical": resolve_model_physical(ASSESS_MODEL), "policy_version": ERA_RESEARCH_POLICY_VERSION, }, "world_facts_block": world_facts_block, } def normalize_era_reference(meta: Optional[Dict[str, Any]] ) -> Optional[Dict[str, Any]]: """★조사 결과 → **정규화 한 벌.** 부작용 없다(네트워크·캐시·생성 0). 왜 뽑았나 — `still_recipe_service` 의 groupbg·confined·plate 가 각각 `meta["subject"]` 로 역할문을 만들고 `meta["path"]` 로 경로를 짓고 `sha256` 을 계보에 붙인다. **세 곳에 같은 조립이 흩어져 있으면** 한 곳만 고쳐지고, 시험도 그 자리를 못 태운다(Codex: 「소스 문자열 count 는 보조일 뿐 통과 근거가 아니다」). ★`None`(비대상·실패)이면 `None` 을 돌려준다 — late 계약은 **비차단**이다. ★★**세 소비자를 한 mutation helper 로 뭉치지 않는다** (Codex). 셋의 **효과가 다르다** — groupbg 는 프롬프트+참조+지문, confined 는 refs 자리+계보, plate 는 프롬프트+경로+meta 다. 정규화만 공통으로 두고 **projection 은 셋으로** 나눈다. Returns: ``{"subject", "role_text", "path", "file", "sha256"}`` 또는 ``None`` """ if not meta: return None path = str(meta.get("path") or "") subject = str(meta.get("subject") or "") if not path or not subject: # ★반쪽 산출을 참조로 쓰지 않는다 — 역할문이나 경로가 없으면 붙일 수 # 없고, 조용히 붙이면 「참조가 있다」가 거짓이 된다. return None return { "subject": subject, "role_text": build_ref_role(subject), "path": Path(path), "file": str(meta.get("file") or ""), "sha256": str(meta.get("sha256") or ""), } def project_groupbg_era(prompt: str, env: Optional[Dict[str, Any]]) -> tuple: """groupbg — **프롬프트 · 참조 목록 · 지문 조각**을 함께 낸다. ★지문 조각을 같이 내는 이유: 참조가 바뀌면 groupbg 지문도 움직여야 한다. 따로 두면 참조만 갈리고 지문은 그대로여서 **옛 배경이 봉인**된다 (그 사고가 `still_recipe_service:2159` 주석에 적혀 있다). Returns: ``(prompt, era_refs, fingerprint_fragment)`` """ if not env: return prompt, None, None # ★★조각은 **`era_research_sha` 하나뿐**이다. `era_subject` 를 더하면 # outer meta bytes 가 바뀌어 **기존 groupbg 가 전부 한 번 재생성**된다 — # 참조가 바뀐 것도 아닌데 무효가 되는 것이라 값이 없다(비회귀 선택). # ★subject 는 `env` 로도 계보로도 이미 남는다. return (f"{prompt}\n\n{env['role_text']}", [env["path"]], {"era_research_sha": env["sha256"]}) def project_confined_era(refs: List[Any], env: Optional[Dict[str, Any]] ) -> tuple: """confined — **새 refs 와 계보 한 줄**을 함께 낸다. ★자리는 **1번**이다(도면 다음). 도면이 배치 권위이고 era 참조는 실물 look 앵커라 그 뒤에 온다. ★계보 줄은 asset UUID 가 없으므로 `role+file+sha` 병기다 — `_attach` 가 **이 줄만** 소비한다. Returns: ``(refs, lineage_row)`` — 둘 다 새 값. 입력을 안 건드린다. """ if not env: return list(refs), None out = list(refs) out.insert(1, (env["role_text"], env["path"])) return out, {"kind": "era_ref", "subject": env["subject"], "file": env["file"], "sha256": env["sha256"]} def project_plate_era(prompt: str, env: Optional[Dict[str, Any]], raw_meta: Optional[Dict[str, Any]] = None) -> tuple: """plate — **프롬프트 · `era_ref_path` · `era_meta`** 를 함께 낸다. ★`_run_bgfirst_bg` 가 경로와 meta 를 **둘 다** 받는다. 하나만 넘기면 참조는 실렸는데 계보가 없거나 그 반대가 된다. Returns: ``(prompt, era_ref_path, era_meta)`` """ if not env: return prompt, None, None return (f"{prompt}\n\n{env['role_text']}", env["path"], raw_meta if raw_meta is not None else env) def assess_cache_key(*, subject_text: str, world_facts_block: str, canonical_scope_id: Optional[str] = None, canonical_scope_role: Optional[str] = None, canonical_scope_sha: Optional[str] = None, pack: Optional[str] = None) -> str: """판별 캐시 키. ★**한 곳에서만** 만든다. 앞쪽 판별이 「이건 이미 있으니 공짜다」를 **사기 전에** 알아야 하는데, 키 조립을 그쪽에 베껴 두면 두 곳이 갈린다 — 갈리면 「공짜인 줄 알고 상한 밖에서 불렀다가 실제로는 샀다」가 된다. ★뿌리가 `subject_text`(샷마다 흔들림) → **정본 세 칸**. fallback 일 때만 옛 텍스트를 쓴다. """ scope_ok, _note = _identity_of(canonical_scope_id, canonical_scope_role, canonical_scope_sha) parts = ((str(canonical_scope_id), str(canonical_scope_role), str(canonical_scope_sha)) if scope_ok else (subject_text,)) return "era_assess::" + _cache_sha( *parts, world_facts_block, ERA_RESEARCH_POLICY_VERSION, pack or resolve_era_pack(), era_pack_content_hash(), ASSESS_MODEL, resolve_model_physical(ASSESS_MODEL)) def assess_plan_cached( *, step_tag: str, subject_text: str, world_facts_block: str, cache_get: Any, cache_put: Any, project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, failed_memo: Optional[set] = None, outcome: Optional[Dict[str, Any]] = None, canonical_scope_id: Optional[str] = None, canonical_scope_role: Optional[str] = None, canonical_scope_sha: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """★**판별만 한다.** 「무엇을 찾아야 하는지」까지고, 사지는 않는다. 앞쪽(`entity_filter` 전)이 이것만 부르고, 중앙 획득(19.x)이 `acquire_from_plan_cached` 를 부른다. 소유하는 것: `era_assess::` 캐시 · 판별 실패 기록 · `failed_memo` 의 판별 몫. **빈 목록(비대상)도 캐시**한다. Returns: ★**명시 envelope** — `{"status": …, "plan": …}`. `status` 는 `PLAN_NO_SUBJECT`(정당한 era 없음) · `PLAN_FAILED`(판별이 못 섰다) · `PLAN_OK` 중 하나다. ★둘을 `None` 하나로 접으면 앞쪽 caller 가 **실패를 정상 빈 목록으로 오독**한다(Codex BLOCK-4). legacy wrapper 만 둘을 `None` 으로 접어 기존 비차단을 유지한다. """ pack = resolve_era_pack() scope_ok, identity_note = _identity_of( canonical_scope_id, canonical_scope_role, canonical_scope_sha) if outcome is not None: outcome.update(identity_note) fail_put = _make_fail_put(cache_get, cache_put) def _mark_failed() -> None: if outcome is not None: outcome["failed"] = True a_key = assess_cache_key( subject_text=subject_text, world_facts_block=world_facts_block, canonical_scope_id=canonical_scope_id, canonical_scope_role=canonical_scope_role, canonical_scope_sha=canonical_scope_sha, pack=pack) a_sha = a_key.split("::", 1)[1] if failed_memo is not None and a_key in failed_memo: _mark_failed() return {"status": PLAN_FAILED, "plan": None, "reason": "same_walk_memo"} cached = cache_get(a_key) if isinstance(cached, dict) and "subjects" in cached: subjects = [x for x in (cached.get("subjects") or []) if isinstance(x, dict)] else: try: subjects = assess_subjects( step_tag=step_tag, subject_text=subject_text, world_facts_block=world_facts_block, project_config=project_config, opik_metadata=opik_metadata, ) except Exception as exc: # noqa: BLE001 — 판별 실패 비차단 logger.warning("era_research[%s]: 판별 실패 — %r", step_tag, exc) if failed_memo is not None: failed_memo.add(a_key) fail_put(a_sha, {"stage": "assess", "error": repr(exc)}) _mark_failed() return {"status": PLAN_FAILED, "plan": None, "reason": f"assess_error: {exc!r}"[:200]} # 판별에 **무엇을 넣었는지**도 남긴다 — 산출만 남기면 같은 판별을 # 재현할 수 없다. cache_put(a_key, {"subjects": subjects, "subject_text": subject_text, **identity_note}) if not subjects: # 비대상 — **실패 아님**(정당한 era 없음, 캐시됨) return {"status": PLAN_NO_SUBJECT, "plan": None} if not str((subjects[0] or {}).get("subject_native") or "").strip(): _mark_failed() # 판별 산출 결손 — 비대상이 아니라 미성립 return {"status": PLAN_FAILED, "plan": None, "reason": "subject_native_empty"} plan = build_assess_plan( subjects=subjects, a_key=a_key, a_sha=a_sha, identity_note=identity_note, world_facts_block=world_facts_block, pack=pack) # ★정본 여부를 계획에 실어 둔다 — 획득 쪽이 `_ref_parts` 를 고를 때 쓴다. plan["scope_ok"] = scope_ok plan["canonical_scope"] = {"id": canonical_scope_id, "role": canonical_scope_role, "sha": canonical_scope_sha} if outcome is not None: outcome["assess_plan"] = plan return {"status": PLAN_OK, "plan": plan} def acquire_from_plan_cached( plan: Dict[str, Any], *, step_tag: str, out_dir: Path, cache_get: Any, cache_put: Any, max_rounds: int, project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, openai_client: Any = None, failed_memo: Optional[set] = None, outcome: Optional[Dict[str, Any]] = None, ) -> Optional[Dict[str, Any]]: """★**그 계획으로 산다.** 판별은 안 한다. 소유하는 것: `era_ref::` 캐시 · 내용 주소 파일 · 획득 실패 기록 · `failed_memo` 의 획득 몫. ★`max_rounds` 는 **필수**다. 기본값을 두면 「이 호출이 실제로 몇 번 찾는가」와 키가 갈린다 — 그러면 「2라운드로 찾은 참조」와 「1라운드로 찾은 참조」가 같은 키를 갖는다(Codex BLOCK-1). ★★**지금은 `1` 만 받는다.** 이 함수에는 아직 라운드 loop 가 없어서 `2` 를 주면 **키는 2라운드인데 실행은 1라운드**가 된다 — 같은 결함을 이름만 바꿔 되풀이하는 것이다(Codex BLOCK-2). 실제 loop 는 중앙 획득 (step 5)에서 넣고, 그때 `2` 를 연다. """ if type(max_rounds) is not int or max_rounds != 1: raise ValueError( "max_rounds 는 지금 1 만 받는다 — 라운드 loop 가 아직 없다. " f"2 는 중앙 획득(step 5)에서 연다 (받은 값 {max_rounds!r})") from app.modules.pipeline.reference_acquisition import ( acquisition_contract_sha) from app.modules.pipeline.search_grounded_ref import search_contract_sha subject = (plan.get("subjects") or [{}])[0] name = str(subject.get("subject_native") or "").strip() if not name: if outcome is not None: outcome["failed"] = True return None pack = plan.get("assess_contract", {}).get("pack") or resolve_era_pack() world_facts_block = plan.get("world_facts_block") or "" scope_ok = bool(plan.get("scope_ok")) canon = plan.get("canonical_scope") or {} fail_put = _make_fail_put(cache_get, cache_put) def _mark_failed() -> None: if outcome is not None: outcome["failed"] = True # ★정본이면 정본 세 칸, 아니면 **옛 조립 그대로**(저작된 대상·질의·잠금문). # 여기서 `subject_text` 로 바꾸면 샷마다 다른 문장이 같은 대상을 다시 # 사게 만든다. # ★★fallback 조립은 **옛것 그대로**여야 한다. 내가 `"\n".join` 을 # `"|".join` 으로 바꿔서 **coarse 계약과 무관하게** fallback 키가 또 # 바뀔 뻔했다(Codex BLOCK-3). refactor 가 키 재료를 바꾸면 안 된다. ref_parts = ((str(canon.get("id")), str(canon.get("role")), str(canon.get("sha"))) if scope_ok else (name, "\n".join(str(t) for t in (subject.get("search_terms_native") or [])), str(subject.get("language_lock_native") or ""))) r_sha = _cache_sha( *ref_parts, world_facts_block, ERA_RESEARCH_POLICY_VERSION, pack, era_pack_content_hash(), search_contract_sha(), PICK_MODEL, resolve_model_physical(PICK_MODEL), # ★★**이 호출이 실제로 하는 라운드 수**를 접는다. 기본값 금지. acquisition_contract_sha(rounds=max_rounds)) r_key = f"era_ref::{r_sha}" out_path = out_dir / f"eraref_{r_sha}.png" if failed_memo is not None and r_key in failed_memo: _mark_failed() return None rc = cache_get(r_key) if (isinstance(rc, dict) and rc.get("sha256") and out_path.is_file() and out_path.stat().st_size > 0 and hashlib.sha256(out_path.read_bytes()).hexdigest() == rc.get("sha256")): return {**rc, "path": str(out_path)} research_audit: Dict[str, Any] = {} meta = research_reference( subject=subject, world_facts_block=world_facts_block, out_path=out_path, step_tag=step_tag, project_config=project_config, openai_client=openai_client, opik_metadata=opik_metadata, audit=research_audit, ) if not meta: # 같은 걷기에서는 이 대상 재시도 금지 + 실패 감사 영속 if failed_memo is not None: failed_memo.add(r_key) fail_put(r_sha, {"stage": "research", "subject": name, **research_audit}) _mark_failed() return None cache_put(r_key, meta) return {**meta, "path": str(out_path)} def assess_and_research_cached( *, step_tag: str, subject_text: str, world_facts_block: str, out_dir: Path, cache_get: Any, cache_put: Any, project_config: Optional[Dict[str, Any]] = None, openai_client: Any = None, failed_memo: Optional[set] = None, outcome: Optional[Dict[str, Any]] = None, canonical_scope_id: Optional[str] = None, canonical_scope_role: Optional[str] = None, canonical_scope_sha: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """판별+조사를 records 사이드카에 캐시 — 같은 장소의 샷들이 1회만 지출한다 (#119①: 샷별 판 98개가 장소별로 겹치는 실측 대응). ## canonical_scope_id — 신원을 저작에서 떼어낸다 (2026-08-27, 감사 2-C) ★그 「1회만」이 **한 번도 안 됐다.** 신원이 LLM 자유 저작 (`subject_native`·`search_terms`·`language_lock`)에 매여 있어 같은 장소를 조금 다르게 부르면 다른 키가 됐다. records 전수 실측: **저작 대상 179건 중 서로 다른 문자열 177종 — 적중 1.1%.** 3씬 시나리오(물리 장소 2곳)에서 조사가 6번 돌았고 자전거 수리점 하나가 네 번이었다. 그래서 호출측이 **정본 신원 세 칸**을 준다: canonical_scope_id `scene_director.primary_location` — 실물은 `L01` 같은 **short_id** 다(UUID 아님). `background_classify.members[].loc_id` 와 **같은 공간**이라 변환이 필요 없다. canonical_scope_role `location_interior` / `location_exterior`. **코드가 정한다** — `background_classify` 의 `is_indoor`(그 loc_id 행 하나)에서 읽는다. 모델에게 종류를 고르게 하지 않는다. canonical_scope_sha 그 id 에서 읽어 **실제로 판별에 보내는** 정본 내용(이름·서술)의 해시. 정본이 고쳐지면 옛 참조가 남지 않는다. ★**같은 장소라도 안과 밖은 다른 참조다.** 역할을 안 나누면 같은 `L01` 의 실내 뷰와 실외 뷰가 한 참조로 합쳐진다. ★**셋 중 하나라도 없으면 합치지 않는다.** 옛 자유 키로 떨어지고 `identity_fallback` 과 사유를 기록에 남긴다. 잘못 합쳐 **다른 장소의 참조를 재사용**하는 것이 중복 조사보다 나쁘다. ★**입력도 같이 맞춰야 한다.** 키에서만 빼면 「첫 샷의 문안이 그 장소 전체의 결과가 되는」 순서 의존 캐시가 된다 — 호출측이 같은 id 에서 푼 정본 이름·서술을 `subject_text` 로 넘긴다(샷별 문안 아님). failed_memo (Codex BLOCK-1): 호출측이 걷기(1바퀴) 단위로 소유하는 실패 sentinel — 같은 걷기에서 실패한 키는 재시도하지 않는다(같은 장소 다음 샷마다 유료 사슬 재구매 차단). 메모리 전용이라 다음 resume 걷기는 자연 재시도. 실패 감사는 era_fail:: record 로 영속(게이트로 읽지 않는다). cache_get(key)->dict|None / cache_put(key, dict) 는 호출측(records) 소유 — 이 모듈은 저장소를 모른다. 캐시 키 = 대상 텍스트·세계관·정책· 팩의 sha16. **판별 결과는 빈 목록(비대상)도 캐시**해 재판별 지출을 막고, **조사 실패는 캐시하지 않는다**(비차단 계약 그대로 — 다음 방문이 재시도한다). 참조 파일은 out_dir/eraref_.png 로 내용 주소화되어 샷 간 공유된다. 반환 meta 에는 "path"(절대 경로)가 붙는다. """ # ★★**호환 wrapper 다** — 둘을 순서대로 부른다. 공개 signature 와 반환 # 모양은 그대로다. late caller(groupbg·confined·plate)가 이것을 쓴다. # # ★`max_rounds=LATE_PATH_ROUNDS`(=1) — **재검색은 중앙 획득에만** 넣는다. # late 는 이관 대상이라 비용을 늘릴 이유가 없다(Codex). 키도 **실제로 # 하는 것**을 접는다. plan = assess_plan_cached( step_tag=step_tag, subject_text=subject_text, world_facts_block=world_facts_block, cache_get=cache_get, cache_put=cache_put, project_config=project_config, failed_memo=failed_memo, outcome=outcome, canonical_scope_id=canonical_scope_id, canonical_scope_role=canonical_scope_role, canonical_scope_sha=canonical_scope_sha, ) # ★legacy wrapper **만** 둘을 `None` 으로 접는다 — 기존 비차단 유지. if (plan or {}).get("status") != PLAN_OK: return None return acquire_from_plan_cached( plan["plan"], step_tag=step_tag, out_dir=out_dir, cache_get=cache_get, cache_put=cache_put, max_rounds=LATE_PATH_ROUNDS, project_config=project_config, openai_client=openai_client, failed_memo=failed_memo, outcome=outcome, )