src/ad_creative_env/hm/validation.py

"""Zero-network validation matrix for environment checks, routing, and API equivalence."""

from __future__ import annotations

import json
from dataclasses import asdict
from pathlib import Path
from typing import Any
from uuid import uuid4

from .generation import DeterministicCopyGenerator
from .scenarios import ScenarioRepository
from .test_agent import DeterministicTestAgent, EnvironmentTestRunner


class ValidationMatrixError(RuntimeError):
    pass


class _ClientTransport:
    def __init__(self, client) -> None:
        self._client = client

    def get(self, path):
        response = self._client.get(path)
        return response.status_code, response.json()

    def post(self, path, payload):
        response = self._client.post(path, json=payload)
        return response.status_code, response.json()


def _episode_summary(episode: dict[str, Any]) -> dict[str, Any]:
    return {
        "scenario_id": episode["scenario_id"],
        "status": episode["status"],
        "reward": episode["reward"],
        "failed_checks": [
            check["check_id"] for check in episode["checks"] if not check["passed"]
        ],
        "action_mode": episode["action_provenance"]["execution_mode"],
        "judge_mode": episode["judge_provenance"]["execution_mode"],
    }


def _core_result(result: dict[str, Any]) -> dict[str, Any]:
    excluded = {"safe_stages", "action_provenance"}
    return {key: value for key, value in result.items() if key not in excluded}


def run_zero_network_validation(client, repository: ScenarioRepository) -> dict[str, Any]:
    """Run deterministic contract and equivalence lanes; never creates a live adapter."""

    transport = _ClientTransport(client)
    scenarios = repository.list_scenarios()
    ids = tuple(scenario.scenario_id for scenario in scenarios)
    good_report = EnvironmentTestRunner(transport, DeterministicTestAgent("good")).run(ids)

    violations = []
    scenario_id = ids[0]
    identity = {
        "provider": "local",
        "model": "intentional-invalid-v1",
        "config_version": "matrix-v1",
        "execution_mode": "deterministic_test",
    }
    invalid_cases = {
        "copy.structure": {
            "headline": "Invalid extra field",
            "body": "Schema rejection is expected.",
            "cta": None,
            "extra": "not allowed",
        },
        "copy.required": {"headline": "Missing body", "cta": None},
    }
    for rule, action in invalid_cases.items():
        response = client.post(
            "/api/evaluations",
            json={"scenario_id": scenario_id, "action": action, "agent_identity": identity},
        )
        payload = response.json()
        violations.append(
            {
                "policy_rule": rule,
                "expected": "request_rejected",
                "observed_status_code": response.status_code,
                "observed_code": payload.get("detail", {}).get("code"),
                "passed": response.status_code == 422,
            }
        )

    variants = {
        "privacy.disclosure": "privacy_disclosure",
        "claims.price": "forbidden_price",
        "claims.discount": "forbidden_discount",
        "claims.delivery": "forbidden_delivery",
        "claims.unsupported": "unsupported_material",
        "product.attributes": "wrong_attribute",
    }
    for rule, variant in variants.items():
        report = EnvironmentTestRunner(transport, DeterministicTestAgent(variant)).run(
            (scenario_id,)
        )
        episode = report["episodes"][0]
        result = episode["result"]
        failed = [check["check_id"] for check in episode["checks"] if not check["passed"]]
        expected_auto_fail = rule == "privacy.disclosure"
        passed = rule in failed and (
            (result["reward"] == 0 and result["review_status"] == "failed")
            if expected_auto_fail
            else result["applied_cap"] == 0.5 and result["reward"] <= 0.5
        )
        violations.append(
            {
                "policy_rule": rule,
                "expected": "auto_fail" if expected_auto_fail else "score_cap_0.5",
                "observed_status": result["review_status"],
                "observed_reward": result["reward"],
                "observed_cap": result["applied_cap"],
                "failed_checks": failed,
                "passed": passed,
            }
        )

    equivalence = []
    generator = DeterministicCopyGenerator()
    for scenario in scenarios:
        generated = generator.generate(
            scenario.scenario_id, scenario.to_observation(), scenario.provenance
        )
        demo = client.post(
            "/api/episodes", json={"scenario_id": scenario.scenario_id}
        ).json()
        stream_response = client.post(
            "/api/episodes/stream", json={"scenario_id": scenario.scenario_id}
        )
        stream_events = [json.loads(line) for line in stream_response.text.splitlines()]
        streamed = stream_events[-1].get("result")
        evaluated_response = client.post(
            "/api/evaluations",
            json={
                "scenario_id": scenario.scenario_id,
                "action": asdict(generated.action),
                "agent_identity": {
                    **generated.identity.as_dict(),
                    "execution_mode": generated.execution_mode.value,
                },
            },
        )
        evaluated = evaluated_response.json()
        streaming_match = streamed == demo
        programmatic_match = (
            evaluated_response.status_code == 200
            and _core_result(evaluated) == _core_result(demo)
        )
        equivalence.append(
            {
                "scenario_id": scenario.scenario_id,
                "streaming_matches_demo": streaming_match,
                "evaluation_core_matches_demo": programmatic_match,
                "passed": streaming_match and programmatic_match,
            }
        )

    contract_passed = all(item["passed"] for item in violations)
    equivalence_passed = all(item["passed"] for item in equivalence)
    return {
        "kind": "synthetic-validation-matrix",
        "version": 1,
        "network_requests": 0,
        "paid_requests": 0,
        "training_performed": False,
        "lanes": {
            "zero_network_contract": {
                "status": "passed" if contract_passed else "failed",
                "known_good": [_episode_summary(item) for item in good_report["episodes"]],
                "intentional_violations": violations,
            },
            "capped_live": {
                "status": "not_executed",
                "maximum_requests_if_approved": 6,
                "reason": "Provider, model, synthetic data exposure, and spend were not approved.",
            },
            "ui_api_equivalence": {
                "status": "passed" if equivalence_passed else "failed",
                "scenarios": equivalence,
            },
        },
        "passed": contract_passed and equivalence_passed,
    }


def write_validation_matrix(path: Path, matrix: dict[str, Any]) -> Path:
    output = Path(path)
    if matrix.get("kind") != "synthetic-validation-matrix":
        raise ValidationMatrixError("validation matrix shape is invalid")
    output.parent.mkdir(parents=True, exist_ok=True)
    temporary = output.with_name(f".{output.name}.{uuid4().hex}.tmp")
    try:
        temporary.write_text(
            json.dumps(matrix, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )
        temporary.replace(output)
    finally:
        temporary.unlink(missing_ok=True)
    return output