ml-services/evaluation/v2/runtime.py

from __future__ import annotations

import hashlib
import json
from collections.abc import Callable
from enum import Enum
from pathlib import Path
from typing import Any, Literal

from pydantic import Field, model_validator

from .acoustic_dynamics import derive_acoustic_dynamics
from .adapters import build_signal_bundle
from .decisions import DECISION_POLICY_VERSION, build_call_decision
from .domain_profiles import (
    DomainProfile,
    ProfileSelection,
    load_general_service_profile,
    resolve_domain_plan,
)
from .findings import (
    RequirementAssessmentBatch,
    derive_findings,
)
from .presentation import CallEvaluationView, project_call_evaluation
from .schemas import (
    CallDecision,
    ContractModel,
    DecisionStatus,
    SourceProvenance,
)
from .semantic_assessment import assess_requirements
from .validation import validate_decision_references

RUNTIME_VERSION = "0.1.0"
HERE = Path(__file__).resolve().parent
PROFILE_PATH = HERE / "profiles" / "banking_v1.json"
SELECTIONS_PATH = HERE / "profiles" / "banking_call_selections.json"


class ShadowRunStatus(str, Enum):
    SUCCEEDED = "succeeded"
    UNSUPPORTED_DOMAIN = "unsupported_domain"
    PROFILE_SELECTION_UNAVAILABLE = "profile_selection_unavailable"
    FAILED = "failed"


class LegacyAttentionProxy(ContractModel):
    evaluator_version: str = Field(min_length=1)
    attention_required: bool
    basis: Literal["failed_compliance_or_escalation"]
    failed_compliance_count: int = Field(ge=0)
    escalation_tier: str = Field(min_length=1)


class DecisionComparison(ContractModel):
    comparable: bool
    reason: str = Field(min_length=1)
    attention_agreement: bool | None = None

    @model_validator(mode="after")
    def validate_agreement(self):
        if self.comparable != (self.attention_agreement is not None):
            raise ValueError(
                "attention agreement is present only when runs are comparable"
            )
        return self


class EvaluationV2Run(ContractModel):
    schema_version: Literal["1.0"] = "1.0"
    runtime_version: str = Field(min_length=1)
    run_id: str = Field(min_length=1)
    call_id: str = Field(min_length=1)
    mode: Literal["shadow"] = "shadow"
    status: ShadowRunStatus
    evaluator_version: str = Field(min_length=1)
    profile_id: str | None = None
    decision_sha256: str | None = Field(
        default=None,
        pattern=r"^[0-9a-f]{64}$",
    )
    decision: CallDecision | None = None
    presentation: CallEvaluationView | None = None
    legacy_proxy: LegacyAttentionProxy | None = None
    comparison: DecisionComparison | None = None
    limitations: list[str] = Field(default_factory=list)
    failure_reason: str | None = None
    provenance: SourceProvenance

    @model_validator(mode="after")
    def validate_run(self):
        succeeded = self.status == ShadowRunStatus.SUCCEEDED
        if succeeded != (
            self.decision is not None and self.presentation is not None
        ):
            raise ValueError(
                "successful shadow runs require decision and presentation"
            )
        if succeeded != (self.decision_sha256 is not None):
            raise ValueError(
                "successful shadow runs require a decision hash"
            )
        if self.status == ShadowRunStatus.FAILED:
            if not self.failure_reason:
                raise ValueError("failed shadow runs require a reason")
        elif self.failure_reason is not None:
            raise ValueError(
                "only failed shadow runs may include a failure reason"
            )
        if self.comparison is not None and self.legacy_proxy is None:
            raise ValueError("comparison requires a legacy proxy")
        return self


def _source(method: str) -> SourceProvenance:
    return SourceProvenance(
        producer="evaluator_v2.runtime",
        producer_version=RUNTIME_VERSION,
        method=method,
    )


def _sha256(payload: Any) -> str:
    encoded = json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _run_id(call_id: str, marker: str) -> str:
    digest = hashlib.sha256(
        f"{call_id}:{RUNTIME_VERSION}:{marker}".encode()
    ).hexdigest()[:16]
    return f"{call_id}:v2:{digest}"


def _profile() -> DomainProfile:
    return DomainProfile.model_validate_json(
        PROFILE_PATH.read_text(encoding="utf-8")
    )


def _general_selection(call_id: str) -> ProfileSelection:
    return ProfileSelection(
        call_id=call_id,
        intent_ids=["service.general_request"],
        selection_method="general_service_fallback",
    )


def _known_selections() -> dict[str, ProfileSelection]:
    payload = json.loads(SELECTIONS_PATH.read_text(encoding="utf-8"))
    return {
        item["call_id"]: ProfileSelection.model_validate(item)
        for item in payload["calls"]
    }


def _selection(
    call_id: str,
    explicit: ProfileSelection | dict[str, Any] | None,
) -> ProfileSelection | None:
    if explicit is not None:
        selection = (
            explicit
            if isinstance(explicit, ProfileSelection)
            else ProfileSelection.model_validate(explicit)
        )
        if selection.call_id != call_id:
            raise ValueError("profile selection call_id does not match")
        return selection
    return _known_selections().get(call_id)


def legacy_attention_proxy(
    evaluation: dict[str, Any] | None,
) -> LegacyAttentionProxy | None:
    if not evaluation:
        return None
    compliance = evaluation.get("compliance") or {}
    failed = sum(
        isinstance(item, dict) and item.get("passed") is False
        for item in compliance.values()
    )
    escalation = str(
        (evaluation.get("escalation") or {}).get("risk_level")
        or "none"
    ).lower()
    return LegacyAttentionProxy(
        evaluator_version=str(
            (evaluation.get("_evaluator") or {}).get("version")
            or evaluation.get("rubric_version")
            or "v1"
        ),
        attention_required=failed > 0 or escalation in {
            "review",
            "escalate",
        },
        basis="failed_compliance_or_escalation",
        failed_compliance_count=failed,
        escalation_tier=escalation,
    )


def _comparison(
    decision: CallDecision,
    proxy: LegacyAttentionProxy | None,
) -> DecisionComparison | None:
    if proxy is None:
        return None
    if decision.decision_status != DecisionStatus.COMPLETE:
        return DecisionComparison(
            comparable=False,
            reason=(
                "v2 requirement coverage is incomplete; attention values "
                "must not be treated as equivalent."
            ),
        )
    return DecisionComparison(
        comparable=True,
        reason=(
            "Both evaluators produced a complete attention value under "
            "their named policies."
        ),
        attention_agreement=(
            proxy.attention_required == decision.attention_required
        ),
    )


def _unavailable(
    *,
    call_id: str,
    status: ShadowRunStatus,
    limitation: str,
    legacy_proxy: LegacyAttentionProxy | None,
) -> EvaluationV2Run:
    return EvaluationV2Run(
        runtime_version=RUNTIME_VERSION,
        run_id=_run_id(call_id, status.value),
        call_id=call_id,
        status=status,
        evaluator_version=f"v2-policy-{DECISION_POLICY_VERSION}",
        legacy_proxy=legacy_proxy,
        limitations=[limitation],
        provenance=_source("feature_flagged_shadow_unavailable"),
    )


def run_shadow_evaluation(
    *,
    transcript: dict[str, Any],
    sentiment: dict[str, Any] | None = None,
    legacy_evaluation: dict[str, Any] | None = None,
    profile_selection: ProfileSelection | dict[str, Any] | None = None,
    assessments: RequirementAssessmentBatch | dict[str, Any] | None = None,
    run_semantic_assessor: bool = True,
    transcript_source: str | None = None,
    sentiment_source: str | None = None,
    progress: Callable[[str], None] | None = None,
) -> EvaluationV2Run:
    report = progress or (lambda _stage: None)
    report("evaluating_v2_prepare")
    call_id = str(transcript.get("call_id") or "").strip()
    if not call_id:
        raise ValueError("transcript call_id is required")
    domain = str(transcript.get("domain") or "").strip().lower()
    proxy = legacy_attention_proxy(legacy_evaluation)
    selection = _selection(call_id, profile_selection)
    uses_general_profile = domain != "banking" or selection is None
    if uses_general_profile:
        profile = load_general_service_profile()
        selection = _general_selection(call_id)
    else:
        profile = _profile()
    plan = resolve_domain_plan(profile, selection)
    report("evaluating_v2_signals")
    bundle = build_signal_bundle(
        transcript,
        sentiment,
        transcript_source=transcript_source,
        sentiment_source=sentiment_source,
    )
    if sentiment:
        bundle = derive_acoustic_dynamics(bundle)
    assessment_batch = None
    if assessments is not None:
        assessment_batch = (
            assessments
            if isinstance(assessments, RequirementAssessmentBatch)
            else RequirementAssessmentBatch.model_validate(assessments)
        )
    elif run_semantic_assessor:
        report("evaluating_v2_requirements")
        assessment_batch = assess_requirements(bundle, plan)
    report("evaluating_v2_findings")
    derivation = derive_findings(bundle, plan, assessment_batch)
    report("evaluating_v2_decision")
    decision = build_call_decision(bundle, derivation)
    validate_decision_references(decision, bundle)
    report("evaluating_v2_presentation")
    presentation = project_call_evaluation(decision, bundle=bundle)
    limitations = [
        "research_domain_profile",
        "bounded_supervisor_not_run",
    ]
    if uses_general_profile:
        limitations.append("general_service_profile_not_domain_compliance")
    if assessment_batch is None:
        limitations.append("semantic_requirement_assessments_missing")
    if not sentiment:
        limitations.append("acoustic_input_unavailable")
    marker = presentation.decision_sha256
    return EvaluationV2Run(
        runtime_version=RUNTIME_VERSION,
        run_id=_run_id(call_id, marker),
        call_id=call_id,
        status=ShadowRunStatus.SUCCEEDED,
        evaluator_version=decision.evaluator_version,
        profile_id=plan.profile_id,
        decision_sha256=presentation.decision_sha256,
        decision=decision,
        presentation=presentation,
        legacy_proxy=proxy,
        comparison=_comparison(decision, proxy),
        limitations=limitations,
        provenance=_source("feature_flagged_shadow_evaluation"),
    )


def safe_run_shadow_evaluation(**kwargs) -> EvaluationV2Run:
    try:
        return run_shadow_evaluation(**kwargs)
    # Shadow execution must record any failure without failing the v1 job.
    except Exception as exc:  # noqa: BLE001
        transcript = kwargs.get("transcript") or {}
        call_id = str(transcript.get("call_id") or "unknown-call")
        proxy = legacy_attention_proxy(kwargs.get("legacy_evaluation"))
        reason = f"{type(exc).__name__}: {exc}"
        return EvaluationV2Run(
            runtime_version=RUNTIME_VERSION,
            run_id=_run_id(call_id, _sha256(reason)),
            call_id=call_id,
            status=ShadowRunStatus.FAILED,
            evaluator_version=f"v2-policy-{DECISION_POLICY_VERSION}",
            legacy_proxy=proxy,
            limitations=["shadow_runtime_failure"],
            failure_reason=reason[:500],
            provenance=_source("feature_flagged_shadow_failure"),
        )