diff --git a/app/policy/shadow.py b/app/policy/shadow.py new file mode 100644 index 0000000..04638a2 --- /dev/null +++ b/app/policy/shadow.py @@ -0,0 +1,90 @@ +"""ShadowLogger — Protocol + in-memory implementation. + +Live 전환 전 1주 shadow 기간에 "만약 이 정책이면 어디로 라우팅했을지" 를 기록. +실제 DB writer (DBShadowLogger) 는 PR-B 의 책임. PR-A 는: + 1. Protocol 로 인터페이스 확정. + 2. InMemoryShadowLogger 로 테스트 가능한 fake 제공. + +PR-B 가 Protocol 시그니처를 변경하지 않는 것이 불변식. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Protocol, runtime_checkable + +from app.policy.routing import RoutingDecision + + +@dataclass(frozen=True) +class ShadowRecord: + """단일 shadow 이벤트 — InMemoryShadowLogger 가 dict 로 보관.""" + + doc_id: str + decision: RoutingDecision + actual_model_used: str + prompt_version: str + policy_version: str + recorded_at: datetime + extra: dict[str, Any] = field(default_factory=dict) + + +@runtime_checkable +class ShadowLogger(Protocol): + """PR-A 가 정의하는 shadow 기록 인터페이스. + + PR-B 가 DBShadowLogger(ShadowLogger) 로 구현할 때 이 시그니처를 그대로 준수. + """ + + async def record_would_route( + self, + *, + doc_id: str, + decision: RoutingDecision, + actual_model_used: str, + prompt_version: str, + policy_version: str, + extra: dict[str, Any] | None = None, + ) -> None: + ... + + +class InMemoryShadowLogger: + """테스트 전용 구현. PR-B 의 DBShadowLogger 와 시그니처 호환.""" + + def __init__(self) -> None: + self._records: list[ShadowRecord] = [] + + async def record_would_route( + self, + *, + doc_id: str, + decision: RoutingDecision, + actual_model_used: str, + prompt_version: str, + policy_version: str, + extra: dict[str, Any] | None = None, + ) -> None: + self._records.append( + ShadowRecord( + doc_id=doc_id, + decision=decision, + actual_model_used=actual_model_used, + prompt_version=prompt_version, + policy_version=policy_version, + recorded_at=datetime.now(timezone.utc), + extra=dict(extra or {}), + ) + ) + + # --- Inspection helpers (테스트 전용) ---------------------------------- + @property + def records(self) -> tuple[ShadowRecord, ...]: + return tuple(self._records) + + def clear(self) -> None: + self._records.clear() + + def count(self) -> int: + return len(self._records)