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