复用：按照历史 ArtifactGraph 判断 Segment 是否仍可 REUSE

Segment 复用身份继续覆盖计划业务输入、选中素材、Shot、输出规格与共享锚点，但不再因未使用的可选 Accepted 素材池变化而失效。

REUSE 前严格重验旧 PASS Bundle，并从历史 ArtifactGraph 找出实际消费的 Global Data 或 Accepted Segment Artifact；未使用素材变化允许复用，实际输入缺失或内容变化则转为 EXECUTE。

合法 Plan 业务输入变化不再误报为来源篡改，来源 Package、Delivery、Report 或 Artifact 被修改时仍然 fail closed；补充可选池增长、真实上游变化与篡改回归。

production_build_agents/contracts/production_planning_evaluation.py

@@ -1128,7 +1128,14 @@ def materialize_segment_package(
                 accepted_segments[dependency].validation_report_ref
             ),
         )
-        for dependency in segment.depends_on
+        for dependency in (
+            item.segment_id
+            for item in plan.segments
+            if (
+                item.segment_id != segment.segment_id
+                and item.segment_id in accepted_segments
+            )
+        )
     ]
     input_ids = set(segment.production_input_ids)
     artifact_ids = [item.artifact_id for item in segment.artifact_inputs]
@@ -1218,7 +1225,11 @@ def evaluate_segment_package(
 def segment_reuse_identity_sha256(
     package: SegmentPackage,
 ) -> str:
-    """覆盖全部业务执行输入，但排除路径、版本和绝对组装位置。"""
+    """覆盖计划业务输入，但排除路径、版本和未使用的 Accepted 素材池。
+
+    历史 Delivery 实际消费的动态上游由运行适配层依据 ArtifactGraph 另行
+    重验，避免可选池仅增加一个未使用素材就无条件重做 Segment。
+    """
 
     payload = {
         "plan_id": package.plan_id,
@@ -1260,18 +1271,6 @@ def segment_reuse_identity_sha256(
             "size_bytes": package.shared_visual_anchor.size_bytes,
             "mime_type": package.shared_visual_anchor.mime_type,
         },
-        "dependencies": [
-            {
-                "segment_id": item.segment_id,
-                "segment_delivery_sha256": (
-                    item.segment_delivery_ref.document_sha256
-                ),
-                "validation_report_sha256": (
-                    item.validation_report_ref.document_sha256
-                ),
-            }
-            for item in package.dependencies
-        ],
         "shots": [item.model_dump(mode="json") for item in package.shots],
     }
     return canonical_json_sha256(payload)

production_build_agents/production/segment_acceptance.py

@@ -13,7 +13,7 @@ from ..contracts.production_execution_models import (
     SegmentArtifactRef,
     SegmentPackage,
 )
-from ..contracts.production_models import ProductionArtifact
+from ..contracts.production_models import ProductionArtifact, SegmentDelivery
 from ..contracts.production_planning_evaluation import (
     segment_reuse_identity_sha256,
 )
@@ -21,6 +21,11 @@ from .segment_verification import (
     SegmentVerificationError,
     load_verified_segment_bundle,
 )
+from ..run.segment_inputs import (
+    SegmentInputIntegrityError,
+    VerifiedSegmentInputs,
+    load_verified_segment_inputs,
+)
 
 
 class SegmentAcceptanceError(ValueError):
@@ -119,6 +124,19 @@ def verify_accepted_segment_ref(
 ) -> AcceptedSegmentRef:
     """重验历史 Bundle，并确认其执行身份仍适用于当前 Package。"""
 
+    _verify_accepted_source_bundle(current_package, accepted)
+    current_identity = segment_reuse_identity_sha256(current_package)
+    if accepted.segment_reuse_identity_sha256 != current_identity:
+        raise SegmentAcceptanceError("AcceptedSegmentRef 执行输入身份已变化")
+    return accepted
+
+
+def _verify_accepted_source_bundle(
+    current_package: SegmentPackage,
+    accepted: AcceptedSegmentRef,
+) -> SegmentPackage:
+    """严格重验历史来源，但不要求它仍可用于当前 Package。"""
+
     source_package = _load_ref(
         accepted.segment_package_ref,
         SegmentPackage,
@@ -138,10 +156,70 @@ def verify_accepted_segment_ref(
     )
     if expected != accepted:
         raise SegmentAcceptanceError("AcceptedSegmentRef 或来源 Bundle 已变化")
-    current_identity = segment_reuse_identity_sha256(current_package)
-    if accepted.segment_reuse_identity_sha256 != current_identity:
-        raise SegmentAcceptanceError("AcceptedSegmentRef 执行输入身份已变化")
-    return accepted
+    return source_package
+
+
+def segment_reuse_inputs_compatible(
+    current_package: SegmentPackage,
+    accepted: AcceptedSegmentRef,
+    *,
+    current_inputs: VerifiedSegmentInputs | None = None,
+) -> bool:
+    """判断历史 Delivery 实际消费的上游在当前素材池中是否仍保持身份。
+
+    旧 Bundle 或其来源素材被篡改会抛出 ``SegmentAcceptanceError``；
+    当前合法素材池只是增删了未使用素材时返回 ``True``；旧 Delivery 实际
+    使用过的素材缺失或内容身份变化时返回 ``False``，由调用者选择 EXECUTE。
+    """
+
+    source_package = _verify_accepted_source_bundle(
+        current_package,
+        accepted,
+    )
+    if (
+        accepted.segment_reuse_identity_sha256
+        != segment_reuse_identity_sha256(current_package)
+    ):
+        return False
+    try:
+        source_inputs = load_verified_segment_inputs(source_package)
+        active_inputs = current_inputs or load_verified_segment_inputs(
+            current_package
+        )
+    except SegmentInputIntegrityError as exc:
+        raise SegmentAcceptanceError(str(exc)) from exc
+    source_delivery = _load_ref(
+        accepted.segment_delivery_ref,
+        SegmentDelivery,
+    )
+    produced_ids = {
+        item.artifact_id for item in source_delivery.artifacts
+    }
+    actual_upstream_ids = {
+        artifact_id
+        for lineage in source_delivery.artifact_graph.lineages
+        for artifact_id in lineage.input_artifact_ids
+        if artifact_id not in produced_ids
+    }
+    source_by_id = source_inputs.artifact_by_id(source_package)
+    current_by_id = active_inputs.artifact_by_id(current_package)
+    for artifact_id in actual_upstream_ids:
+        source = source_by_id.get(artifact_id)
+        if source is None:
+            raise SegmentAcceptanceError(
+                "历史 PASS Bundle 的实际上游不属于其封印素材池："
+                f"{artifact_id}"
+            )
+        current = current_by_id.get(artifact_id)
+        if current is None:
+            return False
+        if (
+            source.content_sha256 != current.content_sha256
+            or source.size_bytes != current.size_bytes
+            or source.mime_type != current.mime_type
+        ):
+            return False
+    return True
 
 
 def load_verified_accepted_segments(

tests/contracts/test_production_planning.py

@@ -25,6 +25,7 @@ from production_build_agents.contracts.production_evaluation import (
 )
 from production_build_agents.contracts.production_execution_models import (
     ImmutableJsonRef,
+    SegmentDependencyRef,
 )
 from production_build_agents.contracts.production_models import (
     ArtifactInputBinding,
@@ -1235,6 +1236,29 @@ class ProductionPlanningContractTest(unittest.TestCase):
         )
         self.assertEqual(materialized.output_profile, output_profile())
 
+    def test_segment_package_snapshots_all_accepted_not_only_dag_dependencies(
+        self,
+    ) -> None:
+        independent = self._plan(dependencies=((), ()))
+        accepted = segment_bundle(1, plan_version=1)[-1].model_copy(
+            update={"source_plan_id": "production-plan"}
+        )
+        materialized = materialize_segment_package(
+            self.brief,
+            self.delivery,
+            self.catalog,
+            self.package,
+            independent,
+            planning_package_ref=self.planning_ref,
+            run_id="production-run",
+            segment_id="Segment2",
+            accepted_segments={"Segment1": accepted},
+        )
+        self.assertEqual(
+            [item.segment_id for item in materialized.dependencies],
+            ["Segment1"],
+        )
+
     def test_segment_package_is_exact_recomputation_of_current_inputs(
         self,
     ) -> None:
@@ -1357,6 +1381,33 @@ class ProductionPlanningContractTest(unittest.TestCase):
             segment_reuse_identity_sha256(changed_profile),
         )
 
+    def test_reuse_identity_ignores_optional_accepted_pool_growth(
+        self,
+    ) -> None:
+        package = materialize_segment_package(
+            self.brief,
+            self.delivery,
+            self.catalog,
+            self.package,
+            self.plan,
+            planning_package_ref=self.planning_ref,
+            run_id="production-run",
+            segment_id="Segment1",
+            accepted_segments={},
+        )
+        dependency = SegmentDependencyRef(
+            segment_id="Segment2",
+            segment_delivery_ref=_ref("/accepted/Segment2-delivery.json"),
+            validation_report_ref=_ref("/accepted/Segment2-report.json"),
+        )
+        expanded = package.model_copy(
+            update={"dependencies": [dependency]}
+        )
+        self.assertEqual(
+            segment_reuse_identity_sha256(package),
+            segment_reuse_identity_sha256(expanded),
+        )
+
     def test_directive_is_computed_per_ready_segment(self) -> None:
         package = materialize_segment_package(
             self.brief,

tests/production/test_segment_acceptance.py

@@ -5,7 +5,16 @@ import unittest
 from pathlib import Path
 
 from production_build_agents.contracts.models import (
+    Artifact,
     CandidateArtifact,
+    GlobalDataStageDelivery,
+)
+from production_build_agents.contracts.document_identity import (
+    exact_document_sha256,
+)
+from production_build_agents.contracts.production_execution_models import (
+    ImmutableJsonRef,
+    SegmentDependencyRef,
 )
 from production_build_agents.run.artifacts import (
     ArtifactIntegrityError,
@@ -23,11 +32,22 @@ from production_build_agents.run.layout import (
 from production_build_agents.run.records import write_once_json
 from production_build_agents.production.segment_acceptance import (
     build_accepted_segment_ref,
+    segment_reuse_inputs_compatible,
 )
 from tests.support.production_fixtures import segment_bundle
+from tests.support.production_planner_fixtures import (
+    build_production_planner_scenario,
+)
 
 
-def _persist_pass_bundle(root: Path):
+def _persist_pass_bundle(
+    root: Path,
+    *,
+    include_unselected_actual_input: bool = False,
+    segment_index: int = 1,
+    run_id: str = "source-run",
+):
+    root.mkdir(parents=True, exist_ok=True)
     (
         package,
         candidate,
@@ -35,8 +55,8 @@ def _persist_pass_bundle(root: Path):
         validator_candidate,
         report,
         _,
-    ) = segment_bundle(1, run_id="source-run")
-    run_dir = root / "source-run"
+    ) = segment_bundle(segment_index, run_id=run_id)
+    run_dir = root / run_id
     source_path = root / "source.png"
     source_path.write_bytes(b"source")
     old_source_uri = package.artifacts[0].uri
@@ -57,6 +77,40 @@ def _persist_pass_bundle(root: Path):
             "shared_visual_anchor": shared_anchor,
         }
     )
+    global_artifacts = [source_artifact]
+    optional_artifact = None
+    if include_unselected_actual_input:
+        optional_path = root / "optional-audio.wav"
+        optional_path.write_bytes(b"optional-audio")
+        optional_artifact = Artifact(
+            artifact_id="Task2-v1-artifact-1",
+            artifact_type="audio",
+            uri=str(optional_path.resolve()),
+            source_uri=None,
+            description="Planner 未选择但 Executor 实际使用的素材",
+            **content_identity_for_path(optional_path),
+        )
+        global_artifacts.append(optional_artifact)
+    global_delivery = build_production_planner_scenario(
+        run_id="source-global"
+    ).global_data_delivery.model_copy(
+        update={"active_artifacts": global_artifacts}
+    )
+    global_delivery_path = root / "global_data_stage_delivery.json"
+    global_delivery_path.write_text(
+        global_delivery.model_dump_json(),
+        encoding="utf-8",
+    )
+    package = package.model_copy(
+        update={
+            "global_data_delivery_ref": ImmutableJsonRef(
+                uri=str(global_delivery_path.resolve()),
+                document_sha256=exact_document_sha256(
+                    global_delivery_path.read_bytes()
+                ),
+            )
+        }
+    )
 
     old_to_new = {
         old_source_uri: str(source_path.resolve()),
@@ -99,10 +153,18 @@ def _persist_pass_bundle(root: Path):
                         "input_artifact_uris": [
                             old_to_new.get(uri, uri)
                             for uri in claim.input_artifact_uris
-                        ],
+                        ]
+                        + (
+                            [optional_artifact.uri]
+                            if (
+                                optional_artifact is not None
+                                and index == 0
+                            )
+                            else []
+                        ),
                     }
                 )
-                for claim in candidate.lineage_claims
+                for index, claim in enumerate(candidate.lineage_claims)
             ],
             "shot_outputs": [
                 claim.model_copy(
@@ -171,9 +233,19 @@ def _persist_pass_bundle(root: Path):
                                     for artifact_id
                                     in lineage.input_artifact_ids
                                 ]
+                                + (
+                                    [optional_artifact.artifact_id]
+                                    if (
+                                        optional_artifact is not None
+                                        and index == 0
+                                    )
+                                    else []
+                                )
                             }
                         )
-                        for lineage in delivery.artifact_graph.lineages
+                        for index, lineage in enumerate(
+                            delivery.artifact_graph.lineages
+                        )
                     ]
                 }
             ),
@@ -207,6 +279,17 @@ def _persist_pass_bundle(root: Path):
 
 
 class SegmentAcceptanceTest(unittest.TestCase):
+    def _accepted(self, package, run_dir):
+        return build_accepted_segment_ref(
+            package,
+            run_dir=run_dir,
+            package_record_path=segment_package_path(
+                run_dir,
+                package.segment_id,
+                package.plan_version,
+            ),
+        )
+
     def test_complete_current_pass_bundle_can_be_accepted(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             package, run_dir = _persist_pass_bundle(Path(temp_dir))
@@ -261,6 +344,231 @@ class SegmentAcceptanceTest(unittest.TestCase):
                     ),
                 )
 
+    def test_optional_global_pool_growth_does_not_break_reuse(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            package, run_dir = _persist_pass_bundle(root)
+            accepted = self._accepted(package, run_dir)
+            delivery_path = Path(package.global_data_delivery_ref.uri)
+            global_delivery = GlobalDataStageDelivery.model_validate_json(
+                delivery_path.read_bytes()
+            )
+            optional_path = root / "new-optional.wav"
+            optional_path.write_bytes(b"new optional")
+            optional = Artifact(
+                artifact_id="Task2-v1-artifact-1",
+                artifact_type="audio",
+                uri=str(optional_path.resolve()),
+                source_uri=None,
+                description="新增但未使用的可选素材",
+                **content_identity_for_path(optional_path),
+            )
+            current_delivery = global_delivery.model_copy(
+                update={
+                    "active_artifacts": [
+                        *global_delivery.active_artifacts,
+                        optional,
+                    ]
+                }
+            )
+            current_delivery_path = root / "current-global-delivery.json"
+            current_delivery_path.write_text(
+                current_delivery.model_dump_json(),
+                encoding="utf-8",
+            )
+            current = package.model_copy(
+                update={
+                    "plan_version": 2,
+                    "global_data_delivery_ref": ImmutableJsonRef(
+                        uri=str(current_delivery_path.resolve()),
+                        document_sha256=exact_document_sha256(
+                            current_delivery_path.read_bytes()
+                        ),
+                    ),
+                }
+            )
+            self.assertTrue(
+                segment_reuse_inputs_compatible(current, accepted)
+            )
+
+    def test_authorized_business_input_change_forces_execute(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            package, run_dir = _persist_pass_bundle(root)
+            accepted = self._accepted(package, run_dir)
+            current = package.model_copy(
+                update={
+                    "plan_version": 2,
+                    "shots": [
+                        shot.model_copy(
+                            update={"objective": "授权修订后的镜头目标"}
+                        )
+                        for shot in package.shots
+                    ],
+                }
+            )
+
+            self.assertFalse(
+                segment_reuse_inputs_compatible(current, accepted)
+            )
+
+    def test_unused_accepted_segment_pool_growth_does_not_break_reuse(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            package, run_dir = _persist_pass_bundle(root / "current")
+            accepted = self._accepted(package, run_dir)
+            dependency_package, dependency_run_dir = _persist_pass_bundle(
+                root / "dependency",
+                segment_index=2,
+                run_id="dependency-run",
+            )
+            dependency_accepted = self._accepted(
+                dependency_package,
+                dependency_run_dir,
+            )
+            current = package.model_copy(
+                update={
+                    "plan_version": 2,
+                    "dependencies": [
+                        SegmentDependencyRef(
+                            segment_id=dependency_accepted.segment_id,
+                            segment_delivery_ref=(
+                                dependency_accepted.segment_delivery_ref
+                            ),
+                            validation_report_ref=(
+                                dependency_accepted.validation_report_ref
+                            ),
+                        )
+                    ],
+                }
+            )
+            self.assertTrue(
+                segment_reuse_inputs_compatible(current, accepted)
+            )
+
+    def test_missing_actually_used_optional_artifact_forces_execute(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            package, run_dir = _persist_pass_bundle(
+                root,
+                include_unselected_actual_input=True,
+            )
+            accepted = self._accepted(package, run_dir)
+            source_delivery = GlobalDataStageDelivery.model_validate_json(
+                Path(package.global_data_delivery_ref.uri).read_bytes()
+            )
+            reduced = source_delivery.model_copy(
+                update={
+                    "active_artifacts": [
+                        source_delivery.active_artifacts[0]
+                    ]
+                }
+            )
+            reduced_path = root / "reduced-global-delivery.json"
+            reduced_path.write_text(
+                reduced.model_dump_json(),
+                encoding="utf-8",
+            )
+            current = package.model_copy(
+                update={
+                    "plan_version": 2,
+                    "global_data_delivery_ref": ImmutableJsonRef(
+                        uri=str(reduced_path.resolve()),
+                        document_sha256=exact_document_sha256(
+                            reduced_path.read_bytes()
+                        ),
+                    ),
+                }
+            )
+            self.assertFalse(
+                segment_reuse_inputs_compatible(current, accepted)
+            )
+            changed_path = root / "changed-optional-audio.wav"
+            changed_path.write_bytes(b"changed optional audio")
+            changed_optional = source_delivery.active_artifacts[1].model_copy(
+                update={
+                    "uri": str(changed_path.resolve()),
+                    **content_identity_for_path(changed_path),
+                }
+            )
+            changed_delivery = source_delivery.model_copy(
+                update={
+                    "active_artifacts": [
+                        source_delivery.active_artifacts[0],
+                        changed_optional,
+                    ]
+                }
+            )
+            changed_delivery_path = root / "changed-global-delivery.json"
+            changed_delivery_path.write_text(
+                changed_delivery.model_dump_json(),
+                encoding="utf-8",
+            )
+            changed_current = package.model_copy(
+                update={
+                    "plan_version": 2,
+                    "global_data_delivery_ref": ImmutableJsonRef(
+                        uri=str(changed_delivery_path.resolve()),
+                        document_sha256=exact_document_sha256(
+                            changed_delivery_path.read_bytes()
+                        ),
+                    ),
+                }
+            )
+            self.assertFalse(
+                segment_reuse_inputs_compatible(
+                    changed_current,
+                    accepted,
+                )
+            )
+
+    def test_reuse_compatibility_fails_closed_on_source_bundle_tamper(
+        self,
+    ) -> None:
+        for target in ("artifact", "report"):
+            with self.subTest(target=target):
+                with tempfile.TemporaryDirectory() as temp_dir:
+                    root = Path(temp_dir)
+                    package, run_dir = _persist_pass_bundle(root)
+                    accepted = self._accepted(package, run_dir)
+                    current = package.model_copy(
+                        update={"plan_version": 2}
+                    )
+                    if target == "artifact":
+                        from production_build_agents.contracts.production_models import (
+                            SegmentDelivery,
+                        )
+
+                        delivery = SegmentDelivery.model_validate_json(
+                            Path(
+                                accepted.segment_delivery_ref.uri
+                            ).read_bytes()
+                        )
+                        artifact = next(
+                            item
+                            for item in delivery.artifacts
+                            if item.artifact_id
+                            == delivery.subtitle_artifact_id
+                        )
+                        Path(artifact.uri).write_text(
+                            "tampered",
+                            encoding="utf-8",
+                        )
+                    else:
+                        Path(accepted.validation_report_ref.uri).write_text(
+                            "{}",
+                            encoding="utf-8",
+                        )
+                    with self.assertRaises(ValueError):
+                        segment_reuse_inputs_compatible(
+                            current,
+                            accepted,
+                        )
+
 
 if __name__ == "__main__":
     unittest.main()