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()