"""Reve 2.1 (fal.ai queue) 이미지 편집 클라이언트 (2026-08-25). 최종 i2i 변환 provider 교체 — `GeminiImageClient` 를 상속해 컨텍스트·로깅· 캡처 기계만 그대로 쓰고 **운반층은 독립 구현**한다. `generate_image` 슬롯이 `GrokImageClient` 와 같은 모양이라 호출부는 client 만 갈아끼우면 된다. ★왜 Grok 운반 코드를 재사용하지 않는가 — 근거가 **돈**이다. `GrokImageClient` 는 재시도 루프 안에서 **매 HTTP 시도 직전에** 예산을 예약한다(`grok_image_client.py:186`). 동기 왕복이라 그것이 맞다. 그런데 fal 은 **queue** 다 — submit 1회가 유료 작업 1건이고 poll 은 새 호출이 아니다. Grok 의 「timeout 이면 새 요청」을 그대로 옮기면 fal 이 이미 접수·과금한 작업을 로컬 timeout 뒤 **다시 submit** 한다. $0.25 짜리를. 계약 (fal 공식 문서, 2026-08-25 확인): - submit POST https://queue.fal.run/{model} → request_id·status_url·response_url - status GET .../requests/{id}/status → IN_QUEUE | IN_PROGRESS | COMPLETED - result GET .../requests/{id} → images[].url - 오류 모델 오류 = `{"detail":[{"type": ...}]}` · 요청 오류 = `{"error_type": ...}` 422 는 전부 재시도 금지. 재시도 가부 판단은 `X-Fal-Needs-Retry` 우선. - `sync_mode` 는 **쓰지 않는다** — 결과가 data URI 로 오는 대신 request history 에 안 남아 크래시 뒤 `request_id` 로 다시 가져오기할 수 없다. ★request_id 를 받은 뒤로는 **새 submit 이 없다.** timeout 은 「poll 중단」일 뿐 새 지출 근거가 아니다. submit 응답 자체가 유실돼 접수 여부가 모호하면 자동 재제출하지 않고 실패로 끝낸다(fail-closed) — 다음 걷기가 재시도한다. ★`on_submit` 콜백은 request_id 를 **호출측이 durable 하게 남기라고** 준다. 파일과 기록 사이엔 원자성이 없으므로, 접수된 작업의 신원을 먼저 남겨야 크래시 뒤 그 작업을 다시 가져오기할 수 있다. """ import base64 import json import logging import time import urllib.error import urllib.request from typing import Any, Callable, Dict, List, Optional, Tuple from app.core.config import settings as _settings from app.core.image_call_budget import ( ImageCallBudgetExceeded, reserve_current_call, ) from app.modules.llm.gemini_image_client import GeminiImageClient from app.modules.llm.image_format import ensure_png_bytes from app.modules.llm.image_send_state import ( NEVER_SENT_ERRORS, ImageSubmissionUnknown, classify_send_failure, ) from app.modules.llm.llm_logger import log_llm_call from app.services.image_capture.sink import capture_generated_image logger = logging.getLogger(__name__) FAL_QUEUE_BASE = "https://queue.fal.run" # ── 오류 유형 = fal 이 주는 **정확한 이름** ───────────────────────────── # ★문자열 부분 일치로 판단하지 않는다. 일부 endpoint 가 아직 이 구조로 # 이행 중이라 type 이 없을 수 있는데, 그때는 「알 수 없는 4xx terminal」로 # 기록하고 **검열로 추정하지 않는다**(추정하면 포기 셈이 오염된다). FAL_POLICY_TYPE = "content_policy_violation" FAL_NO_MEDIA_TYPE = "no_media_generated" # 인프라 실패 — 다시 보내면 될 수도 있다(요청 단위 오류). FAL_RETRYABLE_TYPES = frozenset({ "request_timeout", "startup_timeout", "runner_scheduling_failure", "runner_connection_timeout", "runner_disconnected", "runner_connection_refused", "runner_connection_error", "runner_incomplete_response", "runner_server_error", }) # 모델 오류 중 재시도 가부가 갈리는 것 — 헤더가 말해 줄 때만 다시 보낸다. FAL_VARIABLE_TYPES = frozenset({ "internal_server_error", "generation_timeout", "downstream_service_error", "downstream_service_unavailable", }) # queue 상태값 _STATUS_TERMINAL = "COMPLETED" # poll 간격(초) — 접수 직후는 촘촘히, 길어지면 성기게. _POLL_FIRST = 2.0 _POLL_MAX = 8.0 # 변환 시도 횟수 — 첫 시도 + 재시도 2회 (2026-08-26 사용자 지시). _CINE_MAX_ATTEMPTS = 3 _CINE_RETRY_WAIT = 15.0 # 다시 하는 실패 유형 = **fal 인프라**뿐이다. 검열·결과 없음·검증 실패는 # 여기 없다 — 같은 입력에 같은 답이 오므로 돈만 쓴다. # # ★여기서 「다시 한다」는 **이미 받은 접수 번호로 결과를 다시 조회한다**는 # 뜻이다. 접수 번호를 아직 못 받았는데 5xx 가 왔으면 요청이 서버에 닿은 # 것이므로 요금이 나갔을 수 있다 — 그때는 다시 보내지 않고 멈춘다 # (`ReveSubmissionUnknown`). 돈 쪽으로 닫아 둔 것이 의도다. _CINE_RETRY_TYPES = frozenset(FAL_RETRYABLE_TYPES | FAL_VARIABLE_TYPES) class ReveSubmissionUnknown(ImageSubmissionUnknown): """접수됐는지 **모르는** 채로 끝났다 — 다시 보내면 요금이 두 번 나간다. ★2026-08-26: 공용 `ImageSubmissionUnknown` 의 **하위**가 됐다. 그 전에는 둘이 남남이라, 소비자가 클래스 **이름**으로 판정하던 자리 (`cine_transform`)가 Grok·Gemini 가 던지는 공용 예외를 못 알아봤다 — grok 이 기본 변환 제공자이므로 **기본 경로에서 재요청 차단이 열려 있었다.** 지금은 `isinstance` 하나로 둘 다 잡힌다. ## 왜 따로 두나 접수 번호(`request_id`)가 없으면 종전에는 「접수 자체가 안 됐다」로 읽고 다시 보냈다. 그런데 번호가 없는 경우가 둘이다: - **연결이 아예 안 됐다** (이름 못 찾음·연결 거부) → 요금 안 나갔다 - **보냈는데 답이 오다가 끊겼다** (연결 끊김·읽기 시간 초과) → **fal 이 받아서 요금이 나갔을 수 있다** 아래 경우에 다시 보내면 같은 이미지에 $0.25 가 두 번 나가고, 조용해서 로그에도 안 보인다. 그래서 **모르면 멈춘다.** 자동으로 다시 보내는 것은 위의 「연결이 아예 안 됐다」뿐이다. ★멈춘 샷은 기록에 `submission_unknown` 으로 남는다. 다음 걷기가 자동으로 다시 보내지 않는다 — 사람이 fal 대시보드에서 접수 여부를 보고 정한다. ★`__init__` 을 따로 두지 않는다 — 부모와 시그니처가 같다. 자식이 `self.cause` 를 먼저 넣고 `super().__init__(message)` 만 부르면 **부모가 기본값 `""` 로 그것을 덮어쓴다**(2026-08-26 실측: 상속으로 바꾼 직후 `cause` 가 전부 빈 문자열이 됐다). """ # ★이 판정은 2026-08-26 에 **공용 자리로 올라갔다** — Grok·Gemini 도 같은 # 계약을 써야 하기 때문이다(감사 0-A). 여기 이름은 기존 호출부·테스트를 # 위해 남긴 별칭이고, 계약 본문은 `image_send_state` 가 소유한다. _NEVER_SENT: tuple = NEVER_SENT_ERRORS classify_submit_failure = classify_send_failure class ReveTerminalError(RuntimeError): """다시 보내면 안 되는 실패 — 검열·검증·결과 없음. ★`content_policy_violation` 은 메시지에 그 이름을 그대로 담는다 — 공용 판별(`image_moderation.MODERATION_MARKERS`)이 `content_policy` 를 표식으로 쓰므로 상위 포기 셈이 이것을 검열로 읽는다. 반대로 `no_media_generated` 에는 그 표식이 없어야 한다: **검열과 다른 실패**이고 섞으면 포기 기준이 엉뚱한 곳에서 발화한다. """ def __init__(self, message: str, *, error_type: str = "", http_status: int = 0, request_id: str = "") -> None: self.error_type = error_type self.http_status = int(http_status) self.request_id = request_id super().__init__(message) def _first_detail_type(payload: Any) -> str: """모델 오류 본문에서 **첫 detail 의 type** — 없으면 빈 문자열.""" if not isinstance(payload, dict): return "" detail = payload.get("detail") if isinstance(detail, list): for item in detail: if isinstance(item, dict) and item.get("type"): return str(item["type"]) return "" # 요청 단위 오류는 평평한 모양 — `{"detail": "...", "error_type": "..."}` if payload.get("error_type"): return str(payload["error_type"]) return "" def _detail_message(payload: Any) -> str: if not isinstance(payload, dict): return "" detail = payload.get("detail") if isinstance(detail, str): return detail if isinstance(detail, list): msgs = [str(d.get("msg") or "") for d in detail if isinstance(d, dict)] return " · ".join(m for m in msgs if m) return "" class ReveImageClient(GeminiImageClient): # ★★능력 계약을 **덮어쓴다** — `reve/2.1/edit` 의 입력이 `image_url` # **단수**라 참조는 **정확히 1장**이다(아래 `generate_image` 가 그렇게 # 못박고 있고, 여러 장이 오면 조용히 안 버리고 크게 실패시킨다). def reference_input_capability(self): base = super().reference_input_capability() base.update({"supports_labeled_refs": True, "min_images": 1, "max_images": 1}) return base """fal.ai queue 경유 Reve 2.1 클라이언트 — cine gen_fn 슬롯 호환. 상속 재사용: set_context/_ctx/_trace_step/_log_ctx/_opik_meta. 오버라이드: 키 소스(FAL_KEY)와 generate_image 운반층. """ def __init__( self, api_key: str = None, model: str = None, on_submit: Optional[Callable[[Dict[str, Any]], None]] = None, ) -> None: super().__init__( api_key=api_key, model=model or _settings.reve_image_model, ) # 접수된 작업의 신원을 호출측이 durable 하게 남기는 통로. self._on_submit = on_submit # 마지막으로 접수된 작업 — 예외 갈래에서도 읽을 수 있게 남긴다. self.last_request_id: str = "" # fal 이 접수 응답으로 알려 주는 조회 주소. self.last_status_url: str = "" self.last_response_url: str = "" def set_submit_hook( self, fn: Optional[Callable[[Dict[str, Any]], None]], ) -> "ReveImageClient": """접수 기록 통로를 나중에 끼운다 — 기록을 쥔 쪽은 호출부다. client 는 만들어지는 자리(스텝 초입)와 쓰이는 자리(샷 루프)가 달라, `records` 를 쥐고 있는 샷 루프가 여기에 콜백을 건다. """ self._on_submit = fn return self def _get_api_key(self) -> str: if self._fixed_api_key: return self._fixed_api_key key = (getattr(_settings, "fal_key", "") or "").strip() if not key: raise RuntimeError("FAL_KEY 가 설정에 없다") return key # ── HTTP 원시층 ─────────────────────────────────────────────────── def _headers(self) -> Dict[str, str]: return { "Authorization": f"Key {self._get_api_key()}", "Content-Type": "application/json", # 플랫폼의 숨은 재시도·동등 endpoint 대체를 끈다 — 한 submit 이 # 한 지출이어야 provider 지문과 실제 실행이 일치한다. "x-fal-no-retry": "1", "x-app-fal-disable-fallback": "1", } def _request(self, url: str, *, method: str = "GET", body: Optional[Dict[str, Any]] = None, timeout: int = 120) -> Tuple[int, Any, Dict[str, str]]: data = (json.dumps(body).encode("utf-8") if body is not None else None) req = urllib.request.Request( url, data=data, headers=self._headers(), method=method) with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8", errors="replace") headers = {k.lower(): v for k, v in resp.headers.items()} try: return resp.status, json.loads(raw) if raw else {}, headers except json.JSONDecodeError: return resp.status, {"_raw": raw}, headers def _raise_http(self, exc: urllib.error.HTTPError, *, request_id: str = "") -> None: """제공자 오류를 **유형 그대로** 분류해 올린다. 재시도 가부는 `X-Fal-Needs-Retry` 헤더가 우선이고, 헤더가 없으면 유형표를 본다. 유형도 없으면 「알 수 없는 4xx terminal」이다 — 검열로 추정하지 않는다. """ raw = exc.read().decode("utf-8", errors="replace") headers = {k.lower(): v for k, v in (exc.headers or {}).items()} try: payload = json.loads(raw) if raw else {} except json.JSONDecodeError: payload = {"_raw": raw[:500]} etype = (_first_detail_type(payload) or headers.get("x-fal-error-type", "")) msg = _detail_message(payload) or raw[:300] needs_retry = (headers.get("x-fal-needs-retry", "") or "").lower() label = f"reve {exc.code}" if etype: label += f" {etype}" text = f"{label}: {msg}" if exc.code == 405 and not msg.strip(): # ★405 는 「메서드가 틀렸다」로 읽히지만 fal 대기열에서는 사실상 # **그런 접수 경로가 없다**는 뜻이다. 몸통이나 인증을 뒤지게 # 만드는 코드라, 무엇을 봤는지 여기서 그대로 적어 준다. # # ★**무엇을 고치라고 권하지 않는다** (2026-08-26 두 번째 교훈). # 전에 이 자리는 "fal 의 endpoint id 는 'fal-ai/…' 로 시작한다" # 고 단정했다. 그 안내를 따라 멀쩡한 `reve/2.1/edit` 를 바꿨다가 # 주행을 잃었다 — fal 이 직접 호스팅하는 모델만 그 접두를 쓰고 # 제휴 모델은 제공자 이름으로 시작한다(`config.reve_image_model` # 주석). 그 모델 id 로 실제 변환이 성공한 기록이 있다. # 진단은 **본 것만** 남기고 판단은 읽는 사람에게 넘긴다. _seen = {k: v for k, v in headers.items() if k.startswith("x-fal-")} text += (f"(빈 응답) — 접수 경로가 없다는 뜻일 수 있다. " f"보낸 곳: {FAL_QUEUE_BASE}/{self._model} · " f"status={exc.code} · 응답 헤더={_seen or '없음'} · " f"응답 본문={raw[:200]!r}") if needs_retry == "true" or ( not needs_retry and etype in (FAL_RETRYABLE_TYPES | FAL_VARIABLE_TYPES)): # 다시 해도 되는 실패 — terminal 로 표시하지 않는다. # # ★「다시 해도 된다」가 **어느 단계냐에 따라 다르다** (2026-08-26 # Codex 2차 재리뷰). 여기서 올리는 RuntimeError 는 두 자리에서 # 잡힌다. # # ⓐ **접수 전**(request_id 를 아직 못 받았다): `generate_image` # 의 일반 except 가 접수 여부를 모른다고 보고 `멈춘다`. 서버가 # 5xx 를 돌려줬다는 것은 요청이 닿았다는 뜻이므로, 요금이 # 나갔을 수 있다 — 다시 보내지 않는다. # ⓑ **접수 후**(request_id 를 이미 받았다): 같은 번호로 결과만 # 다시 조회한다. 새 요청이 아니라 요금이 안 나간다. # # 즉 이 유형 분류는 **조회 단계의 재시도**를 열어 주는 것이지, # 접수를 다시 하라는 뜻이 아니다. 돈 쪽으로 닫아 둔 것이 의도다. raise RuntimeError(text) raise ReveTerminalError( text, error_type=etype, http_status=exc.code, request_id=request_id) # ── queue 왕복 ──────────────────────────────────────────────────── def _submit(self, body: Dict[str, Any]) -> Dict[str, Any]: """작업 접수 — **예산은 여기서 딱 한 번** 예약한다. ★몸통은 **평평하게** 보낸다 — fal 공식 예제의 `arguments={...}` 가 그대로 몸통이 된다. 2026-08-26 실측으로 확인했다(평평한 몸통 → 200 IN_QUEUE). 종전에 `{"input": {...}}` 로 감싼 것은 405 의 원인을 잘못 짚어 넣은 것이었고, 감싸도 접수는 200 이라 증상이 안 바뀌어 오진이 오래 갔다. 감싸면 작업자가 인자를 못 읽는다. """ url = f"{FAL_QUEUE_BASE}/{self._model}" reserve_current_call(source="reve_image_client.submit") try: _status, payload, _h = self._request( url, method="POST", body=body, timeout=120) except urllib.error.HTTPError as exc: self._raise_http(exc) raise # pragma: no cover — _raise_http 가 반드시 올린다 rid = str((payload or {}).get("request_id") or "") if not rid: # ★접수 여부가 모호한 경우 **중에서도 가장 위험한 쪽**이다 — # 요청이 서버에 닿아 답까지 왔는데(200) 번호만 없다. 「연결이 안 # 됐다」보다 오히려 요금이 나갔을 가능성이 높다. # # 2026-08-26 Codex 재리뷰 BLOCK-2: 종전에는 `ReveTerminalError` # 라 이 호출 안에서만 멈추고 **다음 걷기가 같은 지문으로 다시 # 보냈다.** `cine_transform` 은 `ReveSubmissionUnknown` 일 때만 # `submission_unknown` 표식을 남기기 때문이다. 유형이 달라 # 표식이 안 붙었고, 그래서 같은 그림에 요금이 두 번 나갔다. raise ReveSubmissionUnknown( "reve submit 응답에 request_id 가 없다 — 서버가 답은 했는데 " "번호를 안 줬다. 접수됐을 수 있어 다시 보내지 않는다", cause="submit_ack_missing") self.last_request_id = rid # ★결과 조회 주소는 **fal 이 알려 준 것**을 쓴다. 직접 조립하면 틀린다 — # 접수 주소에는 동작 이름이 붙지만(`reve/2.1/edit`) 조회 주소에는 # 안 붙는다(`reve/2.1/requests/{id}`). 그것을 모르고 # `{model}/{id}` 로 만들어 **405** 를 받았고, 405 가 「메서드가 # 틀렸다」는 코드라 원인을 몸통·인증에서 찾느라 두 판을 날렸다. self.last_status_url = str((payload or {}).get("status_url") or "") self.last_response_url = str( (payload or {}).get("response_url") or "") if self._on_submit: try: # ★`status_url` 도 함께 넘긴다 (2026-08-26 Codex 재리뷰). # 프로세스가 죽었다 살아나면 인스턴스에 남은 주소가 없어 # 조회 주소를 직접 조립해야 하는데, 그 조립을 틀려 **405** 를 # 받고 두 판을 날린 적이 있다. fal 이 준 주소를 둘 다 남겨 # 두면 되찾을 때 조립에 기대지 않는다. self._on_submit({ "provider": "reve", "endpoint": self._model, "request_id": rid, "status_url": str((payload or {}).get("status_url") or ""), "response_url": str((payload or {}).get("response_url") or ""), }) except Exception: # noqa: BLE001 # ★기록 실패를 **삼키면 안 된다** (2026-08-26 Codex PR#4 # BLOCK-4). 여기까지 왔다는 것은 fal 이 받아서 **요금이 이미 # 나갔다**는 뜻이다. 그런데 번호를 디스크에 못 남긴 채 # 74~95초 폴링을 계속하다 프로세스가 죽으면, 다음 걷기는 # 그 번호를 모르므로 **같은 그림에 요금이 또 나간다.** # # 올려서 폴링 전에 멈춘다. 새로 보내지는 않는다 — # `self.last_request_id` 가 이미 세팅돼 있어 재시도 갈래가 # `_finish` 만 부른다. 호출부의 `submitted_info` 도 저장 # 실패 **전에** 갱신되므로, 마무리가 그 번호로 기록을 다시 # 남길 수 있다. logger.error( "reve: 접수 기록 콜백이 실패했다 — 요금이 나간 작업의 " "번호를 못 남겼다. 폴링도 재시도도 하지 않고 올린다 " "(request_id=%s)", rid, exc_info=True) # ★재시도 대상이 **아니다.** 재시도하면 다시 폴링에 들어가 # 기록 없는 74~95초 창이 그대로 열린다. 그 사이 프로세스가 # 죽으면 번호가 디스크에 없어 다음 걷기가 새로 보낸다. # # 여기서 멈추면 번호는 인스턴스와 호출부의 `submitted_info` # 에 남으므로, 마무리가 그것으로 pending 기록을 다시 시도한다. # 그 기록이 되면 다음 걷기가 **무료로** 결과를 가져온다. raise ReveTerminalError( f"reve 접수 기록 실패 — 번호는 {rid} 다. 이 번호로 결과를 " f"가져오면 추가 요금이 없다", error_type="submit_record_failed", request_id=rid) from None return payload or {} def _queue_base(self) -> str: """결과 조회 주소의 앞부분 — 접수 주소에서 **동작 이름을 뗀다**. 접수는 `reve/2.1/edit` 이지만 조회는 `reve/2.1/requests/{id}` 다. fal 이 접수 응답으로 주소를 알려 주므로 평소에는 그것을 쓰고, 이 함수는 **request_id 만 남은 상황**(크래시 뒤 그것으로만 결과를 가져올 때)의 대비책이다. """ parts = [p for p in self._model.split("/") if p] stem = "/".join(parts[:-1]) if len(parts) > 1 else self._model return f"{FAL_QUEUE_BASE}/{stem}/requests" # ★fal 이 준 주소는 **그 요청의 것일 때만** 쓴다. # # 종전에는 넘어온 `request_id` 를 보지도 않고 인스턴스에 남은 주소를 # 먼저 썼다. 이 클라이언트는 샷 루프 바깥에서 한 번 만들어져 공유되므로, # 앞 샷의 주소로 **다음 샷의 결과를 조회**하게 된다 — 앞 샷 그림을 자기 # 결과로 가져오던 그 결함과 같은 뿌리다(2026-08-26). # # 신원이 안 맞으면 주소를 직접 조립한다. 조립 규칙은 아래 `_queue_base` # 에 있고, 405 를 겪고 고친 그 규칙이다. def _url_for(self, request_id: str, saved: str, suffix: str = "") -> str: if saved and request_id and request_id == self.last_request_id: return saved return f"{self._queue_base()}/{request_id}{suffix}" def _status_url(self, request_id: str) -> str: return self._url_for(request_id, self.last_status_url, "/status") def _response_url(self, request_id: str) -> str: return self._url_for(request_id, self.last_response_url) def _poll(self, request_id: str, *, deadline: float) -> None: """COMPLETED 까지 기다린다 — **poll 은 새 지출이 아니다.** ★주소는 fal 이 접수 응답으로 준 `status_url` 을 그대로 쓴다. 직접 조립했다가 **405** 를 받았다(2026-08-26): 접수 주소에는 동작 이름이 붙지만 조회 주소에는 안 붙는데, `{model}/{id}` 로 만들어 `reve/2.1/edit/{id}` 를 요청했다. 이 파일 머리말에는 올바른 계약 (`.../requests/{id}/status`)이 처음부터 적혀 있었다 — 문서와 구현이 갈렸고, 405 가 「메서드가 틀렸다」는 코드라 원인을 몸통·인증에서 찾느라 두 판을 날렸다. """ url = self._status_url(request_id) wait = _POLL_FIRST while True: if time.monotonic() >= deadline: raise TimeoutError( f"reve 작업 대기 마감 초과 — request_id={request_id} " "(접수된 작업이므로 다시 보내지 않는다)") try: _s, payload, _h = self._request(url, timeout=60) except urllib.error.HTTPError as exc: self._raise_http(exc, request_id=request_id) raise # pragma: no cover status = str((payload or {}).get("status") or "") if status == _STATUS_TERMINAL: err = (payload or {}).get("error") if err: etype = str((payload or {}).get("error_type") or "") raise ReveTerminalError( f"reve 작업 실패 {etype}: {err}", error_type=etype, request_id=request_id) return time.sleep(wait) wait = min(_POLL_MAX, wait * 1.5) def _result(self, request_id: str) -> Dict[str, Any]: url = self._response_url(request_id) try: _s, payload, _h = self._request(url, timeout=120) except urllib.error.HTTPError as exc: self._raise_http(exc, request_id=request_id) raise # pragma: no cover return payload or {} # ── 공개 슬롯 ───────────────────────────────────────────────────── def generate_image( self, prompt: str, reference_images: Optional[List[bytes]] = None, aspect_ratio: str = "16:9", labeled_references: Optional[List[Tuple[str, bytes]]] = None, ) -> Tuple[bytes, int]: """reve 2.1 편집 — (PNG bytes, ms) 반환. ★참조는 **한 장**이다. `reve/2.1/edit` 의 입력 칸은 `image_url` 단수라 여러 장을 받을 자리가 없다 — 여러 장이 오면 조립층이 잘못된 provider 를 고른 것이므로 **조용히 버리지 않고** 크게 실패시킨다(잘려 나간 참조는 그림에서만 드러난다). aspect_ratio 는 보내지 않는다 — 편집은 소스 비율을 잇는 것이 맞고 (`auto` 가 기본), 비율을 지정하면 원본 구도가 잘린다. """ refs: List[Tuple[str, bytes]] = [] if labeled_references: refs = [(lbl or "", b) for lbl, b in labeled_references] elif reference_images: refs = [("", b) for b in reference_images] if len(refs) != 1: raise ValueError( f"reve/2.1/edit 는 참조 1장만 받는다 — {len(refs)}장이 왔다") ref_label, ref_bytes = refs[0] ref_image_ids = [ref_label] if ref_label else [] body = { "prompt": prompt, "image_url": ("data:image/png;base64," + base64.b64encode(ref_bytes).decode("ascii")), "num_images": 1, "output_format": "png", } start_time = time.monotonic() deadline = start_time + _settings.llm_timeout_image_gen last_exc: Optional[Exception] = None # ★★★ 새 호출은 **접수 전** 상태로 시작한다. # # `last_request_id` 는 아래 재시도 루프가 「이미 샀으니 결과만 다시 # 가져온다」를 판단하는 값인데, **인스턴스 칸**이다. 그런데 이 # 클라이언트는 샷 루프 **바깥에서 한 번** 만들어져 모든 샷이 공유한다 # (`still_recipe_service.py` 의 `cine_client`). 여기서 안 비우면 # 앞 샷이 남긴 신원을 다음 샷이 물어 **앞 샷의 그림을 자기 결과로 # 가져온다.** # # 2026-08-26 실측으로 실제 발생: S2sh1 이 성공한 뒤 S3sh2 가 3.7초 # 만에 「완료」했고(다른 샷은 74~95초), 산출 바이트가 S2sh1 것과 # 해시까지 같았다. 검사가 「장소·인물·사물이 다르다」로 잡아서 # 드러났다 — 검사가 없었으면 S3sh2 가 남의 그림으로 확정됐다. # # ★**세 칸을 다 비운다.** 처음엔 `last_request_id` 하나만 비웠는데, # 조회 주소 두 칸이 그대로 남아 `_status_url`/`_response_url` 이 # 앞 샷 주소를 계속 썼다. 신원은 세 칸이 함께 이룬다. self.last_request_id = "" self.last_status_url = "" self.last_response_url = "" # ★재시도 (2026-08-26 사용자 지시 "재시도 붙여 꼭 2번 정도 더") — # fal 쪽 일시 장애로 한 샷이 변환 없이 지나가던 것을 막는다. # 실측: 02:13 에 `504 downstream_service_unavailable` 로 한 샷이 # 원본 그대로 남았다. 우리 코드 문제가 아니라 fal 하류가 잠시 # 응답을 못 한 것이다. # # ★다시 **보내는** 것과 다시 **가져오는** 것을 가른다: # request_id 있다 → 이미 요금이 나갔다 → **같은 신원으로 결과만 # 다시 가져온다** (추가 요금 없음) # request_id 없다 → 아래 판단으로 한 번 더 가른다 # # ★★번호가 없는 경우가 **둘**이다 (2026-08-26 Codex 지적): # 연결이 아예 안 됐다 → 요금 안 나갔다 → 다시 보낸다 # 답이 오다가 끊겼다 → **요금이 나갔을 수 있다** → 멈춘다 # 종전에는 둘을 안 가르고 전부 다시 보냈다. 아래 경우에 다시 보내면 # 같은 이미지에 요금이 두 번 나가고, 조용해서 로그에도 안 보인다. for attempt in range(1, _CINE_MAX_ATTEMPTS + 1): try: if self.last_request_id: # 이미 접수된 작업 — 예산을 다시 예약하지 않는다. return self._finish( self.last_request_id, prompt=prompt, ref_image_ids=ref_image_ids, start_time=start_time, deadline=deadline) submitted = self._submit(body) return self._finish( str(submitted.get("request_id") or ""), prompt=prompt, ref_image_ids=ref_image_ids, start_time=start_time, deadline=deadline) except ImageCallBudgetExceeded: # 예산 상한은 재시도 대상이 아니다 — 더 쓰지 말라는 뜻이다. raise except ReveSubmissionUnknown: # 접수 여부를 모른다 — 아래 `except Exception` 이 한 번 더 # 감싸면 이유가 「ReveSubmissionUnknown」으로 뭉개진다. # 그대로 올려 원래 사유를 살린다. raise except ReveTerminalError as exc: # 검열·결과 없음·422 는 다시 보내도 같은 답이 온다. # ★fal **인프라** 유형(5xx 계열)만 다시 시도한다. if exc.error_type not in _CINE_RETRY_TYPES: raise last_exc = exc except TimeoutError: # 마감 초과 — 작업은 살아 있을 수 있다. 다시 보내지 않고 # 그대로 올린다(같은 것에 요금이 두 번 나가지 않게). raise except Exception as exc: # noqa: BLE001 — 그 밖의 일시 실패 # ★번호를 못 받은 채 실패했다면 **접수됐는지 모른다.** # 연결이 아예 안 된 경우만 다시 보내고, 나머지는 멈춘다. if not self.last_request_id: 갈래 = classify_submit_failure(exc) if 갈래 != "never_sent": self._fail_log( prompt, ref_image_ids, f"접수 여부 불명 — 다시 보내지 않는다 " f"({type(exc).__name__}: {exc})"[:500], start_time) raise ReveSubmissionUnknown( f"reve 접수 여부를 모른다 — 요금이 나갔을 수 " f"있어 다시 보내지 않는다 " f"({type(exc).__name__}: {exc})"[:300], cause=type(exc).__name__, ) from exc last_exc = exc if attempt >= _CINE_MAX_ATTEMPTS: break logger.warning( "reve 변환 %d/%d 실패 — %s초 뒤 다시 시도한다 " "(request_id=%s · %s)", attempt, _CINE_MAX_ATTEMPTS, _CINE_RETRY_WAIT, self.last_request_id or "(접수 전)", f"{type(last_exc).__name__}: {last_exc}"[:200]) time.sleep(_CINE_RETRY_WAIT) self._fail_log( prompt, ref_image_ids, f"{_CINE_MAX_ATTEMPTS}회 시도 모두 실패 — " f"{type(last_exc).__name__}: {last_exc}"[:500], start_time) raise last_exc if last_exc else RuntimeError("reve 변환 실패") def fetch_submitted( self, request_id: str, *, prompt: str = "", labeled_references: Optional[List[Tuple[str, bytes]]] = None, status_url: str = "", response_url: str = "", ) -> Tuple[bytes, int]: """이미 접수된 작업의 결과를 **다시 요청하지 않고** 가져온다. 크래시나 응답 유실로 결말을 못 본 작업을 다시 가져오는 자리다. 예산을 예약하지 않는다 — 지출은 submit 때 이미 일어났고, 여기서 또 세면 한 작업이 두 번 계산된다. ★`status_url`·`response_url` 은 접수 때 fal 이 준 주소다. 프로세스가 죽었다 살아나면 인스턴스에 아무것도 안 남아 조회 주소를 직접 조립해야 하는데, 그 조립을 틀려 **405** 를 받은 적이 있다. 기록에 남겨 둔 주소를 여기로 넘기면 조립에 기대지 않는다. """ if not request_id: raise ValueError("가져올 request_id 가 없다") if status_url or response_url: # `_url_for` 가 「그 요청의 주소일 때만」 쓰도록 신원을 맞춘다. self.last_request_id = request_id self.last_status_url = status_url self.last_response_url = response_url start_time = time.monotonic() ref_image_ids = [lbl for lbl, _ in (labeled_references or []) if lbl] return self._finish( request_id, prompt=prompt, ref_image_ids=ref_image_ids, start_time=start_time, deadline=start_time + _settings.llm_timeout_image_gen) # ── 결말 수습 ───────────────────────────────────────────────────── def _fail_log(self, prompt: str, ref_image_ids: List[str], err_text: str, start_time: float) -> None: from app.modules.llm.image_tracer import get_image_tracer elapsed = int((time.monotonic() - start_time) * 1000) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=err_text, duration_ms=elapsed, reference_image_ids=ref_image_ids, **self._log_ctx, ) get_image_tracer().log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=err_text, duration_ms=elapsed, extra_metadata=self._opik_meta, ) def _finish( self, request_id: str, *, prompt: str, ref_image_ids: List[str], start_time: float, deadline: float, ) -> Tuple[bytes, int]: """접수된 작업의 결말을 받아 PNG 로 — **여기서는 돈을 쓰지 않는다.**""" from app.modules.llm.image_tracer import get_image_tracer try: self._poll(request_id, deadline=deadline) payload = self._result(request_id) except Exception as exc: # noqa: BLE001 self._fail_log(prompt, ref_image_ids, f"{type(exc).__name__}: {exc}"[:500], start_time) raise url = "" for im in ((payload or {}).get("images") or []): if isinstance(im, dict) and im.get("url"): url = str(im["url"]) break if not url: err = f"reve 응답에 그림이 없다 (request_id={request_id})" self._fail_log(prompt, ref_image_ids, err, start_time) raise ReveTerminalError( err, error_type=FAL_NO_MEDIA_TYPE, request_id=request_id) try: with urllib.request.urlopen(url, timeout=120) as dl: raw_png = dl.read() except Exception as exc: # noqa: BLE001 # 그림은 이미 샀는데 받아오지 못했다 — 다시 사지 않도록 # request_id 를 오류에 실어 남긴다(그 신원으로 다시 가져온다). err = (f"reve 산출 내려받기 실패 (request_id={request_id}): " f"{type(exc).__name__}: {exc}") self._fail_log(prompt, ref_image_ids, err[:500], start_time) raise RuntimeError(err) from exc elapsed_ms = int((time.monotonic() - start_time) * 1000) call_id = log_llm_call( model_name=self._model, user_prompt=prompt, output_text="[image generated]", status="success", duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) get_image_tracer().log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], output_image_id=self._ctx.get("image_id"), duration_ms=elapsed_ms, params={"backend": "reve", "request_id": request_id}, extra_metadata=self._opik_meta, ) img_bytes = ensure_png_bytes( raw_png, context=f"{self._model}/" f"{self._ctx.get('operation_type') or '?'}", ) _capture_meta = { k: self._ctx.get(k) for k in ( "project_id", "episode_id", "operation_type", "step", "scene_index", "shot_index", "still_id", "entity_id", "image_id", ) if self._ctx.get(k) is not None } if ref_image_ids: _capture_meta["reference_labels"] = ref_image_ids _capture_meta["provider_request_id"] = request_id capture_generated_image( img_bytes, role=self._ctx.get("operation_type") or "reve_image", prompt=prompt, generation_call_id=call_id, pipeline_metadata=_capture_meta, ) return img_bytes, elapsed_ms