video-image-production-build/tests/agents/test_task_validator.py

from __future__ import annotations

import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from langchain_core.messages import AIMessage

from production_build_agents.capabilities import CAPABILITIES
from production_build_agents.agents.executor.context import load_task_contract
from production_build_agents.agents.validator.skills.registry import (
    ValidatorSkillError,
    load_validator_skill,
)
from production_build_agents.agents.validator.task_agent import (
    run_validator_agent,
)
from production_build_agents.agents.validator.task_context import (
    ValidatorContextError,
    build_validator_user_message,
)
from production_build_agents.contracts.evaluation import (
    task_expectation_pairs,
)
from production_build_agents.contracts.identifiers import validator_run_id_for
from production_build_agents.contracts.models import (
    Artifact,
    ArtifactExpectation,
    ArtifactExpectationBinding,
    CandidateArtifact,
    ExecutorCandidate,
    ExecutorDelivery,
    ToolCallRecord,
    ValidationCriterionResult,
    ValidatorCandidate,
)
from production_build_agents.tools.registry import ToolRegistry
from production_build_agents.run.artifacts import content_identity_for_path
from tests.support.fake_models import ToolAwareFakeChatModel
from tests.support.planner_fixtures import (
    build_planned_task,
    build_task_package,
)


BRIEF_PATH = Path(__file__).parents[1] / "fixtures" / "minimal_brief.json"


def _task(
    root: Path,
    *,
    skill_id: str = "structured-analysis",
    deliverable_type: str = "structured_data",
    importance: str = "critical",
):
    task = build_task_package(
        BRIEF_PATH,
        run_id="validator-test-run",
        planned_task=build_planned_task(
            skill_id=skill_id,
            deliverable_type=deliverable_type,
        ),
        plan_path=root / "plan.json",
    )
    if importance != "critical":
        _, plan, _ = load_task_contract(task)
        requirement = plan.stage_requirements[0].model_copy(
            update={"importance": importance}
        )
        Path(task.plan_uri).write_text(
            plan.model_copy(
                update={"stage_requirements": [requirement]}
            ).model_dump_json(indent=2),
            encoding="utf-8",
        )
    return task


def _delivery(
    root: Path,
    task,
    *,
    artifact_uri: str | None = None,
) -> ExecutorDelivery:
    _, plan, planned_task = load_task_contract(task)
    expectation = task_expectation_pairs(plan, planned_task)[0][1]
    manifest = root / "executor_candidate.json"
    materialized_artifact = root / "delivery_artifact.json"
    media = expectation.artifact_type in {"image", "video", "audio"}
    final_uri = artifact_uri or (
        f"https://example.test/final.{expectation.artifact_type}"
        if media
        else str(materialized_artifact)
    )
    candidate = ExecutorCandidate(
        run_id=task.run_id,
        plan_id=task.plan_id,
        task_id=task.task_id,
        plan_version=task.plan_version,
        deliverable_type=planned_task.deliverable_type,
        payload=(
            {"constraints": ["统一视觉"]}
            if planned_task.deliverable_type == "structured_data"
            else {}
        ),
        artifacts=(
            [
                CandidateArtifact(
                    artifact_type=expectation.artifact_type,
                    uri=final_uri,
                    description="最终测试产物",
                )
            ]
            if media
            else []
        ),
        findings=[],
        unresolved=[],
        summary="已完成当前 Task",
    )
    manifest.write_text(candidate.model_dump_json(), encoding="utf-8")
    if not media:
        materialized_artifact.write_text(
            json.dumps(
                {
                    "deliverable_type": planned_task.deliverable_type,
                    "payload": candidate.payload,
                    "findings": [],
                    "summary": candidate.summary,
                },
                ensure_ascii=False,
            ),
            encoding="utf-8",
        )
    artifact_id = (
        f"{task.task_id}-v{task.plan_version}-artifact-1"
    )
    evidence_required = bool(
        set(expectation.verification_capabilities).intersection(
            {
                "source_identity",
                "technical_integrity",
                "visual_content",
                "external_source",
            }
        )
    )
    tool_calls = (
        [
            ToolCallRecord(
                tool_call_id="evidence-call-1",
                tool_name="view_images",
                success=True,
                output_refs=[final_uri],
            )
        ]
        if evidence_required
        else []
    )
    return ExecutorDelivery(
        run_id=task.run_id,
        plan_id=task.plan_id,
        task_id=task.task_id,
        plan_version=task.plan_version,
        executor_run_id=f"executor-{task.task_id}-v{task.plan_version}-stable",
        skill_id=planned_task.skill_id,
        deliverable_type=planned_task.deliverable_type,
        manifest_uri=str(manifest),
        artifacts=[
            Artifact(
                artifact_id=artifact_id,
                artifact_type=expectation.artifact_type,
                uri=final_uri,
                **(
                    {}
                    if media
                    else content_identity_for_path(materialized_artifact)
                ),
                description="最终测试产物",
            )
        ],
        artifact_expectation_bindings=[
            ArtifactExpectationBinding(
                expectation_id=expectation.expectation_id,
                artifact_id=artifact_id,
                evidence_tool_call_ids=(
                    ["evidence-call-1"] if evidence_required else []
                ),
            )
        ],
        findings=[],
        unresolved=[],
        tool_calls=tool_calls,
        attempt_count=1,
        summary="已完成当前 Task",
    )


def _candidate(
    task,
    delivery: ExecutorDelivery,
    *,
    verdict: str = "PASS",
    expectation_id: str | None = None,
) -> ValidatorCandidate:
    _, plan, planned_task = load_task_contract(task)
    results = [
        ValidationCriterionResult(
            expectation_id=expectation_id or expectation.expectation_id,
            verification_capability=capability,
            verdict=verdict,
            evidence=[f"{expectation.expectation_id} 的可复查证据"],
            reason=f"{expectation.expectation_id} 检查完成",
        )
        for _, expectation in task_expectation_pairs(plan, planned_task)
        for capability in expectation.verification_capabilities
    ]
    return ValidatorCandidate(
        run_id=task.run_id,
        plan_id=task.plan_id,
        plan_version=task.plan_version,
        task_id=task.task_id,
        executor_run_id=delivery.executor_run_id,
        criterion_results=results,
        summary="逐项检查完成",
    )


def _model(*candidates: ValidatorCandidate) -> ToolAwareFakeChatModel:
    return ToolAwareFakeChatModel(
        responses=[
            AIMessage(content=candidate.model_dump_json())
            for candidate in candidates
        ]
    )


class ValidatorSkillTest(unittest.TestCase):
    def test_executor_skill_and_deliverable_map_to_one_validator(self) -> None:
        cases = (
            ("structured-analysis", "structured_data", "structured-validation"),
            ("external-research", "research_result", "research-validation"),
            ("image-production", "image", "image-validation"),
            ("video-production", "video", "video-validation"),
            (
                "reference-inspection",
                "reference_collection",
                "reference-validation",
            ),
        )
        for executor_skill, deliverable, validator_skill in cases:
            with self.subTest(executor_skill=executor_skill):
                skill = load_validator_skill(executor_skill, deliverable)
                self.assertEqual(skill.skill_id, validator_skill)

    def test_invalid_skill_deliverable_pair_is_rejected(self) -> None:
        with self.assertRaises(ValidatorSkillError):
            load_validator_skill("structured-analysis", "image")
        with self.assertRaises(ValidatorSkillError):
            load_validator_skill("unknown-skill", "document")


class GeneralValidatorTest(unittest.TestCase):
    def test_context_uses_plan_expectations_and_compact_output_contract(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            task = _task(root)
            delivery = _delivery(root, task)
            _, original_plan, _ = load_task_contract(task)
            requirement = original_plan.stage_requirements[0]
            sibling = ArtifactExpectation(
                expectation_id="Requirement1-Expectation2",
                artifact_type="document",
                minimum_count=1,
                usage_scope="兄弟 Task 的说明文档",
                verification_capabilities=["document_content"],
            )
            Path(task.plan_uri).write_text(
                original_plan.model_copy(
                    update={
                        "stage_requirements": [
                            requirement.model_copy(
                                update={
                                    "artifact_expectations": [
                                        *requirement.artifact_expectations,
                                        sibling,
                                    ]
                                }
                            )
                        ]
                    }
                ).model_dump_json(indent=2),
                encoding="utf-8",
            )
            brief, plan, planned_task = load_task_contract(task)
            Path(delivery.manifest_uri).unlink()
            message = build_validator_user_message(
                validator_run_id=validator_run_id_for(delivery),
                task=task,
                delivery=delivery,
                brief=brief,
                plan=plan,
                planned_task=planned_task,
                skill=load_validator_skill(
                    delivery.skill_id,
                    delivery.deliverable_type,
                ),
                tools=[],
            )
        payload = json.loads(message["content"][0]["text"])
        self.assertEqual(
            payload["validation_targets"][0]["expectation_id"],
            "Requirement1-Expectation1",
        )
        self.assertEqual(len(payload["validation_targets"]), 1)
        self.assertNotIn(
            "Requirement1-Expectation2",
            json.dumps(payload, ensure_ascii=False),
        )
        self.assertNotIn("executor_candidate", payload)
        self.assertEqual(
            payload["materialized_artifacts"][0]["content"]["payload"],
            {"constraints": ["统一视觉"]},
        )
        self.assertNotIn("requirements", payload["task_package"])
        self.assertNotIn("acceptance_criteria", payload["task_package"])
        self.assertEqual(
            payload["output_json_schema"],
            ValidatorCandidate.model_json_schema(),
        )

    def test_critical_failure_deterministically_fails_report(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            task = _task(root)
            delivery = _delivery(root, task)
            report = run_validator_agent(
                task,
                delivery,
                run_dir=root,
                model=_model(_candidate(task, delivery, verdict="FAIL")),
                tool_registry=ToolRegistry({}),
            )
        self.assertEqual(report.verdict, "FAIL")

    def test_minor_failure_does_not_fail_report(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            task = _task(root, importance="minor")
            delivery = _delivery(root, task)
            report = run_validator_agent(
                task,
                delivery,
                run_dir=root,
                model=_model(_candidate(task, delivery, verdict="FAIL")),
                tool_registry=ToolRegistry({}),
            )
        self.assertEqual(report.verdict, "PASS")

    def test_stale_delivery_is_rejected_before_model_execution(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            task = _task(root)
            delivery = _delivery(root, task).model_copy(
                update={"plan_version": 2}
            )
            with self.assertRaisesRegex(
                ValidatorContextError,
                "plan_version",
            ):
                run_validator_agent(
                    task,
                    delivery,
                    run_dir=root,
                    model=ToolAwareFakeChatModel(
                        responses=[AIMessage(content="不应被调用")]
                    ),
                    tool_registry=ToolRegistry({}),
                )

    def test_missing_structured_artifact_is_rejected_before_model_execution(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            task = _task(root)
            delivery = _delivery(root, task)
            Path(delivery.artifacts[0].uri).unlink()
            model = _model(_candidate(task, delivery))
            with (
                patch.object(
                    model,
                    "_generate",
                    side_effect=AssertionError("Validator 模型不应被调用"),
                ),
                self.assertRaisesRegex(
                    ValidatorContextError,
                    "正式 Artifact 不存在",
                ),
            ):
                run_validator_agent(
                    task,
                    delivery,
                    run_dir=root,
                    model=model,
                    tool_registry=ToolRegistry({}),
                )

    def test_invalid_target_retries_inside_same_validator_run(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            task = _task(root)
            delivery = _delivery(root, task)
            report = run_validator_agent(
                task,
                delivery,
                run_dir=root,
                model=_model(
                    _candidate(
                        task,
                        delivery,
                        expectation_id="Requirement9-Expectation1",
                    ),
                    _candidate(task, delivery),
                ),
                tool_registry=ToolRegistry({}),
            )
        self.assertEqual(report.verdict, "PASS")
        self.assertEqual(
            report.validator_run_id,
            validator_run_id_for(delivery),
        )


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