feat(policy): pydantic schema + yaml loader

app/policy/schema.py — DomainPolicy, SubjectDomain, FallbackDomain,
RiskFlag, ForbiddenRule, Escalation, Observability (pydantic v2, frozen).
suggested_ui_category 는 실측 doc_category enum (document|library|news|memo|
audio|video|law) 만 허용. synthesis_directive 500 chars 제한. cross-reference
validator — default_risk_flags 가 미정의 flag 참조 시 ValidationError.

app/policy/loader.py — load_policy(path) + functools.lru_cache.
env POLICY_PATH override, read_policy_bytes() helper (policy_version hash 용).

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

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

app/policy/__init__.py

@@ -0,0 +1,5 @@
+"""AI policy layer — pure-function judgment engine.
+
+Runtime 동작 변경 없음. 이 패키지를 app/workers 나 app/api 에서 import 하지 말 것
+(PR-A CI gate: import 격리 검증).
+"""

app/policy/loader.py

@@ -0,0 +1,52 @@
+"""domain_policy.yaml loader with lru_cache."""
+
+from __future__ import annotations
+
+import os
+from functools import lru_cache
+from pathlib import Path
+
+import yaml
+
+from app.policy.schema import DomainPolicy
+
+
+DEFAULT_POLICY_PATH = "domain_policy.yaml"
+POLICY_PATH_ENV = "POLICY_PATH"
+
+
+def _resolve_path(path: str | None) -> Path:
+    if path is not None:
+        return Path(path)
+    env_path = os.environ.get(POLICY_PATH_ENV)
+    if env_path:
+        return Path(env_path)
+    # repo root 기준 상대 경로 — working dir 에 따라 결정
+    return Path(DEFAULT_POLICY_PATH)
+
+
+@lru_cache(maxsize=8)
+def _load_cached(resolved: str) -> DomainPolicy:
+    text = Path(resolved).read_text(encoding="utf-8")
+    raw = yaml.safe_load(text)
+    return DomainPolicy.model_validate(raw)
+
+
+def load_policy(path: str | None = None) -> DomainPolicy:
+    """Load policy yaml and validate via pydantic.
+
+    Cache key = resolved absolute path (문자열). 테스트에서 다른 path 주면 별도 캐시.
+    """
+    resolved = str(_resolve_path(path).resolve())
+    return _load_cached(resolved)
+
+
+def clear_cache() -> None:
+    """테스트용 — 연속 호출 시 서로 다른 yaml 을 반영해야 할 때."""
+    _load_cached.cache_clear()
+
+
+def read_policy_bytes(path: str | None = None) -> bytes:
+    """policy_version hash 계산용 — yaml 원본 바이트."""
+    resolved = _resolve_path(path).resolve()
+    return resolved.read_bytes()

app/policy/schema.py

@@ -0,0 +1,133 @@
+"""Pydantic v2 models for domain_policy.yaml.
+
+Loader 가 yaml → DomainPolicy 로 파싱. Schema 위반 시 ValidationError → 배포 차단.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+
+# documents.category enum (migration 143 + 152)
+UICategory = Literal["document", "library", "news", "memo", "audio", "video", "law"]
+
+SelfDeclareSemantics = Literal["additive_trigger_only"]
+
+
+class SubjectDomain(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    description: str
+    suggested_ui_category: UICategory
+    high_impact: bool = False
+    default_risk_flags: tuple[str, ...] = ()
+    deep_summary_risk_flags: tuple[str, ...] = ()
+    keywords: tuple[str, ...] = ()
+    note: str | None = None
+
+
+class FallbackDomain(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    name: str
+    description: str
+    suggested_ui_category: UICategory
+    high_impact: bool = False
+    default_risk_flags: tuple[str, ...] = ()
+    requires_human_review: bool = True
+
+
+class RiskFlag(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    description: str
+    requires_26b: bool
+    synthesis_directive: str | None = None
+    output_mask_required: bool = False
+
+    @field_validator("synthesis_directive")
+    @classmethod
+    def _directive_length(cls, v: str | None) -> str | None:
+        if v is not None and len(v) > 500:
+            raise ValueError("synthesis_directive must be <= 500 chars")
+        return v
+
+
+class ForbiddenRule(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    id: str
+    description: str
+    applies_when_subject_in: tuple[str, ...]
+    detection_patterns: tuple[str, ...] = ()
+
+
+class Escalation(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    confidence_threshold: float = Field(ge=0.0, le=1.0)
+    context_char_cap_4b: int = Field(gt=0)
+    context_char_cap_26b: int = Field(gt=0)
+    escalate_on_multi_doc_count: int = Field(ge=1)
+
+
+class HealthRange(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    min: float | None = None
+    max: float | None = None
+
+
+class Observability(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    required_event_fields: tuple[str, ...]
+    health_ranges: dict[str, HealthRange]
+
+
+class DomainPolicy(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    version: int
+    last_updated: str
+    scope: tuple[str, ...]
+    self_declare_semantics: SelfDeclareSemantics
+
+    subject_domains: dict[str, SubjectDomain]
+    fallback_domain: FallbackDomain
+    risk_flags: dict[str, RiskFlag]
+    forbidden_for_4b: tuple[ForbiddenRule, ...]
+    escalation: Escalation
+    observability: Observability
+
+    @model_validator(mode="after")
+    def _cross_reference_check(self) -> "DomainPolicy":
+        """Cross-field validation — yaml 내부 일관성."""
+        known_flags = set(self.risk_flags.keys())
+
+        # 1. 모든 subject_domain.default_risk_flags 가 risk_flags 에 정의돼 있어야 함
+        for name, dom in self.subject_domains.items():
+            for flag in (*dom.default_risk_flags, *dom.deep_summary_risk_flags):
+                if flag not in known_flags:
+                    raise ValueError(
+                        f"subject_domain '{name}' references unknown risk_flag '{flag}'"
+                    )
+
+        for flag in self.fallback_domain.default_risk_flags:
+            if flag not in known_flags:
+                raise ValueError(
+                    f"fallback_domain references unknown risk_flag '{flag}'"
+                )
+
+        # 2. forbidden_for_4b.applies_when_subject_in 의 도메인이 subject_domains 에 있어야 함
+        known_domains = set(self.subject_domains.keys())
+        for rule in self.forbidden_for_4b:
+            for dom_name in rule.applies_when_subject_in:
+                if dom_name not in known_domains:
+                    raise ValueError(
+                        f"forbidden rule '{rule.id}' references unknown subject_domain '{dom_name}'"
+                    )
+
+        return self