"""중앙 조사 결과 → **이 샷의 sidecar**. ★고르지도 판정하지도 않는다. ## 무엇을 하나 Codex D-inert 3/3 (2026-09-01) — 「중앙 결과 → `members_from_rows` → `write_sidecar`」. 중앙 조사 CP 이 에피소드에서 **무엇을 구했나** 이 샷의 visible 그 중 **이 컷에 보이는 것** → sidecar 그 컷에 붙을 멤버들과 요구 ★★**sidecar 가 없으면 조립은 한 글자도 안 바뀐다** — 그것이 이 설계의 문이다 (`grounding_reference_bundle.attach_from_rpc` 가 멤버 키가 없으면 0 을 낸다). 그래서 중앙 CP 가 없거나 이 컷에 해당 멤버가 없으면 **아무것도 안 적는다**. ## 어느 멤버가 이 컷의 것인가 멤버는 **덮는 의무 목록**(`covers`)을 갖는다. 부모 장소 맥락 한 장이 형제 `LP01`·`LP02` 의 의무를 함께 덮을 수 있다. 그래서 — 이 컷의 visible 신원 중 하나라도 그 멤버의 `covers` 에 있으면 → 이 컷 것 ★이름으로 안 고른다. **정본 ID 로만** 본다. """ from __future__ import annotations import copy from typing import Any, Callable, Dict, List, Optional, Sequence from app.modules.pipeline import grounding_entity_contract as _gc #: ★2026-09-02: 소비 문이 **고증 축**을 함께 본다 — 판을 올린다. #: 옛 판으로 쓴 sidecar 는 확인 안 된 사진을 붙였을 수 있다. SIDECAR_WRITER_CONTRACT_VERSION = "2.202609020800" #: ★이 모듈이 **인물 아웃룩 자리**를 집행한다 (아웃룩). ENFORCES_GATE = _gc.ENFORCE_BY_OUTLOOK_ATTACHMENT class SidecarWriterError(RuntimeError): """sidecar 를 못 만든다. ★조용히 빈손으로 지나가지 않는다.""" def visible_final_ids(visible_entities: Sequence[Any]) -> List[str]: """이 컷이 보는 **정본 ID** 들. ★모양 둘을 다 받는다(dict · 문자열).""" out = [] for e in visible_entities or (): if isinstance(e, dict): got = str(e.get("short_id") or e.get("id") or "") else: got = str(e or "") if got: out.append(got) return sorted(set(out)) def members_for_shot(central_cp: Optional[Dict[str, Any]], visible_entities: Sequence[Any], *, content_sha_of: Callable[[Dict[str, Any]], str], coordinate_of: Callable[[Dict[str, Any]], Dict[str, Any]] ) -> List[Dict[str, Any]]: """중앙 CP + 이 컷의 visible → **이 컷의 멤버들**. 없으면 빈 목록. ★못 구한 의무도 **남긴다** — 이미지 요구는 안 되지만 「무엇이 없었나」가 기록에 있어야 나중 UI 가 고칠 수 있다. """ from app.modules.pipeline import grounding_bundle_projection as bp rows = ((central_cp or {}).get("data") or {}).get("rows") if not rows: return [] # ★옛 길 — 조립이 한 글자도 안 바뀐다 every = bp.members_from_rows(rows, content_sha_of=content_sha_of, coordinate_of=coordinate_of) seen = set(visible_final_ids(visible_entities)) if not seen: return [] return [copy.deepcopy(m) for m in every if seen & set(m.get("covers") or ())] def assert_central_checkpoint(central_cp: Optional[Dict[str, Any]]) -> None: """켠 판에서 중앙 조사 CP 가 **온전한가**. ★공개 끝점 — 아니면 선다. ★판정은 `grounding_central_acquisition` 한 곳이 한다 — 여기서 같은 규칙을 다시 적으면 두 벌이 된다. """ from app.modules.pipeline import grounding_central_acquisition as ca ca.acquisition_projection(central_cp, required=True) class ReferenceCoordinateConflict(RuntimeError): """한 줄에 적힌 두 좌표(`chosen.path` · `acquisition.chosen_path`)가 다르다.""" def resolved_reference_path(row: Dict[str, Any]) -> str: """고른 사진의 **하나의** 상대 경로. ★두 저장 위치를 대조한다 (Codex BLOCK 2026-09-02). CP 의 selected 줄에는 `acquisition.chosen.path` 와 `acquisition.chosen_path` 가 **둘 다** 있다(실측 · 값이 같다). 한쪽만 정본으로 택해 다른 쪽을 무시하면 둘이 갈린 날 그 차이를 아무도 못 본다. 둘 다 있으면 **같아야** 하고, 다르면 붙이기·판정 전에 선다. 한쪽만 있으면 그것을 쓴다. 둘 다 없으면 "" — selected 줄이면 `_selected` 가 「내용 해시가 없다」로 선다. """ acq = row.get("acquisition") or {} a = str(((acq.get("chosen") or {}).get("path")) or "") b = str(acq.get("chosen_path") or "") if a and b and a != b: raise ReferenceCoordinateConflict( f"{row.get('research_subject_id')!r} 의 좌표가 갈린다 — chosen.path={a!r} " f"· chosen_path={b!r}. 어느 파일을 붙일지 모르는 채 안 붙인다") return a or b def _chosen_rel_path(row: Dict[str, Any]) -> str: return resolved_reference_path(row) def row_content_sha256(row: Dict[str, Any], *, root) -> str: """고른 사진의 **내용 해시** — 대조를 지난 resolved path 의 파일 bytes 로 센다. ★한 곳. ★★★실측 (2026-09-02 1단계 live · attempt aec5ce986c4c · 유료 30): production `detail_steps` 가 `chosen.content_sha256` 칸을 읽었는데 `reference_acquisition` CP 의 chosen 에는 그 칸이 **없다**(caption·index·path·source_website_url·url). 그래서 7샷 전부 「selected 인데 내용 해시가 없다」로 실패했다. 사람 판정 (`grounding_fidelity_review`)과 canary probe 는 파일 bytes 로 셌기에 통과했고 — 그것이 「도구가 production 과 다른 입력으로 쟀다」는 결함이다. 이제 셋이 이 함수 하나를 쓴다. 파일이 없으면 "" — 붙이지 않는다. """ import hashlib from pathlib import Path rel = _chosen_rel_path(row) if not rel: return "" p = Path(root) / rel if not p.is_file(): return "" h = hashlib.sha256() with open(p, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def row_file_coordinate(row: Dict[str, Any], *, root) -> Dict[str, Any]: """고른 사진의 **좌표** — 조립(`attach_from_rpc`)이 이것으로 bytes 를 읽는다.""" from pathlib import Path from app.modules.pipeline import grounding_reference_bundle as rb rel = _chosen_rel_path(row) return {"source": rb.SOURCE_FILE, "path": str(Path(root) / rel) if rel else ""} def default_reference_root(): """`chosen.path` 는 `projects//...` 상대 경로 — 그 기준은 projects_dir 의 부모.""" from pathlib import Path from app.core.config import settings return Path(settings.projects_dir).parent def write_for_shot(rpc: Dict[str, Any], central_cp: Optional[Dict[str, Any]], visible_entities: Sequence[Any], *, content_sha_of: Callable[[Dict[str, Any]], str], coordinate_of: Callable[[Dict[str, Any]], Dict[str, Any]] ) -> int: """이 컷의 sidecar 를 RPC 에 **원자적으로** 적는다. Returns: 적은 **요구 수**. 0 이면 RPC 는 **한 글자도 안 바뀐다**. ★멤버와 요구를 한 함수(`write_sidecar`)가 같이 적는다 — 두 벌로 만들면 그 둘이 어긋난 채로 굳는다. """ from app.modules.pipeline import grounding_reference_bundle as rb members = members_for_shot(central_cp, visible_entities, content_sha_of=content_sha_of, coordinate_of=coordinate_of) # ★★아웃룩은 **이 카드가 이미 요구하는 짝**에만 붙는다 — 조사 결과가 # 새 짝을 만들지 않는다 (Codex 2026-09-01). members = members + outlook_members_for_shot( central_cp, rpc, content_sha_of=content_sha_of, coordinate_of=coordinate_of) if not members: return 0 # ★안 적는다 — 무변이 계약이다 return len(rb.write_sidecar(rpc, members)) #: 카드가 요구를 적어 두는 자리. ★이름을 여기서 지어내지 않는다 — #: `render_prompt_card.build_asset_requirements` 가 쓰는 그대로다. CARD_REQUIREMENTS_KEY = "asset_requirements" CARD_REQUIRED_REFS_KEY = "required_refs" def required_outlook_ref_ids(rpc: Dict[str, Any]) -> List[str]: """이 컷의 카드가 요구하는 **인물 아웃룩 신원**들 (`C##O##`). ★★★이것이 「그 복장이 이 컷에서 실제로 쓰이나」의 **정본**이다. 카드는 `O00`(빈 아웃룩)·state variant·`generic_descriptor_allowed` 를 이미 걸러 뒀다 — 그 판단을 여기서 다시 하지 않는다. """ from app.modules.pipeline import grounding_reference_bundle as rb req = ((rpc or {}).get(CARD_REQUIREMENTS_KEY) or {}).get( CARD_REQUIRED_REFS_KEY) if not isinstance(req, list): return [] out = [] for r in req: if not isinstance(r, dict): continue if str(r.get("kind") or "") != rb.OUTLOOK_KIND: continue got = str(r.get("id") or "") if got: out.append(got) return sorted(set(out)) def outlook_members_for_shot(central_cp: Optional[Dict[str, Any]], rpc: Dict[str, Any], *, content_sha_of: Callable[[Dict[str, Any]], str], coordinate_of: Callable[[Dict[str, Any]], Dict[str, Any]] ) -> List[Dict[str, Any]]: """중앙 CP + 이 카드의 요구 → **아웃룩 자리에 붙을 멤버들**.""" from app.modules.pipeline import grounding_bundle_projection as bp rows = ((central_cp or {}).get("data") or {}).get("rows") if not rows: return [] want = required_outlook_ref_ids(rpc) if not want: return [] # ★이 컷은 아웃룩을 안 쓴다 return bp.outlook_members_from_rows( rows, required_ref_ids=want, content_sha_of=content_sha_of, coordinate_of=coordinate_of) def roles_in(rpc: Dict[str, Any]) -> List[str]: """이 sidecar 가 쓸 **역할**들. ★공개 끝점 — 선언 밖이면 조립이 선다.""" from app.modules.pipeline import grounding_reference_bundle as rb members = rb.members_from_rpc(rpc) if not members: return [] plan, _req = rb.plan_bundle(members) return sorted({str(p["role"]) for p in plan}) def assert_roles_declared(rpc: Dict[str, Any], known: Sequence[str]) -> None: """쓸 역할이 **전부 선언돼 있나**. ★아니면 provider 앞에서 선다.""" missing = [r for r in roles_in(rpc) if r not in set(known)] if missing: raise SidecarWriterError( f"선언 안 된 역할 {missing} — `REF_ROLE_VALUES` 는 닫힌 목록이다")