video-image-production-build/tests/graph/test_graph.py

from __future__ import annotations

import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from langchain_core.messages import AIMessage
from langgraph.graph.state import CompiledStateGraph

from production_build_agents.contracts.models import (
    ArtifactRejection,
    ArtifactExpectation,
    GlobalDataPlan,
    GlobalDataRequirement,
)
from production_build_agents.agents.executor.agent import create_executor_agent
from production_build_agents.global_data.graph import create_global_data_graph
from production_build_agents.global_data.nodes.finish_run import (
    finalize_global_data_node,
)
from production_build_agents.global_data.nodes.execute_task import execute_task_node
from production_build_agents.global_data.nodes.prepare_task import (
    prepare_next_task_node,
)
from production_build_agents.run.langgraph_checkpointer import create_in_memory_checkpointer
from production_build_agents.run.operation_journal import OperationOutcomeUnknownError
from production_build_agents.agents.planner.agent import create_planner_agent
from production_build_agents.agents.validator.task_agent import create_validator_agent
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_global_data_plan,
    build_planned_task,
    build_planner_model,
)
from tests.support.validator_fixtures import (
    build_stage_read_call,
    build_validator_candidate,
    build_validator_model,
    build_stage_validator_candidate,
)


INPUT_PATH = (
    Path(__file__).parents[1] / "fixtures" / "minimal_input.json"
)


def _initial_state(
    *,
    run_id: str,
    input_path: Path,
    output_dir: Path,
) -> dict:
    return {
        "run_id": run_id,
        "protocol_version": "0.3",
        "input_path": str(input_path.resolve()),
        "output_dir": str(output_dir.resolve()),
        "status": "PENDING",
        "phase": "PREPROCESS",
        "replan_scope": None,
        "failure_code": None,
        "error": None,
        "global_data_plan_path": None,
        "plan_history": {},
        "task_records": {},
        "current_task_path": None,
        "current_executor_delivery_path": None,
        "current_validation_report_path": None,
        "current_stage_candidate_path": None,
        "current_stage_validation_report_path": None,
        "global_data_delivery_path": None,
        "planner_calls": 0,
        "executor_calls": 0,
        "validator_calls": 0,
        "stage_validator_calls": 0,
        "replan_count": 0,
        "event_log": [],
    }


def _invoke(
    *,
    run_id: str,
    output_dir: Path,
    plans,
    executor_candidates,
    validator_candidates,
    validator_model=None,
) -> dict:
    graph = create_global_data_graph(
        checkpointer=create_in_memory_checkpointer(),
        planner_model=build_planner_model(*plans),
        executor_model=build_executor_model(*executor_candidates),
        validator_model=(
            validator_model
            or build_validator_model(*validator_candidates)
        ),
    )
    return graph.invoke(
        _initial_state(
            run_id=run_id,
            input_path=INPUT_PATH,
            output_dir=output_dir,
        ),
        config={
            "configurable": {"thread_id": run_id},
            "recursion_limit": 200,
        },
    )


class AgentConstructionTest(unittest.TestCase):
    def test_all_three_agent_roles_are_langchain_agents(self) -> None:
        model = build_planner_model(build_global_data_plan())
        agents = [
            create_planner_agent(model),
            create_executor_agent(model, system_prompt="测试 Executor"),
            create_validator_agent(
                model,
                system_prompt="测试 Validator",
                tools=[],
            ),
        ]
        for agent in agents:
            self.assertIsInstance(agent, CompiledStateGraph)
            self.assertIn("model", agent.get_graph().nodes)


class GlobalDataGraphTest(unittest.TestCase):
    def test_graph_has_fixed_outer_loop_and_no_production_stage(self) -> None:
        graph = create_global_data_graph(
            checkpointer=create_in_memory_checkpointer()
        )
        self.assertEqual(
            set(graph.get_graph().nodes),
            {
                "__start__",
                "preprocess",
                "plan_global_data",
                "prepare_next_task",
                "execute_task",
                "validate_task",
                "validate_global_data_stage",
                "replan_global_data",
                "finalize_global_data",
                "__end__",
            },
        )
        self.assertNotIn("production_stage", graph.get_graph().draw_mermaid())

    def test_single_task_completes_and_writes_stage_delivery(self) -> None:
        run_id = "Run-single"
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            result = _invoke(
                run_id=run_id,
                output_dir=output_dir,
                plans=[build_global_data_plan()],
                executor_candidates=[
                    build_executor_candidate(run_id=run_id)
                ],
                validator_candidates=[
                    build_validator_candidate(run_id=run_id)
                ],
            )

            self.assertEqual(result["status"], "COMPLETED")
            self.assertEqual(result["phase"], "FINALIZE")
            self.assertEqual(result["planner_calls"], 1)
            self.assertEqual(result["executor_calls"], 1)
            self.assertEqual(result["validator_calls"], 1)
            self.assertEqual(result["stage_validator_calls"], 1)
            self.assertEqual(result["replan_count"], 0)
            self.assertEqual(
                result["task_records"]["Task1"]["status"],
                "passed",
            )
            delivery_path = Path(result["global_data_delivery_path"])
            self.assertTrue(delivery_path.is_file())
            delivery = json.loads(delivery_path.read_text(encoding="utf-8"))
            self.assertEqual(delivery["status"], "GLOBAL_DATA_COMPLETED")
            self.assertEqual(len(delivery["tasks"]), 1)
            self.assertTrue(
                (output_dir / "validation_results" / "Task1.v1.json").is_file()
            )
            saved = json.loads(
                (output_dir / "run_summary.json").read_text(encoding="utf-8")
            )
            self.assertEqual(saved["status"], "COMPLETED")
            self.assertNotIn("production_brief", saved)
            self.assertFalse((output_dir / "run_state.json").exists())
            for field in (
                "global_data_plan",
                "global_data_mermaid_path",
                "current_task",
            ):
                self.assertNotIn(field, result)
                self.assertNotIn(field, saved)

    def test_prepare_task_rejects_corrupt_plan_reference(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            plan_path = output_dir / "plans" / "global_data_dag.v1.json"
            plan_path.parent.mkdir(parents=True)
            plan_path.write_text("{}\n", encoding="utf-8")
            state = {
                **_initial_state(
                    run_id="Run-corrupt-plan",
                    input_path=INPUT_PATH,
                    output_dir=output_dir,
                ),
                "status": "RUNNING",
                "phase": "PREPARE_TASK",
                "global_data_plan_path": str(plan_path),
            }

            result = prepare_next_task_node(state)

            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["phase"], "PREPARE_TASK")
            self.assertEqual(result["failure_code"], "DAG_STATE_INVALID")
            self.assertTrue((output_dir / "run_summary.json").is_file())

    def test_execute_task_rejects_missing_task_reference(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            state = {
                **_initial_state(
                    run_id="Run-missing-task",
                    input_path=INPUT_PATH,
                    output_dir=output_dir,
                ),
                "status": "RUNNING",
                "phase": "EXECUTE_TASK",
                "current_task_path": str(output_dir / "tasks" / "missing.json"),
                "task_records": {
                    "Task1": {
                        "task_id": "Task1",
                        "status": "ready",
                        "active_plan_version": 1,
                    }
                },
            }

            result = execute_task_node(state)

            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["phase"], "EXECUTE_TASK")
            self.assertEqual(
                result["failure_code"],
                "EXECUTOR_OUTPUT_INVALID",
            )

    def test_minor_stage_gap_is_preserved_as_unresolved(self) -> None:
        run_id = "Run-stage-minor-gap"
        stage_candidate = build_stage_validator_candidate(
            run_id=run_id,
            missing_minor=True,
        )
        validator_model = build_validator_model(
            build_validator_candidate(run_id=run_id),
            stage_candidate=stage_candidate,
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            result = _invoke(
                run_id=run_id,
                output_dir=Path(temp_dir),
                plans=[build_global_data_plan()],
                executor_candidates=[
                    build_executor_candidate(run_id=run_id)
                ],
                validator_candidates=[],
                validator_model=validator_model,
            )
            delivery = json.loads(
                Path(result["global_data_delivery_path"]).read_text(
                    encoding="utf-8"
                )
            )

        self.assertEqual(result["status"], "COMPLETED")
        self.assertTrue(
            any("缺少非关键参考说明" in item for item in delivery["unresolved"])
        )

    def test_stage_missing_requirement_replans_without_rerunning_passed_task(
        self,
    ) -> None:
        run_id = "Run-stage-missing-replan"
        task1 = build_planned_task(task_id="Task1")
        plan1 = build_global_data_plan(tasks=[task1])
        requirement2 = GlobalDataRequirement(
            requirement_id="Requirement2",
            description="补齐阶段验收发现的共享说明",
            importance="critical",
            source_paths=["$.核心制作点[0]"],
            artifact_expectations=[
                ArtifactExpectation(
                    expectation_id="Requirement2-Expectation1",
                    artifact_type="structured_data",
                    minimum_count=1,
                    usage_scope="正式生产",
                    verification_capabilities=["document_content"],
                )
            ],
        )
        task2 = build_planned_task(
            task_id="Task2",
            objective="补齐阶段遗漏的共享说明",
            expectation_ids=["Requirement2-Expectation1"],
        )
        plan2 = GlobalDataPlan(
            plan_id=plan1.plan_id,
            plan_version=2,
            goal=plan1.goal,
            stage_requirements=[
                *plan1.stage_requirements,
                requirement2,
            ],
            tasks=[task1, task2],
            revision_summary="保留已验收资料，并补齐阶段缺口。",
        )
        stage_fail = build_stage_validator_candidate(
            run_id=run_id,
            plan_version=1,
            missing_critical=True,
        )
        stage_pass = build_stage_validator_candidate(
            run_id=run_id,
            plan_version=2,
            requirement_ids=("Requirement1", "Requirement2"),
        )
        validator_model = ToolAwareFakeChatModel(
            responses=[
                AIMessage(
                    content=build_validator_candidate(
                        run_id=run_id,
                        task_id="Task1",
                        plan_version=1,
                    ).model_dump_json()
                ),
                build_stage_read_call("stage-read-v1"),
                AIMessage(content=stage_fail.model_dump_json()),
                AIMessage(
                    content=build_validator_candidate(
                        run_id=run_id,
                        task_id="Task2",
                        plan_version=2,
                        expectation_id="Requirement2-Expectation1",
                    ).model_dump_json()
                ),
                build_stage_read_call("stage-read-v2"),
                AIMessage(content=stage_pass.model_dump_json()),
            ]
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            result = _invoke(
                run_id=run_id,
                output_dir=Path(temp_dir),
                plans=[plan1, plan2],
                executor_candidates=[
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task1",
                        plan_version=1,
                    ),
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task2",
                        plan_version=2,
                        expectation_id="Requirement2-Expectation1",
                    ),
                ],
                validator_candidates=[],
                validator_model=validator_model,
            )
            current_plan = GlobalDataPlan.model_validate_json(
                Path(result["global_data_plan_path"]).read_text(
                    encoding="utf-8"
                )
            )

        self.assertEqual(result["status"], "COMPLETED")
        self.assertEqual(result["replan_count"], 1)
        self.assertEqual(result["executor_calls"], 2)
        self.assertEqual(result["validator_calls"], 2)
        self.assertEqual(result["stage_validator_calls"], 2)
        self.assertEqual(
            result["task_records"]["Task1"]["accepted_plan_version"],
            1,
        )
        self.assertEqual(
            len(current_plan.stage_requirements),
            2,
        )

    def test_stage_rejected_artifact_is_superseded_by_replan(self) -> None:
        run_id = "Run-stage-artifact-replacement"
        task1 = build_planned_task(task_id="Task1")
        plan1 = build_global_data_plan(tasks=[task1])
        task2 = build_planned_task(
            task_id="Task2",
            objective="替换阶段验收拒绝的全局资料",
        ).model_copy(
            update={"replaces_artifact_ids": ["Task1-v1-artifact-1"]}
        )
        plan2 = GlobalDataPlan(
            plan_id=plan1.plan_id,
            plan_version=2,
            goal=plan1.goal,
            stage_requirements=plan1.stage_requirements,
            tasks=[task1, task2],
            revision_summary="保留旧产物审计记录并交付替代版本。",
        )
        validator_model = ToolAwareFakeChatModel(
            responses=[
                AIMessage(
                    content=build_validator_candidate(
                        run_id=run_id,
                    ).model_dump_json()
                ),
                build_stage_read_call("stage-replacement-read-v1"),
                AIMessage(
                    content=build_stage_validator_candidate(
                        run_id=run_id,
                        artifact_rejections=[
                            ArtifactRejection(
                                artifact_id="Task1-v1-artifact-1",
                                reason_code="semantic_rejection",
                                reason="阶段语义验收拒绝该产物",
                            )
                        ],
                    ).model_dump_json()
                ),
                AIMessage(
                    content=build_validator_candidate(
                        run_id=run_id,
                        task_id="Task2",
                        plan_version=2,
                    ).model_dump_json()
                ),
                build_stage_read_call("stage-replacement-read-v2"),
                AIMessage(
                    content=build_stage_validator_candidate(
                        run_id=run_id,
                        plan_version=2,
                    ).model_dump_json()
                ),
            ]
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            result = _invoke(
                run_id=run_id,
                output_dir=Path(temp_dir),
                plans=[plan1, plan2],
                executor_candidates=[
                    build_executor_candidate(run_id=run_id),
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task2",
                        plan_version=2,
                    ),
                ],
                validator_candidates=[],
                validator_model=validator_model,
            )
            delivery = json.loads(
                Path(result["global_data_delivery_path"]).read_text(
                    encoding="utf-8"
                )
            )

        self.assertEqual(result["status"], "COMPLETED")
        self.assertEqual(
            [item["artifact_id"] for item in delivery["active_artifacts"]],
            ["Task2-v2-artifact-1"],
        )
        self.assertEqual(
            [
                item["artifact_id"]
                for item in delivery["superseded_artifacts"]
            ],
            ["Task1-v1-artifact-1"],
        )

    def test_finalizer_rejects_non_passed_task(self) -> None:
        run_id = "Run-finalizer-guard"
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            result = _invoke(
                run_id=run_id,
                output_dir=output_dir,
                plans=[build_global_data_plan()],
                executor_candidates=[
                    build_executor_candidate(run_id=run_id)
                ],
                validator_candidates=[
                    build_validator_candidate(run_id=run_id)
                ],
            )
            (output_dir / "run_summary.json").unlink()
            result["task_records"]["Task1"]["status"] = "delivered"
            guarded = finalize_global_data_node(result)
            self.assertEqual(guarded["status"], "FAILED")
            self.assertEqual(
                guarded["failure_code"],
                "FINALIZATION_INVALID",
            )

    def test_finalizer_rejects_missing_plan_history_file(self) -> None:
        run_id = "Run-finalizer-plan-history"
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            result = _invoke(
                run_id=run_id,
                output_dir=output_dir,
                plans=[build_global_data_plan()],
                executor_candidates=[
                    build_executor_candidate(run_id=run_id)
                ],
                validator_candidates=[
                    build_validator_candidate(run_id=run_id)
                ],
            )
            (output_dir / "run_summary.json").unlink()
            Path(
                result["plan_history"]["1"]["json_uri"]
            ).unlink()
            guarded = finalize_global_data_node(result)
            self.assertEqual(guarded["status"], "FAILED")
            self.assertEqual(
                guarded["failure_code"],
                "FINALIZATION_INVALID",
            )

    def test_finalizer_recomputes_saved_validation_report(self) -> None:
        run_id = "Run-finalizer-report-check"
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            result = _invoke(
                run_id=run_id,
                output_dir=output_dir,
                plans=[build_global_data_plan()],
                executor_candidates=[
                    build_executor_candidate(run_id=run_id)
                ],
                validator_candidates=[
                    build_validator_candidate(run_id=run_id)
                ],
            )
            (output_dir / "run_summary.json").unlink()
            Path(result["global_data_delivery_path"]).unlink()
            report_path = Path(
                result["task_records"]["Task1"][
                    "validation_report_path"
                ]
            )
            report = json.loads(report_path.read_text(encoding="utf-8"))
            report["criterion_results"][0]["verdict"] = "FAIL"
            report_path.write_text(
                json.dumps(report, ensure_ascii=False),
                encoding="utf-8",
            )

            guarded = finalize_global_data_node(result)

            self.assertEqual(guarded["status"], "FAILED")
            self.assertEqual(
                guarded["failure_code"],
                "FINALIZATION_INVALID",
            )

    def test_task_package_conflict_becomes_stable_terminal_failure(
        self,
    ) -> None:
        run_id = "Run-task-package-conflict"
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            conflict = output_dir / "tasks" / "Task1.v1.json"
            conflict.parent.mkdir(parents=True)
            conflict.write_text("{}\n", encoding="utf-8")
            result = _invoke(
                run_id=run_id,
                output_dir=output_dir,
                plans=[build_global_data_plan()],
                executor_candidates=[
                    build_executor_candidate(run_id=run_id)
                ],
                validator_candidates=[
                    build_validator_candidate(run_id=run_id)
                ],
            )
            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["failure_code"], "VERSION_CONFLICT")

    def test_unknown_tool_outcome_terminates_without_validator(self) -> None:
        run_id = "Run-tool-unknown"
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            graph = create_global_data_graph(
                checkpointer=create_in_memory_checkpointer(),
                planner_model=build_planner_model(
                    build_global_data_plan()
                ),
            )
            with patch(
                "production_build_agents.global_data.nodes.execute_task."
                "run_executor_agent",
                side_effect=OperationOutcomeUnknownError(
                    "executor:operation-1"
                ),
            ):
                result = graph.invoke(
                    _initial_state(
                        run_id=run_id,
                        input_path=INPUT_PATH,
                        output_dir=output_dir,
                    ),
                    config={"configurable": {"thread_id": run_id}},
                )
            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["phase"], "EXECUTE_TASK")
            self.assertIsNone(result["replan_scope"])
            self.assertEqual(
                result["failure_code"],
                "TOOL_OUTCOME_UNKNOWN",
            )
            self.assertEqual(result["validator_calls"], 0)

    def test_unknown_validator_tool_outcome_is_stable_terminal(self) -> None:
        run_id = "Run-validator-tool-unknown"
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            graph = create_global_data_graph(
                checkpointer=create_in_memory_checkpointer(),
                planner_model=build_planner_model(
                    build_global_data_plan()
                ),
                executor_model=build_executor_model(
                    build_executor_candidate(run_id=run_id)
                ),
            )
            with patch(
                "production_build_agents.global_data.nodes.validate_task."
                "run_validator_agent",
                side_effect=OperationOutcomeUnknownError(
                    "validator:operation-1"
                ),
            ):
                result = graph.invoke(
                    _initial_state(
                        run_id=run_id,
                        input_path=INPUT_PATH,
                        output_dir=output_dir,
                    ),
                    config={"configurable": {"thread_id": run_id}},
                )
            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["phase"], "VALIDATE_TASK")
            self.assertIsNone(result["replan_scope"])
            self.assertEqual(
                result["failure_code"],
                "TOOL_OUTCOME_UNKNOWN",
            )
            self.assertEqual(
                result["task_records"]["Task1"]["status"],
                "delivered",
            )

    def test_multi_task_dag_unlocks_only_after_pass(self) -> None:
        run_id = "Run-multi"
        tasks = [
            build_planned_task(task_id="Task1", priority=20),
            build_planned_task(
                task_id="Task2",
                objective="基于前置资料整理补充约束",
                depends_on=["Task1"],
                priority=10,
            ),
        ]
        with tempfile.TemporaryDirectory() as temp_dir:
            result = _invoke(
                run_id=run_id,
                output_dir=Path(temp_dir),
                plans=[build_global_data_plan(tasks=tasks)],
                executor_candidates=[
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task1",
                    ),
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task2",
                    ),
                ],
                validator_candidates=[
                    build_validator_candidate(
                        run_id=run_id,
                        task_id="Task1",
                    ),
                    build_validator_candidate(
                        run_id=run_id,
                        task_id="Task2",
                    ),
                ],
            )

            self.assertEqual(result["status"], "COMPLETED")
            self.assertEqual(result["executor_calls"], 2)
            self.assertEqual(result["validator_calls"], 2)
            self.assertTrue(
                all(
                    record["status"] == "passed"
                    for record in result["task_records"].values()
                )
            )
            task2_package = json.loads(
                (
                    Path(temp_dir) / "tasks" / "Task2.v1.json"
                ).read_text(encoding="utf-8")
            )
            self.assertEqual(
                task2_package["dependency_deliveries"][0]["task_id"],
                "Task1",
            )
            self.assertTrue(
                task2_package["dependency_deliveries"][0][
                    "delivery_uri"
                ].endswith("Task1.v1.json")
            )

    def test_failed_validation_replans_once_then_passes(self) -> None:
        run_id = "Run-replan-once"
        v1 = build_global_data_plan(plan_version=1)
        v2 = build_global_data_plan(
            tasks=[
                build_planned_task(
                    objective="按失败证据修正全局约束"
                )
            ],
            plan_version=2,
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            result = _invoke(
                run_id=run_id,
                output_dir=Path(temp_dir),
                plans=[v1, v2],
                executor_candidates=[
                    build_executor_candidate(
                        run_id=run_id,
                        plan_version=1,
                    ),
                    build_executor_candidate(
                        run_id=run_id,
                        plan_version=2,
                    ),
                ],
                validator_candidates=[
                    build_validator_candidate(
                        run_id=run_id,
                        plan_version=1,
                        verdict="FAIL",
                    ),
                    build_validator_candidate(
                        run_id=run_id,
                        plan_version=2,
                    ),
                ],
            )

            self.assertEqual(result["status"], "COMPLETED")
            self.assertEqual(result["replan_count"], 1)
            self.assertEqual(result["planner_calls"], 2)
            self.assertEqual(result["executor_calls"], 2)
            self.assertEqual(set(result["plan_history"]), {"1", "2"})
            self.assertEqual(
                result["task_records"]["Task1"]["accepted_plan_version"],
                2,
            )

    def test_two_replans_preserve_passed_task_and_change_remaining_dag(
        self,
    ) -> None:
        run_id = "Run-replan-twice"
        task1_v2 = build_planned_task(
            task_id="Task1",
            objective="修正后的第一项资料",
        )
        task2_v2 = build_planned_task(
            task_id="Task2",
            objective="使用第一项资料完成第二项",
            depends_on=["Task1"],
        )
        v1 = build_global_data_plan(plan_version=1)
        v2 = build_global_data_plan(
            tasks=[task1_v2, task2_v2],
            plan_version=2,
        )
        v3 = build_global_data_plan(
            tasks=[
                task1_v2,
                task2_v2.model_copy(
                    update={"objective": "按第二次失败证据修正第二项"}
                ),
                build_planned_task(
                    task_id="Task3",
                    objective="汇总修正后的两项资料",
                    depends_on=["Task1", "Task2"],
                ),
            ],
            plan_version=3,
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            result = _invoke(
                run_id=run_id,
                output_dir=Path(temp_dir),
                plans=[v1, v2, v3],
                executor_candidates=[
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task1",
                        plan_version=1,
                    ),
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task1",
                        plan_version=2,
                    ),
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task2",
                        plan_version=2,
                    ),
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task2",
                        plan_version=3,
                    ),
                    build_executor_candidate(
                        run_id=run_id,
                        task_id="Task3",
                        plan_version=3,
                    ),
                ],
                validator_candidates=[
                    build_validator_candidate(
                        run_id=run_id,
                        task_id="Task1",
                        plan_version=1,
                        verdict="FAIL",
                    ),
                    build_validator_candidate(
                        run_id=run_id,
                        task_id="Task1",
                        plan_version=2,
                    ),
                    build_validator_candidate(
                        run_id=run_id,
                        task_id="Task2",
                        plan_version=2,
                        verdict="FAIL",
                    ),
                    build_validator_candidate(
                        run_id=run_id,
                        task_id="Task2",
                        plan_version=3,
                    ),
                    build_validator_candidate(
                        run_id=run_id,
                        task_id="Task3",
                        plan_version=3,
                    ),
                ],
            )

            self.assertEqual(result["status"], "COMPLETED")
            self.assertEqual(result["replan_count"], 2)
            self.assertEqual(result["executor_calls"], 5)
            self.assertEqual(
                result["task_records"]["Task1"]["accepted_plan_version"],
                2,
            )
            self.assertFalse(
                (Path(temp_dir) / "tasks" / "Task1.v3.json").exists()
            )
            self.assertEqual(
                result["task_records"]["Task3"]["accepted_plan_version"],
                3,
            )

    def test_sixth_failure_exhausts_replan_budget_without_v7(self) -> None:
        run_id = "Run-replan-budget"
        plans = [
            build_global_data_plan(
                tasks=[
                    build_planned_task(
                        objective=f"第 {version} 版待验收资料"
                    )
                ],
                plan_version=version,
            )
            for version in range(1, 7)
        ]
        with tempfile.TemporaryDirectory() as temp_dir:
            result = _invoke(
                run_id=run_id,
                output_dir=Path(temp_dir),
                plans=plans,
                executor_candidates=[
                    build_executor_candidate(
                        run_id=run_id,
                        plan_version=version,
                    )
                    for version in range(1, 7)
                ],
                validator_candidates=[
                    build_validator_candidate(
                        run_id=run_id,
                        plan_version=version,
                        verdict="FAIL",
                    )
                    for version in range(1, 7)
                ],
            )

            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["phase"], "REPLAN")
            self.assertEqual(result["replan_scope"], "task")
            self.assertEqual(
                result["failure_code"],
                "REPLAN_BUDGET_EXHAUSTED",
            )
            self.assertEqual(result["replan_count"], 5)
            self.assertEqual(result["planner_calls"], 6)
            self.assertEqual(result["executor_calls"], 6)
            self.assertFalse(
                (Path(temp_dir) / "plans" / "global_data_dag.v7.json").exists()
            )

    def test_planner_failure_is_persisted(self) -> None:
        invalid_model = ToolAwareFakeChatModel(
            responses=[AIMessage(content="{}"), AIMessage(content="{}")]
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            graph = create_global_data_graph(
                checkpointer=create_in_memory_checkpointer(),
                planner_model=invalid_model,
            )
            config = {"configurable": {"thread_id": "planner-failed"}}
            result = graph.invoke(
                _initial_state(
                    run_id="Run-planner-failed",
                    input_path=INPUT_PATH,
                    output_dir=output_dir,
                ),
                config=config,
            )
            saved = dict(graph.get_state(config).values)
            summary = json.loads(
                (output_dir / "run_summary.json").read_text(encoding="utf-8")
            )
            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["phase"], "PLAN")
            self.assertEqual(saved["status"], "FAILED")
            self.assertEqual(summary["status"], "FAILED")
            self.assertEqual(result["planner_calls"], 1)

    def test_executor_failure_is_persisted(self) -> None:
        run_id = "Run-executor-failed"
        invalid = build_executor_candidate(run_id=run_id).model_copy(
            update={"payload": {}},
        )
        executor_model = ToolAwareFakeChatModel(
            responses=[
                AIMessage(content=invalid.model_dump_json()),
                AIMessage(content=invalid.model_dump_json()),
            ]
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            output_dir = Path(temp_dir)
            graph = create_global_data_graph(
                checkpointer=create_in_memory_checkpointer(),
                planner_model=build_planner_model(build_global_data_plan()),
                executor_model=executor_model,
            )
            result = graph.invoke(
                _initial_state(
                    run_id=run_id,
                    input_path=INPUT_PATH,
                    output_dir=output_dir,
                ),
                config={"configurable": {"thread_id": run_id}},
            )
            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["phase"], "EXECUTE_TASK")
            self.assertEqual(
                result["task_records"]["Task1"]["status"],
                "failed",
            )
            self.assertEqual(result["executor_calls"], 1)

    def test_preprocess_error_ends_before_agents(self) -> None:
        invalid_payload = {
            "帖子类型": "video",
            "制作表": {
                "段落结构": [],
                "形式结果": [],
                "关系结果": [],
            },
        }
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            input_path = root / "invalid.json"
            output_dir = root / "output"
            input_path.write_text(
                json.dumps(invalid_payload, ensure_ascii=False),
                encoding="utf-8",
            )
            graph = create_global_data_graph(
                checkpointer=create_in_memory_checkpointer()
            )
            result = graph.invoke(
                _initial_state(
                    run_id="Run-preprocess-failed",
                    input_path=input_path,
                    output_dir=output_dir,
                ),
                config={"configurable": {"thread_id": "preprocess-failed"}},
            )
            self.assertEqual(result["status"], "FAILED")
            self.assertEqual(result["phase"], "PREPROCESS")
            self.assertEqual(result["planner_calls"], 0)
            self.assertEqual(result["executor_calls"], 0)
            self.assertFalse((output_dir / "plans").exists())


if __name__ == "__main__":
    unittest.main()