"""Typed HTTP boundary for G3 rupture/repair ledgers.""" from __future__ import annotations import secrets from collections.abc import AsyncIterator from datetime import datetime from typing import Annotated, Literal from uuid import UUID import asyncpg from fastapi import APIRouter, Depends, Header, HTTPException, status from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from ..contracts.rupture_repair import RuptureLifecycleState, RuptureType from ..config import Settings, get_settings from ..deps import AIView, CurrentPrincipal, db_for_ai_view from ..services import rupture_repair_store router = APIRouter(tags=["rupture-repairs"]) INTERNAL_TOKEN_HEADER = "X-Vignette-Rupture-Token" MIN_INTERNAL_TOKEN_LENGTH = 32 _evaluator_db_provider = db_for_ai_view(AIView.EVALUATOR) async def rupture_internal_evaluator_db( settings: Annotated[Settings, Depends(get_settings)], presented_token: Annotated[ str | None, Header(alias=INTERNAL_TOKEN_HEADER), ] = None, ) -> AsyncIterator[asyncpg.Connection]: """Authenticate before acquiring any evaluator-view DB connection.""" configured_token = settings.rupture_internal_token.get_secret_value() if len(configured_token) < MIN_INTERNAL_TOKEN_LENGTH: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="internal rupture ingestion is unavailable", ) if presented_token is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="internal authentication required", ) if not secrets.compare_digest(presented_token, configured_token): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="internal authentication failed", ) async for conn in _evaluator_db_provider(): yield conn EvaluatorDB = Annotated[ asyncpg.Connection, Depends(rupture_internal_evaluator_db), ] class RuptureObservationResponse(BaseModel): model_config = ConfigDict(protected_namespaces=()) observation_id: UUID episode_id: UUID sequence_no: int = Field(ge=1) event_kind: Literal[ "rupture.detected", "rupture.recognized", "rupture.missed", "repair.attempted", "repair.partial", "repair.resolved", "repair.missed", "human.corrected", ] from_state: RuptureLifecycleState | None = None to_state: RuptureLifecycleState rupture_type: RuptureType source_kind: Literal["model_inferred", "observed_runtime", "human_rated"] perspective: Literal[ "independent_observer", "runtime_observation", "supervisor_human" ] ai_view: Literal["evaluator", "supervisor"] confidence: float | None = Field(default=None, ge=0.0, le=1.0) uncertainty: float = Field(ge=0.0, le=1.0) evidence_turn_ids: list[UUID] = Field(min_length=1) counterevidence: list[str] = Field(default_factory=list) model_run_id: UUID | None = None supersedes_observation_id: UUID | None = None correction_reason: str | None = None created_at: datetime class RuptureReconciliationResponse(BaseModel): model_config = ConfigDict(protected_namespaces=()) revision_id: UUID episode_id: UUID revision_no: int = Field(ge=1) supersedes_revision_id: UUID | None = None fast_warning_observation_id: UUID deep_observation_id: UUID | None = None fast_warning_id: str provisional_status: Literal["missed", "partial"] deep_status: Literal[ "missed", "partial", "resolved", "not_applicable", "insufficient_evidence", ] disposition: Literal[ "confirmed", "superseded_resolved", "superseded_partial", "dismissed" ] uncertainty: float = Field(ge=0.0, le=1.0) evidence_turn_ids: list[UUID] = Field(default_factory=list) counterevidence: list[str] = Field(default_factory=list) model_run_id: UUID created_at: datetime class RuptureSafetyReferenceResponse(BaseModel): episode_id: UUID safety_event_id: int turn_id: UUID | None = None ko_risk_level: int | None = None escalated: bool created_at: datetime class RuptureEpisodeResponse(BaseModel): episode_id: UUID session_id: UUID case_id: UUID learner_id: UUID episode_key: str created_at: datetime rupture_type: RuptureType | None = None current_status: Literal[ "onset", "recognized", "repair_attempted", "missed", "partial", "resolved", "not_applicable", "insufficient_evidence", ] | None = None status_source: Literal[ "lifecycle_event", "deep_reconciliation", "human_correction" ] observations: list[RuptureObservationResponse] = Field(default_factory=list) reconciliation_revisions: list[RuptureReconciliationResponse] = Field( default_factory=list ) safety_references: list[RuptureSafetyReferenceResponse] = Field( default_factory=list ) class RuptureRepairReadModelResponse(BaseModel): session_id: UUID requested_view: Literal["counselor", "supervisor"] clinical_claim_allowed: Literal[False] episodes: list[RuptureEpisodeResponse] class InternalRuptureObservationRequest(BaseModel): model_config = ConfigDict(protected_namespaces=()) episode_key: str = Field(min_length=1, max_length=180) idempotency_key: UUID event_kind: Literal[ "rupture.detected", "rupture.recognized", "rupture.missed", "repair.attempted", "repair.partial", "repair.resolved", "repair.missed", ] from_state: RuptureLifecycleState | None = None to_state: RuptureLifecycleState rupture_type: RuptureType source_kind: Literal["model_inferred", "observed_runtime"] perspective: Literal["independent_observer", "runtime_observation"] ai_view: Literal["evaluator"] confidence: float | None = Field(default=None, ge=0.0, le=1.0) uncertainty: float = Field(ge=0.0, le=1.0) evidence_turn_ids: list[UUID] = Field(min_length=1) counterevidence: list[str] = Field(default_factory=list) model_run_id: UUID | None = None safety_event_ids: list[int] = Field(default_factory=list) visible_to: list[ Literal["counselor", "evaluator", "supervisor", "research"] ] = Field( default_factory=lambda: [ "counselor", "evaluator", "supervisor", "research", ], min_length=1, ) @field_validator("episode_key") @classmethod def strip_episode_key(cls, value: str) -> str: stripped = value.strip() if not stripped: raise ValueError("episode_key must not be blank") return stripped @model_validator(mode="after") def require_provenance_pair(self) -> "InternalRuptureObservationRequest": if self.source_kind == "model_inferred": if self.perspective != "independent_observer" or self.model_run_id is None: raise ValueError( "model_inferred requires independent_observer and model_run_id" ) elif self.perspective != "runtime_observation": raise ValueError( "observed_runtime requires runtime_observation perspective" ) if len(set(self.evidence_turn_ids)) != len(self.evidence_turn_ids): raise ValueError("evidence_turn_ids must be unique") if len(set(self.safety_event_ids)) != len(self.safety_event_ids): raise ValueError("safety_event_ids must be unique") if len(set(self.visible_to)) != len(self.visible_to): raise ValueError("visible_to must be unique") if "evaluator" not in self.visible_to: raise ValueError("visible_to must include evaluator") return self class InternalRuptureObservationResponse(BaseModel): episode_id: UUID observation_id: UUID class InternalReconciliationRequest(BaseModel): model_config = ConfigDict(protected_namespaces=()) idempotency_key: UUID fast_warning_observation_id: UUID deep_observation_id: UUID | None = None fast_warning_id: str = Field(min_length=1, max_length=180) provisional_status: Literal["missed", "partial"] deep_status: Literal[ "missed", "partial", "resolved", "not_applicable", "insufficient_evidence", ] disposition: Literal[ "confirmed", "superseded_resolved", "superseded_partial", "dismissed" ] uncertainty: float = Field(ge=0.0, le=1.0) evidence_turn_ids: list[UUID] = Field(default_factory=list) counterevidence: list[str] = Field(default_factory=list) model_run_id: UUID ai_view: Literal["evaluator"] visible_to: list[ Literal["counselor", "evaluator", "supervisor", "research"] ] = Field( default_factory=lambda: [ "counselor", "evaluator", "supervisor", "research", ], min_length=1, ) @field_validator("fast_warning_id") @classmethod def strip_warning_id(cls, value: str) -> str: stripped = value.strip() if not stripped: raise ValueError("fast_warning_id must not be blank") return stripped @model_validator(mode="after") def validate_disposition(self) -> "InternalReconciliationRequest": valid = ( (self.disposition == "confirmed" and self.deep_status == self.provisional_status) or (self.disposition == "superseded_resolved" and self.deep_status == "resolved") or (self.disposition == "superseded_partial" and self.deep_status == "partial") or (self.disposition == "dismissed" and self.deep_status == "not_applicable") ) if not valid: raise ValueError("reconciliation disposition does not match deep_status") if len(set(self.evidence_turn_ids)) != len(self.evidence_turn_ids): raise ValueError("evidence_turn_ids must be unique") if len(set(self.visible_to)) != len(self.visible_to): raise ValueError("visible_to must be unique") if "evaluator" not in self.visible_to: raise ValueError("visible_to must include evaluator") return self class InternalReconciliationResponse(BaseModel): episode_id: UUID revision_id: UUID revision_no: int = Field(ge=1) class HumanRuptureCorrectionRequest(BaseModel): idempotency_key: UUID supersedes_observation_id: UUID rupture_type: RuptureType corrected_status: Literal["missed", "partial", "resolved"] uncertainty: float = Field(ge=0.0, le=1.0) evidence_turn_ids: list[UUID] = Field(min_length=1) counterevidence: list[str] = Field(default_factory=list) correction_reason: str = Field(min_length=1, max_length=1000) @field_validator("correction_reason") @classmethod def strip_reason(cls, value: str) -> str: stripped = value.strip() if not stripped: raise ValueError("correction_reason must not be blank") return stripped @field_validator("evidence_turn_ids") @classmethod def unique_evidence(cls, value: list[UUID]) -> list[UUID]: if len(set(value)) != len(value): raise ValueError("evidence_turn_ids must be unique") return value class HumanRuptureCorrectionResponse(BaseModel): episode_id: UUID observation_id: UUID def _http_error(exc: Exception) -> HTTPException: if isinstance(exc, rupture_repair_store.RuptureRepairNotFoundError): return HTTPException(status.HTTP_404_NOT_FOUND, detail=str(exc)) if isinstance(exc, rupture_repair_store.RuptureRepairConflictError): return HTTPException(status.HTTP_409_CONFLICT, detail=str(exc)) if isinstance(exc, rupture_repair_store.RuptureRepairStateError): return HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) raise exc @router.get( "/sessions/{session_id}/ruptures", response_model=RuptureRepairReadModelResponse, ) async def get_rupture_repairs( session_id: UUID, principal: CurrentPrincipal, ) -> RuptureRepairReadModelResponse: try: payload = await rupture_repair_store.read_rupture_repairs( principal=principal, session_id=session_id, ) except ( rupture_repair_store.RuptureRepairNotFoundError, rupture_repair_store.RuptureRepairConflictError, rupture_repair_store.RuptureRepairStateError, ) as exc: raise _http_error(exc) from exc return RuptureRepairReadModelResponse.model_validate(payload) @router.post( "/internal/sessions/{session_id}/ruptures/observations", response_model=InternalRuptureObservationResponse, status_code=status.HTTP_201_CREATED, ) async def create_internal_rupture_observation( session_id: UUID, body: InternalRuptureObservationRequest, conn: EvaluatorDB, ) -> InternalRuptureObservationResponse: try: payload = await rupture_repair_store.append_evaluator_observation( conn=conn, session_id=session_id, **body.model_dump(), ) except ( rupture_repair_store.RuptureRepairNotFoundError, rupture_repair_store.RuptureRepairConflictError, rupture_repair_store.RuptureRepairStateError, ) as exc: raise _http_error(exc) from exc return InternalRuptureObservationResponse.model_validate(payload) @router.post( "/internal/sessions/{session_id}/ruptures/{episode_id}/reconciliations", response_model=InternalReconciliationResponse, status_code=status.HTTP_201_CREATED, ) async def create_internal_reconciliation( session_id: UUID, episode_id: UUID, body: InternalReconciliationRequest, conn: EvaluatorDB, ) -> InternalReconciliationResponse: try: payload = await rupture_repair_store.append_reconciliation_revision( conn=conn, session_id=session_id, episode_id=episode_id, **body.model_dump(), ) except ( rupture_repair_store.RuptureRepairNotFoundError, rupture_repair_store.RuptureRepairConflictError, rupture_repair_store.RuptureRepairStateError, ) as exc: raise _http_error(exc) from exc return InternalReconciliationResponse.model_validate(payload) @router.post( "/sessions/{session_id}/ruptures/{episode_id}/corrections", response_model=HumanRuptureCorrectionResponse, status_code=status.HTTP_201_CREATED, ) async def create_human_rupture_correction( session_id: UUID, episode_id: UUID, body: HumanRuptureCorrectionRequest, principal: CurrentPrincipal, ) -> HumanRuptureCorrectionResponse: try: observation_id = await rupture_repair_store.append_human_correction( principal=principal, session_id=session_id, episode_id=episode_id, **body.model_dump(), ) except ( rupture_repair_store.RuptureRepairNotFoundError, rupture_repair_store.RuptureRepairConflictError, rupture_repair_store.RuptureRepairStateError, ) as exc: raise _http_error(exc) from exc return HumanRuptureCorrectionResponse( episode_id=episode_id, observation_id=observation_id, ) __all__ = ["router"]