vignette/apps/api/app/contracts/continuous_improvement.py
Yun Chan 16e791e044 G0~G8 성과·동맹 측정 OS 작업 일괄 고정
8월 7일까지 워킹트리에만 남아 있던 미커밋 작업을 커밋한다. 여러 사본
폴더(worktree·clone)에 흩어져 있던 중간 스냅샷을 정리하기 전에 원본을
git 이력으로 고정하는 것이 목적이다.

- contracts/routes/services: measurement, outcome_trajectory, rupture_repair,
  deliberate_practice, calibration_transfer, supervision_research,
  multimodal_alliance, continuous_improvement 계열 신규 모듈과 테스트
- infra/db/init: 07~16 마이그레이션(측정 기반~calibration transfer 실행)
- apps/web: 세션 리뷰 카드·관리 화면·E2E 스펙 추가
- docs/ops: G0~G8 라이브 통합·배포·롤백 증거 문서와 evidence JSON/PNG
- scripts: smoke·ledger·릴리스 에이전트·NAS 프리뷰 운영 스크립트

engine.public 로그 .bak과 apps/web/test-results 산출물은 커밋에서 제외했다.
2026-08-08 01:30:53 +09:00

255 lines
9.5 KiB
Python

"""G8 자율 콘텐츠 생성·적대 검토·승격·운영 환류 계약."""
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
ReviewDimension = Literal[
"safety",
"identity",
"answer_leakage",
"cultural_bias",
"difficulty",
"pii",
"grounding",
]
FindingSeverity = Literal["blocker", "high", "moderate", "low"]
FindingState = Literal["open", "resolved", "accepted_risk"]
ModelChangeDecision = Literal["promote", "rollback", "quarantine"]
class ContentSourceArtifact(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
source_id: str = Field(pattern=r"^oas-g8-source-[a-z0-9-]+$")
version: str = Field(min_length=1, max_length=80)
content_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
provenance_uri: str = Field(pattern=r"^(repo|db|audit)://[a-zA-Z0-9_./:-]+$")
usage_status: Literal["approved", "restricted", "rejected"]
citation_label: str = Field(min_length=1, max_length=300)
class GeneratedContentDraft(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
draft_id: str = Field(pattern=r"^oas-g8-draft-[a-z0-9-]+$")
content_kind: Literal["case", "rupture", "practice", "benchmark"]
source_refs: tuple[str, ...] = Field(min_length=1)
generation_model: str = Field(min_length=1, max_length=180)
prompt_version: str = Field(min_length=1, max_length=80)
prompt_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
payload_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
synthetic_identity_id: str = Field(pattern=r"^synthetic-identity-[a-z0-9-]+$")
difficulty_level: int = Field(ge=1, le=5)
hidden_answer_fingerprint: str = Field(pattern=r"^[a-f0-9]{64}$")
visible_answer_overlap_tokens: int = Field(ge=0)
pii_findings: int = Field(ge=0)
unsupported_clinical_claims: int = Field(ge=0)
class RedTeamFinding(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
finding_id: str = Field(pattern=r"^oas-g8-finding-[a-z0-9-]+$")
dimension: ReviewDimension
severity: FindingSeverity
state: FindingState
evidence_ref: str = Field(min_length=1, max_length=220)
remediation_ref: str | None = Field(default=None, max_length=220)
@model_validator(mode="after")
def require_resolution_evidence(self) -> "RedTeamFinding":
if self.state == "resolved" and not self.remediation_ref:
raise ValueError("resolved red-team finding requires remediation evidence")
if self.state == "accepted_risk" and self.severity in {"blocker", "high"}:
raise ValueError("blocker/high finding cannot be accepted as residual risk")
return self
class IndependentRedTeamReview(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
review_id: str = Field(pattern=r"^oas-g8-review-[a-z0-9-]+$")
draft_id: str
reviewer_agent_id: str = Field(min_length=1, max_length=180)
dimensions: tuple[ReviewDimension, ...] = Field(min_length=3)
findings: tuple[RedTeamFinding, ...]
reviewed_payload_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
@model_validator(mode="after")
def require_unique_coverage(self) -> "IndependentRedTeamReview":
if len(set(self.dimensions)) != len(self.dimensions):
raise ValueError("red-team review dimensions must be unique")
if any(item.dimension not in self.dimensions for item in self.findings):
raise ValueError("red-team finding must belong to a reviewed dimension")
return self
class ContentBenchmarkQualification(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
benchmark_id: str = Field(pattern=r"^oas-g8-benchmark-[a-z0-9-]+$")
draft_id: str
variant_count: int = Field(ge=3)
variant_pass_rate: float = Field(ge=0.0, le=1.0)
answer_leakage_count: int = Field(ge=0)
pii_finding_count: int = Field(ge=0)
unsupported_claim_count: int = Field(ge=0)
safety_failure_count: int = Field(ge=0)
reward_hacking_count: int = Field(ge=0)
evidence_refs: tuple[str, ...] = Field(min_length=1)
@property
def qualified(self) -> bool:
return (
self.variant_pass_rate >= 0.85
and self.answer_leakage_count == 0
and self.pii_finding_count == 0
and self.unsupported_claim_count == 0
and self.safety_failure_count == 0
and self.reward_hacking_count == 0
)
class ApprovedCatalogEntry(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
catalog_entry_id: str = Field(pattern=r"^oas-g8-catalog-[a-z0-9-]+$")
draft_id: str
payload_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
source_refs: tuple[str, ...] = Field(min_length=1)
review_ids: tuple[str, ...] = Field(min_length=2)
benchmark_id: str
status: Literal["approved"] = "approved"
clinical_claim_allowed: Literal[False] = False
class ModelCalibrationSnapshot(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
snapshot_id: str = Field(pattern=r"^oas-g8-model-snapshot-[a-z0-9-]+$")
model: str = Field(min_length=1, max_length=180)
prompt_version: str = Field(min_length=1, max_length=80)
benchmark_version: str = Field(min_length=1, max_length=80)
task_accuracy: float = Field(ge=0.0, le=1.0)
critical_miss_count: int = Field(ge=0)
leakage_count: int = Field(ge=0)
pii_count: int = Field(ge=0)
calibration_error: float = Field(ge=0.0, le=1.0)
subgroup_max_gap: float = Field(ge=0.0, le=1.0)
class ModelChangeGateResult(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
baseline_snapshot_id: str
candidate_snapshot_id: str
decision: ModelChangeDecision
reasons: tuple[str, ...] = Field(min_length=1)
rollback_target_snapshot_id: str | None = None
@model_validator(mode="after")
def require_rollback_target(self) -> "ModelChangeGateResult":
if self.decision == "rollback" and not self.rollback_target_snapshot_id:
raise ValueError("rollback decision requires a target snapshot")
if self.decision != "rollback" and self.rollback_target_snapshot_id:
raise ValueError("non-rollback decision cannot carry rollback target")
return self
class OperationalIncident(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
incident_id: str = Field(pattern=r"^oas-g8-incident-[a-z0-9-]+$")
error_fingerprint: str = Field(pattern=r"^[a-f0-9]{64}$")
affected_contract: str = Field(min_length=1, max_length=180)
evidence_refs: tuple[str, ...] = Field(min_length=1)
pii_included: Literal[False] = False
class RegressionBacklogNode(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
node_id: str = Field(pattern=r"^oas-g8-node-[a-z0-9-]+$")
node_type: Literal["reproduction_test", "implementation", "e2e", "runtime_proof"]
depends_on: tuple[str, ...]
evidence_ref: str | None = Field(default=None, max_length=220)
status: Literal["pending", "passed", "failed"]
class IncidentRegressionDag(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
incident_id: str
nodes: tuple[RegressionBacklogNode, ...] = Field(min_length=4, max_length=4)
@model_validator(mode="after")
def require_ordered_closed_loop(self) -> "IncidentRegressionDag":
ids = [item.node_id for item in self.nodes]
if len(set(ids)) != len(ids):
raise ValueError("incident DAG node ids must be unique")
by_type = {item.node_type: item for item in self.nodes}
if set(by_type) != {
"reproduction_test",
"implementation",
"e2e",
"runtime_proof",
}:
raise ValueError(
"incident DAG requires reproduction, implementation, E2E, runtime"
)
known: set[str] = set()
for item in self.nodes:
if any(parent not in known for parent in item.depends_on):
raise ValueError("incident DAG dependencies must point backward")
known.add(item.node_id)
return self
class AgenticReleaseManifest(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
release_id: str = Field(pattern=r"^oas-g8-release-[a-z0-9-]+$")
red_green_passed: bool
contract_passed: bool
e2e_passed: bool
runtime_proof_passed: bool
public_proof_passed: bool
ssot_synced: bool
evidence_refs: tuple[str, ...]
@property
def releasable(self) -> bool:
return all(
(
self.red_green_passed,
self.contract_passed,
self.e2e_passed,
self.runtime_proof_passed,
self.public_proof_passed,
self.ssot_synced,
)
) and bool(self.evidence_refs)
__all__ = [
"AgenticReleaseManifest",
"ApprovedCatalogEntry",
"ContentBenchmarkQualification",
"ContentSourceArtifact",
"FindingSeverity",
"FindingState",
"GeneratedContentDraft",
"IncidentRegressionDag",
"IndependentRedTeamReview",
"ModelCalibrationSnapshot",
"ModelChangeDecision",
"ModelChangeGateResult",
"OperationalIncident",
"RedTeamFinding",
"RegressionBacklogNode",
"ReviewDimension",
]