"""발송 장부 — **호출 경계**에서 시도를 적는다 (2026-09-20). ## 왜 따로 두는가 지금까지 지출은 「record 묶음이 움직였나」로 쟀다. 그것은 **기록 변경** 감지이지 **발송**의 증거가 아니다. 그리고 기존 로그를 사후에 세는 것도 근거가 못 된다: · `generate_image` 함수 입구 **1회 ≠ 실제 전송 1회** — 재시도 루프가 함수 **안**에 있다 · Router 에도 자체 `num_retries` 가 있다 · ★**「로그 행 없음 = 전송 0」이 아니다** — 기록 실패를 삼키는 자리가 있고, 재시도 `continue` 가 실패 로그보다 **앞**이다 그래서 **보내기 직전**에 적는다. ★**이번 범위는 메모리 장부 + 집계 보고까지다.** `attempt_id` 는 지금 메모리에만 있고 로그·Opik 으로 **나가지 않는다** — 「같은 attempt id 로 잇는다」는 **앞으로의 계약**이지 지금 된 것이 아니다. 잇게 되더라도 두 기록을 **합산하지 않는다**(같은 발송을 두 번 센다). ## 관측 단위를 **스스로 밝힌다** 경계마다 볼 수 있는 것이 다르다. 그래서 시도마다 `granularity` 를 적는다: request_attempt 원시 요청을 **보내려 한** 자리 — 재시도 루프 **안**. ★「물리 전송」이라고 부르지 않는다 — 이 지점 뒤의 DNS·연결 실패도 여기 세어진다(보냈는지 모른다). slot_entry 키 슬롯·Router 진입 — 그 **아래**에 자체 재시도가 있다 sdk_entry SDK 진입 — 내부 재시도는 안 보인다 ★어느 단위도 **「물리 전송 총수」가 아니다**. 그래서 집계는 단위별로 갈라서 돌려주고, 합산 키를 **만들지 않는다**. ## 이 장부는 **제동이 아니다** 보고값이다. 기존 예산·상한·래치는 그대로 둔다 — 아직 못 재는 경로를 0 으로 치고 제동을 푸는 것이 이 판의 병이었다. """ from __future__ import annotations import threading import uuid from dataclasses import dataclass, field from typing import Any, Dict, List, Optional #: 관측 단위 — 그 경계에서 **무엇을 볼 수 있는가**. GRAIN_RAW = "request_attempt" # 요청을 **보내려 한** 자리(재시도 루프 안) GRAIN_SLOT = "slot_entry" # 키 슬롯·Router 진입 (아래에 재시도 있음) GRAIN_SDK = "sdk_entry" # SDK 진입 (내부 재시도 안 보임) #: 종착 — 「결과 불명」을 **실패와 가른다**. 불명은 돈이 나갔을 수 있다. STATE_SENT = "sent" STATE_OK = "ok" STATE_FAILED = "failed" STATE_UNKNOWN = "unknown" @dataclass class SendAttempt: """발송 시도 **한 번**. ★`attempt_id` 는 지금 **메모리에만** 있다 — 접수 확인·결과를 잇는 것은 **앞으로의 감사 연결**이지 지금 된 것이 아니다. """ attempt_id: str #: "image" | "llm" — ★`llm` 은 **텍스트도 지나는 공통 경로**라 #: 「이미지 관찰 횟수」가 아니다. 좁혀 부르지 않는다. kind: str granularity: str source: str model: str = "" work: str = "" # 본체 tag · `tag::child` · `groupbg::…` visit: str = "" # 이번 샷 방문 신원 state: str = STATE_SENT detail: str = "" def ok(self) -> None: self.state = STATE_OK def failed(self, detail: str = "") -> None: self.state = STATE_FAILED self.detail = str(detail)[:400] def unknown(self, detail: str = "") -> None: """★**돈이 나갔는지 모른다** — 실패로 적으면 안 된다.""" self.state = STATE_UNKNOWN self.detail = str(detail)[:400] @dataclass class SendLedger: """한 걷기(또는 한 샷)의 발송 시도 모음.""" attempts: List[SendAttempt] = field(default_factory=list) _lock: Any = field(default_factory=threading.Lock, repr=False) def record( self, *, kind: str, granularity: str, source: str, model: str = "", work: str = "", visit: str = "", ) -> SendAttempt: att = SendAttempt( attempt_id=uuid.uuid4().hex, kind=kind, granularity=granularity, source=source, model=str(model or ""), work=str(work or ""), visit=str(visit or ""), ) with self._lock: self.attempts.append(att) return att def counts(self) -> Dict[str, Dict[str, int]]: """**단위별로 갈라** 센다 — 합산해서 「전송 총수」라고 하지 않는다.""" out: Dict[str, Dict[str, int]] = {} with self._lock: rows = list(self.attempts) for a in rows: key = f"{a.kind}/{a.granularity}" bucket = out.setdefault( key, {STATE_SENT: 0, STATE_OK: 0, STATE_FAILED: 0, STATE_UNKNOWN: 0}) bucket[STATE_SENT] += 1 if a.state != STATE_SENT: bucket[a.state] = bucket.get(a.state, 0) + 1 return out def summary(self) -> str: parts = [] for key in sorted(self.counts()): c = self.counts()[key] seg = f"{key} 발송 {c[STATE_SENT]}" if c[STATE_FAILED]: seg += f"·실패 {c[STATE_FAILED]}" if c[STATE_UNKNOWN]: seg += f"·**결과불명 {c[STATE_UNKNOWN]}**" parts.append(seg) return " / ".join(parts) # ── 발송 맥락 — **소유자가 정해서 설치한다** ─────────────────────── # # ★관측이 **스스로 캐게 하지 않는다** (2026-09-20 Codex). `record_send` # 가 Opik 로그·DB·문안을 뒤져 「이게 어느 샷이지」를 추측하면, 그 추측이 # 틀려도 아무도 모른다. 샷의 **소유자**(걷기 루프)가 한 번 정해 넘긴다. # ★모르면 **미상으로 남긴다** — trace 가 꺼져 방문 신원이 없으면 빈 # 문자열이다. 숫자나 `still_id` 로 **메우지 않는다**. @dataclass(frozen=True) class SendContext: """이 발송이 **무엇을 위한 것인가** — 소유자가 정한 불변 값. work: canonical 샷 tag · `tag::child` · `groupbg::…` 같은 **구조 키** visit: 방문 신원. 모르면 **빈 문자열**(미상). """ work: str = "" visit: str = "" _EMPTY_CTX = SendContext() def get_current_send_context() -> SendContext: return getattr(_local, "send_ctx", _EMPTY_CTX) def install_send_context(ctx: SendContext) -> SendContext: prev = get_current_send_context() _local.send_ctx = ctx return prev class send_context: # noqa: N801 """소유자가 맥락을 **씌운다**. 자식·공유는 `work` 만 덮었다 복원한다.""" def __init__(self, *, work: str = "", visit: Optional[str] = None) -> None: self._work = str(work or "") self._visit = visit self._prev: SendContext = _EMPTY_CTX def __enter__(self) -> SendContext: self._prev = get_current_send_context() # ★`visit` 을 안 주면 **바깥 것을 잇는다** — 자식·공유가 `work` 만 # 덮는 쓰임이다. 새로 캐지 않는다. visit = self._prev.visit if self._visit is None else str( self._visit or "") install_send_context(SendContext(work=self._work, visit=visit)) return get_current_send_context() def __exit__(self, exc_type, exc, tb) -> bool: install_send_context(self._prev) return False # ── 현재 장부 (스레드 지역 — 예산·정지와 같은 관례) ──────────────── _local = threading.local() def get_current_ledger() -> Optional[SendLedger]: return getattr(_local, "ledger", None) def install_ledger(ledger: Optional[SendLedger]) -> Optional[SendLedger]: """이 스레드의 장부를 갈아 끼우고 **옛 것을 돌려준다**(복원용).""" prev = get_current_ledger() _local.ledger = ledger return prev def bind_current_ledger(fn: Any) -> Any: """지금 스레드의 장부를 **다른 스레드에서도** 보이게 감싼다. ★롤 생성은 팬아웃이다 — 안 실으면 그 안의 발송이 통째로 안 잡힌다 (`bind_current_budget` 과 같은 이유, 같은 관례). """ captured = get_current_ledger() # ★**맥락도 같이** 나른다 — 장부만 실으면 팬아웃 안의 발송이 「어느 # 샷인지 모름」으로 남는다(예산·trace 와 같은 이유). captured_ctx = get_current_send_context() def _wrapped(*args: Any, **kwargs: Any) -> Any: prev = install_ledger(captured) prev_ctx = install_send_context(captured_ctx) try: return fn(*args, **kwargs) finally: install_send_context(prev_ctx) install_ledger(prev) return _wrapped class ledger_scope: # noqa: N801 """설치 → **성공·예외 어느 쪽이든 복원**, 그리고 중단까지의 수를 보고. ★2026-09-20 Codex BLOCK. 종전에는 정상 흐름에서만 복원·보고했다. 걷기 도중 상한 예외가 나가면 ①앞 샷들의 발송 보고가 **통째로 사라지고** ②끝난 걷기의 장부가 **스레드에 남아** 다음 작업이 거기 들어간다. ★보고 중 오류가 **원래 중단 예외를 덮지 않는다** — 보고는 본 작업의 실패를 가릴 자격이 없다. """ def __init__(self, ledger: "SendLedger", report: Any = None) -> None: self._ledger = ledger self._report = report self._prev: Optional[SendLedger] = None def __enter__(self) -> "SendLedger": self._prev = install_ledger(self._ledger) return self._ledger def __exit__(self, exc_type, exc, tb) -> bool: install_ledger(self._prev) if self._report is not None and self._ledger.attempts: try: self._report(self._ledger) except Exception: # noqa: BLE001 pass # ★보고 실패가 중단 예외를 덮지 않는다 return False # 예외는 그대로 올린다 def record_send( *, kind: str, granularity: str, source: str, model: str = "", work: str = "", visit: str = "", ) -> Optional[SendAttempt]: """**보내기 직전에** 한 번 적는다. 장부가 없으면 아무 일도 안 한다. ★반환값이 `None` 일 수 있다 — 호출부는 그대로 진행한다(관측이 본 작업을 막지 않는다). """ led = get_current_ledger() if led is None: return None # ★맥락은 **소유자가 설치한 것**을 받아 쓴다. 호출부가 명시로 준 값이 # 있으면 그것이 우선이고, 없으면 설치된 맥락이다 — 관측이 스스로 # 캐는 자리는 **없다**. ctx = get_current_send_context() return led.record(kind=kind, granularity=granularity, source=source, model=model, work=work or ctx.work, visit=visit or ctx.visit)