script-build-and-framework-tmp/visualization/backend/tests/test_main_agent_decision_projection.py

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"]