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