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

from __future__ import annotations

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

from langchain_core.messages import AIMessage
from langchain_core.tools import StructuredTool
from PIL import Image

from production_build_agents.agents.segment.run import (
    run_segment_production,
)
from production_build_agents.agents.segment.executor import (
    SegmentExecutorConfigurationError,
    SegmentExecutorOutputError,
    _remote_tool_call_limits,
    _remote_tool_call_counts,
    _seedance_call_count,
    _seedance_call_limit,
    _resolve_artifacts,
)
from production_build_agents.agents.segment.validator import (
    _expected_narration_text,
)
from production_build_agents.agents.segment.provenance import (
    SegmentProvenanceError,
    _text_recall,
    expected_narration_text,
    normalize_candidate_lineage,
    validate_required_production_trace,
)
from production_build_agents.agents.segment.skills.registry import (
    load_segment_executor_skill,
    load_segment_validator_skill,
)
from production_build_agents.contracts.models import Artifact, ToolCallRecord
from production_build_agents.contracts.document_identity import (
    exact_document_sha256,
)
from production_build_agents.contracts.production_models import (
    SEGMENT_VALIDATION_CRITERIA,
    ArtifactInputBinding,
    ArtifactLineageClaim,
    PlannedShot,
    ProductionArtifact,
    ProductionInput,
    SegmentExecutorCandidate,
    SegmentValidationCriterionResult,
    SegmentValidatorCandidate,
    ShotOutputClaim,
    TimelineEntry,
)
from production_build_agents.contracts.production_execution_models import (
    ImmutableJsonRef,
    SegmentPackage,
)
from production_build_agents.run.artifacts import content_identity_for_path
from production_build_agents.run.artifacts import seal_shared_visual_anchor
from production_build_agents.run.layout import segment_validation_report_path
from production_build_agents.tools.publishing import PublishingToolError
from production_build_agents.tools import publishing
from production_build_agents.tools.registry import ToolRegistry
from tests.support.fake_models import ToolAwareFakeChatModel
from tests.support.production_fixtures import output_profile
from tests.support.production_planner_fixtures import (
    build_production_planner_scenario,
)


def _production_input() -> ProductionInput:
    return ProductionInput(
        input_id="ProductionInput-" + "1" * 64,
        source_path="$.制作表.段落结构[0]",
        content_sha256="2" * 64,
        value={"旁白": "测试旁白"},
        applies_to_source_segment_ids=["段落1"],
    )


def _artifact(
    path: Path,
    *,
    artifact_id: str,
    artifact_type: str,
    source_uri: str | None = None,
) -> Artifact:
    return Artifact(
        artifact_id=artifact_id,
        artifact_type=artifact_type,
        uri=str(path.resolve()),
        source_uri=source_uri,
        description=artifact_id,
        **content_identity_for_path(path),
    )


def _package(root: Path) -> SegmentPackage:
    person = root / "person.png"
    voice = root / "voice.wav"
    Image.new("RGB", (16, 16), "navy").save(person)
    voice.write_bytes(b"sealed voice")
    anchor = root / "shared-anchor.png"
    Image.new("RGB", (16, 16), "white").save(anchor)
    production_input = _production_input()
    artifacts = [
        _artifact(
            person,
            artifact_id="Task1-v1-artifact-1",
            artifact_type="image",
            source_uri="https://mutable.test/person.png",
        ),
        _artifact(
            voice,
            artifact_id="Task2-v1-artifact-1",
            artifact_type="audio",
            source_uri="https://mutable.test/voice.wav",
        ),
    ]
    global_delivery = build_production_planner_scenario(
        run_id="segment-test-global"
    ).global_data_delivery.model_copy(
        update={"active_artifacts": artifacts}
    )
    global_delivery_path = root / "global_data_stage_delivery.json"
    global_delivery_path.write_text(
        global_delivery.model_dump_json(),
        encoding="utf-8",
    )
    return SegmentPackage(
        run_id="segment-test-run",
        plan_id="ProductionPlan",
        plan_version=1,
        segment_id="Segment1",
        source_segment_id="段落1",
        timeline=TimelineEntry(
            segment_id="Segment1",
            order=1,
            start_ms=0,
            end_ms=2000,
        ),
        production_brief_ref=ImmutableJsonRef(
            uri="/global/brief.json",
            document_sha256="3" * 64,
        ),
        global_data_delivery_ref=ImmutableJsonRef(
            uri=str(global_delivery_path.resolve()),
            document_sha256=exact_document_sha256(
                global_delivery_path.read_bytes()
            ),
        ),
        production_input_catalog_ref=ImmutableJsonRef(
            uri="/global/catalog.json",
            document_sha256="5" * 64,
        ),
        production_inputs=[production_input],
        artifact_inputs=[
            ArtifactInputBinding(
                artifact_id=artifact.artifact_id,
                expectation_ids=[
                    f"Requirement{index}-Expectation1"
                ],
                usage="正式输入",
            )
            for index, artifact in enumerate(artifacts, start=1)
        ],
        artifacts=artifacts,
        output_profile=output_profile(),
        shared_visual_anchor=seal_shared_visual_anchor(anchor),
        dependencies=[],
        shots=[
            PlannedShot(
                shot_id="Segment1-Shot1",
                objective="生成测试镜头",
                start_ms=0,
                end_ms=2000,
                production_input_ids=[production_input.input_id],
                artifact_ids=[artifacts[0].artifact_id],
            )
        ],
    )


def _candidate(
    root: Path,
    package: SegmentPackage,
    *,
    artifact_root: Path | None = None,
) -> SegmentExecutorCandidate:
    output_root = artifact_root or root / "run"
    paths = {
        "narration": output_root / "narration.wav",
        "shot": output_root / "shot.mp4",
        "subtitle": output_root / "subtitle.ass",
        "primary": output_root / "primary.mp4",
    }
    return SegmentExecutorCandidate(
        run_id=package.run_id,
        plan_id=package.plan_id,
        plan_version=package.plan_version,
        segment_id=package.segment_id,
        artifacts=[
            {
                "artifact_type": "audio",
                "uri": str(paths["narration"].resolve()),
                "description": "正式旁白",
            },
            {
                "artifact_type": "video",
                "uri": str(paths["shot"].resolve()),
                "description": "镜头输出",
            },
            {
                "artifact_type": "document",
                "uri": str(paths["subtitle"].resolve()),
                "description": "ASS 字幕",
            },
            {
                "artifact_type": "video",
                "uri": str(paths["primary"].resolve()),
                "description": "最终 Segment",
            },
        ],
        lineage_claims=[
            ArtifactLineageClaim(
                artifact_uri=str(paths["narration"].resolve()),
                input_artifact_uris=[package.artifacts[1].uri],
                production_input_ids=[
                    package.production_inputs[0].input_id
                ],
                evidence_tool_call_ids=["produce-segment"],
            ),
            ArtifactLineageClaim(
                artifact_uri=str(paths["shot"].resolve()),
                input_artifact_uris=[
                    package.artifacts[0].uri,
                    package.shared_visual_anchor.uri,
                ],
                production_input_ids=[
                    package.production_inputs[0].input_id
                ],
                evidence_tool_call_ids=["produce-segment"],
            ),
            ArtifactLineageClaim(
                artifact_uri=str(paths["subtitle"].resolve()),
                input_artifact_uris=[package.artifacts[0].uri],
                production_input_ids=[
                    package.production_inputs[0].input_id
                ],
                evidence_tool_call_ids=["produce-segment"],
            ),
            ArtifactLineageClaim(
                artifact_uri=str(paths["primary"].resolve()),
                input_artifact_uris=[
                    str(paths["narration"].resolve()),
                    str(paths["shot"].resolve()),
                    str(paths["subtitle"].resolve()),
                ],
                production_input_ids=[
                    package.production_inputs[0].input_id
                ],
                evidence_tool_call_ids=["produce-segment"],
            ),
        ],
        shot_outputs=[
            ShotOutputClaim(
                shot_id=package.shots[0].shot_id,
                artifact_uri=str(paths["shot"].resolve()),
            )
        ],
        narration_artifact_uri=str(paths["narration"].resolve()),
        subtitle_artifact_uri=str(paths["subtitle"].resolve()),
        primary_artifact_uri=str(paths["primary"].resolve()),
        summary="fake Segment 已完成",
    )


def _validator_candidate(
    package: SegmentPackage,
) -> SegmentValidatorCandidate:
    return SegmentValidatorCandidate(
        run_id=package.run_id,
        plan_id=package.plan_id,
        plan_version=package.plan_version,
        segment_id=package.segment_id,
        executor_run_id=(
            f"{package.run_id}:{package.segment_id}:"
            f"v{package.plan_version}:executor"
        ),
        criterion_results=[
            SegmentValidationCriterionResult(
                criterion=criterion,
                verdict="PASS",
                evidence=[f"{criterion} evidence"],
                reason=f"{criterion} passed",
            )
            for criterion in SEGMENT_VALIDATION_CRITERIA
        ],
        summary="六项验收通过",
    )


def _tool_registry(
    root: Path,
    output_paths: list[Path],
    *,
    primary_has_audio: bool = True,
    narration_transcript: str = "测试旁白",
    final_transcript: str = "测试旁白",
    subtitle_text: str = "测试旁白",
    subtitle_duration: float = 2.0,
    subtitle_overlaps: list[int] | None = None,
    video_width: int = 1080,
) -> ToolRegistry:
    frame = root / "frame.png"
    Image.new("RGB", (16, 16), "white").save(frame)

    def produce(sources: list[str]) -> dict:
        for path in output_paths:
            path.parent.mkdir(parents=True, exist_ok=True)
            if path.suffix == ".ass":
                path.write_text("subtitle", encoding="utf-8")
            else:
                path.write_bytes(f"artifact:{path.name}".encode())
        return {
            "success": True,
            "sources": sources,
            "local_paths": [str(path.resolve()) for path in output_paths],
        }

    def probe_media(source: str) -> dict:
        suffix = Path(source).suffix
        is_audio = suffix == ".wav"
        return {
            "success": True,
            "source": source,
            "local_path": source,
            "media_type": "audio" if is_audio else "video",
            "has_video": not is_audio,
            "has_audio": (
                is_audio
                or (
                    Path(source).name == "primary.mp4"
                    and primary_has_audio
                )
            ),
            "duration_sec": 2.0,
            "width": None if is_audio else video_width,
            "height": None if is_audio else 1920,
        }

    def transcribe_audio(audio: str, language: str = "zh") -> dict:
        return {
            "success": True,
            "source": audio,
            "transcript": (
                final_transcript
                if Path(audio).name == "primary.mp4"
                else narration_transcript
            ),
            "language": language,
            "duration_sec": 2.0,
        }

    def inspect_ass_subtitles(subtitle: str) -> dict:
        return {
            "success": True,
            "source": subtitle,
            "cues": [
                {
                    "start_sec": 0.0,
                    "end_sec": subtitle_duration,
                    "text": subtitle_text,
                }
            ],
            "cue_count": 1,
            "duration_sec": subtitle_duration,
            "overlap_indexes": subtitle_overlaps or [],
        }

    def extract_frames(video: str, num_frames: int = 5) -> dict:
        return {
            "success": True,
            "frames": [
                {
                    "timestamp_sec": index * 0.4,
                    "local_path": str(frame.resolve()),
                }
                for index in range(num_frames)
            ],
        }

    def view_images(image_sources: list[str]) -> dict:
        return {
            "success": True,
            "images": [
                {"source": source, "local_path": source}
                for source in image_sources
            ],
        }

    tools = {
        "video_concat": StructuredTool.from_function(
            produce,
            name="video_concat",
            description="produce fake segment artifacts",
        ),
        "probe_media": StructuredTool.from_function(
            probe_media,
            name="probe_media",
            description="probe fake media",
        ),
        "transcribe_audio": StructuredTool.from_function(
            transcribe_audio,
            name="transcribe_audio",
            description="transcribe fake media",
        ),
        "inspect_ass_subtitles": StructuredTool.from_function(
            inspect_ass_subtitles,
            name="inspect_ass_subtitles",
            description="inspect fake subtitles",
        ),
        "extract_frames": StructuredTool.from_function(
            extract_frames,
            name="extract_frames",
            description="extract fake frames",
        ),
        "view_images": StructuredTool.from_function(
            view_images,
            name="view_images",
            description="view fake images",
        ),
    }
    allowed = set(load_segment_executor_skill().allowed_tools) | set(
        load_segment_validator_skill().allowed_tools
    )
    for name in allowed.difference(tools):
        tools[name] = StructuredTool.from_function(
            lambda: {"success": True},
            name=name,
            description=f"unused fake {name}",
        )
    return ToolRegistry(tools)


def _file_hashes(root: Path) -> dict[str, str]:
    return {
        str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
        for path in sorted(root.rglob("*"))
        if path.is_file()
    }


class SegmentRuntimeTest(unittest.TestCase):
    def test_narration_recall_rejects_missing_sentence_tail(self) -> None:
        expected = "今天真的很想和大家分享一个话题，就是如何真正学会自我接纳"
        truncated = "今天真的很想和大家分享一个话题，就是如何真正学会"

        self.assertLess(_text_recall(expected, truncated), 0.95)
        self.assertEqual(_text_recall(expected, expected), 1.0)

    def test_candidate_lineage_is_rebuilt_in_artifact_order(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            record = ToolCallRecord(
                tool_call_id="deterministic-producer",
                operation_id="segment:1",
                tool_name="video_concat",
                arguments={
                    "sources": [
                        package.artifacts[0].uri,
                        package.shared_visual_anchor.uri,
                    ]
                },
                success=True,
                output_refs=[
                    item.uri for item in candidate.artifacts
                ],
            )

            normalized = normalize_candidate_lineage(
                candidate.model_copy(
                    update={
                        "lineage_claims": list(
                            reversed(candidate.lineage_claims)
                        )
                    }
                ),
                package=package,
                records=[record],
            )

        self.assertEqual(
            [item.artifact_uri for item in normalized.lineage_claims],
            [item.uri for item in normalized.artifacts],
        )
        self.assertTrue(
            all(
                item.evidence_tool_call_ids
                == ["deterministic-producer"]
                for item in normalized.lineage_claims
            )
        )
        shot_uri = normalized.shot_outputs[0].artifact_uri
        shot_lineage = next(
            item
            for item in normalized.lineage_claims
            if item.artifact_uri == shot_uri
        )
        self.assertEqual(
            shot_lineage.production_input_ids,
            package.shots[0].production_input_ids,
        )
        self.assertIn(
            package.artifacts[0].uri,
            shot_lineage.input_artifact_uris,
        )

    def test_candidate_lineage_uses_only_real_tool_inputs(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            narration_uri = candidate.narration_artifact_uri
            shot_uri = candidate.shot_outputs[0].artifact_uri
            subtitle_uri = candidate.subtitle_artifact_uri
            primary_uri = candidate.primary_artifact_uri
            records = [
                ToolCallRecord(
                    tool_call_id="tts",
                    tool_name="synthesize_speech",
                    arguments={"text": "测试旁白"},
                    success=True,
                    output_refs=[narration_uri],
                ),
                ToolCallRecord(
                    tool_call_id="shot",
                    tool_name="make_video",
                    arguments={
                        "reference": package.shared_visual_anchor.uri
                    },
                    success=True,
                    output_refs=[shot_uri],
                ),
                ToolCallRecord(
                    tool_call_id="subtitle",
                    tool_name="create_ass_subtitles",
                    arguments={"text": "测试旁白"},
                    success=True,
                    output_refs=[subtitle_uri],
                ),
                ToolCallRecord(
                    tool_call_id="primary",
                    tool_name="video_mux_audio",
                    arguments={
                        "video": shot_uri,
                        "audio": narration_uri,
                        "subtitle": subtitle_uri,
                    },
                    success=True,
                    output_refs=[primary_uri],
                ),
            ]

            for descriptions in (
                ["普通音频", "试听音频"],
                ["普通图片", "唯一音色参考"],
                ["音色参考", "人声参考"],
            ):
                with self.subTest(descriptions=descriptions):
                    changed_package = package.model_copy(
                        update={
                            "artifacts": [
                                artifact.model_copy(
                                    update={"description": description}
                                )
                                for artifact, description in zip(
                                    package.artifacts,
                                    descriptions,
                                )
                            ]
                        }
                    )
                    normalized = normalize_candidate_lineage(
                        candidate,
                        package=changed_package,
                        records=records,
                    )
                    claims = {
                        item.artifact_uri: item
                        for item in normalized.lineage_claims
                    }

                    self.assertEqual(
                        claims[narration_uri].input_artifact_uris,
                        [],
                    )
                    self.assertEqual(
                        claims[subtitle_uri].input_artifact_uris,
                        [],
                    )
                    self.assertEqual(
                        claims[shot_uri].input_artifact_uris,
                        [package.shared_visual_anchor.uri],
                    )
                    self.assertNotIn(
                        package.artifacts[0].uri,
                        claims[shot_uri].input_artifact_uris,
                    )
                    self.assertEqual(
                        claims[primary_uri].input_artifact_uris,
                        [shot_uri, narration_uri, subtitle_uri],
                    )

    def test_unplanned_bgm_is_optional_but_traced_when_used(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            bgm_path = root / "optional-bgm.wav"
            bgm_path.write_bytes(b"optional-bgm")
            bgm = Artifact(
                artifact_id="Task2-v1-artifact-1",
                artifact_type="audio",
                uri=str(bgm_path.resolve()),
                source_uri=None,
                description="Planner 未选择但 Executor 实际使用的 BGM",
                **content_identity_for_path(bgm_path),
            )
            mixed_uri = str((root / "mixed.wav").resolve())
            narration_uri = candidate.narration_artifact_uri
            shot_uri = candidate.shot_outputs[0].artifact_uri
            subtitle_uri = candidate.subtitle_artifact_uri
            primary_uri = candidate.primary_artifact_uri
            records = [
                ToolCallRecord(
                    tool_call_id="tts",
                    tool_name="synthesize_speech",
                    arguments={"text": "测试旁白"},
                    success=True,
                    output_refs=[narration_uri],
                ),
                ToolCallRecord(
                    tool_call_id="shot",
                    tool_name="make_video",
                    arguments={
                        "reference": package.shared_visual_anchor.uri
                    },
                    success=True,
                    output_refs=[shot_uri],
                ),
                ToolCallRecord(
                    tool_call_id="subtitle",
                    tool_name="create_ass_subtitles",
                    arguments={"text": "测试旁白"},
                    success=True,
                    output_refs=[subtitle_uri],
                ),
                ToolCallRecord(
                    tool_call_id="mix",
                    tool_name="mix_audio_tracks",
                    arguments={
                        "tracks": [
                            {"audio": narration_uri},
                            {"audio": bgm.uri},
                        ]
                    },
                    success=True,
                    output_refs=[mixed_uri],
                ),
                ToolCallRecord(
                    tool_call_id="primary",
                    tool_name="video_mux_audio",
                    arguments={
                        "video": shot_uri,
                        "audio": mixed_uri,
                        "subtitle": subtitle_uri,
                    },
                    success=True,
                    output_refs=[primary_uri],
                ),
            ]

            normalized = normalize_candidate_lineage(
                candidate,
                package=package,
                records=records,
                dependency_artifacts=[bgm],
            )

        mixed = next(
            item
            for item in normalized.lineage_claims
            if item.artifact_uri == mixed_uri
        )
        self.assertEqual(
            mixed.input_artifact_uris,
            [narration_uri, bgm.uri],
        )

    def test_accepted_image_audio_and_video_can_be_actual_inputs(
        self,
    ) -> None:
        for artifact_type, suffix in (
            ("image", ".png"),
            ("audio", ".wav"),
            ("video", ".mp4"),
        ):
            with (
                self.subTest(artifact_type=artifact_type),
                tempfile.TemporaryDirectory() as temp_dir,
            ):
                root = Path(temp_dir)
                package = _package(root)
                candidate = _candidate(root, package)
                upstream_path = root / f"accepted{suffix}"
                upstream_path.write_bytes(
                    f"accepted-{artifact_type}".encode()
                )
                upstream = ProductionArtifact(
                    artifact_id="Segment9-v1-artifact-1",
                    artifact_type=artifact_type,
                    uri=str(upstream_path.resolve()),
                    source_uri=None,
                    description=f"Accepted Segment {artifact_type}",
                    **content_identity_for_path(upstream_path),
                )
                record = ToolCallRecord(
                    tool_call_id=f"use-accepted-{artifact_type}",
                    tool_name="video_concat",
                    arguments={"upstream": upstream.uri},
                    success=True,
                    output_refs=[
                        item.uri for item in candidate.artifacts
                    ],
                )

                normalized = normalize_candidate_lineage(
                    candidate,
                    package=package,
                    records=[record],
                    dependency_artifacts=[upstream],
                )
                shot_uri = normalized.shot_outputs[0].artifact_uri
                shot = next(
                    item
                    for item in normalized.lineage_claims
                    if item.artifact_uri == shot_uri
                )
                self.assertEqual(
                    shot.input_artifact_uris,
                    [upstream.uri],
                )

    def test_failed_publish_alias_cannot_create_lineage(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            remote_reference = "https://failed.test/reference.png"
            shot_uri = candidate.shot_outputs[0].artifact_uri
            records = [
                ToolCallRecord(
                    tool_call_id="failed-publish",
                    tool_name="publish_media_reference",
                    arguments={
                        "local_path": package.artifacts[0].uri,
                    },
                    success=False,
                    output_refs=[remote_reference],
                ),
                ToolCallRecord(
                    tool_call_id="produce-shot",
                    tool_name="make_video",
                    arguments={"reference": remote_reference},
                    success=True,
                    output_refs=[shot_uri],
                ),
                ToolCallRecord(
                    tool_call_id="produce-narration",
                    tool_name="synthesize_speech",
                    arguments={"text": "测试旁白"},
                    success=True,
                    output_refs=[candidate.narration_artifact_uri],
                ),
                ToolCallRecord(
                    tool_call_id="produce-subtitle",
                    tool_name="create_ass_subtitles",
                    arguments={"text": "测试旁白"},
                    success=True,
                    output_refs=[candidate.subtitle_artifact_uri],
                ),
                ToolCallRecord(
                    tool_call_id="produce-primary",
                    tool_name="video_mux_audio",
                    arguments={
                        "video": shot_uri,
                        "audio": candidate.narration_artifact_uri,
                        "subtitle": candidate.subtitle_artifact_uri,
                    },
                    success=True,
                    output_refs=[candidate.primary_artifact_uri],
                ),
            ]

            with self.assertRaisesRegex(
                SegmentProvenanceError,
                "无法绑定到封印素材",
            ):
                normalize_candidate_lineage(
                    candidate,
                    package=package,
                    records=records,
                )

    def test_candidate_lineage_uses_latest_successful_producer(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            original = ToolCallRecord(
                tool_call_id="original-producer",
                tool_name="video_concat",
                arguments={"source": package.artifacts[0].uri},
                success=True,
                output_refs=[item.uri for item in candidate.artifacts],
            )
            corrected = ToolCallRecord(
                tool_call_id="corrected-producer",
                tool_name="video_concat",
                arguments={"source": package.artifacts[1].uri},
                success=True,
                output_refs=[candidate.primary_artifact_uri],
            )

            normalized = normalize_candidate_lineage(
                candidate,
                package=package,
                records=[original, corrected],
            )
            primary = next(
                item
                for item in normalized.lineage_claims
                if item.artifact_uri == candidate.primary_artifact_uri
            )

        self.assertEqual(
            primary.evidence_tool_call_ids,
            ["corrected-producer"],
        )

    def test_formal_shared_anchor_is_used_as_upstream_not_output(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_dir = root / "run"
            anchor_path = run_dir / "production_inputs" / "anchor.png"
            anchor_path.parent.mkdir(parents=True)
            Image.new("RGB", (16, 16), "white").save(anchor_path)
            package = _package(root).model_copy(
                update={
                    "shared_visual_anchor": seal_shared_visual_anchor(
                        anchor_path
                    )
                }
            )
            candidate = _candidate(root, package)
            anchor_url = "https://input.test/formal-anchor.png"
            records = [
                ToolCallRecord(
                    tool_call_id="publish-anchor",
                    tool_name="publish_media_reference",
                    arguments={"local_path": str(anchor_path.resolve())},
                    success=True,
                    output_refs=[anchor_url],
                ),
                ToolCallRecord(
                    tool_call_id="produce",
                    tool_name="video_concat",
                    arguments={"reference_images": [anchor_url]},
                    success=True,
                    output_refs=[
                        item.uri for item in candidate.artifacts
                    ],
                ),
            ]

            normalized = normalize_candidate_lineage(
                candidate,
                package=package,
                records=records,
            )

        shot = next(
            item
            for item in normalized.lineage_claims
            if item.artifact_uri
            == normalized.shot_outputs[0].artifact_uri
        )
        self.assertIn(
            package.shared_visual_anchor.uri,
            shot.input_artifact_uris,
        )
        self.assertNotIn(
            package.shared_visual_anchor.uri,
            {item.uri for item in normalized.artifacts},
        )

    def test_formal_shared_anchor_forbids_per_segment_seedream(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_dir = root / "run"
            anchor_path = run_dir / "production_inputs" / "anchor.png"
            anchor_path.parent.mkdir(parents=True)
            Image.new("RGB", (16, 16), "white").save(anchor_path)
            package = _package(root).model_copy(
                update={
                    "shared_visual_anchor": seal_shared_visual_anchor(
                        anchor_path
                    )
                }
            )
            candidate = _candidate(root, package)
            records = [
                ToolCallRecord(
                    tool_call_id="forbidden-seedream",
                    tool_name="run_tool",
                    remote_tool_id="seedream_generate_image",
                    success=True,
                )
            ]

            with self.assertRaisesRegex(
                SegmentProvenanceError,
                "不得各自重新生成",
            ):
                validate_required_production_trace(
                    candidate,
                    package=package,
                    records=records,
                    run_dir=run_dir,
                )

    def test_candidate_uses_trimmed_tts_consumed_by_formal_mix(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            trimmed = str((root / "trimmed.wav").resolve())
            records = [
                ToolCallRecord(
                    tool_call_id="base-producer",
                    tool_name="video_concat",
                    success=True,
                    output_refs=[item.uri for item in candidate.artifacts],
                ),
                ToolCallRecord(
                    tool_call_id="tts",
                    tool_name="synthesize_speech",
                    success=True,
                    output_refs=[candidate.narration_artifact_uri],
                ),
                ToolCallRecord(
                    tool_call_id="trim",
                    tool_name="audio_trim",
                    arguments={"audio": candidate.narration_artifact_uri},
                    success=True,
                    output_refs=[trimmed],
                ),
                ToolCallRecord(
                    tool_call_id="mix",
                    tool_name="mix_audio_tracks",
                    arguments={"tracks": [{"audio": trimmed}]},
                    success=True,
                    output_refs=[candidate.primary_artifact_uri],
                ),
            ]

            normalized = normalize_candidate_lineage(
                candidate,
                package=package,
                records=records,
            )

        self.assertEqual(normalized.narration_artifact_uri, trimmed)
        self.assertIn(trimmed, {item.uri for item in normalized.artifacts})

    def test_trace_rejects_seedance_after_final_render(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            records = [
                ToolCallRecord(
                    tool_call_id="render",
                    tool_name="render_ass_subtitles",
                    success=True,
                    output_refs=[candidate.primary_artifact_uri],
                ),
                ToolCallRecord(
                    tool_call_id="test-submit",
                    operation_id="segment:2",
                    tool_name="generate_seedance_video",
                    success=True,
                ),
            ]

            with self.assertRaisesRegex(
                SegmentProvenanceError,
                "禁止继续提交",
            ):
                validate_required_production_trace(
                    candidate,
                    package=package,
                    records=records,
                    run_dir=root / "run",
                )

    def test_trace_rejects_narration_asr_mismatch(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_dir = root / "run"
            run_dir.mkdir()
            package = _package(root)
            package_anchor = run_dir / "shared-anchor.png"
            Image.new("RGB", (16, 16), "white").save(package_anchor)
            package = package.model_copy(
                update={
                    "shared_visual_anchor": seal_shared_visual_anchor(
                        package_anchor
                    )
                }
            )
            candidate = _candidate(root, package)
            anchor_uri = str((run_dir / "anchor.png").resolve())
            tts_uri = str((run_dir / "tts.wav").resolve())
            anchor_url = "https://input.test/anchor.png"
            candidate = candidate.model_copy(
                update={
                    "artifacts": [
                        candidate.artifacts[0].model_copy(
                            update={
                                "artifact_type": "image",
                                "uri": anchor_uri,
                            }
                        ),
                        candidate.artifacts[0].model_copy(
                            update={"uri": tts_uri}
                        ),
                        *candidate.artifacts,
                    ],
                    "lineage_claims": [
                        ArtifactLineageClaim(
                            artifact_uri=anchor_uri,
                            input_artifact_uris=[
                                package.artifacts[0].uri
                            ],
                            production_input_ids=[
                                package.production_inputs[0].input_id
                            ],
                            evidence_tool_call_ids=["seedream"],
                        ),
                        ArtifactLineageClaim(
                            artifact_uri=tts_uri,
                            input_artifact_uris=[
                                package.artifacts[1].uri
                            ],
                            production_input_ids=[
                                package.production_inputs[0].input_id
                            ],
                            evidence_tool_call_ids=["tts"],
                        ),
                        *[
                            (
                                claim.model_copy(
                                    update={
                                        "input_artifact_uris": [tts_uri]
                                    }
                                )
                                if claim.artifact_uri
                                == candidate.narration_artifact_uri
                                else claim
                            )
                            for claim in candidate.lineage_claims
                        ],
                    ],
                }
            )
            records = [
                ToolCallRecord(
                    tool_call_id="seedream",
                    tool_name="run_tool",
                    remote_tool_id="seedream_generate_image",
                    success=True,
                    arguments={"prompt": "彩铅转绘"},
                    artifact_mappings=[
                        {
                            "source": "https://provider.test/anchor.png",
                            "local_path": anchor_uri,
                        }
                    ],
                    output_refs=[anchor_uri],
                ),
                ToolCallRecord(
                    tool_call_id="publish-anchor",
                    tool_name="publish_media_reference",
                    success=True,
                    arguments={"local_path": anchor_uri},
                    output_refs=[anchor_uri, anchor_url],
                ),
                ToolCallRecord(
                    tool_call_id="submit-shot",
                    tool_name="generate_seedance_video",
                    success=True,
                    arguments={"first_frame_url": anchor_url},
                ),
                ToolCallRecord(
                    tool_call_id="detect-anchor",
                    tool_name="detect_faces",
                    success=True,
                    arguments={"source": anchor_uri},
                    output_excerpt=json.dumps(
                        {
                            "decision": "NO_FACE",
                            "local_path": anchor_uri,
                            "input_sha256": "a" * 64,
                        }
                    ),
                ),
                ToolCallRecord(
                    tool_call_id="tts",
                    tool_name="synthesize_speech",
                    success=True,
                    arguments={
                        "text": expected_narration_text(package)
                    },
                    output_refs=[tts_uri],
                ),
                ToolCallRecord(
                    tool_call_id="asr",
                    tool_name="transcribe_audio",
                    success=True,
                    arguments={
                        "audio": candidate.narration_artifact_uri
                    },
                    output_excerpt=json.dumps(
                        {"transcript": "。"},
                        ensure_ascii=False,
                    ),
                ),
            ]

            with self.assertRaisesRegex(
                SegmentProvenanceError,
                "ASR 与权威完整旁白不一致",
            ):
                validate_required_production_trace(
                    candidate,
                    package=package,
                    records=records,
                    run_dir=run_dir,
                )

    def test_segment_skill_calls_seedance_provider_directly(self) -> None:
        skill = load_segment_executor_skill()

        self.assertIn("generate_seedance_video", skill.allowed_tools)
        self.assertIn("detect_faces", skill.allowed_tools)
        self.assertNotIn(
            "seedance_generate_video",
            skill.allowed_remote_tool_ids,
        )
        self.assertIn("直接调用顶层", skill.instructions)
        self.assertIn("三态结论", skill.instructions)

    def test_segment_remote_budget_is_configurable_per_shot(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            package = _package(Path(temp_dir))
            second_shot = package.shots[0].model_copy(
                update={
                    "shot_id": "Segment1-Shot2",
                    "start_ms": 1000,
                    "end_ms": 2000,
                }
            )
            package = package.model_copy(
                update={"shots": [package.shots[0], second_shot]}
            )
            with patch.dict(
                "os.environ",
                {
                    "SEGMENT_SEEDREAM_MAX_CALLS": "3",
                    "SEGMENT_SEEDANCE_MAX_CALLS_PER_SHOT": "6",
                },
                clear=False,
            ):
                limits = _remote_tool_call_limits()
                video_limit = _seedance_call_limit(package)

        self.assertEqual(limits["seedream_generate_image"], 3)
        self.assertEqual(video_limit, 12)

    def test_segment_remote_budget_rejects_invalid_config(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            package = _package(Path(temp_dir))
            for value in ("0", "-1", "not-an-int"):
                with (
                    self.subTest(value=value),
                    patch.dict(
                        "os.environ",
                        {"SEGMENT_SEEDREAM_MAX_CALLS": value},
                        clear=False,
                    ),
                    self.assertRaises(
                        SegmentExecutorConfigurationError
                    ),
                ):
                    _remote_tool_call_limits()

    def test_remote_budget_counts_only_journaled_remote_calls(self) -> None:
        records = [
            ToolCallRecord(
                tool_call_id="paid",
                operation_id="segment:1",
                tool_name="run_tool",
                remote_tool_id="seedance_generate_video",
                success=False,
            ),
            ToolCallRecord(
                tool_call_id="preflight-failure",
                tool_name="run_tool",
                remote_tool_id="seedance_generate_video",
                success=False,
            ),
            ToolCallRecord(
                tool_call_id="local",
                operation_id="segment:2",
                tool_name="video_concat",
                success=True,
            ),
        ]

        self.assertEqual(
            _remote_tool_call_counts(records),
            {"seedance_generate_video": 1},
        )

    def test_video_budget_counts_only_journaled_submissions(self) -> None:
        records = [
            ToolCallRecord(
                tool_call_id="submitted",
                operation_id="segment:1",
                tool_name="generate_seedance_video",
                success=True,
            ),
            ToolCallRecord(
                tool_call_id="preflight-failure",
                tool_name="generate_seedance_video",
                success=False,
            ),
        ]

        self.assertEqual(_seedance_call_count(records), 1)

    def test_expected_narration_merges_adjacent_overlap(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            package = _package(Path(temp_dir))
            production_input = package.production_inputs[0].model_copy(
                update={
                    "value": {
                        "子段落": [
                            {
                                "内容类型": "旁白",
                                "内容实质": "就是难道真的就",
                            },
                            {
                                "内容类型": "旁白",
                                "内容实质": "就是难道真的就接纳自己",
                            },
                            {
                                "内容类型": "旁白",
                                "内容实质": "接纳自己继续行动",
                            },
                        ]
                    }
                }
            )
            package = package.model_copy(
                update={"production_inputs": [production_input]}
            )

            self.assertEqual(
                _expected_narration_text(package),
                "就是难道真的就接纳自己继续行动",
            )

    def test_expected_narration_includes_dialogue_and_deduplicates_tail(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            package = _package(Path(temp_dir))
            production_input = package.production_inputs[0].model_copy(
                update={
                    "value": {
                        "子段落": [
                            {
                                "内容类型": "对话",
                                "内容实质": (
                                    "前半段完整独白，其实你也可以把它当做一个经历，"
                                    "然后慢慢地铸成内心更强大的自己。"
                                ),
                            },
                            {
                                "内容类型": "旁白",
                                "内容实质": (
                                    "其实你也可以把它当做一个经历，"
                                    "然后慢慢的铸成内心更强大的自己。"
                                ),
                            },
                        ]
                    }
                }
            )
            package = package.model_copy(
                update={"production_inputs": [production_input]}
            )

            self.assertEqual(
                _expected_narration_text(package),
                (
                    "前半段完整独白，其实你也可以把它当做一个经历，"
                    "然后慢慢地铸成内心更强大的自己。"
                ),
            )

    def _models(
        self,
        package: SegmentPackage,
        candidate: SegmentExecutorCandidate,
        *,
        validator_responses: list[AIMessage] | None = None,
    ) -> tuple[ToolAwareFakeChatModel, ToolAwareFakeChatModel]:
        return (
            ToolAwareFakeChatModel(
                responses=[
                    AIMessage(
                        content="",
                        tool_calls=[
                            {
                                "name": "video_concat",
                                "args": {
                                    "sources": [
                                        package.artifacts[0].uri,
                                        package.shared_visual_anchor.uri,
                                    ]
                                },
                                "id": "produce-segment",
                                "type": "tool_call",
                            }
                        ],
                    ),
                    AIMessage(
                        content="",
                        tool_calls=[
                            {
                                "name": "video_concat",
                                "args": {
                                    "sources": [
                                        candidate.shot_outputs[0].artifact_uri,
                                        candidate.narration_artifact_uri,
                                        candidate.subtitle_artifact_uri,
                                    ]
                                },
                                "id": "assemble-segment",
                                "type": "tool_call",
                            }
                        ],
                    ),
                    AIMessage(content=candidate.model_dump_json()),
                    AIMessage(content=candidate.model_dump_json()),
                    AIMessage(content=candidate.model_dump_json()),
                ]
            ),
            ToolAwareFakeChatModel(
                responses=(
                    validator_responses
                    if validator_responses is not None
                    else [
                        AIMessage(
                            content=_validator_candidate(
                                package
                            ).model_dump_json()
                        )
                    ]
                )
            ),
        )

    def test_fake_segment_runs_then_same_thread_is_exact_noop(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(root, output_paths)
            executor_model, validator_model = self._models(
                package,
                candidate,
            )

            first = run_segment_production(
                package,
                run_dir=root / "run",
                executor_model=executor_model,
                validator_model=validator_model,
                tool_registry=registry,
            )
            self.assertEqual(first.verdict, "PASS")
            before = _file_hashes(root / "run")
            executor_calls = executor_model.i
            validator_calls = validator_model.i

            with patch.object(
                publishing._SESSION,
                "get",
                side_effect=AssertionError("no-op 不得读取旧远程 URL"),
            ):
                second = run_segment_production(
                    package,
                    run_dir=root / "run",
                    executor_model=executor_model,
                    validator_model=validator_model,
                    tool_registry=registry,
                )
            after = _file_hashes(root / "run")

            self.assertEqual(second, first)
            self.assertEqual(after, before)
            self.assertEqual(executor_model.i, executor_calls)
            self.assertEqual(validator_model.i, validator_calls)

    def test_multi_input_package_only_requires_shot_commitments(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            extra_inputs = [
                ProductionInput(
                    input_id=f"ProductionInput-{index:064x}",
                    source_path=f"$.核心制作点[{index}]",
                    content_sha256=f"{index + 1:064x}",
                    value={"要求": f"完整输入池中的全局要求{index}"},
                    applies_to_source_segment_ids=[
                        package.source_segment_id
                    ],
                )
                for index in range(2, 7)
            ]
            package = package.model_copy(
                update={
                    "production_inputs": [
                        *package.production_inputs,
                        *extra_inputs,
                    ]
                }
            )
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(root, output_paths)
            executor_model, validator_model = self._models(
                package,
                candidate,
            )

            report = run_segment_production(
                package,
                run_dir=root / "run",
                executor_model=executor_model,
                validator_model=validator_model,
                tool_registry=registry,
            )

        self.assertEqual(report.verdict, "PASS")

    def test_validator_candidate_checkpoint_builds_report_without_tools(
        self,
    ) -> None:
        """Candidate 已封印而 Report 尚未提交时只能确定性补 Report。"""

        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_dir = root / "run"
            package = _package(root)
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(root, output_paths)
            executor_model, validator_model = self._models(
                package,
                candidate,
            )
            first = run_segment_production(
                package,
                run_dir=run_dir,
                executor_model=executor_model,
                validator_model=validator_model,
                tool_registry=registry,
            )
            before = _file_hashes(run_dir)
            report_path = segment_validation_report_path(
                run_dir,
                package.segment_id,
                package.plan_version,
            )
            report_path.unlink()

            with patch(
                "production_build_agents.agents.segment.run."
                "run_segment_validator",
                side_effect=AssertionError(
                    "已有 Validator Candidate 不得重调模型或证据工具"
                ),
            ):
                second = run_segment_production(
                    package,
                    run_dir=run_dir,
                    executor_model=executor_model,
                    validator_model=validator_model,
                    tool_registry=registry,
                )

            self.assertEqual(second, first)
            self.assertEqual(_file_hashes(run_dir), before)

    def test_missing_oss_stops_before_executor_agent(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            executor_model, validator_model = self._models(
                package,
                candidate,
            )
            empty_oss = {
                "ALIYUN_OSS_ACCESS_KEY_ID": "",
                "ALIYUN_OSS_ACCESS_KEY_SECRET": "",
                "ALIYUN_OSS_ENDPOINT": "",
                "ALIYUN_OSS_BUCKET": "",
                "ALIYUN_OSS_CDN_BASE_URL": "",
            }

            with (
                patch.dict("os.environ", empty_oss, clear=False),
                patch(
                    "production_build_agents.agents.segment.run."
                    "run_segment_executor"
                ) as executor,
                self.assertRaises(PublishingToolError),
            ):
                run_segment_production(
                    package,
                    run_dir=root / "run",
                    executor_model=executor_model,
                    validator_model=validator_model,
                )

            executor.assert_not_called()

    def test_executor_rejects_formal_artifacts_outside_run_dir(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(
                root,
                package,
                artifact_root=root / "outside",
            )
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            for path in output_paths:
                path.parent.mkdir(parents=True, exist_ok=True)
                path.write_bytes(b"outside")
            record = ToolCallRecord(
                tool_call_id="produce-segment",
                tool_name="video_concat",
                success=True,
                output_refs=[str(path.resolve()) for path in output_paths],
            )

            with self.assertRaisesRegex(
                SegmentExecutorOutputError,
                "必须位于当前 Run 目录",
            ):
                _resolve_artifacts(
                    candidate,
                    package,
                    [record],
                    run_dir=root / "run",
                )

    def test_completed_noop_rejects_tampered_primary_before_agents(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(root, output_paths)
            executor_model, validator_model = self._models(
                package,
                candidate,
            )
            run_segment_production(
                package,
                run_dir=root / "run",
                executor_model=executor_model,
                validator_model=validator_model,
                tool_registry=registry,
            )
            Path(candidate.primary_artifact_uri).write_bytes(b"tampered")

            with self.assertRaisesRegex(ValueError, "内容已变化"):
                run_segment_production(
                    package,
                    run_dir=root / "run",
                    executor_model=executor_model,
                    validator_model=validator_model,
                    tool_registry=registry,
                )

    def test_completed_noop_rejects_rewritten_validator_report(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_dir = root / "run"
            package = _package(root)
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(root, output_paths)
            executor_model, validator_model = self._models(
                package,
                candidate,
            )
            run_segment_production(
                package,
                run_dir=run_dir,
                executor_model=executor_model,
                validator_model=validator_model,
                tool_registry=registry,
            )
            report_path = segment_validation_report_path(
                run_dir,
                package.segment_id,
                package.plan_version,
            )
            payload = json.loads(report_path.read_text(encoding="utf-8"))
            payload["criterion_results"][0]["verdict"] = "FAIL"
            payload["verdict"] = "FAIL"
            report_path.write_text(
                json.dumps(payload, ensure_ascii=False),
                encoding="utf-8",
            )

            with self.assertRaisesRegex(
                ValueError,
                "Validator Candidate",
            ):
                run_segment_production(
                    package,
                    run_dir=run_dir,
                    executor_model=executor_model,
                    validator_model=validator_model,
                    tool_registry=registry,
                )

    def test_missing_final_audio_is_rejected_before_validator_model(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(
                root,
                output_paths,
                primary_has_audio=False,
            )
            executor_model, validator_model = self._models(
                package,
                candidate,
                validator_responses=[],
            )

            with self.assertRaisesRegex(ValueError, "视频流和音频流"):
                run_segment_production(
                    package,
                    run_dir=root / "run",
                    executor_model=executor_model,
                    validator_model=validator_model,
                    tool_registry=registry,
                )

    def test_wrong_final_transcript_is_rejected_before_model(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(
                root,
                output_paths,
                final_transcript="完全错误的旁白",
            )
            executor_model, validator_model = self._models(
                package,
                candidate,
                validator_responses=[],
            )

            with self.assertRaisesRegex(ValueError, "ASR 与正式旁白不一致"):
                run_segment_production(
                    package,
                    run_dir=root / "run",
                    executor_model=executor_model,
                    validator_model=validator_model,
                    tool_registry=registry,
                )

    def test_consistently_wrong_audio_and_subtitle_cannot_fake_pass(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(
                root,
                output_paths,
                narration_transcript="完全错误的旁白",
                final_transcript="完全错误的旁白",
                subtitle_text="完全错误的旁白",
            )
            executor_model, validator_model = self._models(
                package,
                candidate,
                validator_responses=[],
            )

            with self.assertRaisesRegex(ValueError, "权威旁白不一致"):
                run_segment_production(
                    package,
                    run_dir=root / "run",
                    executor_model=executor_model,
                    validator_model=validator_model,
                    tool_registry=registry,
                )

    def test_wrong_shot_resolution_is_rejected_before_model(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package = _package(root)
            candidate = _candidate(root, package)
            output_paths = [Path(item.uri) for item in candidate.artifacts]
            registry = _tool_registry(
                root,
                output_paths,
                video_width=720,
            )
            executor_model, validator_model = self._models(
                package,
                candidate,
                validator_responses=[],
            )

            with self.assertRaisesRegex(ValueError, "1080x1920"):
                run_segment_production(
                    package,
                    run_dir=root / "run",
                    executor_model=executor_model,
                    validator_model=validator_model,
                    tool_registry=registry,
                )

    def test_wrong_or_overlapping_ass_is_rejected_before_model(self) -> None:
        scenarios = (
            (
                {"subtitle_text": "完全错误字幕"},
                "字幕与正式旁白内容明显不一致",
            ),
            (
                {"subtitle_overlaps": [1]},
                "字幕存在时间重叠",
            ),
            (
                {"subtitle_duration": 2.5},
                "ASS 字幕 时长不匹配",
            ),
        )
        for registry_kwargs, message in scenarios:
            with self.subTest(message=message), tempfile.TemporaryDirectory() as temp_dir:
                root = Path(temp_dir)
                package = _package(root)
                candidate = _candidate(root, package)
                output_paths = [Path(item.uri) for item in candidate.artifacts]
                registry = _tool_registry(
                    root,
                    output_paths,
                    **registry_kwargs,
                )
                executor_model, validator_model = self._models(
                    package,
                    candidate,
                    validator_responses=[],
                )

                with self.assertRaisesRegex(ValueError, message):
                    run_segment_production(
                        package,
                        run_dir=root / "run",
                        executor_model=executor_model,
                        validator_model=validator_model,
                        tool_registry=registry,
                    )


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