diff --git a/app/policy/__init__.py b/app/policy/__init__.py new file mode 100644 index 0000000..4ff5982 --- /dev/null +++ b/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 격리 검증). +""" diff --git a/app/policy/loader.py b/app/policy/loader.py new file mode 100644 index 0000000..60204d6 --- /dev/null +++ b/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() diff --git a/app/policy/schema.py b/app/policy/schema.py new file mode 100644 index 0000000..cec257e --- /dev/null +++ b/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