"""Grok Imagine (xAI, OpenRouter 경유) 이미지 생성 클라이언트 (2026-08-13). 최종 스틸 생성 엔진 교체(#103, 사용자 확정 "grok 2 로 가자") 전용 — GeminiImageClient 를 상속해 컨텍스트·로깅·캡처 기계를 그대로 쓰고 운반층만 OpenRouter chat/completions(modalities=["image"]) 로 바꾼다. 운영 실측 계약 (artifact/20260813_MAI이미지25_실측/calls.json): - modalities 는 ["image"] 단독 — ["image","text"] 는 404. - 참조는 콘텐츠 parts 로 라벨 텍스트+이미지 쌍 다중 첨부 (5장 실측 통과). - 텍스트 총량(프롬프트+라벨) 8,000바이트 상한 — 초과는 xAI 400. 잘라 보내는 일은 없다: 여기서는 사전 검증으로 크게 실패시키고, 조립층(still_recipe v17 컴팩트 지도 절)이 예산을 맞춘다. - 응답 이미지: choices[0].message.images[].image_url.url (data URL). """ import base64 import concurrent.futures as _futures import json import logging import time import urllib.error import urllib.request from typing import List, Optional, Tuple from app.core.config import settings as _settings from app.core.image_call_budget import ( ImageCallBudgetExceeded, reserve_current_call, ) from app.core.send_ledger import GRAIN_RAW as _GRAIN_RAW from app.core.send_ledger import record_send as _record_send from app.modules.llm.gemini_image_client import GeminiImageClient from app.modules.llm.image_format import ensure_png_bytes from app.modules.llm.image_moderation import is_moderation_text from app.modules.llm.image_send_state import ( ImageSubmissionUnknown, classify_http_status, 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__) OPENROUTER_CHAT_URL = "https://openrouter.ai/api/v1/chat/completions" # xAI 이미지 계열 텍스트 상한 실측 400 문언: "maximum allowed length of # 8000". 라벨 텍스트도 합산된다. # # 2026-08-18: 목표와 거절 상한을 나눈다(Codex BLOCK 수용). 하나로 쓰면 # 여유분 100B 안에 든 롤(실측 7,938B 1건)이 **보내 보지도 않고** 죽는데, # 모델이 실제로 거절하는 값은 8,000B 이고 거절은 400 이라 돈이 안 든다. # 즉 여유분을 "덜어내기 목표"로만 쓰고, 더 못 덜어도 진짜 상한 안이면 # 보낸다 — 못 보내는 것보다 낫고 잃는 것은 없다. GROK_TEXT_BYTE_LIMIT = 7900 # 덜어내기 목표(여유분 포함) GROK_TEXT_BYTE_HARD_LIMIT = 8000 # 모델이 선언한 실제 상한 — 넘으면 400 # ★**이 상수는 더 이상 재시도 판정에 쓰이지 않는다** (2026-08-26 감사 0-A). # 판정은 `image_send_state.classify_http_status(code, via="openrouter")` 가 # 한다 — 429 만 재시도하고 5xx 는 「상류가 이미 작업을 시작했을 수 있다」로 # 멈춘다. 여기 남겨 둔 것은 **그 시절 계약을 잠그는 시험**이 이 이름을 # 참조하기 때문이고, 새 코드에서 쓰면 안 된다. # # 옛 사연(2026-08-19): xAI 는 OpenRouter 를 거치며 상태 200 에 오류를 본문 # 으로 담아 보낼 때가 있는데 그 갈래에 재시도가 없어 S88sh2 가 죽었고 # 걷기 전체가 두 시간 다시 돌았다. 지금은 갈래가 하나로 합쳐졌다. GROK_RETRYABLE_BODY_CODES = frozenset({429, 500, 502, 503, 504}) def _body_error_code(err: object) -> int | None: """본문 오류의 상태 코드 — 정수로 실려 올 때만 읽는다.""" if not isinstance(err, dict): return None code = err.get("code") if isinstance(code, bool) or not isinstance(code, int): return None return code class GrokPromptOverBudget(RuntimeError): """텍스트 총량이 xAI 상한을 넘는다 — 조립층에서 고칠 문제라 크게 실패.""" class GrokImageClient(GeminiImageClient): """OpenRouter 경유 xAI Grok Imagine 클라이언트 — nb2 gen_fn 슬롯 호환. 상속 재사용: set_context/_ctx/_trace_step/_log_ctx/_opik_meta. 오버라이드: 키 소스(OPENROUTER_API_KEY)와 generate_image 운반층. """ def __init__(self, api_key: str = None, model: str = None) -> None: super().__init__(api_key=api_key, model=model or _settings.grok_image_model) def _get_api_key(self) -> str: if self._fixed_api_key: return self._fixed_api_key key = (getattr(_settings, "openrouter_api_key", "") or "").strip() if not key: raise RuntimeError("OPENROUTER_API_KEY 가 설정에 없다") return key def generate_image( self, prompt: str, reference_images: Optional[List[bytes]] = None, aspect_ratio: Optional[str] = None, labeled_references: Optional[List[Tuple[str, bytes]]] = None, ) -> Tuple[bytes, int]: """OpenRouter chat/completions 이미지 생성 — (PNG bytes, ms) 반환. ★aspect_ratio 는 **호출자가 줄 때만** `image_config` 로 싣는다 (2026-09-19). 종전 문구는 「API 파라미터가 없어 전송하지 않는다 — 참조 비율을 상속한다」였는데 둘 다 틀렸다: · OpenRouter 는 이 모델에 `aspect_ratio` 를 받는다 (`/api/v1/images/models` 의 supported_parameters). · 참조가 여러 장이면 **어느 참조를 따를지 모른다** — 실측 S49sh14: 가로 배경(1536×864) 한 장 + 세로 인물 참조(896×1200) 세 장을 받고 세로 880×1184 로 나왔다(nb2 거절 → 교차 백엔드 갈래). 안 주면 종전 그대로(안 보냄) — 바탕 지도의 비율을 따라가야 하는 도면 지도(`marker_map_engine`)는 비율을 넘기지 않는다. """ content: list = [{"type": "text", "text": prompt}] ref_image_ids = [] text_bytes = len(prompt.encode("utf-8")) if labeled_references: for i, (label, img_bytes) in enumerate(labeled_references, 1): lbl = label or f"Reference image {i}:" content.append({"type": "text", "text": lbl}) content.append({ "type": "image_url", "image_url": { "url": "data:image/png;base64," + base64.b64encode(img_bytes).decode("ascii"), }, }) text_bytes += len(lbl.encode("utf-8")) if label: ref_image_ids.append(label) elif reference_images: for img_bytes in reference_images: content.append({ "type": "image_url", "image_url": { "url": "data:image/png;base64," + base64.b64encode(img_bytes).decode("ascii"), }, }) shed: List[str] = [] if text_bytes > GROK_TEXT_BYTE_LIMIT: # 컴팩트 스템을 다 적용해도 긴 롤은 상한을 넘는다(직전 판 실측 # 378롤 중 61롤). 그대로 두면 롤이 죽고 롤이 죽으면 샷이 죽으므로 # **일반 규칙 절만** 정해진 순서로 덜어 맞춘다. 샷 고유 재료는 # 목록에 없다. 조립 권위가 무엇을 덜지 정하므로 헬퍼는 # still_recipe 에 있고 여기서는 상한을 아는 자리에서 부른다. from app.modules.pipeline.still_recipe import fit_grok_prompt label_bytes = text_bytes - len(prompt.encode("utf-8")) fitted, shed = fit_grok_prompt( prompt, label_bytes, GROK_TEXT_BYTE_LIMIT) if shed: before_bytes = text_bytes prompt = fitted content[0] = {"type": "text", "text": prompt} text_bytes = len(prompt.encode("utf-8")) + label_bytes # 덜어낸 내역의 durable 기록은 이 줄이 아니라 **실제로 보낸 # 프롬프트**다 — 아래 log_llm_call·Opik·capture 가 재대입된 # prompt 를 그대로 남긴다(records.json 은 조립 명목판). # 이 줄은 운전 중 눈으로 보기 위한 것이다. logger.warning( "grok 상한 대응 — 일반 규칙 절 %d개 덜어냄 %s " "(%dB → %dB, 목표 %dB, 모델 상한 %dB)", len(shed), shed, before_bytes, text_bytes, GROK_TEXT_BYTE_LIMIT, GROK_TEXT_BYTE_HARD_LIMIT, ) if text_bytes > GROK_TEXT_BYTE_HARD_LIMIT: raise GrokPromptOverBudget( f"grok 텍스트 총량 {text_bytes}B > " f"{GROK_TEXT_BYTE_HARD_LIMIT}B(모델 상한) — 컴팩트 조립" "(guidance v17)·일반 규칙 절 덜어내기로도 안 맞는다" ) body = { "model": self._model, "messages": [{"role": "user", "content": content}], "modalities": ["image"], } if aspect_ratio: body["image_config"] = {"aspect_ratio": aspect_ratio} from app.modules.llm.image_tracer import get_image_tracer _tracer = get_image_tracer() req_bytes = json.dumps(body).encode("utf-8") headers = { "Authorization": f"Bearer {self._get_api_key()}", "Content-Type": "application/json", } start_time = time.monotonic() last_error: Exception | None = None _send = None # 이번 방문의 마지막 발송 표 timeout = _settings.llm_timeout_image_gen max_retries = _settings.llm_max_retries for attempt in range(1, max_retries + 2): try: reserve_current_call(source="grok_image_client.urlopen") # ★**요청을 보내려는 자리**다 — 재시도 루프 **안**이라 429 # 로 다시 보내면 여기도 다시 찍힌다. 함수 입구에서 세면 # 「보낸 횟수」가 아니라 「부른 횟수」가 된다(2026-09-20). # ★「물리 전송」이라고 부르지 않는다 — 이 뒤의 DNS·연결 # 실패도 여기 세어진다(보냈는지 모른다). _send = _record_send( kind="image", granularity=_GRAIN_RAW, source="grok_image_client.urlopen", model=self._model, # ★`work` 를 여기서 **안 준다** — `still_id` 는 # canonical tag 가 아니라 본체/자식/공유가 안 갈린다. # 소유자가 설치한 맥락이 채운다(Codex (δ)). ) req = urllib.request.Request( OPENROUTER_CHAT_URL, data=req_bytes, headers=headers, method="POST", ) # ★총 마감시간 (2026-08-14 카나리아 실측): urllib 의 # timeout 은 **개별 소켓 연산** 한도라, 서버가 응답을 # 찔끔찔끔 흘리며 연결을 붙잡으면 read 가 영원히 안 # 끝난다 — S3sh4 cine 호출이 ESTABLISHED 소켓 하나로 # 54분+ 매달려 llm_call_log 에 아무 기록도 없이 걷기 # 전체를 세웠다(lsof 로 확정). 왕복 전체를 timeout+60s # 벽 안에 가둔다 — 초과는 여느 네트워크 오류처럼 # 재시도 사다리를 탄다. def _roundtrip(): with urllib.request.urlopen( req, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) # ★shutdown(wait=False): 벽을 넘긴 좀비 스레드를 기다리면 # 행이 그대로 되돌아온다 — 스레드는 버리고 호출자만 푼다 # (소켓은 프로세스 수명/서버 종단에서 정리된다). _ex = _futures.ThreadPoolExecutor(max_workers=1) _fut = _ex.submit(_roundtrip) try: payload = _fut.result(timeout=timeout + 60) finally: _ex.shutdown(wait=False, cancel_futures=True) # ★HTTP 200 인데 본문에 오류가 담겨 오는 갈래를 여기서 # 가른다(2026-08-19). 루프 밖에서 다루면 이미 break 한 # 뒤라 다시 보낼 방법이 없다 — 그것이 S88sh2 를 죽인 # 모양이다. 검열은 다시 보내지 않는다(요금만 나간다). _body_err = ( payload.get("error") if isinstance(payload, dict) else None) if _body_err is not None: _text = json.dumps(_body_err, ensure_ascii=False) _code = _body_error_code(_body_err) # ★본문에 담겨 온 오류도 **같은 제공자 정책**을 쓴다 # (2026-08-26 Codex BLOCK-2). 종전에는 여기가 # {429,500,502,503,504} 를 통째로 다시 보내, 바로 아래 # HTTPError 갈래의 `via="openrouter"` 정책을 우회했다 — # 500·502·503·504 는 이 판이 다시 보내지 않기로 한 바로 # 그 경우다. # # ★**코드를 못 읽으면 「모른다」다** (2026-08-26 Codex # 재리뷰). 상태 200 을 받았다는 것은 요청이 OpenRouter # 까지 **닿았다**는 뜻이고, 본문에 오류가 있는데 코드가 # 없거나 예상 밖 형식(문자열 `"502"` 등)이면 상류가 # 처리했는지·요금이 나갔는지 알 수 없다. terminal 로 # 두면 `possible_charge` 가 사라져 다음 방문이 오독한다. # 검열 문구는 아래 `is_moderation_text` 가 걸러 terminal # 로 남는다 — 그쪽은 무슨 일이 있었는지 아는 경우다. _bv = (classify_http_status(_code, via="openrouter") if _code is not None else "submission_unknown") if (_bv == "retryable" and not is_moderation_text(_text) and attempt <= max_retries): last_error = RuntimeError( f"Grok image API error: {_text}") logger.warning( "grok 본문 오류 %s — 다시 보낸다 (%d/%d)", _code, attempt, max_retries) time.sleep(2 * attempt) continue if _bv == "submission_unknown" and not is_moderation_text( _text): # 상류가 이미 작업을 시작한 뒤 실패했을 수 있다 — # 멈추고 요금 표시를 남긴다. _shown = _code if _code is not None else "코드없음" # ★**결과 불명은 실패와 다르다** — 돈이 나갔을 수 있다. if _send is not None: _send.unknown('submission_unknown') raise self._raise_submission_unknown( cause=f"body_{_shown}", detail=(f"Grok image API 본문 오류 {_shown} — 상류가 " f"이미 작업을 시작했을 수 있다: {_text[:200]}"), attempt_no=attempt, req_bytes=req_bytes, prompt=prompt, ref_image_ids=ref_image_ids, start_time=start_time, tracer=_tracer, ) break except _futures.TimeoutError as exc: # ★**다시 보내지 않는다** (2026-08-26 감사 0-A). # # 종전에는 여기서 재시도 사다리를 탔다. 그런데 이 시점에 # worker 는 이름 찾기·연결·TLS·업로드·읽기 중 **어디에 있는지 # 알 수 없고**(`Future.result` 는 그것을 말해 주지 않는다), # 도는 스레드는 취소되지도 않는다. body 가 이미 나갔다면 # xAI 는 그림을 만들고 있고, 다시 보내면 같은 이미지에 요금이 # 두 번 나간다. 그러고도 기록에는 마지막 한 건만 남아 첫 # 시도의 비용과 결과 신원을 잃는다. # # `reve_image_client` 가 먼저 세운 계약과 같다 — 모르면 멈춘다. # ★**결과 불명은 실패와 다르다** — 돈이 나갔을 수 있다. if _send is not None: _send.unknown('submission_unknown') raise self._raise_submission_unknown( cause="wall_timeout", detail=(f"grok 왕복 총 마감시간 초과({timeout + 60}s) — " "서버가 연결을 붙잡은 채 응답을 안 끝냄"), attempt_no=attempt, req_bytes=req_bytes, prompt=prompt, ref_image_ids=ref_image_ids, start_time=start_time, tracer=_tracer, ) from exc except ImageCallBudgetExceeded: raise except ImageSubmissionUnknown: # ★이미 「보냈는지 모른다」로 판정해 기록까지 남긴 예외다. # 아래 `except Exception` 이 이것을 다시 잡으면 **기록이 두 번** # 남고 메시지가 겹겹이 감싸진다(2026-08-26 실측). raise except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError( f"Grok image API error {exc.code}: {error_text}") # ★다시 보내도 되는 상태만 다시 보낸다 (2026-08-26 감사 0-A). # 종전에는 {429, 500, 502, 503, 504} 를 한 묶음으로 재전송했다. # 500·502·504 는 상류가 이미 요청을 받아 **작업을 시작한 뒤** # 실패했을 수 있어, 그때 다시 보내면 요금이 두 번 나간다. # 429·503 은 「받지 않았다」가 표준 의미라 그대로 재시도한다. _verdict = classify_http_status(exc.code, via="openrouter") if _verdict == "retryable" and attempt <= max_retries: time.sleep(2 * attempt) continue if _verdict == "submission_unknown": # ★상태를 **모르는** 응답 (2026-08-26 Codex BLOCK-1) — # 일반 오류로 남기면 possible_charge 가 사라진다. # ★**결과 불명은 실패와 다르다** — 돈이 나갔을 수 있다. if _send is not None: _send.unknown('submission_unknown') raise self._raise_submission_unknown( cause=f"http_{exc.code}", detail=(f"Grok image API {exc.code} — 상류가 이미 " f"작업을 시작했을 수 있다: {error_text[:200]}"), attempt_no=attempt, req_bytes=req_bytes, prompt=prompt, ref_image_ids=ref_image_ids, start_time=start_time, tracer=_tracer, ) from exc elapsed_ms = int((time.monotonic() - start_time) * 1000) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=str(last_error), duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=str(last_error), duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) raise last_error except Exception as exc: # noqa: BLE001 — 네트워크·타임아웃류 last_error = exc # ★**닿지 못한 것이 확정된 경우만** 다시 보낸다 # (2026-08-26 감사 0-A). 읽기 시간 초과·연결 끊김은 # body 가 이미 나간 뒤일 수 있어 재전송하면 요금이 두 번 # 나간다. 판정은 `image_send_state` 공용 계약. if classify_send_failure(exc) != "never_sent": # ★원래 예외 **타입**을 사유에 남긴다 (자체 리뷰 MEDIUM-5). # 이 자리는 정체를 모르는 예외를 전부 「보냈는지 모른다」로 # 본다(fail-closed). 그러면 `JSONDecodeError` 처럼 응답을 # 받은 것이 분명한 실패도, 평범한 프로그래밍 오류도 같은 # 칸에 담긴다. 열거로 가르지 않는 대신 **무엇이었는지를 # 기록에 남겨** 나중에 갈라 볼 수 있게 한다. # ★**결과 불명은 실패와 다르다** — 돈이 나갔을 수 있다. if _send is not None: _send.unknown('submission_unknown') raise self._raise_submission_unknown( cause=f"network_error_after_send:{type(exc).__name__}", detail=f"Grok image API 응답을 못 받았다: {exc}", attempt_no=attempt, req_bytes=req_bytes, prompt=prompt, ref_image_ids=ref_image_ids, start_time=start_time, tracer=_tracer, ) from exc if attempt <= max_retries: time.sleep(2 * attempt) continue elapsed_ms = int((time.monotonic() - start_time) * 1000) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=str(exc), duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=str(exc), duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) raise else: # pragma: no cover — break 없이 끝나는 일 없음 raise last_error or RuntimeError("unreachable") elapsed_ms = int((time.monotonic() - start_time) * 1000) if isinstance(payload, dict) and payload.get("error"): err = json.dumps(payload["error"], ensure_ascii=False) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=err, duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=err, duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) if is_moderation_text(err): raise RuntimeError(f"moderation blocked: {err}") raise RuntimeError(f"Grok image API error: {err}") choices = (payload or {}).get("choices") or [] msg = (choices[0].get("message") or {}) if choices else {} for im in (msg.get("images") or []): url = ((im.get("image_url") or {}).get("url") if isinstance(im, dict) else "") or "" if not url.startswith("data:"): continue b64 = url.split(",", 1)[1] 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, ) _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": "grok2"}, extra_metadata=self._opik_meta, ) # ★이 방문의 **마지막 발송**이 성공으로 끝났다. 앞선 재시도는 # `sent` 로 남아 「보낸 횟수」에 그대로 센다 — 그것이 이 장부의 # 뜻이다(성공 횟수가 아니다). if _send is not None: _send.ok() # grok 은 jpg 로 준다 — 저장 규약(.png)과 lineage 정합을 위해 # nb2 경로와 같은 정규화를 거친다 (2026-08-06 교훈 동형). img_bytes = ensure_png_bytes( base64.b64decode(b64), 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_generated_image( img_bytes, role=self._ctx.get("operation_type") or "grok_image", prompt=prompt, generation_call_id=call_id, pipeline_metadata=_capture_meta, ) return img_bytes, elapsed_ms # 이미지 없는 200 — 텍스트/빈 응답은 실패로 기록하고 전파 err = f"no image in response (finish={choices[0].get('finish_reason') if choices else None})" log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=err, duration_ms=elapsed_ms, reference_image_ids=ref_image_ids, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_image_ids or [], status="error", error=err, duration_ms=elapsed_ms, extra_metadata=self._opik_meta, ) raise RuntimeError(f"Grok image API: {err}")