ml-services/evaluation/v2/presentation.py

from __future__ import annotations

from enum import Enum
from typing import Literal

from pydantic import Field, model_validator

from .schemas import (
    ActionExecution,
    ActionType,
    CallDecision,
    ContractModel,
    DecisionStatus,
    EvidenceRef,
    Finding,
    FindingCategory,
    Modality,
    ReliabilityStatus,
    SignalBundle,
    SourceProvenance,
    Speaker,
    Visibility,
)
from .supervisor import SupervisorResult, decision_sha256

PRESENTATION_VERSION = "0.1.0"
MAX_PRIMARY_REASONS = 2
MAX_POSITIVE_HIGHLIGHTS = 2
MAX_EVIDENCE_ITEMS = 6


class PageState(str, Enum):
    NEEDS_ATTENTION = "needs_attention"
    NO_ATTENTION_FINDING = "no_attention_finding"
    EVALUATION_INCOMPLETE = "evaluation_incomplete"


class Representation(str, Enum):
    STATUS_TEXT = "status_text"
    REASON_LIST = "reason_list"
    ACTION = "action"
    CHECKLIST = "checklist"
    EVIDENCE_TIMELINE = "evidence_timeline"
    NOTICE = "notice"
    DISCLOSURE = "disclosure"
    NOT_RENDERED = "not_rendered"


class EvidenceKind(str, Enum):
    TRANSCRIPT = "transcript"
    AUDIO_SUPPORT = "audio_support"
    BUSINESS_CONTEXT = "business_context"


class EvidencePurpose(str, Enum):
    REASON = "reason"
    POSITIVE = "positive"


class FindingOutcome(str, Enum):
    EFFECTIVE = "effective"
    INCORRECT = "incorrect"
    MISSED = "missed"
    OBSERVED_CONCERN = "observed_concern"


class ManagerAnswer(str, Enum):
    YES = "yes"
    PARTLY = "partly"
    NO = "no"
    UNCLEAR = "unclear"


class CheckStatus(str, Enum):
    DEMONSTRATED = "demonstrated"
    INCORRECT = "incorrect"
    NOT_DEMONSTRATED = "not_demonstrated"
    UNABLE_TO_DETERMINE = "unable_to_determine"


class AcousticStatus(str, Enum):
    AVAILABLE = "available"
    LIMITED = "limited"
    UNAVAILABLE = "unavailable"


class InventoryItem(ContractModel):
    output_id: str = Field(min_length=1, pattern=r"^[a-z0-9_.]+$")
    source_fields: list[str] = Field(min_length=1)
    visibility: Visibility
    representation: Representation
    visible_when: str = Field(min_length=1)
    rationale: str = Field(min_length=1)


class PresentedFinding(ContractModel):
    finding_id: str = Field(min_length=1)
    category: FindingCategory
    category_label: str = Field(min_length=1)
    title: str = Field(min_length=1)
    summary: str = Field(min_length=1)
    evidence_ids: list[str] = Field(min_length=1)
    counter_evidence_count: int = Field(ge=0)
    reliability_note: str | None = None
    outcome: FindingOutcome = FindingOutcome.OBSERVED_CONCERN


class ManagerQuestion(ContractModel):
    question_id: str = Field(min_length=1, pattern=r"^[a-z0-9_.]+$")
    question: str = Field(min_length=1)
    answer: ManagerAnswer
    answer_label: str = Field(min_length=1)
    summary: str = Field(min_length=1)
    evidence_ids: list[str] = Field(default_factory=list)


class PresentedCheck(ContractModel):
    requirement_id: str = Field(min_length=1)
    title: str = Field(min_length=1)
    category: FindingCategory
    status: CheckStatus
    summary: str = Field(min_length=1)
    evidence_ids: list[str] = Field(default_factory=list)
    promoted: bool = False


class AcousticObservation(ContractModel):
    observation_id: str = Field(min_length=1)
    label: str = Field(min_length=1)
    summary: str = Field(min_length=1)
    start_seconds: float | None = Field(default=None, ge=0.0)
    end_seconds: float | None = Field(default=None, ge=0.0)
    supporting_only: Literal[True] = True

    @model_validator(mode="after")
    def validate_time_range(self):
        if (
            self.start_seconds is not None
            and self.end_seconds is not None
            and self.end_seconds < self.start_seconds
        ):
            raise ValueError("acoustic observation has an invalid time range")
        return self


class AcousticContext(ContractModel):
    status: AcousticStatus
    coverage_label: str = Field(min_length=1)
    conclusion: str = Field(min_length=1)
    observations: list[AcousticObservation] = Field(default_factory=list)


class PresentedEvidence(ContractModel):
    evidence_id: str = Field(min_length=1)
    finding_ids: list[str] = Field(min_length=1)
    purposes: list[EvidencePurpose] = Field(min_length=1)
    kind: EvidenceKind
    speaker: Speaker | None = None
    start_seconds: float | None = Field(default=None, ge=0.0)
    end_seconds: float | None = Field(default=None, ge=0.0)
    text: str | None = None
    seekable: bool
    supporting_only: bool

    @model_validator(mode="after")
    def validate_evidence(self):
        if self.seekable != (self.start_seconds is not None):
            raise ValueError("seekable must match timestamp availability")
        if self.kind == EvidenceKind.AUDIO_SUPPORT and not self.supporting_only:
            raise ValueError("audio evidence must remain supporting-only")
        if len(self.finding_ids) != len(set(self.finding_ids)):
            raise ValueError("duplicate finding reference")
        if len(self.purposes) != len(set(self.purposes)):
            raise ValueError("duplicate evidence purpose")
        return self


class PresentedAction(ContractModel):
    action_type: ActionType
    execution: ActionExecution
    label: str = Field(min_length=1)
    execution_label: str = Field(min_length=1)
    basis_finding_ids: list[str] = Field(min_length=1)
    automation_allowed: bool
    requires_human_approval: bool
    delivery: Literal["email"] = "email"
    audience: Literal["manager", "customer"] = "manager"


class CompletenessNotice(ContractModel):
    status: DecisionStatus
    title: Literal["Evaluation incomplete"] = "Evaluation incomplete"
    message: str = Field(min_length=1)
    affected_requirements: list[str] = Field(default_factory=list)


class PresentationDetails(ContractModel):
    additional_findings: list[PresentedFinding] = Field(
        default_factory=list
    )
    additional_positive_findings: list[PresentedFinding] = Field(
        default_factory=list
    )
    evidence: list[PresentedEvidence] = Field(default_factory=list)
    uncertainty_messages: list[str] = Field(default_factory=list)
    supervisor_context_note: str | None = None
    evaluator_version: str = Field(min_length=1)
    domain_profile_id: str = Field(min_length=1)
    decision_policy_id: str = Field(min_length=1)
    decision_policy_version: str = Field(min_length=1)
    supervisor_fallback_used: bool | None = None


class CallEvaluationView(ContractModel):
    schema_version: Literal["1.0"] = "1.0"
    presentation_version: str = Field(min_length=1)
    call_id: str = Field(min_length=1)
    decision_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    state: PageState
    evaluation_status: DecisionStatus
    attention_required: bool
    headline: str = Field(min_length=1)
    summary: str = Field(min_length=1)
    manager_questions: list[ManagerQuestion] = Field(
        default_factory=list,
        max_length=4,
    )
    checklist: list[PresentedCheck] = Field(default_factory=list)
    acoustic_context: AcousticContext | None = None
    primary_reasons: list[PresentedFinding] = Field(
        default_factory=list,
        max_length=MAX_PRIMARY_REASONS,
    )
    additional_reason_count: int = Field(ge=0)
    positive_highlights: list[PresentedFinding] = Field(
        default_factory=list,
        max_length=MAX_POSITIVE_HIGHLIGHTS,
    )
    additional_positive_count: int = Field(ge=0)
    recommended_action: PresentedAction | None = None
    evidence: list[PresentedEvidence] = Field(
        default_factory=list,
        max_length=MAX_EVIDENCE_ITEMS,
    )
    completeness_notice: CompletenessNotice | None = None
    details: PresentationDetails
    provenance: SourceProvenance

    @model_validator(mode="after")
    def validate_view(self):
        if self.attention_required != (
            self.state == PageState.NEEDS_ATTENTION
        ):
            raise ValueError("page state must preserve attention")
        if not self.attention_required:
            expected_state = (
                PageState.EVALUATION_INCOMPLETE
                if self.evaluation_status
                == DecisionStatus.INSUFFICIENT_EVIDENCE
                else PageState.NO_ATTENTION_FINDING
            )
            if self.state != expected_state:
                raise ValueError(
                    "no-attention page state must preserve completeness"
                )
        if self.attention_required != bool(self.primary_reasons):
            raise ValueError("attention requires a visible primary reason")
        if self.attention_required != (
            self.recommended_action is not None
        ):
            raise ValueError("attention and visible action must match")
        if (
            self.evaluation_status == DecisionStatus.COMPLETE
            and self.completeness_notice is not None
        ):
            raise ValueError(
                "complete evaluation cannot carry an incomplete notice"
            )
        if (
            self.evaluation_status != DecisionStatus.COMPLETE
            and self.completeness_notice is None
        ):
            raise ValueError("incomplete evaluation requires a notice")
        if self.state == PageState.EVALUATION_INCOMPLETE and (
            self.evaluation_status == DecisionStatus.COMPLETE
        ):
            raise ValueError(
                "incomplete page state requires incomplete evaluation"
            )

        finding_groups = (
            self.primary_reasons,
            self.positive_highlights,
            self.details.additional_findings,
            self.details.additional_positive_findings,
        )
        finding_ids = [
            item.finding_id
            for group in finding_groups
            for item in group
        ]
        if len(finding_ids) != len(set(finding_ids)):
            raise ValueError("findings must not be repeated across regions")

        evidence_ids = [item.evidence_id for item in self.evidence]
        if len(evidence_ids) != len(set(evidence_ids)):
            raise ValueError("evidence must be deduplicated")
        known_evidence = set(evidence_ids)
        for finding in (
            self.primary_reasons + self.positive_highlights
        ):
            if not set(finding.evidence_ids).issubset(known_evidence):
                raise ValueError(
                    "visible findings require projected evidence"
                )
        detail_evidence = {
            item.evidence_id for item in self.details.evidence
        }
        if known_evidence.intersection(detail_evidence):
            raise ValueError(
                "primary and detail evidence must not be repeated"
            )
        for finding in (
            self.details.additional_findings
            + self.details.additional_positive_findings
        ):
            if not set(finding.evidence_ids).issubset(
                known_evidence.union(detail_evidence)
            ):
                raise ValueError(
                    "detail findings require projected detail evidence"
                )
        return self


class PresentationCheck(ContractModel):
    attention_is_literal: bool
    reasons_are_bounded: bool
    reasons_have_evidence: bool
    positives_are_bounded: bool
    action_is_unambiguous: bool
    completeness_is_explicit: bool
    content_is_deduplicated: bool
    no_score_fields: bool
    passed: bool


class ScenarioAnswerKey(ContractModel):
    attention_state: PageState
    reasons: list[str] = Field(default_factory=list)
    evidence_locations_seconds: list[float] = Field(default_factory=list)
    positive_handling: list[str] = Field(default_factory=list)
    action: str | None = None
    completeness: DecisionStatus


class PresentationScenarioResult(ContractModel):
    call_id: str = Field(min_length=1)
    answer_key: ScenarioAnswerKey
    check: PresentationCheck


class PresentationValidationReport(ContractModel):
    schema_version: Literal["1.0"] = "1.0"
    presentation_version: str = Field(min_length=1)
    scenario_source: str = Field(min_length=1)
    scenario_count: int = Field(ge=0)
    passed_scenario_count: int = Field(ge=0)
    structural_checks_passed: bool
    human_comprehension_status: Literal["not_run"] = "not_run"
    human_test_questions: list[str] = Field(min_length=1)
    inventory: list[InventoryItem] = Field(min_length=1)
    scenarios: list[PresentationScenarioResult]
    limitations: list[str] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_counts(self):
        if self.scenario_count != len(self.scenarios):
            raise ValueError("scenario count does not match rows")
        passed = sum(item.check.passed for item in self.scenarios)
        if self.passed_scenario_count != passed:
            raise ValueError("passed scenario count does not match rows")
        if self.structural_checks_passed != (
            passed == self.scenario_count
        ):
            raise ValueError("structural pass flag does not match rows")
        return self


OUTPUT_INVENTORY = (
    InventoryItem(
        output_id="decision.attention_state",
        source_fields=["attention_required", "decision_status"],
        visibility=Visibility.PRIMARY,
        representation=Representation.STATUS_TEXT,
        visible_when="always",
        rationale=(
            "Answers whether this call needs attention without implying a "
            "continuous risk score."
        ),
    ),
    InventoryItem(
        output_id="decision.controlling_findings",
        source_fields=["decision_trace.controlling_finding_ids"],
        visibility=Visibility.PRIMARY,
        representation=Representation.REASON_LIST,
        visible_when="attention is required",
        rationale=(
            "At most two evidence-backed reasons explain the decision."
        ),
    ),
    InventoryItem(
        output_id="decision.recommended_action",
        source_fields=["recommended_action"],
        visibility=Visibility.PRIMARY,
        representation=Representation.ACTION,
        visible_when="attention is required",
        rationale=(
            "One constrained next action is more useful than another score."
        ),
    ),
    InventoryItem(
        output_id="findings.positive_highlights",
        source_fields=["positive_findings"],
        visibility=Visibility.PRIMARY,
        representation=Representation.CHECKLIST,
        visible_when="usable positive findings exist",
        rationale=(
            "At most two grounded positives credit handling that went well."
        ),
    ),
    InventoryItem(
        output_id="findings.selected_evidence",
        source_fields=["finding.evidence"],
        visibility=Visibility.EVIDENCE,
        representation=Representation.EVIDENCE_TIMELINE,
        visible_when="a primary reason or positive is shown",
        rationale=(
            "Quotes and timestamps answer where the finding occurred."
        ),
    ),
    InventoryItem(
        output_id="decision.completeness",
        source_fields=["decision_status", "uncertainties"],
        visibility=Visibility.PRIMARY,
        representation=Representation.NOTICE,
        visible_when="the evaluation is partial or insufficient",
        rationale=(
            "An incomplete evaluation must never look like a cleared call."
        ),
    ),
    InventoryItem(
        output_id="findings.additional_context",
        source_fields=[
            "triggered_findings",
            "positive_findings",
            "counter_evidence",
        ],
        visibility=Visibility.DETAILS,
        representation=Representation.DISCLOSURE,
        visible_when="the user opens details",
        rationale=(
            "Non-controlling findings and contradictions remain available "
            "without competing with the decision."
        ),
    ),
    InventoryItem(
        output_id="evaluation.provenance",
        source_fields=[
            "evaluator_version",
            "domain_profile_id",
            "decision_trace.policy_id",
        ],
        visibility=Visibility.DETAILS,
        representation=Representation.DISCLOSURE,
        visible_when="the user opens evaluation details",
        rationale=(
            "Version and policy context support traceability, not scanning."
        ),
    ),
    InventoryItem(
        output_id="signals.raw_model_outputs",
        source_fields=[
            "sentiment probabilities",
            "pitch",
            "volume",
            "energy",
            "speech rate",
            "normalized acoustic values",
        ],
        visibility=Visibility.INTERNAL,
        representation=Representation.NOT_RENDERED,
        visible_when="never on the call evaluator page",
        rationale=(
            "Raw model telemetry is not a business conclusion and may only "
            "support a grounded finding."
        ),
    ),
    InventoryItem(
        output_id="decision.internal_logic",
        source_fields=[
            "decision_trace.qualifications",
            "detection_rule.thresholds",
            "reliability reasons",
            "hashes",
        ],
        visibility=Visibility.INTERNAL,
        representation=Representation.NOT_RENDERED,
        visible_when="never on the call evaluator page",
        rationale=(
            "Policy mechanics belong in diagnostics and exported evidence."
        ),
    ),
    InventoryItem(
        output_id="scores.composites",
        source_fields=[
            "overall score",
            "compliance percentage",
            "workflow percentage",
            "quality rating",
            "friction score",
        ],
        visibility=Visibility.INTERNAL,
        representation=Representation.NOT_RENDERED,
        visible_when="not produced by evaluator v2",
        rationale=(
            "These aggregates mix unlike claims and create false precision."
        ),
    ),
    InventoryItem(
        output_id="timeline.duplicate_lanes",
        source_fields=[
            "compliance markers",
            "workflow markers",
            "quality markers",
            "sentiment markers",
        ],
        visibility=Visibility.INTERNAL,
        representation=Representation.NOT_RENDERED,
        visible_when="not produced by evaluator v2",
        rationale=(
            "One selected evidence timeline replaces overlapping lanes."
        ),
    ),
)


_CATEGORY_LABELS = {
    FindingCategory.REQUIRED_CONTROL: "Required step",
    FindingCategory.PROCESS: "Call handling",
    FindingCategory.ESCALATION: "Customer risk",
    FindingCategory.OUTCOME: "Outcome",
    FindingCategory.AGENT_BEHAVIOR: "Agent handling",
    FindingCategory.DATA_QUALITY: "Evaluation quality",
}


def _source() -> SourceProvenance:
    return SourceProvenance(
        producer="evaluator_v2.presentation",
        producer_version=PRESENTATION_VERSION,
        method="bounded_decision_preserving_page_projection",
    )


def _state(decision: CallDecision) -> PageState:
    if decision.attention_required:
        return PageState.NEEDS_ATTENTION
    if (
        decision.decision_status
        == DecisionStatus.INSUFFICIENT_EVIDENCE
    ):
        return PageState.EVALUATION_INCOMPLETE
    return PageState.NO_ATTENTION_FINDING


def _copy(decision: CallDecision) -> tuple[str, str]:
    state = _state(decision)
    if state == PageState.NEEDS_ATTENTION:
        return (
            "Needs attention",
            "One or more evidence-backed findings require review.",
        )
    if state == PageState.NO_ATTENTION_FINDING:
        if decision.decision_status == DecisionStatus.PARTIAL:
            return (
                "No review finding identified",
                (
                    "Assessed requirements produced no qualifying negative "
                    "finding; one or more requirements remain uncertain."
                ),
            )
        return (
            "No review finding identified",
            "Applicable rules produced no qualifying negative finding.",
        )
    return (
        "Evaluation incomplete",
        (
            "No attention finding is asserted because applicable "
            "requirements remain unassessed."
        ),
    )


def _ordered(findings: list[Finding]) -> list[Finding]:
    return sorted(
        findings,
        key=lambda item: (
            item.display_priority,
            item.category.value,
            item.finding_type,
            item.finding_id,
        ),
    )


def _supervisor_order(
    findings: list[Finding],
    selected_ids: list[str],
) -> list[Finding]:
    by_id = {item.finding_id: item for item in findings}
    selected = [
        by_id[finding_id]
        for finding_id in selected_ids
        if finding_id in by_id
    ]
    selected_set = {item.finding_id for item in selected}
    return selected + [
        item
        for item in _ordered(findings)
        if item.finding_id not in selected_set
    ]


def _positive_order(findings: list[Finding]) -> list[Finding]:
    category_rank = {
        FindingCategory.OUTCOME: 0,
        FindingCategory.REQUIRED_CONTROL: 1,
        FindingCategory.PROCESS: 2,
        FindingCategory.AGENT_BEHAVIOR: 3,
        FindingCategory.ESCALATION: 4,
        FindingCategory.DATA_QUALITY: 5,
    }
    return sorted(
        findings,
        key=lambda item: (
            category_rank[item.category],
            item.display_priority,
            item.finding_id,
        ),
    )


def _supervisor_positive_order(
    findings: list[Finding],
    selected_ids: list[str],
) -> list[Finding]:
    by_id = {item.finding_id: item for item in findings}
    selected = [
        by_id[finding_id]
        for finding_id in selected_ids
        if finding_id in by_id
    ]
    selected_set = {item.finding_id for item in selected}
    return selected + [
        item
        for item in _positive_order(findings)
        if item.finding_id not in selected_set
    ]


def _verify_supervisor(
    decision: CallDecision,
    supervisor: SupervisorResult | None,
) -> None:
    if supervisor is None:
        return
    lock = supervisor.decision_lock
    if (
        supervisor.call_id != decision.call_id
        or lock.call_id != decision.call_id
        or lock.decision_sha256 != decision_sha256(decision)
    ):
        raise ValueError(
            "supervisor result does not match the deterministic decision"
        )


def _evidence_kind(evidence: EvidenceRef) -> EvidenceKind:
    if evidence.modality in (Modality.TRANSCRIPT, Modality.TEXT):
        return EvidenceKind.TRANSCRIPT
    if evidence.modality in (Modality.ACOUSTIC, Modality.MULTIMODAL):
        return EvidenceKind.AUDIO_SUPPORT
    return EvidenceKind.BUSINESS_CONTEXT


def _selected_evidence(
    finding: Finding,
    supervisor_ids: set[str],
    limit: int,
) -> list[EvidenceRef]:
    ranked = sorted(
        finding.evidence,
        key=lambda item: (
            item.evidence_id not in supervisor_ids,
            _evidence_kind(item) == EvidenceKind.AUDIO_SUPPORT,
            item.start_seconds is None,
            item.start_seconds or 0.0,
            item.evidence_id,
        ),
    )
    selected: list[EvidenceRef] = []
    kinds: set[EvidenceKind] = set()
    for evidence in ranked:
        kind = _evidence_kind(evidence)
        if kind in kinds:
            continue
        selected.append(evidence)
        kinds.add(kind)
        if len(selected) == limit:
            break
    if len(selected) < limit:
        selected_ids = {item.evidence_id for item in selected}
        remaining = [
            item
            for item in ranked
            if item.evidence_id not in selected_ids
        ]
        selected.extend(remaining[: limit - len(selected)])
    return selected


def _finding_outcome(finding: Finding) -> FindingOutcome:
    if finding.polarity.value == "positive":
        return FindingOutcome.EFFECTIVE
    verdict = next(
        (
            threshold.value
            for threshold in finding.detection_rule.thresholds
            if threshold.signal_name == "assessment.requirement_verdict"
        ),
        None,
    )
    if verdict == "incorrect":
        return FindingOutcome.INCORRECT
    if verdict == "missed":
        return FindingOutcome.MISSED
    if finding.category == FindingCategory.ESCALATION:
        return FindingOutcome.OBSERVED_CONCERN
    return FindingOutcome.INCORRECT


def _presented_finding(
    finding: Finding,
    evidence: list[EvidenceRef],
) -> PresentedFinding:
    reliability_note = None
    if finding.reliability.status == ReliabilityStatus.LIMITED:
        reliability_note = "Supporting evidence is limited."
    return PresentedFinding(
        finding_id=finding.finding_id,
        category=finding.category,
        category_label=_CATEGORY_LABELS[finding.category],
        title=finding.title,
        summary=finding.summary,
        evidence_ids=[item.evidence_id for item in evidence],
        counter_evidence_count=len(finding.counter_evidence),
        reliability_note=reliability_note,
        outcome=_finding_outcome(finding),
    )


def _presented_evidence(
    evidence: EvidenceRef,
    finding: Finding,
    purpose: EvidencePurpose,
) -> PresentedEvidence:
    kind = _evidence_kind(evidence)
    return PresentedEvidence(
        evidence_id=evidence.evidence_id,
        finding_ids=[finding.finding_id],
        purposes=[purpose],
        kind=kind,
        speaker=evidence.speaker,
        start_seconds=evidence.start_seconds,
        end_seconds=evidence.end_seconds,
        text=evidence.quote or evidence.observation,
        seekable=evidence.start_seconds is not None,
        supporting_only=kind == EvidenceKind.AUDIO_SUPPORT,
    )


def _merge_evidence(
    items: list[PresentedEvidence],
    limit: int | None = None,
) -> list[PresentedEvidence]:
    merged: dict[str, PresentedEvidence] = {}
    for item in items:
        existing = merged.get(item.evidence_id)
        if existing is None:
            merged[item.evidence_id] = item
            continue
        existing.finding_ids = sorted(
            set(existing.finding_ids + item.finding_ids)
        )
        existing.purposes = sorted(
            set(existing.purposes + item.purposes),
            key=lambda value: value.value,
        )
    ordered = sorted(
        merged.values(),
        key=lambda item: (
            item.start_seconds is None,
            item.start_seconds or 0.0,
            item.evidence_id,
        ),
    )
    return ordered if limit is None else ordered[:limit]


def _present_action(decision: CallDecision) -> PresentedAction | None:
    action = decision.recommended_action
    if action.action_type == ActionType.NONE:
        return None
    execution_label = (
        "Automatic manager notification"
        if action.execution == ActionExecution.AUTOMATIC
        else "Manager approval required before email delivery"
    )
    return PresentedAction(
        action_type=action.action_type,
        execution=action.execution,
        label=action.label,
        execution_label=execution_label,
        basis_finding_ids=action.finding_ids,
        automation_allowed=action.automation_allowed,
        requires_human_approval=action.requires_human_approval,
        audience=(
            "customer"
            if action.action_type == ActionType.CUSTOMER_FOLLOW_UP
            else "manager"
        ),
    )


def _checklist(
    decision: CallDecision,
    promoted_ids: set[str],
    evidence_preferences: set[str],
) -> list[PresentedCheck]:
    findings = [
        item
        for item in (
            decision.triggered_findings + decision.positive_findings
        )
        if item.detection_rule.detector
        == "structured_requirement_assessment"
        and item.visibility != Visibility.INTERNAL
    ]
    rows = []
    for finding in _ordered(findings):
        outcome = _finding_outcome(finding)
        status = {
            FindingOutcome.EFFECTIVE: CheckStatus.DEMONSTRATED,
            FindingOutcome.INCORRECT: CheckStatus.INCORRECT,
            FindingOutcome.MISSED: CheckStatus.NOT_DEMONSTRATED,
            FindingOutcome.OBSERVED_CONCERN: (
                CheckStatus.UNABLE_TO_DETERMINE
            ),
        }[outcome]
        evidence = _selected_evidence(
            finding,
            evidence_preferences,
            limit=1,
        )
        rows.append(
            PresentedCheck(
                requirement_id=finding.applicability.rule_id,
                title=finding.title,
                category=finding.category,
                status=status,
                summary=finding.summary,
                evidence_ids=[item.evidence_id for item in evidence],
                promoted=finding.finding_id in promoted_ids,
            )
        )
    return rows


def _signal_value(bundle: SignalBundle, name: str):
    return next(
        (
            signal.value
            for signal in bundle.signals
            if signal.name == name
        ),
        None,
    )


def _acoustic_context(
    bundle: SignalBundle | None,
) -> AcousticContext:
    if bundle is None:
        return AcousticContext(
            status=AcousticStatus.UNAVAILABLE,
            coverage_label="Audio support unavailable",
            conclusion=(
                "The transcript was evaluated without acoustic support."
            ),
        )
    coverage = next(
        (
            item
            for item in bundle.coverage
            if item.modality == Modality.ACOUSTIC
            and item.source == "audio_features"
        ),
        None,
    )
    if coverage is None:
        coverage = next(
            (
                item
                for item in bundle.coverage
                if item.modality == Modality.ACOUSTIC
                and item.source == "emotion_model"
            ),
            None,
        )
    if coverage is None or coverage.usable_units == 0:
        return AcousticContext(
            status=AcousticStatus.UNAVAILABLE,
            coverage_label="Audio support unavailable",
            conclusion=(
                "No reliable acoustic observations were available for this "
                "call."
            ),
        )

    status = (
        AcousticStatus.AVAILABLE
        if coverage.coverage_ratio >= 0.5
        else AcousticStatus.LIMITED
    )
    observations = [
        AcousticObservation(
            observation_id=episode.episode_id,
            label="Sustained customer vocal strain",
            summary=(
                "Multiple acoustic cues rose together during this interval. "
                "This supports review of the surrounding conversation but "
                "does not establish an agent failure on its own."
            ),
            start_seconds=episode.start_seconds,
            end_seconds=episode.end_seconds,
        )
        for episode in bundle.episodes
        if (
            episode.episode_type == "acoustic.customer_elevation"
            and episode.speaker == Speaker.CUSTOMER
            and episode.reliability.status
            not in (
                ReliabilityStatus.UNUSABLE,
                ReliabilityStatus.UNAVAILABLE,
            )
        )
    ]
    direction = _signal_value(
        bundle,
        "acoustic.dynamics.trajectory_direction",
    )
    unresolved = _signal_value(
        bundle,
        "acoustic.dynamics.unresolved_end_candidate",
    )
    if unresolved:
        conclusion = (
            "Customer vocal strain remained elevated near the end of the "
            "call. Read this with the transcript context."
        )
    elif direction == "decreasing":
        conclusion = (
            "Customer vocal strain eased over the call; the audio supports "
            "a recovery pattern."
        )
    elif observations:
        conclusion = (
            "The audio contains sustained customer-strain intervals that "
            "should be read with the transcript."
        )
    else:
        conclusion = (
            "No sustained vocal-friction pattern crossed the provisional "
            "support gate."
        )
    return AcousticContext(
        status=status,
        coverage_label=(
            f"Audio support on {coverage.usable_units} of "
            f"{coverage.expected_units} segments"
        ),
        conclusion=conclusion,
        observations=observations[:3],
    )


def _finding_evidence_ids(findings: list[Finding]) -> list[str]:
    return list(dict.fromkeys(
        evidence.evidence_id
        for finding in findings
        for evidence in finding.evidence
    ))[:3]


def _question_summary(
    findings: list[Finding],
    fallback: str,
) -> str:
    if not findings:
        return fallback
    return findings[0].summary


def _manager_questions(
    decision: CallDecision,
    _acoustic: AcousticContext,
) -> list[ManagerQuestion]:
    all_findings = (
        decision.triggered_findings + decision.positive_findings
    )
    request_ids = {
        "request.intent_confirmed",
        "loan.purpose_and_stage_confirmed",
    }
    request = [
        item
        for item in all_findings
        if item.applicability.rule_id in request_ids
    ]
    process = [
        item
        for item in all_findings
        if (
            item.detection_rule.detector
            == "structured_requirement_assessment"
            and item.applicability.rule_id not in request_ids
            and item.category
            not in (
                FindingCategory.OUTCOME,
                FindingCategory.AGENT_BEHAVIOR,
            )
        )
    ]
    experience = [
        item
        for item in all_findings
        if (
            item.category == FindingCategory.ESCALATION
            or item.finding_type.startswith("experience.")
            or (
                item.category == FindingCategory.AGENT_BEHAVIOR
                and item.detection_rule.detector
                == "structured_requirement_assessment"
            )
        )
    ]
    outcome = [
        item
        for item in all_findings
        if item.category == FindingCategory.OUTCOME
    ]

    def answer(
        findings: list[Finding],
        *,
        clear_when_empty: bool = False,
    ) -> tuple[ManagerAnswer, str]:
        negative = [
            item
            for item in findings
            if item.polarity.value == "negative"
        ]
        positive = [
            item
            for item in findings
            if item.polarity.value == "positive"
        ]
        if any(
            item.severity.value == "critical"
            for item in negative
        ):
            return ManagerAnswer.NO, "No"
        if negative:
            return ManagerAnswer.PARTLY, "Partly"
        if positive or (
            clear_when_empty
            and decision.decision_status == DecisionStatus.COMPLETE
        ):
            return ManagerAnswer.YES, "Yes"
        return ManagerAnswer.UNCLEAR, "Unable to determine"

    request_answer, request_label = answer(request)
    process_answer, process_label = answer(process)
    experience_answer, experience_label = answer(
        experience,
        clear_when_empty=True,
    )
    outcome_answer, outcome_label = answer(outcome)

    request_negative = [
        item for item in request if item.polarity.value == "negative"
    ]
    request_positive = [
        item for item in request if item.polarity.value == "positive"
    ]
    process_negative = [
        item for item in process if item.polarity.value == "negative"
    ]
    experience_negative = [
        item for item in experience if item.polarity.value == "negative"
    ]
    experience_positive = [
        item for item in experience if item.polarity.value == "positive"
    ]
    outcome_negative = [
        item for item in outcome if item.polarity.value == "negative"
    ]
    outcome_positive = [
        item for item in outcome if item.polarity.value == "positive"
    ]
    return [
        ManagerQuestion(
            question_id="call.request",
            question="Was the customer's request understood?",
            answer=request_answer,
            answer_label=request_label,
            summary=_question_summary(
                request_negative or request_positive,
                "The available evidence did not resolve the request.",
            ),
            evidence_ids=_finding_evidence_ids(request),
        ),
        ManagerQuestion(
            question_id="call.process",
            question="Was the applicable process followed?",
            answer=process_answer,
            answer_label=process_label,
            summary=(
                _question_summary(
                    process_negative,
                    "No applicable process checks were available.",
                )
                if process_negative
                else (
                    f"{len(process)} applicable process checks were "
                    "supported by transcript evidence."
                    if process
                    else "No applicable process checks were available."
                )
            ),
            evidence_ids=_finding_evidence_ids(process),
        ),
        ManagerQuestion(
            question_id="call.experience",
            question="Was the customer experience handled appropriately?",
            answer=experience_answer,
            answer_label=experience_label,
            summary=_question_summary(
                experience_negative or experience_positive,
                (
                    "No customer-experience concern requiring review was "
                    "identified in the transcript."
                ),
            ),
            evidence_ids=_finding_evidence_ids(experience),
        ),
        ManagerQuestion(
            question_id="call.outcome",
            question="Was the outcome clear and complete?",
            answer=outcome_answer,
            answer_label=outcome_label,
            summary=_question_summary(
                outcome_negative or outcome_positive,
                "The available evidence did not resolve the call outcome.",
            ),
            evidence_ids=_finding_evidence_ids(outcome),
        ),
    ]


def _completeness_notice(
    decision: CallDecision,
) -> CompletenessNotice | None:
    if decision.decision_status == DecisionStatus.COMPLETE:
        return None
    message = (
        "Some applicable requirements could not be assessed. Visible "
        "findings remain valid, but this is not a complete call assessment."
        if decision.decision_status == DecisionStatus.PARTIAL
        else (
            "Applicable requirements could not be assessed, so this call "
            "has not been cleared."
        )
    )
    return CompletenessNotice(
        status=decision.decision_status,
        message=message,
        affected_requirements=[
            item.message
            for item in decision.uncertainties
            if item.code.startswith("requirement.")
        ],
    )


def project_call_evaluation(
    decision: CallDecision,
    supervisor: SupervisorResult | None = None,
    bundle: SignalBundle | None = None,
) -> CallEvaluationView:
    """Project a decision into the manager-facing call evaluator contract."""
    _verify_supervisor(decision, supervisor)
    controlling_ids = set(
        decision.decision_trace.controlling_finding_ids
    )
    controlling = _ordered(
        [
            item
            for item in decision.triggered_findings
            if item.finding_id in controlling_ids
        ]
    )
    primary_source = controlling[:MAX_PRIMARY_REASONS]
    primary_ids = {item.finding_id for item in primary_source}

    visible_negative = [
        item
        for item in decision.triggered_findings
        if item.visibility != Visibility.INTERNAL
        and item.finding_id not in primary_ids
    ]
    visible_negative = _supervisor_order(
        visible_negative,
        supervisor.supporting_finding_ids if supervisor else [],
    )

    visible_positive = [
        item
        for item in decision.positive_findings
        if item.visibility != Visibility.INTERNAL
    ]
    visible_positive = _supervisor_positive_order(
        visible_positive,
        supervisor.positive_finding_ids if supervisor else [],
    )
    positive_source = visible_positive[:MAX_POSITIVE_HIGHLIGHTS]
    positive_ids = {item.finding_id for item in positive_source}
    promoted_ids = primary_ids | positive_ids
    evidence_preferences = set(
        supervisor.evidence_ids if supervisor else []
    )

    primary: list[PresentedFinding] = []
    positives: list[PresentedFinding] = []
    evidence_items: list[PresentedEvidence] = []
    for finding in primary_source:
        selected = _selected_evidence(
            finding,
            evidence_preferences,
            limit=2,
        )
        primary.append(_presented_finding(finding, selected))
        evidence_items.extend(
            _presented_evidence(
                item,
                finding,
                EvidencePurpose.REASON,
            )
            for item in selected
        )
    for finding in positive_source:
        selected = _selected_evidence(
            finding,
            evidence_preferences,
            limit=1,
        )
        positives.append(_presented_finding(finding, selected))
        evidence_items.extend(
            _presented_evidence(
                item,
                finding,
                EvidencePurpose.POSITIVE,
            )
            for item in selected
        )

    additional_negative = [
        _presented_finding(
            finding,
            _selected_evidence(
                finding,
                evidence_preferences,
                limit=1,
            ),
        )
        for finding in visible_negative
    ]
    additional_positive = [
        _presented_finding(
            finding,
            _selected_evidence(
                finding,
                evidence_preferences,
                limit=1,
            ),
        )
        for finding in visible_positive
        if finding.finding_id not in positive_ids
    ]
    detail_evidence_items: list[PresentedEvidence] = []
    for finding, presented in zip(
        visible_negative,
        additional_negative,
        strict=True,
    ):
        selected_ids = set(presented.evidence_ids)
        detail_evidence_items.extend(
            _presented_evidence(
                item,
                finding,
                EvidencePurpose.REASON,
            )
            for item in finding.evidence
            if item.evidence_id in selected_ids
        )
    additional_positive_source = [
        item
        for item in visible_positive
        if item.finding_id not in positive_ids
    ]
    for finding, presented in zip(
        additional_positive_source,
        additional_positive,
        strict=True,
    ):
        selected_ids = set(presented.evidence_ids)
        detail_evidence_items.extend(
            _presented_evidence(
                item,
                finding,
                EvidencePurpose.POSITIVE,
            )
            for item in finding.evidence
            if item.evidence_id in selected_ids
        )
    primary_evidence = _merge_evidence(
        evidence_items,
        limit=MAX_EVIDENCE_ITEMS,
    )
    primary_evidence_ids = {
        item.evidence_id for item in primary_evidence
    }
    detail_evidence = [
        item
        for item in _merge_evidence(detail_evidence_items)
        if item.evidence_id not in primary_evidence_ids
    ]
    headline, summary = _copy(decision)
    acoustic = _acoustic_context(bundle)
    return CallEvaluationView(
        presentation_version=PRESENTATION_VERSION,
        call_id=decision.call_id,
        decision_sha256=decision_sha256(decision),
        state=_state(decision),
        evaluation_status=decision.decision_status,
        attention_required=decision.attention_required,
        headline=headline,
        summary=summary,
        manager_questions=_manager_questions(decision, acoustic),
        checklist=_checklist(
            decision,
            promoted_ids,
            evidence_preferences,
        ),
        acoustic_context=acoustic,
        primary_reasons=primary,
        additional_reason_count=len(additional_negative),
        positive_highlights=positives,
        additional_positive_count=len(additional_positive),
        recommended_action=_present_action(decision),
        evidence=primary_evidence,
        completeness_notice=_completeness_notice(decision),
        details=PresentationDetails(
            additional_findings=additional_negative,
            additional_positive_findings=additional_positive,
            evidence=detail_evidence,
            uncertainty_messages=[
                item.message
                for item in decision.uncertainties
                if item.visibility != Visibility.INTERNAL
            ],
            supervisor_context_note=(
                supervisor.context_note if supervisor else None
            ),
            evaluator_version=decision.evaluator_version,
            domain_profile_id=decision.domain_profile_id,
            decision_policy_id=decision.decision_trace.policy_id,
            decision_policy_version=(
                decision.decision_trace.policy_version
            ),
            supervisor_fallback_used=(
                supervisor.fallback_used if supervisor else None
            ),
        ),
        provenance=_source(),
    )


def check_presentation(
    view: CallEvaluationView,
) -> PresentationCheck:
    payload = view.model_dump(mode="json")
    serialized_keys: list[str] = []

    def collect_keys(value):
        if isinstance(value, dict):
            for key, child in value.items():
                serialized_keys.append(key)
                collect_keys(child)
        elif isinstance(value, list):
            for child in value:
                collect_keys(child)

    collect_keys(payload)
    forbidden_fragments = ("score", "percentage", "rating", "confidence")
    no_score_fields = not any(
        fragment in key.lower()
        for key in serialized_keys
        for fragment in forbidden_fragments
    )
    visible_ids = [
        item.finding_id
        for item in (
            view.primary_reasons
            + view.positive_highlights
            + view.details.additional_findings
            + view.details.additional_positive_findings
        )
    ]
    evidence_ids = {item.evidence_id for item in view.evidence}
    all_evidence_ids = [
        item.evidence_id
        for item in view.evidence + view.details.evidence
    ]
    reasons_have_evidence = all(
        set(item.evidence_ids).issubset(evidence_ids)
        and bool(item.evidence_ids)
        for item in view.primary_reasons
    )
    action_is_unambiguous = (
        view.recommended_action is not None
    ) == view.attention_required
    completeness_is_explicit = (
        view.evaluation_status == DecisionStatus.COMPLETE
        and view.completeness_notice is None
    ) or (
        view.evaluation_status != DecisionStatus.COMPLETE
        and view.completeness_notice is not None
    )
    checks = {
        "attention_is_literal": (
            view.state.value
            in {
                "needs_attention",
                "no_attention_finding",
                "evaluation_incomplete",
            }
        ),
        "reasons_are_bounded": (
            len(view.primary_reasons) <= MAX_PRIMARY_REASONS
        ),
        "reasons_have_evidence": reasons_have_evidence,
        "positives_are_bounded": (
            len(view.positive_highlights)
            <= MAX_POSITIVE_HIGHLIGHTS
        ),
        "action_is_unambiguous": action_is_unambiguous,
        "completeness_is_explicit": completeness_is_explicit,
        "content_is_deduplicated": (
            len(visible_ids) == len(set(visible_ids))
            and len(all_evidence_ids) == len(set(all_evidence_ids))
        ),
        "no_score_fields": no_score_fields,
    }
    return PresentationCheck(
        **checks,
        passed=all(checks.values()),
    )


def scenario_result(
    view: CallEvaluationView,
) -> PresentationScenarioResult:
    return PresentationScenarioResult(
        call_id=view.call_id,
        answer_key=ScenarioAnswerKey(
            attention_state=view.state,
            reasons=[item.title for item in view.primary_reasons],
            evidence_locations_seconds=[
                item.start_seconds
                for item in view.evidence
                if item.start_seconds is not None
            ],
            positive_handling=[
                item.title for item in view.positive_highlights
            ],
            action=(
                view.recommended_action.label
                if view.recommended_action
                else None
            ),
            completeness=view.evaluation_status,
        ),
        check=check_presentation(view),
    )