Hyungi Ahn
d23ea48223
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>
134 lines
4.1 KiB
Python
134 lines
4.1 KiB
Python
"""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
|