from __future__ import annotations from app.main_agent_decision_projection import ( MainAgentDecisionProjector, parse_goal_items, parse_script_direction, ) from app.runtime_event_index import RuntimeEventIndex from app.runtime_tool_catalog import business_tool_label DIRECTION = """## 选题价值主张 用真实事件解释行业变化。 ## 账号价值主张 保持专业深度与信息密度。 ## 创作总目标 ### 目标 1 - 目标语句：拆解机制并让读者看懂因果链。 - 通过标准：结构清楚。 ### 目标 2 - 目标语句：用可核验数据支撑关键结论。 ## 领域评估维度 1. 数据可信度 2. 适用边界 """ def _event( event_id: int, event_seq: int, event_type: str, event_name: str, **extra, ): return { "id": event_id, "event_seq": event_seq, "event_type": event_type, "event_name": event_name, "status": "ok", "ended_at": f"2026-07-14T10:00:{event_seq:02d}", **extra, } def _main_scope(event_id: int = 1): return _event( event_id, 1, "agent_scope", "main", agent_role="main", agent_depth=0, scope_event_id=event_id, parent_event_id=None, ) def _main_tool(event_id: int, event_seq: int, name: str, **extra): return _event( event_id, event_seq, "tool_call", name, agent_role="main", agent_depth=0, scope_event_id=1, parent_event_id=1, **extra, ) def test_runtime_index_requires_real_main_scope_for_direct_reads(): events = [ _main_scope(), _main_tool(2, 2, "get_topic_detail"), _event( 3, 3, "tool_call", "get_topic_detail", agent_role="script_evaluator", agent_depth=1, scope_event_id=30, parent_event_id=30, ), _event( 4, 4, "tool_call", "get_topic_detail", agent_role="main", agent_depth=0, scope_event_id=999, parent_event_id=999, ), _main_tool(5, 5, "get_domain_info", status="error"), ] index = RuntimeEventIndex(events) assert index.main_scope_ids == {1} assert [event["id"] for event in index.main_direct_reads()] == [2] assert [event["id"] for event in index.main_direct_reads(successful=False)] == [5] assert [event["id"] for event in index.events_for_actor("main")] == [1, 2, 4, 5] assert [event["id"] for event in index.events_for_actor("overall-evaluator")] == [3] assert business_tool_label("get_topic_detail") == "选题" def test_planning_analysis_uses_last_real_main_thought_before_direction_save(): events = [ _main_scope(), _main_tool(2, 2, "think_and_plan", inputData={ "thought_number": 1, "thought_summary": "先形成旧规划", "thought": "旧思考", "plan": "旧计划", "action": "旧下一步", }), _main_tool(3, 3, "think_and_plan", inputData={ "thought_number": 2, "thought_summary": "锁定创作目标与首轮骨架", "thought": "完整分析账号结构与领域下限。", "plan": "保存目标，然后开启首轮。", "action": "写入创作总目标", }), _main_tool(4, 4, "save_script_direction", inputData={"script_direction": DIRECTION}), _main_tool(5, 5, "think_and_plan", inputData={ "thought_summary": "保存后的其他规划", "thought": "不应成为开场规划", "plan": "后续计划", "action": "继续", }), ] planning = MainAgentDecisionProjector().project( RuntimeEventIndex(events), script_direction=DIRECTION, rounds=[], )["planningAnalysis"] assert planning == { "stage": "planning-analysis", "summary": "锁定创作目标与首轮骨架", "thought": "完整分析账号结构与领域下限。", "plan": "保存目标，然后开启首轮。", "action": "写入创作总目标", "eventRef": "event:3", "technicalRefs": ["event:3"], "completeness": "complete", } def test_begin_round_prefers_success_output_then_falls_back_to_round_marker(): events = [ _main_scope(), _main_tool( 2, 2, "begin_round", round_index=0, output_preview='{"success": true, "round_index": 1}', ), _main_tool( 3, 3, "begin_round", round_index=1, output_preview='{"success": true}', ), _event( 4, 4, "round_begin", "2", agent_role="main", agent_depth=0, scope_event_id=1, parent_event_id=1, payload={"round_index": 2, "goal": "第二轮"}, ), ] index = RuntimeEventIndex(events) assert index.resolved_begin_round(events[1]) == 1 assert index.resolved_begin_round(events[2]) == 2 assert {key: [event["id"] for event in value] for key, value in index.successful_begin_rounds().items()} == {1: [2], 2: [3]} def test_evaluator_association_uses_hierarchy_and_structured_round_not_report_text(): events = [ _main_scope(), _event( 10, 2, "agent_invoke", "script_evaluator", agent_role="script_evaluator", agent_depth=1, scope_event_id=10, parent_event_id=1, round_index=1, agentOutputData={"summary": "## 评估报告 round=99\n### 整体结论：部分通过"}, ), _event( 11, 3, "agent_invoke", "script_evaluator", agent_role="script_evaluator", agent_depth=1, scope_event_id=11, parent_event_id=None, round_index=1, agentOutputData={"summary": "### 整体结论：通过"}, ), ] associations = RuntimeEventIndex(events).agent_returns( "script_evaluator", round_index=1 ) assert [(event["id"], relation) for event, relation in associations] == [ (10, "returned-report"), (11, "runtime-associated"), ] assert RuntimeEventIndex(events).agent_returns("script_evaluator", round_index=99) == [] def test_direction_parser_selects_real_goals_instead_of_value_proposition(): parsed = parse_script_direction(DIRECTION) assert parsed["goals"] == [ "拆解机制并让读者看懂因果链。", "用可核验数据支撑关键结论。", ] assert parsed["constraints"] == ["数据可信度", "适用边界", "结构清楚。"] assert parsed["valueJudgements"][0]["label"] == "选题价值主张" assert parsed["goals"][0] != parsed["valueJudgements"][0]["value"] assert parsed["raw"] == DIRECTION.strip() def test_full_projection_preserves_db_truth_and_three_honest_decision_chains(): first_paths = [ { "path_index": 1, "path_type": "内容", "action": "新增", "target": "机制拆解段", "method": "案例对照", "emphasis": "结构可读性", }, { "path_index": 2, "path_type": "领域信息", "action": "核验", "target": "行业数据", "method": "交叉验证", }, ] second_paths = [ {"path_index": 1, "path_type": "内容", "action": "补齐", "target": "结尾"} ] events = [ _main_scope(), _main_tool(2, 2, "get_topic_detail", output_preview='{"topic":{"result":"解释真实机制"},"points":[{"point_result":"测试选题"}]}'), _main_tool(3, 3, "get_account_script_section_patterns", output_preview='{"\u5206\u6bb5\u89c4\u5f8b\u6458\u8981":"成品展示到机制拆解的四段结构"}'), _main_tool(4, 4, "get_account_script_persona_points", output_preview='{"returned_count":82,"records":[{"\u5217":"形式"},{"\u5217":"作用"}]}'), _event( 40, 5, "tool_call", "get_topic_detail", agent_role="script_evaluator", agent_depth=1, scope_event_id=39, parent_event_id=39, output_preview='{"count":1}', ), _main_tool( 5, 6, "save_script_direction", inputData={"script_direction": DIRECTION}, output_preview='{"success": true}', ), _main_tool( 6, 7, "begin_round", round_index=0, inputData={"goal": "搭建机制拆解主结构"}, output_preview='{"success": true, "round_index": 1}', ), _event( 7, 8, "round_begin", "1", agent_role="main", agent_depth=0, scope_event_id=1, parent_event_id=1, payload={"round_index": 1}, ), _main_tool(8, 9, "get_script_snapshot", round_index=1, output_preview='{"count":0}'), _main_tool( 9, 10, "record_multipath_plan", round_index=1, inputData={"paths": first_paths, "mode": "竞争", "note": "一路搭结构，一路核验数据。"}, output_preview='{"success": true, "round_index": 1, "path_count": 2}', ), _main_tool( 90, 11, "record_multipath_plan", round_index=1, inputData={"paths": [], "mode": "竞争"}, status="error", output_preview='{"success": false, "error": "paths required"}', ), _event( 10, 12, "agent_invoke", "script_evaluator", agent_role="script_evaluator", agent_depth=1, scope_event_id=10, parent_event_id=1, round_index=1, agentOutputData={ "summary": "## 评估报告 round=7\n### 整体结论：部分通过\n引领目标：补齐结尾\n遗留问题：缺少行动建议" }, ), _main_tool( 11, 13, "begin_round", round_index=1, inputData={"goal": "补齐结尾。\n具体任务：增加行动建议"}, output_preview='{"success": true, "round_index": 2}', ), _event( 12, 14, "round_begin", "2", agent_role="main", agent_depth=0, scope_event_id=1, parent_event_id=1, payload={"round_index": 2}, ), _main_tool( 13, 15, "record_multipath_plan", round_index=2, inputData={"paths": second_paths, "mode": "分工", "note": "任务已收敛，不需要重复探索。"}, output_preview='{"success": true, "round_index": 2, "path_count": 1}', ), ] rounds = [ { "round_index": 1, "goal": "搭建机制拆解主结构", "multipath_plan": first_paths, "race_or_divide": "竞争", "plan_note": "一路搭结构，一路核验数据。", }, { "round_index": 2, "goal": "补齐结尾。\n具体任务：增加行动建议", "multipath_plan": second_paths, "race_or_divide": "分工", "plan_note": "任务已收敛，不需要重复探索。", }, ] projection = MainAgentDecisionProjector().project( RuntimeEventIndex(events), script_direction=DIRECTION, rounds=rounds, multipath_decisions=[ { "id": 501, "round_index": 1, "decision": "采用结构方案，保留数据核验结果。", "reasoning": "两路结果可组合。", "created_at": "2026-07-14T10:00:11", } ], ) objective = projection["objective"] assert objective["decisionItems"][0] == "拆解机制并让读者看懂因果链。" assert [item["label"] for item in objective["inputs"]] == [ "选题", "账号段落模式", "账号人设", ] assert [item["summary"] for item in objective["inputs"]] == [ "测试选题：解释真实机制", "成品展示到机制拆解的四段结构", "返回 82 个账号表达特征，主要涉及形式、作用", ] assert 40 not in [int(item["detailRef"].split(":")[1]) for item in objective["inputs"]] round_two = projection["roundGoalsByRound"][2] assert [item["relation"] for item in round_two["inputs"][:2]] == [ "returned-report", "persisted-record", ] assert round_two["inputs"][0]["summary"] == "部分通过" assert "下一步：补齐结尾" in round_two["constraints"] assert round_two["revisionRefs"] == ["event:11"] first_plan = projection["implementationPlansByRound"][1] assert first_plan["decisionItems"] == [ "方案 1 · 新增：机制拆解段", "方案 2 · 核验：行业数据", ] assert first_plan["inputs"][1]["label"] == "当前主脚本" assert first_plan["currentRevisionRef"] == "event:9" assert first_plan["technicalRefs"] == ["event:90"] assert first_plan["routeItems"][0] == { "pathIndex": 1, "pathType": "内容", "action": "新增", "target": "机制拆解段", "method": "案例对照", "emphasis": "结构可读性", } second_plan = projection["implementationPlansByRound"][2] assert second_plan["displayMode"] == "单路" assert second_plan["rawMode"] == "分工" assert projection["objective"]["sourceDocument"] == DIRECTION.strip() first_goal = projection["roundGoalsByRound"][1] assert first_goal["decisionItems"] == ["搭建机制拆解主结构"] assert first_goal["sourceDocument"] == "搭建机制拆解主结构" assert "goalSections" not in first_goal def test_round_goal_inputs_keep_complete_business_text(): long_goal = "完整创作目标需要保留全部约束和判断依据。" * 24 long_report = "上一轮整体评审的完整结论需要原样传给下一轮。" * 20 long_decision = "上一轮主 Agent 的完整多路决策需要原样保留。" * 20 direction = f"## 创作总目标\n- 目标语句：{long_goal}" events = [ _main_scope(), _main_tool( 2, 2, "save_script_direction", inputData={"script_direction": direction}, output_preview='{"success": true}', ), _main_tool( 3, 3, "begin_round", round_index=0, output_preview='{"success": true, "round_index": 1}', ), _event( 4, 4, "agent_invoke", "script_evaluator", agent_role="script_evaluator", agent_depth=1, scope_event_id=4, parent_event_id=1, round_index=1, agentOutputData={"summary": f"### 整体结论：{long_report}"}, ), _main_tool( 5, 5, "begin_round", round_index=1, output_preview='{"success": true, "round_index": 2}', ), ] projection = MainAgentDecisionProjector().project( RuntimeEventIndex(events), script_direction=direction, rounds=[ {"round_index": 1, "goal": "第一轮", "multipath_plan": []}, {"round_index": 2, "goal": "第二轮", "multipath_plan": []}, ], multipath_decisions=[ { "id": 88, "round_index": 1, "decision": long_decision, "created_at": "2026-07-14T10:00:04", } ], ) first_round_input = projection["roundGoalsByRound"][1]["inputs"][0]["summary"] second_round_inputs = projection["roundGoalsByRound"][2]["inputs"] assert first_round_input == "\n\n".join(projection["objective"]["decisionItems"]) assert len(first_round_input) > 220 assert second_round_inputs[0]["summary"] == long_report assert second_round_inputs[1]["summary"] == long_decision assert all(not str(item["summary"]).endswith("…") for item in second_round_inputs[:2]) def test_db_only_projection_degrades_without_inventing_runtime_evidence(): projection = MainAgentDecisionProjector().project( RuntimeEventIndex([]), script_direction="做一个清晰的行业解释。", rounds=[ { "round_index": 1, "goal": "搭建结构", "multipath_plan": [{"path_index": 1, "target": "开场"}], "race_or_divide": "分工", "plan_note": None, } ], ) assert projection["objective"]["completeness"] == "partial" assert projection["objective"]["inputs"] == [] assert projection["roundGoalsByRound"][1]["completeness"] == "partial" assert projection["implementationPlansByRound"][1]["completeness"] == "partial" assert projection["implementationPlansByRound"][1]["explicitReasoning"] is None def test_multiple_saves_and_plan_versions_preserve_history_and_separate_failures(): old_direction = "## 创作总目标\n- 目标语句：旧目标" current_direction = "## 创作总目标\n- 目标语句：新目标" old_paths = [{"path_index": 1, "action": "新增", "target": "开场"}] current_paths = [{"path_index": 1, "action": "改写", "target": "开场"}] events = [ _main_scope(), _main_tool(2, 2, "save_script_direction", inputData={"script_direction": old_direction}, output_preview='{"success":true}'), _main_tool(3, 3, "save_script_direction", inputData={"script_direction": current_direction}, output_preview='{"success":true}'), _main_tool(4, 4, "save_script_direction", status="error", inputData={"script_direction": "失败版"}, output_preview='{"success":false,"error":"write failed"}'), _main_tool(5, 5, "begin_round", round_index=0, output_preview='{"success":true,"round_index":1}'), _main_tool(6, 6, "record_multipath_plan", round_index=1, inputData={"paths": old_paths, "mode": "竞争", "note": "初版"}, output_preview='{"success":true,"round_index":1}'), _main_tool(7, 7, "record_multipath_plan", round_index=1, inputData={"paths": current_paths, "mode": "分工", "note": "修订版"}, output_preview='{"success":true,"round_index":1}'), ] projection = MainAgentDecisionProjector().project( RuntimeEventIndex(events), script_direction=current_direction, rounds=[{ "round_index": 1, "goal": "修正开场", "multipath_plan": current_paths, "race_or_divide": "分工", "plan_note": "修订版", }], ) assert projection["objective"]["revisionRefs"] == ["event:2", "event:3"] assert projection["objective"]["currentRevisionRef"] == "event:3" assert projection["objective"]["technicalRefs"] == ["event:4"] plan = projection["implementationPlansByRound"][1] assert plan["revisionRefs"] == ["event:6", "event:7"] assert [item["current"] for item in plan["revisions"]] == [False, True] assert plan["currentRevisionRef"] == "event:7" def test_goal_parser_keeps_the_saved_goal_as_one_item(): goal = "本轮要补齐结构\n① 增加开场\n② 补齐数据" assert parse_goal_items(goal) == [goal] def test_plan_revision_matches_business_alias_without_changing_raw_db_mode(): paths = [{"path_index": 1, "action": "改写", "target": "开场"}] events = [ _main_scope(), _main_tool(2, 2, "begin_round", round_index=0, output_preview='{"success":true,"round_index":1}'), _main_tool( 3, 3, "record_multipath_plan", round_index=1, inputData={"paths": paths, "mode": "竞争", "note": "比较两路表达"}, output_preview='{"success":true,"round_index":1}', ), ] plan = MainAgentDecisionProjector().project( RuntimeEventIndex(events), script_direction=None, rounds=[{ "round_index": 1, "goal": "修正开场", "multipath_plan": paths, "race_or_divide": "赛马", "plan_note": "比较两路表达", }], )["implementationPlansByRound"][1] assert plan["rawMode"] == "赛马" assert plan["currentRevisionRef"] == "event:3" def test_failed_begin_round_does_not_reuse_its_old_round_context(): events = [ _main_scope(), _main_tool( 2, 2, "begin_round", round_index=1, status="error", output_preview='{"success":false,"error":"write failed"}', ), ] projection = MainAgentDecisionProjector().project( RuntimeEventIndex(events), script_direction="创作目标", rounds=[ {"round_index": 1, "goal": "第一轮", "multipath_plan": []}, {"round_index": 2, "goal": "第二轮", "multipath_plan": []}, ], ) assert projection["roundGoalsByRound"][1]["technicalRefs"] == [] assert projection["roundGoalsByRound"][2]["technicalRefs"] == [] assert projection["unassigned"][0]["eventId"] == 2 assert projection["unassigned"][0]["association"] == "failed-main-decision-call-unassigned" def test_round_goal_keeps_report_problems_and_rejects_late_db_decisions(): events = [ _main_scope(), _event( 10, 2, "agent_invoke", "script_evaluator", agent_role="script_evaluator", agent_depth=1, scope_event_id=10, parent_event_id=1, round_index=1, agentOutputData={ "summary": """### 整体结论：部分通过 **问题详情**： - **paragraph_id=9**：收尾段落缺失。 **下一步引领目标**： **补齐收尾并形成闭环。** """ }, ), _main_tool( 11, 5, "begin_round", round_index=1, started_at="2026-07-14T10:00:05", output_preview='{"success":true,"round_index":2}', ), ] projection = MainAgentDecisionProjector().project( RuntimeEventIndex(events), script_direction="创作目标", rounds=[ {"round_index": 1, "goal": "第一轮", "multipath_plan": []}, {"round_index": 2, "goal": "补齐收尾", "multipath_plan": []}, ], multipath_decisions=[ { "id": 71, "round_index": 1, "decision": "采用方案 1", "created_at": "2026-07-14T10:00:06", } ], ) goal = projection["roundGoalsByRound"][2] assert "问题：收尾段落缺失。" in goal["constraints"] assert "下一步：补齐收尾并形成闭环。" in goal["constraints"] assert not any(value.endswith("---") for value in goal["constraints"]) assert not any(item["label"] == "上一轮主 Agent 多路决策" for item in goal["inputs"]) assert "multipath-decision:71" in goal["technicalRefs"]