"""D6 §4.8 — endpoint entry preflight: bg_catalog/shot_binding hash freshness. Compares ``background_master_plan`` cp's ``bg_catalog_hash`` / ``shot_binding_hash`` against each consumer's stamped ``consumed_bg_catalog_hash`` / ``consumed_shot_binding_hash``. Any mismatch, missing-while-D6-active consumer manifest, or corrupt manifest raises ``StaleUpstreamError`` (HTTP 422 via ``app_error_handler``). T5-fix iter7 결정: 4 consumer (``floor_plan_prompt`` / ``background_prompt`` / ``background_render`` / ``scene_detail``) 모두 catalog + binding 양쪽 stamp 하므로 preflight 도 4 consumer 전체 양쪽 비교. fail-fast 강화 (silent skip 차단): - manifest parse 실패 → ``StaleUpstreamError`` (corrupt manifest 는 freshness 판단 불가 — 운영 신호 손실 차단). - master_plan 에 D6 hash 가 있는데 consumer manifest 가 missing → ``StaleUpstreamError`` (consumer 가 D6 mode 에서 안 돌았다는 뜻). - master_plan 자체 missing 또는 ``bg_catalog_hash`` 부재 → return silently (D6 미적용 episode — D5 fallback path). Required-consumer invariant (T8a-fix review iter1 BLOCKING 검증): ``bg_mode in {"on", "floor_plan_anchored"}`` 일 때 4 consumer 모두 ``_execute`` 안에서 manifest 를 작성한다 — empty-input path (no fp_jobs / no bg_jobs / no renderable / no shots) 도 T5/T5-fix 의 "empty path 도 hash stamp" 분기로 빈 data + ``consumed_*_hash`` stamp 를 가진 manifest 를 남긴다. 따라서 master_plan 이 D6 hash 를 stamp 하면 (= bg_mode 가 on 이었음) 4 consumer 도 manifest 가 있어야 한다. Reference (각 consumer 의 empty-path stamp 분기): - floor_plan_prompt_step.py:131-147 (no fp_jobs) - background_prompt_step.py:184-198 (no bg_jobs) - background_render_step.py:238-244 (no renderable) - scene_detail (detail_steps.py:836+) — bg_mode 분기 없음, 항상 실행 spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.8 plan: T8 (Phase A — service module + tmp manifest unit tests) """ from __future__ import annotations import json import logging from pathlib import Path from typing import Any, Dict, Optional from app.core.config import settings from app.core.errors import StaleUpstreamError logger = logging.getLogger(__name__) # T5-fix iter7 결정: 4 consumer 모두 catalog + binding 양쪽 stamp. _CONSUMERS = ( "floor_plan_prompt", "background_prompt", "background_render", "scene_detail", ) def _manifest_path(pid: str, eid: str, step: str) -> Path: return ( Path(settings.projects_dir) / pid / "checkpoints" / "episodes" / eid / step / "manifest.json" ) def _read_manifest(pid: str, eid: str, step: str) -> Optional[Dict[str, Any]]: """Return parsed manifest, or ``None`` if file does not exist. Parse failure raises ``StaleUpstreamError`` — silent skip is forbidden. A corrupt manifest cannot serve as the source of truth for freshness, and swallowing the error would let a stale chain dispatch unnoticed. """ p = _manifest_path(pid, eid, step) if not p.exists(): return None try: return json.loads(p.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as exc: raise StaleUpstreamError( upstream=step, expected_bg_catalog_hash="(unknown — manifest corrupt)", observed_bg_catalog_hash="", remediation=( f"manifest at {p} is corrupt — inspect/restore from archive " f"or POST /steps/{step}?mode=force" ), message=( f"STALE_UPSTREAM upstream={step}: manifest parse failed ({exc})" ), ) def check_bg_catalog_freshness( pid: str, eid: str, *, enforce_shot_binding: bool = True, ) -> None: """Compare master_plan hashes against consumer stamps. Args: pid: project_id. eid: episode_id. enforce_shot_binding: ``False`` for the ``custom_prompt`` path. Catalog freshness is always enforced; binding-only stale is operator-overridable when a custom prompt is used (binding mismatch is acceptable because the operator authored a one-off prompt). Raises: StaleUpstreamError: master_plan ↔ consumer hash mismatch, missing consumer manifest under D6 mode, or corrupt manifest. Returns silently when D6 has not been applied (``master_plan`` cp absent or its ``bg_catalog_hash`` empty) — D5 fallback path. """ mp = _read_manifest(pid, eid, "background_master_plan") if mp is None: return # D6 미적용 episode (master_plan 자체 부재) mp_data = mp.get("data") or {} expected_catalog = mp_data.get("bg_catalog_hash") or "" expected_binding = mp_data.get("shot_binding_hash") or "" if not expected_catalog: return # legacy / pre-D6 master_plan — preflight 의미 없음 # First pass — catalog hash. Cache manifests for the binding pass to avoid # double I/O. consumer_manifests: Dict[str, Dict[str, Any]] = {} for step in _CONSUMERS: cm = _read_manifest(pid, eid, step) if cm is None: # master_plan 에 D6 hash 가 있는데 consumer 미실행 → STALE. raise StaleUpstreamError( upstream=step, expected_bg_catalog_hash=expected_catalog, observed_bg_catalog_hash="(manifest missing)", remediation=( f"consumer {step} has not run under D6 — " f"POST /steps/{step}?mode=resume " f"(or full ordered re-run — see docs/runbooks/d6-migration.md)" ), ) observed_catalog = (cm.get("data") or {}).get( "consumed_bg_catalog_hash", "" ) if observed_catalog != expected_catalog: raise StaleUpstreamError( upstream=step, expected_bg_catalog_hash=expected_catalog, observed_bg_catalog_hash=observed_catalog, remediation=( f"POST /steps/{step}?mode=force " f"(catalog drift — see docs/runbooks/d6-migration.md)" ), ) consumer_manifests[step] = cm # Second pass — binding hash (custom_prompt path skips this). if enforce_shot_binding and expected_binding: for step in _CONSUMERS: cm = consumer_manifests[step] # 첫 pass 에서 보장됨 observed_binding = (cm.get("data") or {}).get( "consumed_shot_binding_hash", "" ) if observed_binding != expected_binding: raise StaleUpstreamError( upstream=step, # catalog 는 이미 OK 로 검증됨 expected_bg_catalog_hash=expected_catalog, observed_bg_catalog_hash=expected_catalog, expected_shot_binding_hash=expected_binding, observed_shot_binding_hash=observed_binding, remediation=( f"POST /steps/{step}?mode=force " f"(shot grouping changed — " f"see docs/runbooks/d6-migration.md)" ), )