from typing import Any, Dict, List, Optional from fastapi import Request from fastapi.responses import JSONResponse _RESERVED_DETAIL_KEYS = ("code", "message") class AppError(Exception): """Application-level error mapped to JSON response by ``app_error_handler``. ``details`` (optional, kw-only) is merged into the ``error`` payload by the handler — used by subclasses (e.g. ``StaleUpstreamError``) that need to surface structured remediation metadata. Existing call sites that omit ``details`` are unaffected: the response shape stays ``{"error": {"code", "message"}}``. Reserved keys (``code`` / ``message``) cannot appear in ``details`` — fail-fast at construction time. The handler also enforces base-fields-win on merge as defense in depth (review iter1 IMPORTANT). """ def __init__( self, code: str, message: str, status_code: int = 400, *, details: Optional[Dict[str, Any]] = None, ): if details: invalid = [k for k in details if k in _RESERVED_DETAIL_KEYS] if invalid: raise ValueError( f"AppError details cannot use reserved keys " f"{_RESERVED_DETAIL_KEYS} — got {invalid!r}" ) self.code = code self.message = message self.status_code = status_code self.details = dict(details) if details else {} super().__init__(message) class StaleUpstreamError(AppError): """D6 §4.8 — preflight detection: upstream cp is stale. Raised by ``app.services.dispatcher_preflight.check_bg_catalog_freshness`` when ``background_master_plan`` and consumer manifests disagree on ``bg_catalog_hash`` / ``shot_binding_hash``, when a consumer manifest is missing under D6 mode, or when a manifest file is corrupt. HTTP 422. Direct attributes (``upstream`` / ``expected_*`` / ``observed_*`` / ``missing_in_render`` / ``remediation``) are also surfaced via ``self.details`` for the handler to serialize. """ def __init__( self, *, upstream: str, expected_bg_catalog_hash: str, observed_bg_catalog_hash: str, expected_shot_binding_hash: Optional[str] = None, observed_shot_binding_hash: Optional[str] = None, missing_in_render: Optional[List[str]] = None, remediation: str = "", message: str = "", ): self.upstream = upstream self.expected_bg_catalog_hash = expected_bg_catalog_hash self.observed_bg_catalog_hash = observed_bg_catalog_hash self.expected_shot_binding_hash = expected_shot_binding_hash self.observed_shot_binding_hash = observed_shot_binding_hash self.missing_in_render = list(missing_in_render or []) self.remediation = remediation details: Dict[str, Any] = { "upstream": upstream, "expected_bg_catalog_hash": expected_bg_catalog_hash, "observed_bg_catalog_hash": observed_bg_catalog_hash, "remediation": remediation, } if expected_shot_binding_hash is not None: details["expected_shot_binding_hash"] = expected_shot_binding_hash details["observed_shot_binding_hash"] = observed_shot_binding_hash or "" if self.missing_in_render: details["missing_in_render"] = list(self.missing_in_render) super().__init__( code="STALE_UPSTREAM", message=message or f"STALE_UPSTREAM upstream={upstream}", status_code=422, details=details, ) class VisibleStagingDriftError(AppError): """shot_director.visible_entity_ids ↔ shot_staging.camera_direction dual SOT drift 가 scene_detail 진입 직전에 감지되었음을 알리는 fail-fast. shot_staging 의 camera_direction NL 에 명시적 off-camera/off-screen/화면 밖 phrase 가 있는데 그 인접의 인물 이름이 director.visible 에 남아 있으면 drift. auto-fix 안 함 (consumer-side SOT 변조 위험) — 운영자가 shot_director force 재실행 또는 shot_staging force 재실행 으로 명시적 reconciliation 결정. HTTP 422. Direct attributes (``shot_label`` / ``visible`` / ``camera_direction`` / ``drift_entities`` / ``remediation``) 도 ``self.details`` 로 serialize. """ def __init__( self, *, shot_label: str, visible: List[str], camera_direction: str, drift_entities: Dict[str, str], remediation: str = "", ): self.shot_label = shot_label self.visible = list(visible) self.camera_direction = camera_direction self.drift_entities = dict(drift_entities) self.remediation = remediation or ( "shot_director force 재실행 (description-level gaze/offscreen 제외 " "재산출) 또는 shot_staging force 재실행 (camera_direction 재기획) " "후 scene_detail 재진입." ) details: Dict[str, Any] = { "shot_label": shot_label, "visible": list(visible), "camera_direction": camera_direction, "drift_entities": dict(drift_entities), "remediation": self.remediation, } drift_repr = ", ".join( f"{sid}({nm})" for sid, nm in sorted(drift_entities.items()) ) super().__init__( code="VISIBLE_STAGING_DRIFT", message=( f"{shot_label}: shot_director.visible 와 shot_staging." f"camera_direction 사이 dual SOT drift — staging 이 " f"{drift_repr} 를 명시적 off-camera 처리했으나 visible 에 남아 " f"있음. auto-fix 안 함." ), status_code=422, details=details, ) class ShotStagingOrientationError(AppError): """Patch C / Area A — content_surface / reflective_surface element 의 orientation 이 max_attempts (3회) 모두 빈 값으로 남음. HTTP 422. shot_staging step fail. 운영자가 analysis_dispatch mode=force 로 step 수동 재실행. raise 는 shot_staging.py 의 batch loop 안 `call_structured` try/ except Exception **밖**에서 실행되어야 broad except 에 swallow 되지 않는다 — umbrella spec Area A § 4.2 참조. """ def __init__( self, *, batch_num: int, total_batches: int, attempts: int, violations: List[Dict[str, Any]], ): self.batch_num = batch_num self.total_batches = total_batches self.attempts = attempts self.violations = list(violations) details: Dict[str, Any] = { "batch_num": batch_num, "total_batches": total_batches, "attempts": attempts, "violations": list(violations), } super().__init__( code="SHOT_STAGING_ORIENTATION_MISSING", message=( f"shot_staging batch {batch_num}/{total_batches}: " f"{len(self.violations)} element(s) with content_surface/" f"reflective_surface class missing orientation after " f"{attempts} attempts." ), status_code=422, details=details, ) async def app_error_handler(request: Request, exc: AppError) -> JSONResponse: """Serialize AppError to ``{"error": {...}}`` JSON. Base ``code`` / ``message`` always win over ``details`` to prevent a subclass from accidentally shadowing them via ``details={"code": ..., ...}`` (review iter1 IMPORTANT). ``StaleUpstreamError`` and other subclasses must surface remediation metadata under disjoint keys. """ base = {"code": exc.code, "message": exc.message} if exc.details: # details merge — base 가 우선. 같은 key 충돌 시 details 무시 (defense # in depth — AppError.__init__ 도 reserved key 거부). merged = {k: v for k, v in exc.details.items() if k not in _RESERVED_DETAIL_KEYS} merged.update(base) err: Dict[str, Any] = merged else: err = base return JSONResponse( status_code=exc.status_code, content={"error": err}, )