From b9cc7f0ae18b78b5f2e9aef8b0e8dafaa2058f48 Mon Sep 17 00:00:00 2001
From: Hyungi Ahn <hyungiahn@Hyungiui-MacBookPro.local>
Date: Fri, 24 Apr 2026 09:31:28 +0900
Subject: [PATCH] feat(policy): shadow Protocol + InMemoryShadowLogger
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

ShadowLogger (runtime_checkable Protocol) — PR-B 가 DBShadowLogger 구현
시 준수해야 할 인터페이스. record_would_route(*, doc_id, decision,
actual_model_used, prompt_version, policy_version, extra=None) → None.

InMemoryShadowLogger — 테스트 전용 in-memory 구현. records/count/clear
inspection helpers. Protocol 호환 (isinstance 통과).

PR-B 책임: app/services/policy_shadow_writer.py::DBShadowLogger(ShadowLogger)
구현 — analyze_events 에 INSERT. DB write 실패 시 WARN 로그만, 본 파이프라인
중단 금지 (shadow 기간 제품 영향 0).

plan: ~/.claude/plans/wise-gliding-hippo.md

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 app/policy/shadow.py | 90 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 90 insertions(+)
 create mode 100644 app/policy/shadow.py

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)