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

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