video-image-production-build/tests/production/test_validator.py

from __future__ import annotations

import json
import tempfile
import unittest
from contextlib import contextmanager
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

from langchain_core.messages import AIMessage

from production_build_agents.production.validator import (
    ProductionValidatorOutputError,
    _build_candidate,
    _collect_evidence,
    _parse_assessment,
    _validate_candidate,
    build_production_validation_report,
    run_production_validator,
)
from production_build_agents.contracts.production_models import (
    ProductionInput,
    ProductionInputCatalog,
)
from production_build_agents.contracts.production_planning_models import (
    PlannedSegment,
    ProductionPlan,
)
from production_build_agents.production.segment_acceptance import (
    immutable_json_ref,
)
from tests.support.production_fixtures import (
    ProductionScenario,
    immutable_ref,
    production_input_id,
    readiness_records,
    text_sha256,
    validation_records,
)
from tests.support.fake_models import ToolAwareFakeChatModel


class ProductionValidatorRuntimeTest(unittest.TestCase):
    def setUp(self) -> None:
        case = ProductionScenario()
        self.plan = case.plan
        self.assembly = case.assembly
        self.delivery = case.assembly_delivery
        self.inspection, self.readiness = readiness_records(
            self.delivery,
            self.assembly,
        )
        self.catalog = ProductionInputCatalog(
            production_brief_uri="/runs/production-brief.json",
            production_brief_sha256="0" * 64,
            source_segment_ids=[
                item.source_segment_id for item in self.plan.segments
            ],
            inputs=[
                production_input
                for package in case.packages
                for production_input in package.production_inputs
            ],
        )
        self.candidate, _ = validation_records(self.delivery)

    def test_failed_readiness_stops_before_model_or_evidence_tools(
        self,
    ) -> None:
        failed = self.readiness.model_copy(update={"verdict": "FAIL"})
        with (
            tempfile.TemporaryDirectory() as temp_dir,
            patch(
                "production_build_agents.production.validator."
                "create_real_production_validator_model"
            ) as model_factory,
            patch(
                "production_build_agents.production.validator."
                "_collect_evidence"
            ) as collect,
        ):
            with self.assertRaisesRegex(
                ProductionValidatorOutputError,
                "Readiness 未通过",
            ):
                run_production_validator(
                    self.plan,
                    self.catalog,
                    self.assembly,
                    self.delivery,
                    self.inspection,
                    failed,
                    run_dir=Path(temp_dir),
                    assembly_delivery_ref=immutable_ref(
                        "assembly-delivery"
                    ),
                    preflight_report_ref=immutable_ref(
                        "readiness-report"
                    ),
                )
        model_factory.assert_not_called()
        collect.assert_not_called()

    def test_candidate_must_cover_exact_semantic_criteria_and_boundaries(
        self,
    ) -> None:
        missing = self.candidate.model_copy(
            update={"criterion_results": self.candidate.criterion_results[:-1]}
        )
        with self.assertRaisesRegex(
            ProductionValidatorOutputError,
            "criterion_results",
        ):
            _validate_candidate(
                missing,
                self.delivery,
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                preflight_report_ref=immutable_ref("readiness-report"),
            )

    def test_host_builds_candidate_identity_criteria_and_boundaries(
        self,
    ) -> None:
        assessment = {
            "criterion_results": [
                {
                    "verdict": item.verdict,
                    "evidence": item.evidence,
                    "reason": item.reason,
                }
                for item in self.candidate.criterion_results
            ],
            "boundary_results": [
                {
                    "verdict": item.verdict,
                    "evidence": item.evidence,
                    "reason": item.reason,
                }
                for item in self.candidate.boundary_results
            ],
            "summary": self.candidate.summary,
        }
        model = ToolAwareFakeChatModel(
            responses=[
                AIMessage(
                    content=json.dumps(assessment, ensure_ascii=False)
                )
            ]
        )
        assembly_ref = immutable_ref("assembly-delivery")
        readiness_ref = immutable_ref("readiness-report")
        with (
            tempfile.TemporaryDirectory() as temp_dir,
            patch(
                "production_build_agents.production.validator."
                "_collect_evidence",
                return_value=({}, []),
            ),
        ):
            result = run_production_validator(
                self.plan,
                self.catalog,
                self.assembly,
                self.delivery,
                self.inspection,
                self.readiness,
                run_dir=Path(temp_dir),
                assembly_delivery_ref=assembly_ref,
                preflight_report_ref=readiness_ref,
                model=model,
                tool_registry=MagicMock(),
            )

        self.assertEqual(result.run_id, self.delivery.run_id)
        self.assertEqual(result.plan_id, self.delivery.plan_id)
        self.assertEqual(result.plan_version, self.delivery.plan_version)
        self.assertEqual(result.assembly_delivery_ref, assembly_ref)
        self.assertEqual(result.preflight_report_ref, readiness_ref)
        self.assertEqual(
            [item.criterion for item in result.criterion_results],
            [
                "timeline_continuity",
                "audio_continuity",
                "subtitle_continuity",
                "visual_continuity",
                "semantic_content",
            ],
        )
        self.assertEqual(
            [
                (item.previous_segment_id, item.next_segment_id)
                for item in result.boundary_results
            ],
            [("Segment1", "Segment2")],
        )

    def test_legacy_checkpoint_candidate_is_rebound_by_host(self) -> None:
        legacy = self.candidate.model_copy(
            update={
                "run_id": "legacy-run",
                "plan_id": "LegacyPlan",
                "plan_version": 99,
                "assembly_delivery_ref": immutable_ref("legacy-assembly"),
                "preflight_report_ref": immutable_ref("legacy-readiness"),
            }
        )
        assessment = _parse_assessment(
            legacy.model_dump_json(),
            self.delivery,
        )
        result = _build_candidate(
            assessment,
            self.delivery,
            assembly_delivery_ref=immutable_ref("assembly-delivery"),
            preflight_report_ref=immutable_ref("readiness-report"),
        )

        self.assertEqual(result.run_id, self.delivery.run_id)
        self.assertEqual(result.plan_id, self.delivery.plan_id)
        self.assertEqual(result.plan_version, self.delivery.plan_version)
        self.assertEqual(
            result.assembly_delivery_ref,
            immutable_ref("assembly-delivery"),
        )
        self.assertEqual(
            result.preflight_report_ref,
            immutable_ref("readiness-report"),
        )

    def test_report_references_candidate_and_code_recomputes_verdict(
        self,
    ) -> None:
        failed_result = self.candidate.criterion_results[0].model_copy(
            update={"verdict": "FAIL", "reason": "语义失败"}
        )
        candidate = self.candidate.model_copy(
            update={
                "criterion_results": [
                    failed_result,
                    *self.candidate.criterion_results[1:],
                ]
            }
        )
        report = build_production_validation_report(
            self.assembly,
            self.delivery,
            self.inspection,
            self.readiness,
            candidate,
            assembly_delivery_ref=immutable_ref("assembly-delivery"),
            media_inspection_ref=immutable_ref("media-inspection"),
            preflight_report_ref=immutable_ref("readiness-report"),
            validator_candidate_ref=immutable_ref("validator-candidate"),
        )
        self.assertEqual(report.verdict, "FAIL")
        self.assertEqual(
            report.validator_candidate_ref,
            immutable_ref("validator-candidate"),
        )
        self.assertFalse(hasattr(report, "criterion_results"))

    def test_validator_uses_plan_version_in_agent_identity(
        self,
    ) -> None:
        plan = ProductionPlan(
            plan_id=self.plan.plan_id,
            plan_version=self.plan.plan_version,
            planning_package_ref=immutable_ref("planning-v2"),
            segments=[
                PlannedSegment(
                    **segment.model_dump(),
                )
                for segment in self.plan.segments
            ],
            timeline=self.plan.timeline,
            regenerate_segment_ids=[],
            summary="0.7 DAG",
        )

        @contextmanager
        def checkpointer():
            yield MagicMock()

        with (
            tempfile.TemporaryDirectory() as temp_dir,
            patch(
                "production_build_agents.production.validator."
                "create_default_tool_registry",
                return_value=MagicMock(),
            ) as create_registry,
            patch(
                "production_build_agents.production.validator."
                "_collect_evidence",
                return_value=({}, []),
            ),
            patch(
                "production_build_agents.production.validator."
                "create_sqlite_checkpointer",
                side_effect=lambda *_args, **_kwargs: checkpointer(),
            ),
            patch(
                "production_build_agents.production.validator.create_agent",
                return_value=MagicMock(),
            ),
            patch(
                "production_build_agents.production.validator."
                "_candidate_from_agent",
                return_value=(self.candidate, {}),
            ) as candidate_from_agent,
            patch(
                "production_build_agents.production.validator."
                "record_agent_messages",
            ),
        ):
            result = run_production_validator(
                plan,
                self.catalog,
                self.assembly,
                self.delivery,
                self.inspection,
                self.readiness,
                run_dir=Path(temp_dir),
                assembly_delivery_ref=immutable_ref(
                    "assembly-delivery"
                ),
                preflight_report_ref=immutable_ref(
                    "readiness-report"
                ),
                model=MagicMock(),
            )

        self.assertEqual(result, self.candidate)
        expected_run_id = (
            "production-run:production-validator:"
            f"v{self.delivery.plan_version}"
        )
        self.assertEqual(
            candidate_from_agent.call_args.kwargs["validator_run_id"],
            expected_run_id,
        )
        self.assertEqual(
            create_registry.call_args.kwargs["output_dir"],
            Path(temp_dir).resolve()
            / "production_tool_outputs"
            / f"v{self.delivery.plan_version}",
        )
        self.assertEqual(
            create_registry.call_args.kwargs["executor_run_id"],
            f"{expected_run_id}:evidence",
        )

    def test_evidence_includes_full_inputs_unassigned_requirements_reports_and_shot_frames(
        self,
    ) -> None:
        unassigned = ProductionInput(
            input_id=production_input_id(99),
            source_path="$.制作表.全局要求.未分配",
            content_sha256=text_sha256("未分配要求"),
            value={"要求": "片尾必须出现完整行动号召"},
            applies_to_source_segment_ids=["段落1"],
        )
        catalog = self.catalog.model_copy(
            update={"inputs": [*self.catalog.inputs, unassigned]}
        )
        first_segment = self.plan.segments[0].model_copy(
            update={
                "production_input_ids": [
                    *self.plan.segments[0].production_input_ids,
                    unassigned.input_id,
                ]
            }
        )
        plan = self.plan.model_copy(
            update={"segments": [first_segment, *self.plan.segments[1:]]}
        )

        with tempfile.TemporaryDirectory() as temporary:
            report_dir = Path(temporary)
            accepted = []
            for report, reference in zip(
                ProductionScenario().reports,
                self.assembly.accepted_segments,
            ):
                path = report_dir / f"{report.segment_id}.json"
                path.write_text(report.model_dump_json(), encoding="utf-8")
                accepted.append(
                    reference.model_copy(
                        update={
                            "validation_report_ref": immutable_json_ref(path)
                        }
                    )
                )
            assembly = self.assembly.model_copy(
                update={"accepted_segments": accepted}
            )
            registry = MagicMock()
            registry.resolve.return_value = [
                SimpleNamespace(name=name)
                for name in (
                    "transcribe_audio",
                    "inspect_ass_subtitles",
                    "extract_frames",
                    "view_images",
                )
            ]

            def evidence(tool, arguments):
                if tool.name == "transcribe_audio":
                    return {"text": "完整旁白"}
                if tool.name == "inspect_ass_subtitles":
                    return {"cues": [{"text": "完整字幕"}]}
                if tool.name == "extract_frames":
                    return {
                        "frames": [
                            {
                                "timestamp_sec": timestamp,
                                "local_path": (
                                    f"/tmp/frame-{index}.png"
                                ),
                            }
                            for index, timestamp in enumerate(
                                arguments["timestamps"]
                            )
                        ]
                    }
                raise AssertionError(tool.name)

            with (
                patch(
                    "production_build_agents.production.validator."
                    "invoke_evidence_tool",
                    side_effect=evidence,
                ) as invoke,
                patch(
                    "production_build_agents.production.validator."
                    "view_image_evidence",
                    return_value=[{"type": "image_url"}],
                ),
            ):
                payload, image_blocks = _collect_evidence(
                    plan,
                    catalog,
                    assembly,
                    self.delivery,
                    self.inspection,
                    self.readiness,
                    registry=registry,
                    assembly_delivery_ref=immutable_ref(
                        "assembly-delivery"
                    ),
                    preflight_report_ref=immutable_ref(
                        "readiness-report"
                    ),
                )

        input_context = payload["production_input_context"]
        self.assertEqual(
            input_context["unassigned_to_shots"],
            [unassigned.input_id],
        )
        self.assertEqual(
            input_context["inputs"][-1]["value"],
            unassigned.value,
        )
        self.assertEqual(
            [item["segment_id"] for item in payload[
                "segment_validation_summaries"
            ]],
            ["Segment1", "Segment2"],
        )
        extract_call = next(
            call
            for call in invoke.call_args_list
            if call.args[0].name == "extract_frames"
        )
        self.assertEqual(
            extract_call.args[1]["timestamps"],
            [0.0, 2.5, 4.9, 5.1, 7.5, 9.95],
        )
        frame_purposes = [
            purpose
            for frame in payload["semantic_evidence"]["timeline_frames"]
            for purpose in frame["purposes"]
        ]
        self.assertEqual(
            [
                item["shot_id"]
                for item in frame_purposes
                if item["kind"] == "shot_midpoint"
            ],
            ["Segment1-Shot1", "Segment2-Shot1"],
        )
        self.assertEqual(image_blocks, [{"type": "image_url"}])

    def test_report_persists_same_versioned_validator_identity(
        self,
    ) -> None:
        validator_run_id = (
            "production-run:production-validator:"
            f"v{self.delivery.plan_version}"
        )
        report = build_production_validation_report(
            self.assembly,
            self.delivery,
            self.inspection,
            self.readiness,
            self.candidate,
            assembly_delivery_ref=immutable_ref("assembly-delivery"),
            media_inspection_ref=immutable_ref("media-inspection"),
            preflight_report_ref=immutable_ref("readiness-report"),
            validator_candidate_ref=immutable_ref("validator-candidate"),
            validator_run_id=validator_run_id,
        )
        self.assertEqual(report.validator_run_id, validator_run_id)


if __name__ == "__main__":
    unittest.main()