from __future__ import annotations import hashlib import json import tempfile import unittest from pathlib import Path from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langchain_core.tools import tool from production_build_agents.capabilities import CAPABILITIES from production_build_agents.contracts.brief_paths import ( build_allowed_source_paths, resolve_direct_json_path, ) from production_build_agents.preprocess.brief_io import load_production_brief from production_build_agents.contracts.models import ( CandidateArtifact, ExecutorCandidate, ArtifactBindingClaim, ) from production_build_agents.agents.executor.agent import ( ExecutorOutputError, run_executor_agent, ) from production_build_agents.agents.invocation import ( CORRECTION_PRESERVATION_INSTRUCTION, ) from production_build_agents.agents.executor.context import ( build_executor_system_prompt, build_executor_user_payload, create_executor_run_id, load_task_contract, ) from production_build_agents.agents.executor.provenance import ( ExecutorProvenanceError, validate_candidate_provenance, ) from production_build_agents.agents.tool_trace import build_tool_call_records from production_build_agents.agents.executor.skills.registry import ( ExecutorSkillError, load_executor_skill, ) from production_build_agents.run.records import save_executor_delivery from production_build_agents.run.langgraph_checkpointer import ( checkpoint_messages, create_sqlite_checkpointer, ) from production_build_agents.tools.registry import ToolRegistry, create_default_tool_registry from tests.support.executor_fixtures import ( build_executor_candidate, build_executor_model, ) from tests.support.fake_models import ToolAwareFakeChatModel from tests.support.planner_fixtures import ( build_planned_task, build_task_package, ) BRIEF_PATH = ( Path(__file__).parents[1] / "fixtures" / "minimal_brief.json" ) IMAGE_URL = "https://example.test/generated-reference.png" SOURCE_URL = "https://example.test/research-source" VIDEO_URL = "https://example.test/generated-video.mp4" FRAME_URL = "https://example.test/generated-video-frame.jpg" _FAKE_MEDIA_TEMP = tempfile.TemporaryDirectory() _FAKE_MEDIA_DIR = Path(_FAKE_MEDIA_TEMP.name) def _fake_media_path(source: str, suffix: str) -> Path: path = _FAKE_MEDIA_DIR / ( hashlib.sha256(source.encode("utf-8")).hexdigest() + suffix ) path.write_bytes(source.encode("utf-8")) return path def _fake_tool_registry() -> ToolRegistry: @tool("search_tool") def search_tool(query: str, limit: int = 5) -> dict: """发现测试工具。""" return { "success": True, "results": [{"tool_id": "fake-tool", "summary": query}], "_duration_ms": 3, } @tool("inspect_tool") def inspect_tool(tool_ids: list[str]) -> dict: """返回测试工具参数。""" return { "success": True, "tools": {tool_id: {"input_schema": {}} for tool_id in tool_ids}, "_duration_ms": 4, } @tool("run_tool") def run_tool(tool_id: str, params: dict) -> dict: """执行测试工具。""" url = params.get("result_url") or { "image": IMAGE_URL, "video": VIDEO_URL, }.get(params.get("kind"), SOURCE_URL) return { "success": True, "tool_id": tool_id, "url": url, "_duration_ms": 5, } @tool("view_images") def view_images(image_sources: list[str]) -> dict: """查看测试图片。""" return { "success": True, "images": [ { "source": source, "local_path": str( _fake_media_path(source, ".png") ), "data_url": ( "data:image/png;base64," "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB" ), } for source in image_sources ], "_duration_ms": 6, } @tool("detect_faces") def detect_faces(source: str) -> dict: """返回测试图片的人脸三态判断。""" return { "success": True, "decision": "NO_FACE", "source": source, "local_path": source, "input_sha256": "a" * 64, } @tool("crop_image") def crop_image(image: str, box: list[float]) -> dict: """裁剪测试图片。""" return {"success": True, "image_urls": [image]} @tool("grid_collage") def grid_collage(images: list[str]) -> dict: """拼接测试图片。""" return {"success": True, "image_urls": images[:1]} @tool("overlay_text") def overlay_text(image: str, texts: list[dict]) -> dict: """叠加测试文字。""" return {"success": True, "image_urls": [image]} @tool("extract_frames") def extract_frames(video: str, num_frames: int = 5) -> dict: """抽取测试视频帧。""" return { "success": True, "source": video, "local_path": str(_fake_media_path(video, ".mp4")), "frames": [ { "source": FRAME_URL, "timestamp_sec": 0, } ], "_duration_ms": 7, } @tool("video_trim") def video_trim(video: str, start_sec: float = 0) -> dict: """裁剪测试视频。""" return {"success": True, "video_urls": [video]} @tool("video_concat") def video_concat(videos: list[str]) -> dict: """拼接测试视频。""" return {"success": True, "video_urls": videos[:1]} @tool("video_mux_audio") def video_mux_audio(video: str, audio: str) -> dict: """合入测试音频。""" return {"success": True, "video_urls": [video]} tools = [ search_tool, inspect_tool, run_tool, detect_faces, view_images, crop_image, grid_collage, overlay_text, extract_frames, video_trim, video_concat, video_mux_audio, ] return ToolRegistry({item.name: item for item in tools}) def _tool_call(name: str, args: dict, call_id: str) -> AIMessage: return AIMessage( content="", tool_calls=[ { "name": name, "args": args, "id": call_id, "type": "tool_call", } ], ) class ExecutorSkillTest(unittest.TestCase): def test_generic_skills_have_distinct_tool_boundaries(self) -> None: analysis = load_executor_skill("structured-analysis") research = load_executor_skill("external-research") image = load_executor_skill("image-production") video = load_executor_skill("video-production") reference = load_executor_skill("reference-inspection") self.assertEqual(analysis.allowed_tools, ()) self.assertIn("ProductionBrief", analysis.instructions) self.assertIn("派生阅读视图", analysis.instructions) self.assertEqual( research.allowed_tools, ("search_tool", "inspect_tool", "run_tool"), ) self.assertIn("view_images", image.allowed_tools) self.assertIn("crop_image", image.allowed_tools) self.assertEqual( image.allowed_remote_tool_ids, ( "seedream_generate_image", "gpt-image", "nano_banana_image", ), ) self.assertNotIn("detect_faces", image.allowed_tools) self.assertNotIn("detect_faces", reference.allowed_tools) self.assertIn("detect_faces", video.allowed_tools) self.assertIn("extract_frames", video.allowed_tools) self.assertEqual( video.allowed_remote_tool_ids, ("seedream_generate_image", "seedance_generate_video"), ) self.assertTrue( research.verification_policy.require_inspect_before_run ) self.assertEqual( image.verification_policy.final_artifact_check_tool, "view_images", ) self.assertEqual( video.verification_policy.final_artifact_check_tool, "extract_frames", ) self.assertEqual( reference.allowed_tools, ("probe_media", "view_images", "extract_frames"), ) self.assertEqual( set(CAPABILITIES[reference.skill_id].artifact_types), {"image", "video", "audio"}, ) def test_unknown_skill_is_rejected(self) -> None: with self.assertRaisesRegex(ExecutorSkillError, "未注册"): load_executor_skill("unknown") class ExecutorContextTest(unittest.TestCase): def test_model_receives_only_requested_brief_excerpt(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: task = build_task_package( BRIEF_PATH, planned_task=build_planned_task( source_paths=["$.核心制作点[0]"] ), plan_path=Path(temp_dir) / "plan.json", ) brief, plan, planned_task = load_task_contract(task) expectation = plan.stage_requirements[ 0 ].artifact_expectations[0] sibling = expectation.model_copy( update={ "expectation_id": "Requirement1-Expectation2", "usage_scope": "不属于当前 Task 的兄弟期待", } ) requirement = plan.stage_requirements[0].model_copy( update={ "artifact_expectations": [expectation, sibling] } ) plan = plan.model_copy( update={"stage_requirements": [requirement]} ) Path(task.plan_uri).write_text( plan.model_dump_json(indent=2), encoding="utf-8", ) brief, plan, planned_task = load_task_contract(task) run_id = create_executor_run_id(task) payload = build_executor_user_payload( executor_run_id=run_id, task=task, brief=brief, plan=plan, planned_task=planned_task, ) self.assertEqual( set(payload["production_brief_excerpt"]), {"$.核心制作点[0]"}, ) self.assertNotIn("production_brief", payload) self.assertEqual( payload["allowed_source_paths"], planned_task.source_paths, ) self.assertEqual( payload["resolved_task"], planned_task.model_dump(mode="json"), ) self.assertNotIn("plan", payload) self.assertNotIn("stage_requirements", payload) self.assertNotIn("requirements", payload) self.assertEqual( [ item["expectation"]["expectation_id"] for item in payload["selected_expectations"] ], ["Requirement1-Expectation1"], ) self.assertNotIn( "Requirement1-Expectation2", json.dumps(payload, ensure_ascii=False), ) self.assertEqual( set(payload["task_package"]), { "schema_version", "run_id", "plan_id", "plan_version", "task_id", "production_brief_uri", "plan_uri", "dependency_deliveries", }, ) prompt = build_executor_system_prompt( load_executor_skill(planned_task.skill_id) ) self.assertIn("structured-analysis", prompt) self.assertNotIn("{{SKILL_ID}}", prompt) def test_shared_json_path_reader_keeps_stable_boundaries(self) -> None: _, brief = load_production_brief(BRIEF_PATH) payload = brief.model_dump(mode="json", by_alias=True) self.assertIn("$.制作表.形式结果[0]", build_allowed_source_paths(brief)) self.assertEqual( resolve_direct_json_path(payload, "$.帖子类型"), payload["帖子类型"], ) def test_dependency_manifest_loads_only_summary_and_artifact_refs(self) -> None: first_task = build_task_package(BRIEF_PATH) with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) delivery = run_executor_agent( first_task, run_dir=root, model=build_executor_model(build_executor_candidate()), tool_registry=_fake_tool_registry(), ) manifest = save_executor_delivery(root, delivery) second_task = build_task_package( BRIEF_PATH, planned_task=build_planned_task( task_id="Task2", depends_on=["Task1"], ), dependency_artifacts={"Task1": str(manifest)}, ) brief, plan, planned_task = load_task_contract(second_task) payload = build_executor_user_payload( executor_run_id=create_executor_run_id(second_task), task=second_task, brief=brief, plan=plan, planned_task=planned_task, ) dependency = payload["dependency_artifacts"][0] self.assertEqual(dependency["task_id"], "Task1") self.assertIsInstance(dependency["summary"], str) self.assertEqual(dependency["summary"], delivery.summary) self.assertEqual( dependency["artifacts"][0]["uri"], delivery.artifacts[0].uri, ) self.assertNotIn("tool_calls", dependency) class GeneralExecutorTest(unittest.TestCase): def test_structured_candidate_must_reference_every_task_source_path( self, ) -> None: planned = build_planned_task( source_paths=[ "$.核心制作点[0]", "$.制作表.形式结果[0]", ] ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( source_paths=["$.核心制作点[0]"] ) with self.assertRaisesRegex( ExecutorProvenanceError, "未引用当前 Task 的全部来源路径", ): validate_candidate_provenance( candidate, task, planned, [], skill=load_executor_skill("structured-analysis"), input_refs=set(), ) def test_structured_candidate_counts_nested_payload_source_paths( self, ) -> None: planned = build_planned_task( source_paths=[ "$.核心制作点[0]", "$.制作表.形式结果[0]", ] ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( source_paths=["$.核心制作点[0]"] ).model_copy( update={ "payload": { "sections": [ { "content": "保留形式结果中的精确规则", "source_paths": ["$.制作表.形式结果[0]"], } ] } } ) validate_candidate_provenance( candidate, task, planned, [], skill=load_executor_skill("structured-analysis"), input_refs=set(), ) def test_same_rule_in_different_scopes_keeps_both_source_paths( self, ) -> None: planned = build_planned_task( source_paths=[ "$.制作表.段落结构[0]", "$.制作表.段落结构[1]", ] ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate().model_copy( update={ "findings": [], "payload": { "sections": [ { "content": "主体居中", "scope": "00:00:00-00:00:05", "source_paths": ["$.制作表.段落结构[0]"], }, { "content": "主体居中", "scope": "00:00:05-00:00:10", "source_paths": ["$.制作表.段落结构[1]"], }, ] }, } ) validate_candidate_provenance( candidate, task, planned, [], skill=load_executor_skill("structured-analysis"), input_refs=set(), ) def test_reference_source_must_match_probe_local_mapping(self) -> None: source_a = "https://example.test/a.png" source_b = "https://example.test/b.png" local_a = str(BRIEF_PATH.resolve()) planned = build_planned_task( skill_id="reference-inspection", deliverable_type="reference_collection", ) task = build_task_package(BRIEF_PATH, planned_task=planned) _, _, resolved_task = load_task_contract(task) candidate = ExecutorCandidate( run_id=task.run_id, plan_id=task.plan_id, task_id=task.task_id, plan_version=task.plan_version, deliverable_type="reference_collection", artifacts=[ CandidateArtifact( artifact_type="image", uri=local_a, source_uri=source_b, description="错误绑定的参考图", ) ], artifact_binding_claims=[ ArtifactBindingClaim( expectation_id="Requirement1-Expectation1", artifact_uri=local_a, ) ], summary="错误地把 A 的缓存声明为 B", ) messages = [ AIMessage( content="", tool_calls=[ { "name": "probe_media", "args": {"source": source_a}, "id": "probe-a", "type": "tool_call", }, { "name": "view_images", "args": {"image_sources": [local_a]}, "id": "view-a", "type": "tool_call", }, ], ), ToolMessage( content=json.dumps( { "success": True, "source": source_a, "local_path": local_a, "media_type": "image", } ), tool_call_id="probe-a", name="probe_media", ), ToolMessage( content=json.dumps( { "success": True, "images": [ { "source": local_a, "local_path": local_a, } ], } ), tool_call_id="view-a", name="view_images", ), ] with self.assertRaisesRegex( ExecutorProvenanceError, "probe_media", ): validate_candidate_provenance( candidate, task, resolved_task, build_tool_call_records(messages), skill=load_executor_skill("reference-inspection"), input_refs={source_b}, ) def test_reference_inspection_adopts_real_viewed_image(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) image_path = root / "existing-reference.png" from PIL import Image Image.new("RGB", (24, 24), "green").save(image_path) brief_payload = json.loads(BRIEF_PATH.read_text(encoding="utf-8")) brief_payload["核心制作点"][0]["测试已有媒体"] = str(image_path) brief_path = root / "production_brief.json" brief_path.write_text( json.dumps(brief_payload, ensure_ascii=False), encoding="utf-8", ) planned = build_planned_task( skill_id="reference-inspection", deliverable_type="reference_collection", max_attempts=2, ) task = build_task_package( brief_path, planned_task=planned, ) candidate = ExecutorCandidate( run_id=task.run_id, plan_id=task.plan_id, task_id=task.task_id, plan_version=task.plan_version, deliverable_type="reference_collection", artifacts=[ CandidateArtifact( artifact_type="image", uri=str(image_path.resolve()), source_uri=str(image_path.resolve()), description="已检查并缓存的人物参考图", ) ], artifact_binding_claims=[ ArtifactBindingClaim( expectation_id="Requirement1-Expectation1", artifact_uri=str(image_path.resolve()), evidence_tool_call_ids=["probe-reference"], ) ], summary="已有参考图可用于后续生产", ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "probe_media", {"source": str(image_path.resolve())}, "probe-reference", ), _tool_call( "view_images", {"image_sources": [str(image_path.resolve())]}, "view-reference", ), AIMessage(content=candidate.model_dump_json()), AIMessage(content=candidate.model_dump_json()), ] ) delivery = run_executor_agent( task, run_dir=root / "run", model=model, tool_registry=create_default_tool_registry( output_dir=root / "tool-output", ), ) self.assertEqual(delivery.skill_id, "reference-inspection") self.assertEqual(delivery.artifacts[0].artifact_type, "image") self.assertEqual( delivery.artifacts[0].source_uri, str(image_path.resolve()), ) def test_reference_review_accepts_separate_probe_and_view_caches( self, ) -> None: source_url = "https://example.test/existing-reference.png" # 使用一个真实存在、且与 source_url 不同的路径模拟 probe 缓存； # 文件内容不由该测试读取。 probe_path = str(BRIEF_PATH.resolve()) @tool("probe_media") def probe_media(source: str) -> dict: """探测测试媒体。""" return { "success": True, "source": source, "local_path": probe_path, "media_type": "image", "mime_type": "image/png", "width": 1080, "height": 1920, } @tool("view_images") def view_images(image_sources: list[str]) -> dict: """查看测试图片，并模拟另一套真实缓存路径。""" return { "success": True, "images": [ { "source": source, "local_path": ( "/cache/view/existing-reference.png" ), "data_url": ( "data:image/png;base64," "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB" ), } for source in image_sources ], } @tool("extract_frames") def extract_frames(video: str) -> dict: """占位视频抽帧工具。""" return {"success": True, "video": video, "frames": []} planned = build_planned_task( skill_id="reference-inspection", deliverable_type="reference_collection", max_attempts=1, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = ExecutorCandidate( run_id=task.run_id, plan_id=task.plan_id, task_id=task.task_id, plan_version=task.plan_version, deliverable_type="reference_collection", artifacts=[ CandidateArtifact( artifact_type="image", uri=probe_path, source_uri=source_url, description="已检查的人物参考图", ) ], artifact_binding_claims=[ ArtifactBindingClaim( expectation_id="Requirement1-Expectation1", artifact_uri=probe_path, evidence_tool_call_ids=["probe-reference"], ) ], summary="参考图已经完成探测和查看", ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "probe_media", {"source": source_url}, "probe-reference", ), _tool_call( "view_images", {"image_sources": [source_url]}, "view-reference", ), AIMessage(content=candidate.model_dump_json()), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: delivery = run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=ToolRegistry( { item.name: item for item in ( probe_media, view_images, extract_frames, ) } ), ) self.assertEqual(delivery.artifacts[0].uri, probe_path) self.assertEqual(delivery.artifacts[0].source_uri, source_url) def test_reference_video_review_links_source_to_probe_cache( self, ) -> None: source_url = "https://example.test/existing-reference.mp4" probe_path = str(BRIEF_PATH.resolve()) frame_path = str(BRIEF_PATH.resolve()) frame_url = "https://example.test/extracted-frame.png" @tool("probe_media") def probe_media(source: str) -> dict: """探测测试视频。""" return { "success": True, "source": source, "local_path": probe_path, "media_type": "video", "mime_type": "video/mp4", "duration_sec": 5, "width": 1080, "height": 1920, } @tool("extract_frames") def extract_frames(video: str) -> dict: """从原始来源抽取测试视频帧。""" return { "success": True, "video": video, "frames": [ { "local_path": frame_path, "url": frame_url, } ], } @tool("view_images") def view_images(image_sources: list[str]) -> dict: """查看测试视频帧。""" return { "success": True, "images": [ { "source": source, "local_path": source, "data_url": ( "data:image/png;base64," "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB" ), } for source in image_sources ], } planned = build_planned_task( skill_id="reference-inspection", deliverable_type="reference_collection", max_attempts=1, ) task = build_task_package( BRIEF_PATH, planned_task=planned, expectation_artifact_type="video", ) candidate = ExecutorCandidate( run_id=task.run_id, plan_id=task.plan_id, task_id=task.task_id, plan_version=task.plan_version, deliverable_type="reference_collection", artifacts=[ CandidateArtifact( artifact_type="video", uri=probe_path, source_uri=source_url, description="已检查的动作参考视频", ) ], artifact_binding_claims=[ ArtifactBindingClaim( expectation_id="Requirement1-Expectation1", artifact_uri=probe_path, evidence_tool_call_ids=["probe-reference"], ) ], summary="参考视频已经完成探测、抽帧和查看", ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "probe_media", {"source": source_url}, "probe-reference", ), _tool_call( "extract_frames", {"video": source_url}, "extract-reference", ), _tool_call( "view_images", {"image_sources": [frame_url]}, "view-reference", ), AIMessage(content=candidate.model_dump_json()), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: delivery = run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=ToolRegistry( { item.name: item for item in ( probe_media, view_images, extract_frames, ) } ), ) self.assertEqual(delivery.artifacts[0].uri, probe_path) self.assertEqual(delivery.artifacts[0].source_uri, source_url) def test_analysis_task_writes_local_structured_artifact(self) -> None: task = build_task_package(BRIEF_PATH) candidate = build_executor_candidate().model_copy( update={ "run_id": "model-run", "plan_id": "ModelPlan", "task_id": "Task99", "plan_version": 99, "artifacts": [ CandidateArtifact( artifact_type="structured_data", uri="delivery_artifact.json", description="由 Host 落盘的结构化结果", ) ] } ) self.assertEqual(candidate.schema_version, "0.3") with tempfile.TemporaryDirectory() as temp_dir: delivery = run_executor_agent( task, run_dir=Path(temp_dir), model=build_executor_model(candidate), tool_registry=_fake_tool_registry(), ) self.assertEqual(delivery.skill_id, "structured-analysis") self.assertEqual(delivery.attempt_count, 1) self.assertEqual( delivery.artifacts[0].artifact_id, "Task1-v1-artifact-1", ) self.assertIsInstance(delivery.summary, str) self.assertTrue(Path(delivery.manifest_uri).is_file()) artifact_path = Path(delivery.artifacts[0].uri) self.assertTrue(artifact_path.is_file()) self.assertNotEqual( artifact_path.resolve(), Path(delivery.manifest_uri).resolve(), ) self.assertEqual(delivery.tool_calls, []) saved = ExecutorCandidate.model_validate_json( Path(delivery.manifest_uri).read_text(encoding="utf-8") ) self.assertEqual(saved.artifacts, []) self.assertEqual(saved.run_id, task.run_id) self.assertEqual(saved.plan_id, task.plan_id) self.assertEqual(saved.task_id, task.task_id) self.assertEqual(saved.plan_version, task.plan_version) self.assertEqual(saved.payload["global_constraints"], ["统一主视觉"]) materialized = json.loads(artifact_path.read_text(encoding="utf-8")) self.assertEqual( materialized["payload"]["global_constraints"], ["统一主视觉"], ) def test_analysis_task_rejects_non_host_artifact_uri(self) -> None: planned = build_planned_task(max_attempts=1) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate().model_copy( update={ "artifacts": [ CandidateArtifact( artifact_type="structured_data", uri="another-result.json", description="未经 Host 声明的结构化结果", ) ] } ) with tempfile.TemporaryDirectory() as temp_dir: with self.assertRaisesRegex( ExecutorOutputError, "未出现在工具或依赖产物", ): run_executor_agent( task, run_dir=Path(temp_dir), model=build_executor_model(candidate), tool_registry=_fake_tool_registry(), ) def test_same_task_execution_reuses_stable_agent_run(self) -> None: task = build_task_package(BRIEF_PATH) candidate = build_executor_candidate() model = build_executor_model(candidate, candidate) with tempfile.TemporaryDirectory() as temp_dir: first = run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) second = run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) self.assertEqual(first.executor_run_id, second.executor_run_id) def test_invalid_output_retries_inside_same_task(self) -> None: task = build_task_package(BRIEF_PATH) valid = build_executor_candidate() invalid = valid.model_dump(mode="json") invalid["payload"] = {} model = ToolAwareFakeChatModel( responses=[ AIMessage(content=json.dumps(invalid, ensure_ascii=False)), AIMessage(content=valid.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: run_dir = Path(temp_dir) delivery = run_executor_agent( task, run_dir=run_dir, model=model, tool_registry=_fake_tool_registry(), ) with create_sqlite_checkpointer( run_dir, filename="agent_checkpoints.sqlite", ) as checkpointer: messages = checkpoint_messages( checkpointer, { "configurable": { "thread_id": delivery.executor_run_id } }, ) self.assertEqual(delivery.attempt_count, 2) self.assertIn(task.task_id, delivery.executor_run_id) self.assertTrue( any( isinstance(message, HumanMessage) and isinstance(message.content, str) and CORRECTION_PRESERVATION_INSTRUCTION in message.content for message in messages ) ) def test_host_derives_exact_evidence_call_ids_without_correction( self, ) -> None: planned = build_planned_task( skill_id="external-research", deliverable_type="research_result", max_attempts=2, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="research_result", artifact_uri=SOURCE_URL, evidence_tool_call_ids=["model-invented-id"], ) model = ToolAwareFakeChatModel( responses=[ _tool_call("search_tool", {"query": "资料"}, "call-search"), _tool_call( "inspect_tool", {"tool_ids": ["fake-tool"]}, "call-inspect", ), _tool_call( "run_tool", {"tool_id": "fake-tool", "params": {"kind": "research"}}, "call-run", ), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: run_dir = Path(temp_dir) delivery = run_executor_agent( task, run_dir=run_dir, model=model, tool_registry=_fake_tool_registry(), ) with create_sqlite_checkpointer( run_dir, filename="agent_checkpoints.sqlite", ) as checkpointer: messages = checkpoint_messages( checkpointer, { "configurable": { "thread_id": create_executor_run_id(task) } }, ) saved_candidate = ExecutorCandidate.model_validate_json( Path(delivery.manifest_uri).read_bytes() ) corrections = [ message.content for message in messages if isinstance(message, HumanMessage) and isinstance(message.content, str) and message.content.startswith("上一版未通过") ] self.assertEqual(delivery.attempt_count, 1) self.assertEqual(corrections, []) self.assertEqual( delivery.artifact_expectation_bindings[ 0 ].evidence_tool_call_ids, ["call-run"], ) self.assertEqual( saved_candidate.artifact_binding_claims[ 0 ].evidence_tool_call_ids, ["call-run"], ) def test_research_task_must_use_real_tool_output(self) -> None: planned = build_planned_task( skill_id="external-research", deliverable_type="research_result", ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="research_result", artifact_uri=SOURCE_URL, evidence_tool_call_ids=["call-run"], ) model = ToolAwareFakeChatModel( responses=[ _tool_call("search_tool", {"query": "资料"}, "call-search"), _tool_call( "inspect_tool", {"tool_ids": ["fake-tool"]}, "call-inspect", ), _tool_call( "run_tool", {"tool_id": "fake-tool", "params": {"kind": "research"}}, "call-run", ), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: delivery = run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) self.assertEqual( [item.tool_name for item in delivery.tool_calls], ["search_tool", "inspect_tool", "run_tool"], ) self.assertTrue( [item.duration_ms for item in delivery.tool_calls] == [3, 4, 5] ) self.assertEqual( delivery.findings[0].source_urls, [SOURCE_URL], ) self.assertIn( SOURCE_URL, delivery.tool_calls[-1].output_excerpt or "", ) def test_image_task_calls_generation_and_views_result(self) -> None: planned = build_planned_task( skill_id="image-production", deliverable_type="image", max_attempts=1, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="image", artifact_uri=IMAGE_URL, evidence_tool_call_ids=["run"], ) model = ToolAwareFakeChatModel( responses=[ _tool_call("search_tool", {"query": "图片生成"}, "search"), _tool_call( "inspect_tool", {"tool_ids": ["seedream_generate_image"]}, "inspect", ), _tool_call( "run_tool", { "tool_id": "seedream_generate_image", "params": {"kind": "image"}, }, "run", ), _tool_call( "view_images", {"image_sources": [IMAGE_URL]}, "view", ), AIMessage(content=candidate.model_dump_json()), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) delivery = run_executor_agent( task, run_dir=root, model=model, tool_registry=_fake_tool_registry(), ) config = { "configurable": { "thread_id": create_executor_run_id(task) } } with create_sqlite_checkpointer( root, filename="agent_checkpoints.sqlite", ) as checkpointer: messages = checkpoint_messages(checkpointer, config) self.assertEqual( Path(delivery.artifacts[0].uri), _fake_media_path(IMAGE_URL, ".png").resolve(), ) self.assertEqual(delivery.artifacts[0].source_uri, IMAGE_URL) self.assertFalse( delivery.artifacts[0].uri.startswith(("http://", "https://")) ) self.assertEqual(delivery.tool_calls[-1].tool_name, "view_images") self.assertTrue( any( isinstance(message.content, list) and any( isinstance(block, dict) and block.get("type") == "image_url" for block in message.content ) for message in messages ) ) def test_research_run_requires_prior_matching_inspect(self) -> None: planned = build_planned_task( skill_id="external-research", deliverable_type="research_result", max_attempts=1, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="research_result", artifact_uri=SOURCE_URL, ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "run_tool", {"tool_id": "fake-tool", "params": {"kind": "research"}}, "run", ), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: with self.assertRaisesRegex( ExecutorOutputError, "run_tool 前必须成功 inspect", ): run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) def test_image_final_artifact_must_be_viewed(self) -> None: planned = build_planned_task( skill_id="image-production", deliverable_type="image", max_attempts=1, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="image", artifact_uri=IMAGE_URL, ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "inspect_tool", {"tool_ids": ["seedream_generate_image"]}, "inspect", ), _tool_call( "run_tool", { "tool_id": "seedream_generate_image", "params": {"kind": "image"}, }, "run", ), _tool_call( "view_images", {"image_sources": [SOURCE_URL]}, "view", ), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: with self.assertRaisesRegex( ExecutorOutputError, "最终 Artifact 必须经过 view_images", ): run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) def test_view_images_does_not_prove_image_origin(self) -> None: planned = build_planned_task( skill_id="image-production", deliverable_type="image", max_attempts=1, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="image", artifact_uri=IMAGE_URL, ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "view_images", {"image_sources": [IMAGE_URL]}, "view", ), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: with self.assertRaisesRegex( ExecutorOutputError, "未出现在工具或依赖产物", ): run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) def test_video_task_calls_generation_and_extracts_final_frames(self) -> None: planned = build_planned_task( skill_id="video-production", deliverable_type="video", ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="video", artifact_uri=VIDEO_URL, evidence_tool_call_ids=["run"], ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "inspect_tool", {"tool_ids": ["seedance_generate_video"]}, "inspect", ), _tool_call( "run_tool", { "tool_id": "seedance_generate_video", "params": {"kind": "video"}, }, "run", ), _tool_call( "extract_frames", {"video": VIDEO_URL, "num_frames": 3}, "extract", ), _tool_call( "view_images", {"image_sources": [FRAME_URL]}, "view-frames", ), AIMessage(content=candidate.model_dump_json()), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) delivery = run_executor_agent( task, run_dir=root, model=model, tool_registry=_fake_tool_registry(), ) config = { "configurable": { "thread_id": create_executor_run_id(task) } } with create_sqlite_checkpointer( root, filename="agent_checkpoints.sqlite", ) as checkpointer: messages = checkpoint_messages(checkpointer, config) self.assertEqual( Path(delivery.artifacts[0].uri), _fake_media_path(VIDEO_URL, ".mp4").resolve(), ) self.assertEqual(delivery.artifacts[0].source_uri, VIDEO_URL) self.assertEqual(delivery.tool_calls[-1].tool_name, "view_images") self.assertTrue( any( isinstance(message.content, list) and any( isinstance(block, dict) and block.get("type") == "image_url" for block in message.content ) for message in messages ) ) def test_media_review_and_format_repairs_share_attempt_budget(self) -> None: planned = build_planned_task( skill_id="image-production", deliverable_type="image", max_attempts=2, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="image", artifact_uri=IMAGE_URL, evidence_tool_call_ids=["run"], ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "inspect_tool", {"tool_ids": ["seedream_generate_image"]}, "inspect", ), _tool_call( "run_tool", { "tool_id": "seedream_generate_image", "params": {"kind": "image"}, }, "run", ), _tool_call( "view_images", {"image_sources": [IMAGE_URL]}, "view", ), AIMessage(content="{}"), AIMessage(content=candidate.model_dump_json()), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: delivery = run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) self.assertEqual(delivery.attempt_count, 2) def test_media_artifact_changes_cannot_bypass_attempt_budget(self) -> None: image_b = "https://example.test/generated-reference-b.png" image_c = "https://example.test/generated-reference-c.png" planned = build_planned_task( skill_id="image-production", deliverable_type="image", max_attempts=2, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate_a = build_executor_candidate( deliverable_type="image", artifact_uri=IMAGE_URL, evidence_tool_call_ids=["run-a"], ) candidate_b = build_executor_candidate( deliverable_type="image", artifact_uri=image_b, evidence_tool_call_ids=["run-b"], ) candidate_c = build_executor_candidate( deliverable_type="image", artifact_uri=image_c, evidence_tool_call_ids=["run-c"], ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "inspect_tool", {"tool_ids": ["seedream_generate_image"]}, "inspect", ), _tool_call( "run_tool", { "tool_id": "seedream_generate_image", "params": { "kind": "image", "result_url": IMAGE_URL, }, }, "run-a", ), _tool_call( "view_images", {"image_sources": [IMAGE_URL]}, "view-a", ), AIMessage(content=candidate_a.model_dump_json()), _tool_call( "run_tool", { "tool_id": "seedream_generate_image", "params": { "kind": "image", "result_url": image_b, }, }, "run-b", ), _tool_call( "view_images", {"image_sources": [image_b]}, "view-b", ), AIMessage(content=candidate_b.model_dump_json()), _tool_call( "run_tool", { "tool_id": "seedream_generate_image", "params": { "kind": "image", "result_url": image_c, }, }, "run-c", ), _tool_call( "view_images", {"image_sources": [image_c]}, "view-c", ), AIMessage(content=candidate_c.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: with self.assertRaisesRegex( ExecutorOutputError, "耗尽修正次数", ): run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) def test_fabricated_remote_uri_is_rejected(self) -> None: planned = build_planned_task( skill_id="external-research", deliverable_type="research_result", max_attempts=1, ) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate( deliverable_type="research_result", artifact_uri="https://fabricated.test/source", ) model = ToolAwareFakeChatModel( responses=[ _tool_call( "run_tool", {"tool_id": "fake-tool", "params": {"kind": "research"}}, "run", ), AIMessage(content=candidate.model_dump_json()), ] ) with tempfile.TemporaryDirectory() as temp_dir: with self.assertRaisesRegex( ExecutorOutputError, "未出现在工具", ): run_executor_agent( task, run_dir=Path(temp_dir), model=model, tool_registry=_fake_tool_registry(), ) def test_analysis_payload_cannot_hide_a_fabricated_url(self) -> None: planned = build_planned_task(max_attempts=1) task = build_task_package(BRIEF_PATH, planned_task=planned) candidate = build_executor_candidate() candidate.payload["fake_reference"] = "https://fabricated.test/image.png" with tempfile.TemporaryDirectory() as temp_dir: with self.assertRaisesRegex( ExecutorOutputError, "未出现在工具", ): run_executor_agent( task, run_dir=Path(temp_dir), model=build_executor_model(candidate), tool_registry=_fake_tool_registry(), ) if __name__ == "__main__": unittest.main()