feat(production): 补齐整片验收的完整需求与镜头证据

- 向 Whole Production Validator 注入完整 ProductionInput 原值、来源路径、Segment/Shot 分配关系和未分配要求

- 从不可变 Accepted 引用读取并校验 SegmentValidationReport，仅提供紧凑 PASS 摘要

- 在整片首尾和 Segment 边界之外增加每个 PlannedShot 中点抽帧，并携带抽样用途

- 保持 Production 0.7 正式 Candidate、Report、Graph、State 和持久化 schema 不变

production_build_agents/contracts/production_evaluation.py

@@ -172,6 +172,72 @@ def add_production_issue(
     )
 
 
+def required_shot_production_input_ids(
+    package: SegmentPackage,
+) -> list[str]:
+    """返回当前 Segment 的 Shot 已明确承诺实现的输入，保持 Package 顺序。"""
+
+    selected = {
+        input_id
+        for shot in package.shots
+        for input_id in shot.production_input_ids
+    }
+    return [
+        item.input_id
+        for item in package.production_inputs
+        if item.input_id in selected
+    ]
+
+
+def build_production_input_context(
+    catalog: ProductionInputCatalog,
+    plan: ProductionPlanContent,
+) -> dict[str, Any]:
+    """为整片语义验收展开完整输入值及其 Plan/Shot 分配关系。
+
+    该结构只是运行时派生证据，不是新的持久化合同。输入按 Catalog 顺序、
+    分配按 Plan/Shot 顺序稳定输出，便于 Validator 直接定位仍未分配到 Shot
+    的正式要求。
+    """
+
+    segment_assignments: dict[str, list[str]] = {
+        item.input_id: [] for item in catalog.inputs
+    }
+    shot_assignments: dict[str, list[dict[str, str]]] = {
+        item.input_id: [] for item in catalog.inputs
+    }
+    for segment in plan.segments:
+        for input_id in segment.production_input_ids:
+            if input_id in segment_assignments:
+                segment_assignments[input_id].append(segment.segment_id)
+        for shot in segment.shots:
+            for input_id in shot.production_input_ids:
+                if input_id in shot_assignments:
+                    shot_assignments[input_id].append(
+                        {
+                            "segment_id": segment.segment_id,
+                            "shot_id": shot.shot_id,
+                        }
+                    )
+
+    inputs = [
+        {
+            **item.model_dump(mode="json"),
+            "assigned_segment_ids": segment_assignments[item.input_id],
+            "assigned_shots": shot_assignments[item.input_id],
+        }
+        for item in catalog.inputs
+    ]
+    return {
+        "inputs": inputs,
+        "unassigned_to_shots": [
+            item["input_id"]
+            for item in inputs
+            if not item["assigned_shots"]
+        ],
+    }
+
+
 def evaluate_production_input_catalog(
     brief: ProductionBrief,
     catalog: ProductionInputCatalog,

production_build_agents/production/nodes/production_validation.py

@@ -13,6 +13,7 @@ from ...contracts.production_assembly_models import (
     ProductionAssemblyPackage,
 )
 from ...contracts.production_evaluation import ProductionContractError
+from ...contracts.production_models import ProductionInputCatalog
 from ...contracts.production_planning_models import ProductionPlan
 from ...contracts.production_validation_evaluation import (
     evaluate_production_validation_report,
@@ -61,6 +62,10 @@ def validate_production_node(
             state["current_production_plan_path"],
             ProductionPlan,
         )
+        production_inputs = load_model(
+            state["production_input_catalog_path"],
+            ProductionInputCatalog,
+        )
         assembly = load_model(
             state["current_assembly_package_path"],
             ProductionAssemblyPackage,
@@ -104,6 +109,7 @@ def validate_production_node(
             validator_invoked = True
             candidate = run_production_validator(
                 plan,
+                production_inputs,
                 assembly,
                 delivery,
                 inspection,

production_build_agents/production/validator.py

@@ -29,7 +29,13 @@ from ..contracts.production_assembly_models import (
     ProductionAssemblyDelivery,
     ProductionAssemblyPackage,
 )
+from ..contracts.document_identity import exact_document_sha256
 from ..contracts.production_execution_models import ImmutableJsonRef
+from ..contracts.production_evaluation import build_production_input_context
+from ..contracts.production_models import (
+    ProductionInputCatalog,
+    SegmentValidationReport,
+)
 from ..contracts.production_planning_models import ProductionPlan
 from ..contracts.production_validation_evaluation import (
     evaluate_production_validator_candidate,
@@ -74,9 +80,12 @@ def _system_prompt() -> str:
     return (
         "你是独立 Production Validator。代码已经完成身份、Artifact 内容、"
         "输出规格、时间线、音频、字幕、血缘和全局覆盖的确定性就绪检查；不得"
-        "否认或覆盖这些结果。你必须根据最终视频的真实 ASR、ASS 和抽帧，"
-        "按固定五项语义 criteria 以及全部相邻 Segment boundary 逐项验收。"
-        "evidence 必须引用输入中的具体事实或时间点。只输出完整 "
+        "否认或覆盖这些结果。你必须根据完整 ProductionInput 原值、未分配到 "
+        "Shot 的正式要求、各 Segment 的 PASS 摘要，以及最终视频的真实 "
+        "ASR、ASS、边界和 Shot 中点抽帧，按固定五项语义 criteria 以及全部"
+        "相邻 Segment boundary 逐项验收。Segment PASS 只代表局部历史结论，"
+        "不得代替整片独立判断。evidence 必须引用输入 ID、来源路径、报告事实"
+        "或具体时间点。只输出完整 "
         "ProductionValidatorCandidate JSON，不输出总体 verdict 或 markdown。"
         "候选合同 JSON Schema："
         + json.dumps(
@@ -101,8 +110,129 @@ def _subtitle_uri(delivery: ProductionAssemblyDelivery) -> str:
     )
 
 
+def _frame_sampling_plan(
+    plan: ProductionPlan,
+    assembly: ProductionAssemblyPackage,
+    *,
+    duration_sec: float,
+) -> list[dict[str, Any]]:
+    """派生首尾、Segment 边界和每个 Shot 中点的稳定抽帧计划。"""
+
+    upper_bound = max(0.0, duration_sec - 0.05)
+    purposes_by_timestamp: dict[float, list[dict[str, str]]] = {}
+
+    def add(timestamp_sec: float, **purpose: str) -> None:
+        timestamp = min(upper_bound, max(0.0, timestamp_sec))
+        purposes_by_timestamp.setdefault(timestamp, []).append(purpose)
+
+    add(0.0, kind="production_start")
+    add(upper_bound, kind="production_end")
+    for previous, following in zip(
+        assembly.timeline,
+        assembly.timeline[1:],
+    ):
+        boundary = previous.end_ms / 1000
+        add(
+            boundary - 0.1,
+            kind="segment_boundary_before",
+            previous_segment_id=previous.segment_id,
+            next_segment_id=following.segment_id,
+        )
+        add(
+            boundary + 0.1,
+            kind="segment_boundary_after",
+            previous_segment_id=previous.segment_id,
+            next_segment_id=following.segment_id,
+        )
+
+    timeline_by_segment = {
+        item.segment_id: item for item in assembly.timeline
+    }
+    for segment in plan.segments:
+        timeline = timeline_by_segment.get(segment.segment_id)
+        if timeline is None:
+            raise ValidatorEvidenceError(
+                f"Assembly 缺少 {segment.segment_id} 时间线"
+            )
+        for shot in segment.shots:
+            add(
+                (
+                    timeline.start_ms
+                    + (shot.start_ms + shot.end_ms) / 2
+                )
+                / 1000,
+                kind="shot_midpoint",
+                segment_id=segment.segment_id,
+                shot_id=shot.shot_id,
+            )
+
+    return [
+        {
+            "timestamp_sec": timestamp,
+            "purposes": purposes_by_timestamp[timestamp],
+        }
+        for timestamp in sorted(purposes_by_timestamp)
+    ]
+
+
+def _segment_validation_summaries(
+    assembly: ProductionAssemblyPackage,
+) -> list[dict[str, Any]]:
+    """从 Accepted 引用读取并校验正式 PASS Report 的紧凑摘要。"""
+
+    summaries: list[dict[str, Any]] = []
+    for accepted in assembly.accepted_segments:
+        ref = accepted.validation_report_ref
+        path = Path(ref.uri).resolve()
+        if not path.is_file():
+            raise ValidatorEvidenceError(
+                f"SegmentValidationReport 不存在：{path}"
+            )
+        content = path.read_bytes()
+        if exact_document_sha256(content) != ref.document_sha256:
+            raise ValidatorEvidenceError(
+                f"SegmentValidationReport 不可变引用已变化：{path}"
+            )
+        try:
+            report = SegmentValidationReport.model_validate_json(content)
+        except ValueError as exc:
+            raise ValidatorEvidenceError(
+                f"SegmentValidationReport 合同无效：{path}: {exc}"
+            ) from exc
+        if (
+            report.run_id != accepted.source_run_id
+            or report.plan_id != accepted.source_plan_id
+            or report.plan_version != accepted.source_plan_version
+            or report.segment_id != accepted.segment_id
+            or report.verdict != "PASS"
+        ):
+            raise ValidatorEvidenceError(
+                "Accepted Segment 的 ValidationReport 身份或 verdict 不一致："
+                f"{accepted.segment_id}"
+            )
+        summaries.append(
+            {
+                "segment_id": report.segment_id,
+                "source_plan_version": report.plan_version,
+                "verdict": report.verdict,
+                "criterion_results": [
+                    {
+                        "criterion": item.criterion,
+                        "verdict": item.verdict,
+                        "reason": item.reason,
+                    }
+                    for item in report.criterion_results
+                ],
+                "summary": report.summary,
+                "validation_report_ref": ref.model_dump(mode="json"),
+            }
+        )
+    return summaries
+
+
 def _collect_evidence(
     plan: ProductionPlan,
+    production_inputs: ProductionInputCatalog,
     assembly: ProductionAssemblyPackage,
     delivery: ProductionAssemblyDelivery,
     inspection: ProductionMediaInspection,
@@ -133,32 +263,36 @@ def _collect_evidence(
         {"subtitle": _subtitle_uri(delivery)},
     )
     duration_sec = inspection.duration_ms / 1000
-    timestamps = {0.0, max(0.0, duration_sec - 0.05)}
-    for boundary_ms in (
-        item.end_ms for item in assembly.timeline[:-1]
-    ):
-        boundary = boundary_ms / 1000
-        timestamps.add(max(0.0, boundary - 0.1))
-        timestamps.add(min(duration_sec - 0.05, boundary + 0.1))
+    sampling_plan = _frame_sampling_plan(
+        plan,
+        assembly,
+        duration_sec=duration_sec,
+    )
+    timestamps = [item["timestamp_sec"] for item in sampling_plan]
     frames_result = invoke_evidence_tool(
         tools["extract_frames"],
-        {"video": primary_uri, "timestamps": sorted(timestamps)},
+        {"video": primary_uri, "timestamps": timestamps},
     )
     frames = [
         {
-            "timestamp_sec": item.get("timestamp_sec"),
+            "requested_timestamp_sec": sample["timestamp_sec"],
+            "observed_timestamp_sec": item.get("timestamp_sec"),
+            "purposes": sample["purposes"],
             "source": item.get("local_path") or item.get("url"),
         }
-        for item in frames_result.get("frames") or []
+        for sample, item in zip(
+            sampling_plan,
+            frames_result.get("frames") or [],
+        )
         if isinstance(item, dict)
         and isinstance(item.get("local_path") or item.get("url"), str)
     ]
     if len(frames) != len(timestamps):
-        raise ValidatorEvidenceError("最终成片边界抽帧证据不完整")
+        raise ValidatorEvidenceError("最终成片时间线抽帧证据不完整")
     image_blocks = view_image_evidence(
         tools["view_images"],
         [str(item["source"]) for item in frames],
-        incomplete_prefix="最终成片抽帧证据不完整",
+        incomplete_prefix="最终成片时间线抽帧证据不完整",
     )
     return (
         {
@@ -173,6 +307,13 @@ def _collect_evidence(
                     preflight_report_ref.model_dump(mode="json")
                 ),
             },
+            "production_input_context": build_production_input_context(
+                production_inputs,
+                plan,
+            ),
+            "segment_validation_summaries": (
+                _segment_validation_summaries(assembly)
+            ),
             "production_plan": plan.model_dump(mode="json"),
             "assembly_package": assembly.model_dump(mode="json"),
             "assembly_delivery": delivery.model_dump(mode="json"),
@@ -181,7 +322,7 @@ def _collect_evidence(
             "semantic_evidence": {
                 "final_video_asr": transcript,
                 "final_subtitles": subtitles,
-                "boundary_frames": frames,
+                "timeline_frames": frames,
             },
         },
         image_blocks,
@@ -304,6 +445,7 @@ def _candidate_from_agent(
 
 def run_production_validator(
     plan: ProductionPlan,
+    production_inputs: ProductionInputCatalog,
     assembly: ProductionAssemblyPackage,
     delivery: ProductionAssemblyDelivery,
     inspection: ProductionMediaInspection,
@@ -339,6 +481,7 @@ def run_production_validator(
     )
     evidence, image_blocks = _collect_evidence(
         plan,
+        production_inputs,
         assembly,
         delivery,
         inspection,

tests/production/test_production_graph.py

@@ -47,6 +47,7 @@ from production_build_agents.contracts.production_models import (
     SEGMENT_VALIDATION_CRITERIA,
     ArtifactLineageClaim,
     SegmentDelivery,
+    ProductionInputCatalog,
     SegmentValidationReport,
     SegmentValidationCriterionResult,
     SegmentValidatorCandidate,
@@ -2327,6 +2328,12 @@ class ProductionGraphTest(unittest.TestCase):
                 update = validate_production_node(state)
 
             run_validator.assert_called_once()
+            catalog = ProductionInputCatalog.model_validate_json(
+                Path(
+                    str(state["production_input_catalog_path"])
+                ).read_bytes()
+            )
+            self.assertEqual(run_validator.call_args.args[1], catalog)
             self.assertEqual(update["status"], "RUNNING")
             self.assertEqual(update["phase"], "FINALIZE")
             self.assertTrue(candidate_path.is_file())

tests/production/test_validator.py

@@ -4,22 +4,33 @@ import tempfile
 import unittest
 from contextlib import contextmanager
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 from production_build_agents.production.validator import (
     ProductionValidatorOutputError,
+    _collect_evidence,
     _validate_candidate,
     build_production_validation_report,
     run_production_validator,
 )
+from production_build_agents.contracts.production_models import (
+    ProductionInput,
+    ProductionInputCatalog,
+)
 from production_build_agents.contracts.production_planning_models import (
     PlannedSegment,
     ProductionPlan,
 )
+from production_build_agents.production.segment_acceptance import (
+    immutable_json_ref,
+)
 from tests.support.production_fixtures import (
     ProductionScenario,
     immutable_ref,
+    production_input_id,
     readiness_records,
+    text_sha256,
     validation_records,
 )
 
@@ -34,6 +45,18 @@ class ProductionValidatorRuntimeTest(unittest.TestCase):
             self.delivery,
             self.assembly,
         )
+        self.catalog = ProductionInputCatalog(
+            production_brief_uri="/runs/production-brief.json",
+            production_brief_sha256="0" * 64,
+            source_segment_ids=[
+                item.source_segment_id for item in self.plan.segments
+            ],
+            inputs=[
+                production_input
+                for package in case.packages
+                for production_input in package.production_inputs
+            ],
+        )
         self.candidate, _ = validation_records(self.delivery)
 
     def test_failed_readiness_stops_before_model_or_evidence_tools(
@@ -57,6 +80,7 @@ class ProductionValidatorRuntimeTest(unittest.TestCase):
             ):
                 run_production_validator(
                     self.plan,
+                    self.catalog,
                     self.assembly,
                     self.delivery,
                     self.inspection,
@@ -176,6 +200,7 @@ class ProductionValidatorRuntimeTest(unittest.TestCase):
         ):
             result = run_production_validator(
                 plan,
+                self.catalog,
                 self.assembly,
                 self.delivery,
                 self.inspection,
@@ -210,6 +235,149 @@ class ProductionValidatorRuntimeTest(unittest.TestCase):
             f"{expected_run_id}:evidence",
         )
 
+    def test_evidence_includes_full_inputs_unassigned_requirements_reports_and_shot_frames(
+        self,
+    ) -> None:
+        unassigned = ProductionInput(
+            input_id=production_input_id(99),
+            source_path="$.制作表.全局要求.未分配",
+            content_sha256=text_sha256("未分配要求"),
+            value={"要求": "片尾必须出现完整行动号召"},
+            applies_to_source_segment_ids=["段落1"],
+        )
+        catalog = self.catalog.model_copy(
+            update={"inputs": [*self.catalog.inputs, unassigned]}
+        )
+        first_segment = self.plan.segments[0].model_copy(
+            update={
+                "production_input_ids": [
+                    *self.plan.segments[0].production_input_ids,
+                    unassigned.input_id,
+                ]
+            }
+        )
+        plan = self.plan.model_copy(
+            update={"segments": [first_segment, *self.plan.segments[1:]]}
+        )
+
+        with tempfile.TemporaryDirectory() as temporary:
+            report_dir = Path(temporary)
+            accepted = []
+            for report, reference in zip(
+                ProductionScenario().reports,
+                self.assembly.accepted_segments,
+            ):
+                path = report_dir / f"{report.segment_id}.json"
+                path.write_text(report.model_dump_json(), encoding="utf-8")
+                accepted.append(
+                    reference.model_copy(
+                        update={
+                            "validation_report_ref": immutable_json_ref(path)
+                        }
+                    )
+                )
+            assembly = self.assembly.model_copy(
+                update={"accepted_segments": accepted}
+            )
+            registry = MagicMock()
+            registry.resolve.return_value = [
+                SimpleNamespace(name=name)
+                for name in (
+                    "transcribe_audio",
+                    "inspect_ass_subtitles",
+                    "extract_frames",
+                    "view_images",
+                )
+            ]
+
+            def evidence(tool, arguments):
+                if tool.name == "transcribe_audio":
+                    return {"text": "完整旁白"}
+                if tool.name == "inspect_ass_subtitles":
+                    return {"cues": [{"text": "完整字幕"}]}
+                if tool.name == "extract_frames":
+                    return {
+                        "frames": [
+                            {
+                                "timestamp_sec": timestamp,
+                                "local_path": (
+                                    f"/tmp/frame-{index}.png"
+                                ),
+                            }
+                            for index, timestamp in enumerate(
+                                arguments["timestamps"]
+                            )
+                        ]
+                    }
+                raise AssertionError(tool.name)
+
+            with (
+                patch(
+                    "production_build_agents.production.validator."
+                    "invoke_evidence_tool",
+                    side_effect=evidence,
+                ) as invoke,
+                patch(
+                    "production_build_agents.production.validator."
+                    "view_image_evidence",
+                    return_value=[{"type": "image_url"}],
+                ),
+            ):
+                payload, image_blocks = _collect_evidence(
+                    plan,
+                    catalog,
+                    assembly,
+                    self.delivery,
+                    self.inspection,
+                    self.readiness,
+                    registry=registry,
+                    assembly_delivery_ref=immutable_ref(
+                        "assembly-delivery"
+                    ),
+                    preflight_report_ref=immutable_ref(
+                        "readiness-report"
+                    ),
+                )
+
+        input_context = payload["production_input_context"]
+        self.assertEqual(
+            input_context["unassigned_to_shots"],
+            [unassigned.input_id],
+        )
+        self.assertEqual(
+            input_context["inputs"][-1]["value"],
+            unassigned.value,
+        )
+        self.assertEqual(
+            [item["segment_id"] for item in payload[
+                "segment_validation_summaries"
+            ]],
+            ["Segment1", "Segment2"],
+        )
+        extract_call = next(
+            call
+            for call in invoke.call_args_list
+            if call.args[0].name == "extract_frames"
+        )
+        self.assertEqual(
+            extract_call.args[1]["timestamps"],
+            [0.0, 2.5, 4.9, 5.1, 7.5, 9.95],
+        )
+        frame_purposes = [
+            purpose
+            for frame in payload["semantic_evidence"]["timeline_frames"]
+            for purpose in frame["purposes"]
+        ]
+        self.assertEqual(
+            [
+                item["shot_id"]
+                for item in frame_purposes
+                if item["kind"] == "shot_midpoint"
+            ],
+            ["Segment1-Shot1", "Segment2-Shot1"],
+        )
+        self.assertEqual(image_blocks, [{"type": "image_url"}])
+
     def test_report_persists_same_versioned_validator_identity(
         self,
     ) -> None: