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

from copy import deepcopy
from datetime import datetime, timezone

from app.runtime_event_projection import RuntimeEventProjector


def _implementer_events():
    return [
        {"id": 10, "event_seq": 1, "event_type": "agent_invoke", "event_name": "script_implementer", "round_index": 1, "branch_id": 7, "status": "ok", "started_at": "2026-07-13T10:00:00", "ended_at": "2026-07-13T10:02:00"},
        {"id": 11, "event_seq": 2, "event_type": "tool_call", "event_name": "get_topic_detail", "agent_role": "script_implementer", "scope_event_id": 10, "round_index": 1, "branch_id": 7, "status": "ok", "started_at": "2026-07-13T10:00:01", "ended_at": "2026-07-13T10:00:02"},
        {"id": 20, "event_seq": 3, "event_type": "agent_invoke", "event_name": "retrieve_data_decode_case", "parent_event_id": 10, "scope_event_id": 10, "round_index": 1, "branch_id": 7, "status": "ok", "inputData": {"task": "查找叙事案例"}, "agentOutputData": {"summary": "初筛出两个案例"}, "started_at": "2026-07-13T10:00:10", "ended_at": "2026-07-13T10:00:30"},
        {"id": 21, "event_seq": 4, "event_type": "tool_call", "event_name": "think_and_plan", "scope_event_id": 20, "parent_event_id": 20, "status": "ok", "started_at": "2026-07-13T10:00:11", "ended_at": "2026-07-13T10:00:12"},
        {"id": 22, "event_seq": 5, "event_type": "tool_call", "event_name": "search_script_decode_case", "scope_event_id": 20, "parent_event_id": 20, "status": "ok", "inputData": {"keyword": "叙事"}, "output_preview": '{"count": 2}', "started_at": "2026-07-13T10:00:13", "ended_at": "2026-07-13T10:00:17"},
        {"id": 23, "event_seq": 6, "event_type": "tool_call", "event_name": "search_script_decode_case", "scope_event_id": 20, "parent_event_id": 20, "status": "ok", "inputData": {"keyword": "构图"}, "output_preview": '{"count": 0}', "started_at": "2026-07-13T10:00:18", "ended_at": "2026-07-13T10:00:22"},
    ]


def test_explicit_event_tree_separates_direct_tools_agent_queries_and_screening():
    projection = RuntimeEventProjector().project(_implementer_events(), valid_rounds={1}, valid_branches={7})
    stage = projection["retrievalStagesByBranch"][7]

    assert stage["observedMode"] == "sequential"
    assert stage["directToolGroups"][0]["sources"] == [{"businessLabel": "选题", "callCount": 1}]
    assert len(stage["agentRuns"]) == 1
    run = stage["agentRuns"][0]
    assert run["id"] == "retrieval-agent:20"
    assert run["taskPreview"] == "查找叙事案例"
    assert run["screening"]["preview"] == "初筛出两个案例"
    assert run["querySummary"] == {"total": 2, "hit": 1, "empty": 1, "failure": 0, "unknown": 0}
    assert [item["eventId"] for item in run["attempts"]] == [22, 23]


def test_same_agent_type_invoked_twice_stays_two_instances():
    events = _implementer_events()
    second = deepcopy(events[2])
    second.update({"id": 30, "event_seq": 7, "started_at": "2026-07-13T10:00:40", "ended_at": "2026-07-13T10:00:50"})
    events.append(second)
    projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
    assert [item["id"] for item in projection["retrievalStagesByBranch"][7]["agentRuns"]] == ["retrieval-agent:20", "retrieval-agent:30"]


def test_missing_parent_is_unassigned_instead_of_text_matched():
    event = {"id": 3, "event_type": "agent_invoke", "event_name": "retrieve_data_knowledge", "round_index": 1, "branch_id": 7, "inputData": {"task": "文字里写了方案 7"}}
    projection = RuntimeEventProjector().project([event], valid_rounds={1}, valid_branches={7})
    assert projection["retrievalStagesByBranch"] == {}
    assert projection["unassigned"][0]["eventId"] == 3
    assert projection["unassigned"][0]["association"] == "missing-parent-implementer"


def test_parallel_and_mixed_modes_come_from_time_overlap():
    events = _implementer_events()
    first_agent = events[2]
    direct = events[1]
    direct["ended_at"] = "2026-07-13T10:00:20"
    first_agent["started_at"] = "2026-07-13T10:00:10"
    projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
    assert projection["retrievalStagesByBranch"][7]["observedMode"] == "parallel"

    later = deepcopy(first_agent)
    later.update({"id": 40, "event_seq": 9, "started_at": "2026-07-13T10:00:40", "ended_at": "2026-07-13T10:00:50"})
    events.append(later)
    projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
    assert projection["retrievalStagesByBranch"][7]["observedMode"] == "mixed"


def test_unknown_time_does_not_claim_parallelism():
    events = _implementer_events()
    events[2]["started_at"] = None
    projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
    stage = projection["retrievalStagesByBranch"][7]
    assert stage["observedMode"] == "unknown"
    assert stage["waves"] == []


def test_failed_direct_read_makes_stage_partial():
    events = _implementer_events()
    events[1]["status"] = "error"
    projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
    stage = projection["retrievalStagesByBranch"][7]
    assert stage["directToolGroups"][0]["failureCount"] == 1
    assert stage["status"] == "partial"


def test_running_implementer_without_retrieval_calls_is_preparing_not_missing():
    events = [{
        "id": 10, "event_seq": 1, "event_type": "agent_invoke",
        "event_name": "script_implementer", "round_index": 1,
        "branch_id": 7, "status": "running",
        "started_at": "2026-07-14T12:00:00", "ended_at": None,
    }]
    projection = RuntimeEventProjector().project(
        events,
        valid_rounds={1},
        valid_branches={7},
        captured_at=datetime(2026, 7, 14, 4, 0, 5, tzinfo=timezone.utc),
    )
    stage = projection["retrievalStagesByBranch"][7]
    assert stage["status"] == "running"
    assert stage["observedMode"] == "unknown"


def test_terminal_run_reclassifies_unfinished_retrieval_as_interrupted():
    events = _implementer_events()
    events[0].update({"status": "running", "ended_at": None})
    events[1].update({"status": "running", "ended_at": None})
    events[2].update({"status": "running", "ended_at": None})
    events[4].update({"status": "running", "ended_at": None})
    projection = RuntimeEventProjector().project(
        events,
        valid_rounds={1},
        valid_branches={7},
        run_status="failed",
    )
    stage = projection["retrievalStagesByBranch"][7]

    assert stage["status"] == "partial"
    assert stage["directToolGroups"][0]["calls"][0]["status"] == "interrupted"
    assert stage["agentRuns"][0]["status"] == "interrupted"
    assert stage["agentRuns"][0]["attempts"][0]["status"] == "interrupted"


def test_running_naive_database_time_uses_asia_shanghai_before_wave_projection():
    events = _implementer_events()
    events[0].update({"status": "running", "ended_at": None})
    events[1].update({"status": "running", "started_at": "2026-07-14T12:00:00", "ended_at": None})
    events[2].update({"status": "running", "started_at": "2026-07-14T12:00:02", "ended_at": None})
    projection = RuntimeEventProjector().project(
        events,
        valid_rounds={1},
        valid_branches={7},
        captured_at=datetime(2026, 7, 14, 4, 0, 5, tzinfo=timezone.utc),
    )
    stage = projection["retrievalStagesByBranch"][7]
    assert stage["observedMode"] == "parallel"
    assert stage["startedAt"] == "2026-07-14T04:00:00+00:00"
    assert stage["endedAt"] == "2026-07-14T04:00:05+00:00"


def test_query_output_error_precedes_empty_count_and_unstructured_object_is_unknown():
    events = _implementer_events()
    events[4]["output_preview"] = '{"error":"knowledge unavailable","items":[]}'
    events[5]["output_preview"] = '{"message":"query accepted"}'
    stage = RuntimeEventProjector().project(
        events, valid_rounds={1}, valid_branches={7}
    )["retrievalStagesByBranch"][7]
    assert [attempt["status"] for attempt in stage["agentRuns"][0]["attempts"]] == ["failure", "unknown"]
    assert stage["agentRuns"][0]["querySummary"] == {
        "total": 2, "hit": 0, "empty": 0, "failure": 1, "unknown": 1,
    }


def test_agent_card_screening_preview_uses_business_cleaning():
    events = _implementer_events()
    events[2]["agentOutputData"] = {
        "summary": "## 委托任务完成\n\n### 初筛结果\n- 找到可用案例 post_id=abc123\n\n**执行统计**:\n- Tokens: 1200\n- 成本: $0.1"
    }
    stage = RuntimeEventProjector().project(
        events, valid_rounds={1}, valid_branches={7}
    )["retrievalStagesByBranch"][7]
    preview = stage["agentRuns"][0]["screening"]["preview"]
    assert "可用案例" in preview
    assert "post_id" not in preview
    assert "abc123" not in preview
    assert "Tokens" not in preview
    assert "成本" not in preview


def test_creative_business_projection_translates_runtime_tool_names():
    events = [
        {
            "id": 10,
            "event_seq": 1,
            "event_type": "agent_invoke",
            "event_name": "script_implementer",
            "round_index": 1,
            "branch_id": 7,
            "status": "ok",
            "inputData": {
                "task": "路序号：1 action：增 target：烹饪贴士 path_type：内容 目标：补齐烹饪贴士与总结。"
            },
        },
        {
            "id": 11,
            "event_seq": 2,
            "event_type": "tool_call",
            "event_name": "think_and_plan",
            "agent_role": "script_implementer",
            "scope_event_id": 10,
            "round_index": 1,
            "branch_id": 7,
            "status": "ok",
            "inputData": {
                "thought": "缺少直接证据，先保留为待验证推导。",
                "action": "record_data_decision",
                "plan": "1. 调用 get_topic_detail。\n2. 委派 retrieve_data_knowledge。\n3. 调用 batch_update_script_paragraphs。",
            },
        },
    ]
    decision = RuntimeEventProjector().project(
        events, valid_rounds={1}, valid_branches={7}
    )["creativeByBranch"][7]

    assert decision["body"]["task"] == "补齐烹饪贴士与总结。"
    assert decision["body"]["actions"] == ["记录数据取舍"]
    assert decision["body"]["steps"] == [
        {
            "stepIndex": 1,
            "action": "记录数据取舍",
            "summary": None,
            "plan": "1. 读取选题。\n2. 委派知识 Agent取数。\n3. 更新候选脚本内容。",
            "reasoning": "缺少直接证据，先保留为待验证推导。",
            "eventRef": "event:11",
        }
    ]
    process_block = decision["detail"]["blocks"][0]
    assert process_block["type"] == "creative-process"
    assert process_block["steps"][0]["action"] == "记录数据取舍"
    business = str(decision["detail"]["blocks"])
    assert "record_data_decision" not in business
    assert "get_topic_detail" not in business
    assert "batch_update_script_paragraphs" not in business
    assert "path_type" not in business


def test_creative_projection_keeps_upstream_and_actual_mutation_refs_separate():
    events = [
        {
            "id": 100,
            "event_seq": 1,
            "event_type": "agent_invoke",
            "event_name": "script_implementer",
            "round_index": 1,
            "branch_id": 7,
            "status": "ok",
            "inputData": {"task": "目标：补齐正文。"},
        },
        {
            "id": 101,
            "event_seq": 2,
            "event_type": "tool_call",
            "event_name": "get_script_snapshot",
            "agent_role": "script_implementer",
            "scope_event_id": 100,
            "round_index": 1,
            "branch_id": 7,
            "status": "ok",
        },
        {
            "id": 102,
            "event_seq": 3,
            "event_type": "tool_call",
            "event_name": "think_and_plan",
            "agent_role": "script_implementer",
            "scope_event_id": 100,
            "round_index": 1,
            "branch_id": 7,
            "status": "ok",
            "inputData": {"action": "更新候选正文", "thought": "先根据已有结构补写。"},
        },
        {
            "id": 103,
            "event_seq": 4,
            "event_type": "tool_call",
            "event_name": "batch_update_script_paragraphs",
            "agent_role": "script_implementer",
            "scope_event_id": 100,
            "round_index": 1,
            "branch_id": 7,
            "status": "ok",
        },
    ]

    decision = RuntimeEventProjector().project(
        events, valid_rounds={1}, valid_branches={7}
    )["creativeByBranch"][7]

    assert decision["eventRefs"] == ["event:101", "event:102", "event:103"]
    assert decision["body"]["availableUpstream"][0]["decisionUse"] == "not-recorded"
    assert decision["body"]["runtimeActions"][0]["label"] == "更新候选脚本内容"
    blocks = {item["title"]: item for item in decision["detail"]["blocks"]}
    upstream_item = blocks["可确认的上游信息"]["items"][0]
    assert "无法确认是否被本次创作采用" in upstream_item["note"]
    assert blocks["实际脚本改动"]["items"][0]["detailRef"] == "event:103"


def test_exact_evaluator_names_create_two_different_business_stages():
    events = [
        {
            "id": 80, "event_seq": 1, "event_type": "agent_invoke",
            "event_name": "script_multipath_evaluator", "round_index": 1,
            "status": "ok", "inputData": {"task": "比较 branch_id=7, branch_id=8"},
            "agentOutputData": {"summary": "### 分支评估 branch_id=7\n**该支结论**：通过\n**目标推进**：结构完整。\n\n### 分支评估 branch_id=8\n**该支结论**：部分通过\n**目标推进**：仍需补充。\n\n### 对比与建议\n**采纳建议**：优先采用 branch_id=7。"},
            "started_at": "2026-07-13T10:03:00", "ended_at": "2026-07-13T10:03:10",
        },
        {
            "id": 81, "event_seq": 2, "event_type": "agent_invoke",
            "event_name": "script_evaluator", "round_index": 1,
            "status": "ok", "agentOutputData": {"summary": "### 整体结论：部分通过"},
            "started_at": "2026-07-13T10:04:00", "ended_at": "2026-07-13T10:04:10",
        },
        {
            "id": 82, "event_seq": 3, "event_type": "agent_invoke",
            "event_name": "some_future_evaluator", "round_index": 1,
            "status": "ok", "started_at": "2026-07-13T10:05:00", "ended_at": "2026-07-13T10:05:10",
        },
    ]
    projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7, 8})

    review = projection["multipathReviewsByRound"][1][0]
    assert review["branchIds"] == [7, 8]
    assert [item["conclusion"] for item in review["branchConclusions"]] == ["通过", "部分通过"]
    assert "优先采用" in review["recommendation"]
    assert projection["overallEvaluationsByRound"][1][0]["eventId"] == 81
    assert "evaluationsByBranch" not in projection
    assert "evaluationsByRound" not in projection


def test_multipath_evaluator_parses_one_labelled_comma_separated_branch_list():
    event = {
        "id": 3687,
        "event_seq": 1,
        "event_type": "agent_invoke",
        "event_name": "script_multipath_evaluator",
        "round_index": 2,
        "status": "ok",
        "inputData": {
            "task": "评估对象：branch_id = 4, 5, 6, 7（分工，各填充一段）"
        },
        "agentOutputData": {
            "summary": "\n".join(
                f"### 分支评估 branch_id={branch_id}\n**该支结论**：通过"
                for branch_id in (4, 5, 6, 7)
            )
        },
        "started_at": "2026-07-14T17:17:09",
        "ended_at": "2026-07-14T17:18:06",
    }

    projection = RuntimeEventProjector().project(
        [event], valid_rounds={2}, valid_branches={4, 5, 6, 7}
    )

    assert projection["multipathReviewsByRound"][2][0]["branchIds"] == [4, 5, 6, 7]


def test_facade_builds_one_index_and_merges_projector_outputs():
    received = []

    class RetrievalSpy:
        def project(self, index, **kwargs):
            received.append(index)
            return {
                "retrievalStagesByBranch": {7: {"id": "retrieval"}},
                "unassigned": [{"id": "retrieval-unassigned"}],
            }

    class EvaluationSpy:
        def project(self, index, **kwargs):
            received.append(index)
            return {
                "multipathReviewsByRound": {1: [{"id": "review"}]},
                "overallEvaluationsByRound": {1: [{"id": "evaluation"}]},
                "unassigned": [{"id": "evaluation-unassigned"}],
            }

    class MainAgentSpy:
        def project(self, index, **kwargs):
            received.append(index)
            return {
                "objective": {"stage": "objective"},
                "roundGoalsByRound": {},
                "implementationPlansByRound": {},
            }

    projection = RuntimeEventProjector(
        retrieval_projector=RetrievalSpy(),
        evaluation_projector=EvaluationSpy(),
        main_agent_projector=MainAgentSpy(),
    ).project(
        [{
            "id": 9,
            "event_seq": 1,
            "event_type": "tool_call",
            "event_name": "record_multipath_plan",
            "round_index": 1,
            "status": "ok",
        }],
        valid_rounds={1},
        valid_branches={7},
    )

    assert len({id(index) for index in received}) == 1
    assert projection["retrievalStagesByBranch"][7]["id"] == "retrieval"
    assert projection["multipathReviewsByRound"][1][0]["id"] == "review"
    assert projection["overallEvaluationsByRound"][1][0]["id"] == "evaluation"
    assert projection["mainAgentDecisions"]["objective"]["stage"] == "objective"
    assert "planRevisionsByRound" not in projection
    assert projection["unassigned"] == [
        {"id": "retrieval-unassigned"},
        {"id": "evaluation-unassigned"},
    ]