awoooi/apps/api/src/services/agent_replay_contract.py

"""
Agent Replay Contract Validator
===============================

Validates that candidate replay outputs line up with candidate-visible replay
inputs before they are normalized and scored.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

from src.services.agent_replay_normalizer import CandidateReplayResult

LABEL_LEAK_KEYS = {
    "evaluation_labels",
    "verification_result",
    "execution_success",
    "execution_error",
    "self_healing_score",
}


@dataclass(frozen=True)
class AgentReplayContractReport:
    """Validation result for one candidate replay output batch."""

    candidate_id: str | None
    inputs: int
    results: int
    valid: bool
    failures: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema_version": "agent_replay_contract_report_v1",
            "candidate_id": self.candidate_id,
            "inputs": self.inputs,
            "results": self.results,
            "valid": self.valid,
            "failures": list(self.failures),
        }


def validate_candidate_replay_contract(
    *,
    candidate_inputs: list[dict[str, Any]],
    candidate_results: list[dict[str, Any]],
    expected_candidate_id: str | None = None,
) -> AgentReplayContractReport:
    """Validate result/input one-to-one alignment and answer-key isolation."""
    failures: list[str] = []
    input_index = _index_inputs(candidate_inputs, failures)
    result_index = _index_results(candidate_results, failures)

    input_ids = set(input_index)
    result_ids = set(result_index)
    missing = sorted(input_ids - result_ids)
    extra = sorted(result_ids - input_ids)
    if missing:
        failures.append(f"missing_results:{','.join(missing)}")
    if extra:
        failures.append(f"unexpected_results:{','.join(extra)}")

    candidate_ids = {
        result.candidate_id
        for result in result_index.values()
        if result.candidate_id
    }
    if expected_candidate_id and candidate_ids != {expected_candidate_id}:
        failures.append(
            "candidate_id_mismatch:"
            f"expected={expected_candidate_id};actual={','.join(sorted(candidate_ids))}"
        )
    elif not expected_candidate_id and len(candidate_ids) > 1:
        failures.append(f"multiple_candidate_ids:{','.join(sorted(candidate_ids))}")

    for incident_id in sorted(input_ids & result_ids):
        expected_run_id = str(input_index[incident_id].get("run_id", ""))
        actual_run_id = result_index[incident_id].run_id
        if expected_run_id != actual_run_id:
            failures.append(
                f"run_id_mismatch:{incident_id}:expected={expected_run_id};actual={actual_run_id}"
            )

    for line_number, payload in enumerate(candidate_results, start=1):
        leaked = sorted(_find_label_leaks(payload))
        if leaked:
            failures.append(
                f"label_leak:result_line_{line_number}:{','.join(leaked)}"
            )

    candidate_id = expected_candidate_id
    if candidate_id is None and len(candidate_ids) == 1:
        candidate_id = next(iter(candidate_ids))

    return AgentReplayContractReport(
        candidate_id=candidate_id,
        inputs=len(candidate_inputs),
        results=len(candidate_results),
        valid=not failures,
        failures=failures,
    )


def _index_inputs(
    candidate_inputs: list[dict[str, Any]],
    failures: list[str],
) -> dict[str, dict[str, Any]]:
    indexed: dict[str, dict[str, Any]] = {}
    for line_number, payload in enumerate(candidate_inputs, start=1):
        incident_id = str(payload.get("incident_id", "")).strip()
        run_id = str(payload.get("run_id", "")).strip()
        if not incident_id or not run_id:
            failures.append(f"invalid_input:line_{line_number}:missing_incident_or_run_id")
            continue
        if incident_id in indexed:
            failures.append(f"duplicate_input:{incident_id}")
            continue
        indexed[incident_id] = payload
    return indexed


def _index_results(
    candidate_results: list[dict[str, Any]],
    failures: list[str],
) -> dict[str, CandidateReplayResult]:
    indexed: dict[str, CandidateReplayResult] = {}
    for line_number, payload in enumerate(candidate_results, start=1):
        try:
            result = CandidateReplayResult.from_dict(payload)
        except Exception as exc:
            failures.append(f"invalid_result:line_{line_number}:{exc}")
            continue
        if result.incident_id in indexed:
            failures.append(f"duplicate_result:{result.incident_id}")
            continue
        indexed[result.incident_id] = result
    return indexed


def _find_label_leaks(
    value: Any,
    *,
    prefix: str = "",
) -> set[str]:
    found: set[str] = set()
    if isinstance(value, dict):
        for key, nested in value.items():
            key_text = str(key)
            path = f"{prefix}.{key_text}" if prefix else key_text
            if key_text in LABEL_LEAK_KEYS:
                found.add(path)
            found.update(_find_label_leaks(nested, prefix=path))
    elif isinstance(value, list):
        for index, nested in enumerate(value):
            path = f"{prefix}[{index}]"
            found.update(_find_label_leaks(nested, prefix=path))
    return found