video-image-production-build/tests/contracts/test_evaluation.py

from __future__ import annotations

import unittest

from pydantic import ValidationError

from production_build_agents.capabilities import CAPABILITIES
from production_build_agents.contracts.evaluation import (
    evaluate_plan,
    evaluate_stage,
    evaluate_task_delivery,
    evaluate_validation_report,
    planned_task_for,
    task_expectation_pairs,
    validate_replan,
)
from production_build_agents.contracts.identifiers import (
    source_asset_id_for,
    validator_run_id_for,
)
from production_build_agents.contracts.models import (
    Artifact,
    ArtifactBindingClaim,
    ArtifactExpectation,
    ArtifactExpectationBinding,
    ArtifactRejection,
    DependencyDeliveryRef,
    ExecutorCandidate,
    ExecutorDelivery,
    GlobalDataDeliveryTask,
    GlobalDataPlan,
    GlobalDataRequirement,
    GlobalDataStageCandidate,
    GlobalDataStageDelivery,
    GlobalDataStageValidationReport,
    GlobalDataStageValidatorCandidate,
    PlannedTask,
    ProductionBrief,
    ProductionTable,
    SourceAsset,
    TaskPackage,
    ToolArtifactMapping,
    ToolCallRecord,
    ValidationCriterionResult,
    ValidationReport,
    ValidatorCandidate,
)
from production_build_agents.preprocess.source_assets import build_source_assets


def _brief(*uris: str) -> ProductionBrief:
    brief = ProductionBrief(
        post_type="video",
        core_production_points=[{"已有图片": list(uris), "用途": "人物身份参考"}],
        production_table=ProductionTable(
            segment_structure=[],
            form_results=[],
            relationship_results=[],
        ),
    )
    return brief.model_copy(update={"source_assets": build_source_assets(brief)})


def _requirement(
    number: int,
    *,
    source_asset_ids: list[str],
) -> GlobalDataRequirement:
    return GlobalDataRequirement(
        requirement_id=f"Requirement{number}",
        description=f"采纳第 {number} 组已有来源",
        importance="critical",
        source_paths=["$.帖子类型", "$.核心制作点[0]"],
        artifact_expectations=[
            ArtifactExpectation(
                expectation_id=f"Requirement{number}-Expectation1",
                artifact_type="image",
                minimum_count=len(source_asset_ids),
                usage_scope=f"来源组 {number}",
                verification_capabilities=[
                    "source_identity",
                    "technical_integrity",
                ],
                source_asset_ids=source_asset_ids,
            )
        ],
    )


def _generated_requirement(
    number: int = 1,
    *,
    artifact_types: tuple[str, ...] = ("image", "image"),
    capabilities: tuple[str, ...] = ("visual_content",),
    importance: str = "critical",
) -> GlobalDataRequirement:
    return GlobalDataRequirement(
        requirement_id=f"Requirement{number}",
        description="生成可同时服务多个明确用途的媒体",
        importance=importance,
        source_paths=["$.帖子类型", "$.核心制作点[0]"],
        artifact_expectations=[
            ArtifactExpectation(
                expectation_id=f"Requirement{number}-Expectation{index}",
                artifact_type=artifact_type,
                minimum_count=1,
                usage_scope=f"生成用途 {index}",
                verification_capabilities=list(capabilities),
            )
            for index, artifact_type in enumerate(artifact_types, start=1)
        ],
    )


def _plan(
    requirements: list[GlobalDataRequirement],
    *,
    skill_id: str = "reference-inspection",
    plan_version: int = 1,
    tasks: list[PlannedTask] | None = None,
) -> GlobalDataPlan:
    expectation_ids = [
        expectation.expectation_id
        for requirement in requirements
        for expectation in requirement.artifact_expectations
    ]
    return GlobalDataPlan(
        plan_id="GlobalDataPlan",
        plan_version=plan_version,
        goal="采纳正式生产需要的已有来源",
        stage_requirements=requirements,
        tasks=tasks
        or [
            PlannedTask(
                task_id="Task1",
                objective="检查并采纳已有来源",
                reason="后续生产需要精确来源身份",
                expectation_ids=expectation_ids,
                skill_id=skill_id,
                source_paths=["$.帖子类型", "$.核心制作点[0]"],
                deliverable_type=(
                    "reference_collection"
                    if skill_id == "reference-inspection"
                    else "image"
                ),
            )
        ],
        revision_summary="显式绑定来源、期待和交付。",
    )


def _package(
    plan: GlobalDataPlan,
    *,
    task_id: str = "Task1",
    dependencies: list[DependencyDeliveryRef] | None = None,
) -> TaskPackage:
    return TaskPackage(
        run_id="Run-policy",
        plan_id=plan.plan_id,
        plan_version=plan.plan_version,
        task_id=task_id,
        production_brief_uri="/fixture/production_brief.json",
        plan_uri=f"/fixture/global_data_dag.v{plan.plan_version}.json",
        dependency_deliveries=dependencies or [],
    )


def _delivery(
    brief: ProductionBrief,
    plan: GlobalDataPlan,
    package: TaskPackage,
    *,
    bindings: list[tuple[str, str, str | None]],
    artifact_type: str = "image",
    skill_id: str | None = None,
    deliverable_type: str | None = None,
    extra_unbound: bool = False,
    evidence: bool = True,
) -> ExecutorDelivery:
    planned = planned_task_for(package, plan)
    source_by_id = {item.source_asset_id: item for item in brief.source_assets}
    artifact_ids = list(dict.fromkeys(artifact_id for _, artifact_id, _ in bindings))
    source_id_by_artifact = {
        artifact_id: source_asset_id
        for _, artifact_id, source_asset_id in bindings
        if source_asset_id is not None
    }
    artifacts = [
        Artifact(
            artifact_id=artifact_id,
            artifact_type=artifact_type,
            uri=f"/cache/{artifact_id}.png",
            content_sha256="0" * 64,
            size_bytes=1,
            source_uri=(
                source_by_id[source_id_by_artifact[artifact_id]].source_uri
                if artifact_id in source_id_by_artifact
                else None
            ),
            description="已检查来源",
        )
        for artifact_id in artifact_ids
    ]
    if extra_unbound:
        artifacts.append(
            Artifact(
                artifact_id=(
                    f"{package.task_id}-v{package.plan_version}-"
                    f"artifact-{len(artifacts) + 1}"
                ),
                artifact_type=artifact_type,
                uri="/cache/unbound.png",
                content_sha256="0" * 64,
                size_bytes=1,
                description="辅助产物",
            )
        )
    tool_calls = [
        ToolCallRecord(
            tool_call_id=f"evidence-{index}",
            tool_name="probe_media",
            success=True,
            output_refs=[
                artifact.uri,
                *([artifact.source_uri] if artifact.source_uri is not None else []),
            ],
            artifact_mappings=[
                ToolArtifactMapping(
                    source=artifact.source_uri or artifact.uri,
                    local_path=artifact.uri,
                )
            ],
        )
        for index, artifact in enumerate(artifacts, start=1)
    ]
    evidence_id_by_artifact = {
        artifact.artifact_id: tool_calls[index].tool_call_id
        for index, artifact in enumerate(artifacts)
    }
    formal_bindings = [
        ArtifactExpectationBinding(
            expectation_id=expectation_id,
            artifact_id=artifact_id,
            source_asset_id=source_asset_id,
            evidence_tool_call_ids=(
                [evidence_id_by_artifact[artifact_id]] if evidence else []
            ),
        )
        for expectation_id, artifact_id, source_asset_id in bindings
    ]
    return ExecutorDelivery(
        run_id=package.run_id,
        plan_id=package.plan_id,
        task_id=package.task_id,
        plan_version=package.plan_version,
        executor_run_id=(
            f"{package.run_id}-executor-{package.task_id}-" f"v{package.plan_version}"
        ),
        skill_id=skill_id or planned.skill_id,
        deliverable_type=deliverable_type or planned.deliverable_type,
        manifest_uri="/fixture/executor_candidate.json",
        artifacts=artifacts,
        artifact_expectation_bindings=formal_bindings,
        tool_calls=tool_calls,
        attempt_count=1,
        summary="完成",
    )


def _report(
    plan: GlobalDataPlan,
    package: TaskPackage,
    delivery: ExecutorDelivery,
    *,
    failed_targets: set[tuple[str, str]] | None = None,
) -> ValidationReport:
    failed = failed_targets or set()
    planned = planned_task_for(package, plan)
    pairs = task_expectation_pairs(plan, planned)
    results = [
        ValidationCriterionResult(
            expectation_id=expectation.expectation_id,
            verification_capability=capability,
            verdict=(
                "FAIL" if (expectation.expectation_id, capability) in failed else "PASS"
            ),
            evidence=["fixture"],
            reason="已核对",
        )
        for _, expectation in pairs
        for capability in expectation.verification_capabilities
    ]
    importance = {
        expectation.expectation_id: requirement.importance
        for requirement, expectation in pairs
    }
    has_critical_failure = any(
        result.verdict == "FAIL" and importance[result.expectation_id] == "critical"
        for result in results
    )
    return ValidationReport(
        run_id=package.run_id,
        plan_id=package.plan_id,
        plan_version=package.plan_version,
        task_id=package.task_id,
        executor_run_id=delivery.executor_run_id,
        validator_run_id=validator_run_id_for(delivery),
        criterion_results=results,
        verdict="FAIL" if has_critical_failure else "PASS",
        summary="失败" if has_critical_failure else "通过",
    )


class ProtocolShapeTest(unittest.TestCase):

    def test_top_level_documents_are_0_3_and_old_task_fields_are_forbidden(
        self,
    ) -> None:
        plan = _plan([_generated_requirement()])
        package = _package(plan)
        top_level_models = (
            ProductionBrief,
            GlobalDataPlan,
            TaskPackage,
            ExecutorCandidate,
            ExecutorDelivery,
            ValidatorCandidate,
            ValidationReport,
            GlobalDataStageCandidate,
            GlobalDataStageValidatorCandidate,
            GlobalDataStageValidationReport,
            GlobalDataStageDelivery,
        )
        for model in top_level_models:
            with self.subTest(model=model.__name__):
                self.assertEqual(model.model_fields["schema_version"].default, "0.3")

        nested_models = (
            SourceAsset,
            ArtifactExpectation,
            DependencyDeliveryRef,
            ArtifactBindingClaim,
            Artifact,
            ArtifactExpectationBinding,
            ValidationCriterionResult,
        )
        for model in nested_models:
            with self.subTest(nested=model.__name__):
                self.assertNotIn("schema_version", model.model_fields)

        for old_field in (
            "expected_output",
            "acceptance_criteria",
            "validation_plan",
            "requirements",
            "source_assets",
        ):
            with self.subTest(old_task_field=old_field):
                payload = package.model_dump(mode="json")
                payload[old_field] = []
                with self.assertRaises(ValidationError):
                    TaskPackage.model_validate(payload)

    def test_stage_delivery_task_is_a_compact_reference(self) -> None:
        self.assertEqual(
            set(GlobalDataDeliveryTask.model_fields),
            {
                "task_id",
                "accepted_plan_version",
                "executor_delivery_uri",
                "validation_report_uri",
            },
        )

    def test_validation_result_identity_is_expectation_and_capability(
        self,
    ) -> None:
        result = ValidationCriterionResult(
            expectation_id="Requirement1-Expectation1",
            verification_capability="visual_content",
            verdict="PASS",
            evidence=["查看了最终图片"],
            reason="匹配",
        )
        self.assertEqual(
            result.model_dump(mode="json"),
            {
                "expectation_id": "Requirement1-Expectation1",
                "verification_capability": "visual_content",
                "verdict": "PASS",
                "evidence": ["查看了最终图片"],
                "reason": "匹配",
            },
        )
        with self.assertRaises(ValidationError):
            ValidationCriterionResult(
                expectation_id="Requirement1-Expectation1",
                verification_capability="visual_content",
                verdict="PASS",
                evidence=[],
                reason="没有证据",
            )
        for old_field in (
            "requirement_id",
            "artifact_type",
            "importance",
            "criterion_id",
            "dimension",
        ):
            with self.subTest(old_criterion_field=old_field):
                payload = result.model_dump(mode="json")
                payload[old_field] = "forged"
                with self.assertRaises(ValidationError):
                    ValidationCriterionResult.model_validate(payload)

    def test_artifact_expectation_rejects_duplicate_capabilities(self) -> None:
        with self.assertRaises(ValidationError):
            ArtifactExpectation(
                expectation_id="Requirement1-Expectation1",
                artifact_type="image",
                usage_scope="人物基准",
                verification_capabilities=["visual_content", "visual_content"],
            )

    def test_source_asset_id_helper_is_the_canonical_hash(self) -> None:
        brief = _brief("https://example.test/a.png")
        self.assertEqual(
            brief.source_assets[0].source_asset_id,
            source_asset_id_for(
                "image",
                "https://example.test/a.png",
            ),
        )

    def test_task_helpers_preserve_task_expectation_order(self) -> None:
        requirement = _generated_requirement()
        plan = _plan([requirement])
        package = _package(plan)
        planned = planned_task_for(package, plan)
        pairs = task_expectation_pairs(plan, planned)
        self.assertEqual(
            [item.expectation_id for _, item in pairs],
            planned.expectation_ids,
        )


class PlanEvaluationTest(unittest.TestCase):

    def test_plan_rejects_source_asset_outside_requirement_scope(
        self,
    ) -> None:
        brief = ProductionBrief(
            post_type="video",
            core_production_points=[
                {"已有图片": ["https://example.test/person.png"]},
                {"已有图片": ["https://example.test/curtain.png"]},
            ],
            production_table=ProductionTable(
                segment_structure=[],
                form_results=[],
                relationship_results=[],
            ),
        )
        brief = brief.model_copy(
            update={"source_assets": build_source_assets(brief)}
        )
        assets_by_uri = {
            item.source_uri: item for item in brief.source_assets
        }
        person = assets_by_uri["https://example.test/person.png"]
        curtain = assets_by_uri["https://example.test/curtain.png"]
        person_requirement = GlobalDataRequirement(
            requirement_id="Requirement1",
            description="采纳出镜女生角色基准图",
            importance="critical",
            source_paths=["$.核心制作点[0]"],
            artifact_expectations=[
                ArtifactExpectation(
                    expectation_id="Requirement1-Expectation1",
                    artifact_type="image",
                    minimum_count=2,
                    usage_scope="人物身份参考",
                    verification_capabilities=[
                        "source_identity",
                        "technical_integrity",
                    ],
                    source_asset_ids=[
                        person.source_asset_id,
                        curtain.source_asset_id,
                    ],
                )
            ],
        )
        curtain_requirement = GlobalDataRequirement(
            requirement_id="Requirement2",
            description="采纳窗帘场景参考图",
            importance="critical",
            source_paths=["$.帖子类型", "$.核心制作点[1]"],
            artifact_expectations=[
                ArtifactExpectation(
                    expectation_id="Requirement2-Expectation1",
                    artifact_type="image",
                    minimum_count=1,
                    usage_scope="场景参考",
                    verification_capabilities=[
                        "source_identity",
                        "technical_integrity",
                    ],
                    source_asset_ids=[curtain.source_asset_id],
                )
            ],
        )
        plan = _plan([person_requirement, curtain_requirement])

        scope_issues = [
            item
            for item in evaluate_plan(brief, plan, CAPABILITIES)
            if item.code == "source_asset_outside_requirement_scope"
        ]

        self.assertEqual(len(scope_issues), 1)
        self.assertEqual(scope_issues[0].requirement_id, "Requirement1")
        self.assertEqual(
            scope_issues[0].expectation_id,
            "Requirement1-Expectation1",
        )
        self.assertEqual(
            scope_issues[0].source_asset_id,
            curtain.source_asset_id,
        )

    def test_plan_accepts_source_asset_with_any_path_in_requirement_scope(
        self,
    ) -> None:
        brief = _brief("https://example.test/person.png")
        asset = brief.source_assets[0].model_copy(
            update={
                "source_paths": [
                    "$.核心制作点[0]",
                    "$.制作表.形式结果[0]",
                ]
            }
        )
        brief = brief.model_copy(update={"source_assets": [asset]})
        plan = _plan(
            [_requirement(1, source_asset_ids=[asset.source_asset_id])]
        )

        codes = {
            item.code for item in evaluate_plan(brief, plan, CAPABILITIES)
        }

        self.assertNotIn(
            "source_asset_outside_requirement_scope",
            codes,
        )

    def test_plan_rejects_undispositioned_and_forged_source_asset(self) -> None:
        brief = _brief("https://example.test/a.png", "https://example.test/b.png")
        forged = brief.source_assets[0].model_copy(
            update={"source_asset_id": "SourceAsset-" + "0" * 64}
        )
        brief = brief.model_copy(
            update={"source_assets": [forged, brief.source_assets[1]]}
        )
        plan = _plan([_requirement(1, source_asset_ids=[forged.source_asset_id])])

        codes = {item.code for item in evaluate_plan(brief, plan, CAPABILITIES)}
        self.assertIn("source_asset_id_mismatch", codes)
        self.assertIn("undispositioned_source_asset", codes)

    def test_plan_business_rules_live_in_evaluate_plan(self) -> None:
        brief = _brief()
        requirement = _generated_requirement()
        plan = _plan([requirement])
        payload = plan.model_dump(mode="json")
        payload["tasks"][0]["expectation_ids"] = ["Requirement9-Expectation1"]
        payload["tasks"][0]["depends_on"] = ["Task9"]
        payload["tasks"][0]["source_paths"].append("$.不存在")
        payload["stage_requirements"][0]["source_paths"] = ["$.不存在"]
        invalid = GlobalDataPlan.model_validate(payload)

        codes = {
            item.code
            for item in evaluate_plan(
                brief,
                invalid,
                CAPABILITIES,
            )
        }
        self.assertIn("unknown_task_expectation", codes)
        self.assertIn("unknown_task_dependency", codes)
        self.assertIn("unassigned_critical_expectation", codes)
        self.assertIn("unknown_requirement_source_path", codes)
        self.assertIn("missing_critical_audit_path", codes)
        self.assertIn("unknown_task_source_path", codes)

    def test_plan_detects_cycle_and_cross_task_replacement_collision(
        self,
    ) -> None:
        brief = _brief()
        requirement = _generated_requirement()
        artifact_id = "Task9-v1-artifact-1"
        tasks = [
            PlannedTask(
                task_id="Task1",
                objective="一",
                reason="一",
                depends_on=["Task2"],
                expectation_ids=["Requirement1-Expectation1"],
                replaces_artifact_ids=[artifact_id],
                skill_id="image-production",
                source_paths=["$.核心制作点[0]"],
                deliverable_type="image",
            ),
            PlannedTask(
                task_id="Task2",
                objective="二",
                reason="二",
                depends_on=["Task1"],
                expectation_ids=["Requirement1-Expectation2"],
                replaces_artifact_ids=[artifact_id],
                skill_id="image-production",
                source_paths=["$.核心制作点[0]"],
                deliverable_type="image",
            ),
        ]
        plan = _plan([requirement], skill_id="image-production", tasks=tasks)
        codes = {item.code for item in evaluate_plan(brief, plan, CAPABILITIES)}
        self.assertIn("task_dependency_cycle", codes)
        self.assertIn("duplicate_artifact_replacement", codes)
        self.assertIn("initial_plan_has_replacements", codes)


class TaskDeliveryEvaluationTest(unittest.TestCase):

    def test_one_artifact_may_bind_multiple_expectations(self) -> None:
        brief = _brief()
        plan = _plan(
            [_generated_requirement()],
            skill_id="image-production",
        )
        package = _package(plan)
        artifact_id = "Task1-v1-artifact-1"
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (expectation_id, artifact_id, None)
                for expectation_id in plan.tasks[0].expectation_ids
            ],
        )

        self.assertEqual(
            evaluate_task_delivery(
                brief,
                plan,
                package,
                delivery,
                CAPABILITIES,
            ),
            [],
        )
        evaluation = evaluate_stage(
            plan,
            [delivery],
            [_report(plan, package, delivery)],
        )
        self.assertEqual(evaluation.contract_issues, [])
        self.assertEqual(evaluation.expectation_gaps, [])
        self.assertEqual(
            [
                item.artifact_ids
                for item in evaluation.requirement_evaluations[
                    0
                ].expectation_evaluations
            ],
            [[artifact_id], [artifact_id]],
        )

    def test_each_shared_artifact_binding_is_independently_typed(self) -> None:
        brief = _brief()
        plan = _plan(
            [_generated_requirement(artifact_types=("image", "video"))],
            skill_id="image-production",
        )
        package = _package(plan)
        artifact_id = "Task1-v1-artifact-1"
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (expectation_id, artifact_id, None)
                for expectation_id in plan.tasks[0].expectation_ids
            ],
        )

        issues = evaluate_task_delivery(
            brief,
            plan,
            package,
            delivery,
            CAPABILITIES,
        )
        self.assertEqual(
            [
                item.expectation_id
                for item in issues
                if item.code == "binding_artifact_type_mismatch"
            ],
            ["Requirement1-Expectation2"],
        )

    def test_one_artifact_cannot_cover_two_distinct_sources(self) -> None:
        brief = _brief("https://example.test/a.png", "https://example.test/b.png")
        first, second = brief.source_assets
        plan = _plan(
            [
                _requirement(1, source_asset_ids=[first.source_asset_id]),
                _requirement(2, source_asset_ids=[second.source_asset_id]),
            ]
        )
        package = _package(plan)
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (
                    "Requirement1-Expectation1",
                    "Task1-v1-artifact-1",
                    first.source_asset_id,
                )
            ],
        )

        codes = {
            item.code
            for item in evaluate_task_delivery(
                brief,
                plan,
                package,
                delivery,
                CAPABILITIES,
            )
        }
        self.assertIn("insufficient_artifact_bindings", codes)
        evaluation = evaluate_stage(
            plan,
            [delivery],
            [_report(plan, package, delivery)],
        )
        self.assertEqual(
            [gap.expectation_id for gap in evaluation.expectation_gaps],
            ["Requirement2-Expectation1"],
        )

    def test_same_path_uris_remain_separate_source_identities(self) -> None:
        brief = _brief("https://example.test/a.png", "https://example.test/b.png")
        first, second = brief.source_assets
        requirement = GlobalDataRequirement(
            requirement_id="Requirement1",
            description="分别采纳同一路径里的两张图",
            importance="critical",
            source_paths=["$.核心制作点[0]"],
            artifact_expectations=[
                ArtifactExpectation(
                    expectation_id="Requirement1-Expectation1",
                    artifact_type="image",
                    minimum_count=1,
                    usage_scope="人物 A",
                    verification_capabilities=[
                        "source_identity",
                        "technical_integrity",
                    ],
                    source_asset_ids=[first.source_asset_id],
                ),
                ArtifactExpectation(
                    expectation_id="Requirement1-Expectation2",
                    artifact_type="image",
                    minimum_count=1,
                    usage_scope="人物 B",
                    verification_capabilities=[
                        "source_identity",
                        "technical_integrity",
                    ],
                    source_asset_ids=[second.source_asset_id],
                ),
            ],
        )
        plan = _plan([requirement])
        package = _package(plan)
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (
                    "Requirement1-Expectation1",
                    "Task1-v1-artifact-1",
                    first.source_asset_id,
                ),
                (
                    "Requirement1-Expectation2",
                    "Task1-v1-artifact-2",
                    second.source_asset_id,
                ),
            ],
        )

        self.assertEqual(
            evaluate_task_delivery(
                brief,
                plan,
                package,
                delivery,
                CAPABILITIES,
            ),
            [],
        )
        evaluation = evaluate_stage(
            plan,
            [delivery],
            [_report(plan, package, delivery)],
        )
        self.assertEqual(evaluation.expectation_gaps, [])
        self.assertEqual(
            [
                item.artifact_ids
                for item in evaluation.requirement_evaluations[
                    0
                ].expectation_evaluations
            ],
            [
                ["Task1-v1-artifact-1"],
                ["Task1-v1-artifact-2"],
            ],
        )

    def test_unbound_artifact_satisfies_nothing(self) -> None:
        brief = _brief()
        plan = _plan(
            [_generated_requirement(artifact_types=("image",))],
            skill_id="image-production",
        )
        package = _package(plan)
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[],
            extra_unbound=True,
        )

        evaluation = evaluate_stage(
            plan,
            [delivery],
            [_report(plan, package, delivery)],
        )
        self.assertEqual(
            evaluation.requirement_evaluations[0].actual_artifact_ids,
            [],
        )
        self.assertEqual(len(evaluation.expectation_gaps), 1)

    def test_generated_skill_cannot_adopt_source_identity(self) -> None:
        brief = _brief("https://example.test/a.png")
        source = brief.source_assets[0]
        plan = _plan(
            [_requirement(1, source_asset_ids=[source.source_asset_id])],
            skill_id="image-production",
        )
        package = _package(plan)
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (
                    "Requirement1-Expectation1",
                    "Task1-v1-artifact-1",
                    source.source_asset_id,
                )
            ],
        )

        codes = {
            item.code
            for item in evaluate_task_delivery(
                brief,
                plan,
                package,
                delivery,
                CAPABILITIES,
            )
        }
        self.assertIn("generated_artifact_cannot_adopt_source", codes)

    def test_evidence_must_exist_succeed_and_link_to_artifact(self) -> None:
        brief = _brief()
        plan = _plan(
            [_generated_requirement(artifact_types=("image",))],
            skill_id="image-production",
        )
        package = _package(plan)
        valid = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (
                    "Requirement1-Expectation1",
                    "Task1-v1-artifact-1",
                    None,
                )
            ],
        )
        binding = valid.artifact_expectation_bindings[0]
        missing = valid.model_copy(
            update={
                "artifact_expectation_bindings": [
                    binding.model_copy(update={"evidence_tool_call_ids": ["missing"]})
                ]
            }
        )
        failed = valid.model_copy(
            update={
                "tool_calls": [
                    valid.tool_calls[0].model_copy(update={"success": False})
                ]
            }
        )
        unlinked = valid.model_copy(
            update={
                "tool_calls": [
                    valid.tool_calls[0].model_copy(
                        update={
                            "output_refs": ["/not/the/artifact"],
                            "artifact_mappings": [],
                        }
                    )
                ]
            }
        )
        absent = valid.model_copy(
            update={
                "artifact_expectation_bindings": [
                    binding.model_copy(update={"evidence_tool_call_ids": []})
                ]
            }
        )
        cases = {
            "unknown_binding_evidence": missing,
            "unsuccessful_binding_evidence": failed,
            "binding_evidence_not_linked": unlinked,
            "missing_binding_evidence": absent,
        }
        for expected_code, delivery in cases.items():
            with self.subTest(expected_code=expected_code):
                codes = {
                    item.code
                    for item in evaluate_task_delivery(
                        brief,
                        plan,
                        package,
                        delivery,
                        CAPABILITIES,
                    )
                }
                self.assertIn(expected_code, codes)

    def test_semantic_only_binding_may_omit_tool_evidence(self) -> None:
        brief = _brief()
        plan = _plan(
            [
                _generated_requirement(
                    artifact_types=("image",),
                    capabilities=("semantic_content",),
                )
            ],
            skill_id="image-production",
        )
        package = _package(plan)
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (
                    "Requirement1-Expectation1",
                    "Task1-v1-artifact-1",
                    None,
                )
            ],
            evidence=False,
        )
        self.assertEqual(
            evaluate_task_delivery(
                brief,
                plan,
                package,
                delivery,
                CAPABILITIES,
            ),
            [],
        )

    def test_task_package_and_artifact_identity_are_checked(self) -> None:
        brief = _brief()
        plan = _plan(
            [_generated_requirement(artifact_types=("image",))],
            skill_id="image-production",
        )
        package = _package(plan)
        valid = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (
                    "Requirement1-Expectation1",
                    "Task1-v1-artifact-1",
                    None,
                )
            ],
        )
        wrong_package = package.model_copy(update={"plan_version": 2})
        wrong_artifact = valid.artifacts[0].model_copy(
            update={"artifact_id": "Task2-v1-artifact-1"}
        )
        wrong_binding = valid.artifact_expectation_bindings[0].model_copy(
            update={"artifact_id": wrong_artifact.artifact_id}
        )
        wrong_delivery = valid.model_copy(
            update={
                "artifacts": [wrong_artifact],
                "artifact_expectation_bindings": [wrong_binding],
            }
        )

        self.assertIn(
            "task_package_plan_mismatch",
            {
                item.code
                for item in evaluate_task_delivery(
                    brief,
                    plan,
                    wrong_package,
                    valid,
                    CAPABILITIES,
                )
            },
        )
        self.assertIn(
            "artifact_id_identity_mismatch",
            {
                item.code
                for item in evaluate_task_delivery(
                    brief,
                    plan,
                    package,
                    wrong_delivery,
                    CAPABILITIES,
                )
            },
        )


class StageAndReplanEvaluationTest(unittest.TestCase):

    def test_stage_rejects_duplicate_artifact_ids_globally(self) -> None:
        brief = _brief()
        requirements = [
            _generated_requirement(
                1,
                artifact_types=("image",),
            ),
            _generated_requirement(
                2,
                artifact_types=("image",),
            ),
        ]
        tasks = [
            PlannedTask(
                task_id="Task1",
                objective="一",
                reason="一",
                expectation_ids=["Requirement1-Expectation1"],
                skill_id="image-production",
                source_paths=["$.核心制作点[0]"],
                deliverable_type="image",
            ),
            PlannedTask(
                task_id="Task2",
                objective="二",
                reason="二",
                expectation_ids=["Requirement2-Expectation1"],
                skill_id="image-production",
                source_paths=["$.核心制作点[0]"],
                deliverable_type="image",
            ),
        ]
        plan = _plan(
            requirements,
            skill_id="image-production",
            tasks=tasks,
        )
        first_package = _package(plan)
        second_package = _package(plan, task_id="Task2")
        duplicate_id = "Task1-v1-artifact-1"
        first = _delivery(
            brief,
            plan,
            first_package,
            bindings=[("Requirement1-Expectation1", duplicate_id, None)],
        )
        second = _delivery(
            brief,
            plan,
            second_package,
            bindings=[
                (
                    "Requirement2-Expectation1",
                    "Task2-v1-artifact-1",
                    None,
                )
            ],
        )
        second_artifact = second.artifacts[0].model_copy(
            update={"artifact_id": duplicate_id}
        )
        second_binding = second.artifact_expectation_bindings[0].model_copy(
            update={"artifact_id": duplicate_id}
        )
        second = second.model_copy(
            update={
                "artifacts": [second_artifact],
                "artifact_expectation_bindings": [second_binding],
            }
        )

        evaluation = evaluate_stage(
            plan,
            [first, second],
            [
                _report(plan, first_package, first),
                _report(plan, second_package, second),
            ],
        )
        self.assertEqual(
            [
                issue.artifact_id
                for issue in evaluation.contract_issues
                if issue.code == "duplicate_global_artifact_id"
            ],
            [duplicate_id],
        )
        self.assertEqual(evaluation.active_artifacts, [])
        self.assertTrue(
            all(
                item.actual_artifact_ids == []
                for item in evaluation.requirement_evaluations
            )
        )

    def test_stage_checks_validation_target_and_verdict_identity(self) -> None:
        brief = _brief()
        plan = _plan(
            [_generated_requirement(artifact_types=("image",))],
            skill_id="image-production",
        )
        package = _package(plan)
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (
                    "Requirement1-Expectation1",
                    "Task1-v1-artifact-1",
                    None,
                )
            ],
        )
        valid_report = _report(plan, package, delivery)
        bad_report = valid_report.model_copy(update={"criterion_results": []})

        evaluation = evaluate_stage(plan, [delivery], [bad_report])
        self.assertIn(
            "validation_targets_mismatch",
            {item.code for item in evaluation.contract_issues},
        )
        self.assertEqual(evaluation.active_artifacts, [])

    def test_validation_report_evaluation_does_not_require_fake_task_package(
        self,
    ) -> None:
        brief = _brief()
        plan = _plan(
            [_generated_requirement(artifact_types=("image",))],
            skill_id="image-production",
        )
        package = _package(plan)
        delivery = _delivery(
            brief,
            plan,
            package,
            bindings=[
                (
                    "Requirement1-Expectation1",
                    "Task1-v1-artifact-1",
                    None,
                )
            ],
        )
        report = _report(plan, package, delivery)

        self.assertEqual(
            evaluate_validation_report(plan, delivery, report),
            [],
        )

    def test_replan_consumes_structured_artifact_rejection(self) -> None:
        requirement = _generated_requirement(artifact_types=("image",))
        previous = _plan(
            [requirement],
            skill_id="image-production",
        )
        rejected_id = "Task1-v1-artifact-1"
        replacement_task = PlannedTask(
            task_id="Task2",
            objective="替换被拒绝产物",
            reason="Stage 语义拒绝",
            expectation_ids=["Requirement1-Expectation1"],
            replaces_artifact_ids=[rejected_id],
            skill_id="image-production",
            source_paths=["$.核心制作点[0]"],
            deliverable_type="image",
        )
        next_plan = previous.model_copy(
            update={
                "plan_version": 2,
                "tasks": [*previous.tasks, replacement_task],
                "revision_summary": "增加替换 Task",
            }
        )
        rejection = ArtifactRejection(
            artifact_id=rejected_id,
            reason_code="semantic_rejection",
            reason="内容不满足正式生产要求",
        )

        self.assertEqual(
            validate_replan(
                previous,
                next_plan,
                [],
                [rejection],
                [],
            ),
            [],
        )
        missing = next_plan.model_copy(
            update={
                "tasks": previous.tasks,
                "revision_summary": "错误地忽略拒绝",
            }
        )
        self.assertIn(
            "required_replacement_missing",
            {
                item.code
                for item in validate_replan(
                    previous,
                    missing,
                    [],
                    [rejection],
                    [],
                )
            },
        )

    def test_replan_cannot_change_goal_or_passed_task(self) -> None:
        previous = _plan(
            [_generated_requirement(artifact_types=("image",))],
            skill_id="image-production",
        )
        changed_task = previous.tasks[0].model_copy(
            update={"objective": "改写已通过工作"}
        )
        next_plan = previous.model_copy(
            update={
                "plan_version": 2,
                "goal": "另一个目标",
                "tasks": [changed_task],
                "revision_summary": "错误改写",
            }
        )
        codes = {
            item.code
            for item in validate_replan(
                previous,
                next_plan,
                {"Task1"},
                [],
                [],
            )
        }
        self.assertIn("replan_goal_changed", codes)
        self.assertIn("passed_task_changed", codes)


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