test_runtime_event_projection.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401
  1. from copy import deepcopy
  2. from datetime import datetime, timezone
  3. from app.runtime_event_projection import RuntimeEventProjector
  4. def _implementer_events():
  5. return [
  6. {"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"},
  7. {"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"},
  8. {"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"},
  9. {"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"},
  10. {"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"},
  11. {"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"},
  12. ]
  13. def test_explicit_event_tree_separates_direct_tools_agent_queries_and_screening():
  14. projection = RuntimeEventProjector().project(_implementer_events(), valid_rounds={1}, valid_branches={7})
  15. stage = projection["retrievalStagesByBranch"][7]
  16. assert stage["observedMode"] == "sequential"
  17. assert stage["directToolGroups"][0]["sources"] == [{"businessLabel": "选题", "callCount": 1}]
  18. assert len(stage["agentRuns"]) == 1
  19. run = stage["agentRuns"][0]
  20. assert run["id"] == "retrieval-agent:20"
  21. assert run["taskPreview"] == "查找叙事案例"
  22. assert run["screening"]["preview"] == "初筛出两个案例"
  23. assert run["querySummary"] == {"total": 2, "hit": 1, "empty": 1, "failure": 0, "unknown": 0}
  24. assert [item["eventId"] for item in run["attempts"]] == [22, 23]
  25. def test_same_agent_type_invoked_twice_stays_two_instances():
  26. events = _implementer_events()
  27. second = deepcopy(events[2])
  28. second.update({"id": 30, "event_seq": 7, "started_at": "2026-07-13T10:00:40", "ended_at": "2026-07-13T10:00:50"})
  29. events.append(second)
  30. projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
  31. assert [item["id"] for item in projection["retrievalStagesByBranch"][7]["agentRuns"]] == ["retrieval-agent:20", "retrieval-agent:30"]
  32. def test_missing_parent_is_unassigned_instead_of_text_matched():
  33. event = {"id": 3, "event_type": "agent_invoke", "event_name": "retrieve_data_knowledge", "round_index": 1, "branch_id": 7, "inputData": {"task": "文字里写了方案 7"}}
  34. projection = RuntimeEventProjector().project([event], valid_rounds={1}, valid_branches={7})
  35. assert projection["retrievalStagesByBranch"] == {}
  36. assert projection["unassigned"][0]["eventId"] == 3
  37. assert projection["unassigned"][0]["association"] == "missing-parent-implementer"
  38. def test_parallel_and_mixed_modes_come_from_time_overlap():
  39. events = _implementer_events()
  40. first_agent = events[2]
  41. direct = events[1]
  42. direct["ended_at"] = "2026-07-13T10:00:20"
  43. first_agent["started_at"] = "2026-07-13T10:00:10"
  44. projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
  45. assert projection["retrievalStagesByBranch"][7]["observedMode"] == "parallel"
  46. later = deepcopy(first_agent)
  47. later.update({"id": 40, "event_seq": 9, "started_at": "2026-07-13T10:00:40", "ended_at": "2026-07-13T10:00:50"})
  48. events.append(later)
  49. projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
  50. assert projection["retrievalStagesByBranch"][7]["observedMode"] == "mixed"
  51. def test_unknown_time_does_not_claim_parallelism():
  52. events = _implementer_events()
  53. events[2]["started_at"] = None
  54. projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
  55. stage = projection["retrievalStagesByBranch"][7]
  56. assert stage["observedMode"] == "unknown"
  57. assert stage["waves"] == []
  58. def test_failed_direct_read_makes_stage_partial():
  59. events = _implementer_events()
  60. events[1]["status"] = "error"
  61. projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7})
  62. stage = projection["retrievalStagesByBranch"][7]
  63. assert stage["directToolGroups"][0]["failureCount"] == 1
  64. assert stage["status"] == "partial"
  65. def test_running_implementer_without_retrieval_calls_is_preparing_not_missing():
  66. events = [{
  67. "id": 10, "event_seq": 1, "event_type": "agent_invoke",
  68. "event_name": "script_implementer", "round_index": 1,
  69. "branch_id": 7, "status": "running",
  70. "started_at": "2026-07-14T12:00:00", "ended_at": None,
  71. }]
  72. projection = RuntimeEventProjector().project(
  73. events,
  74. valid_rounds={1},
  75. valid_branches={7},
  76. captured_at=datetime(2026, 7, 14, 4, 0, 5, tzinfo=timezone.utc),
  77. )
  78. stage = projection["retrievalStagesByBranch"][7]
  79. assert stage["status"] == "running"
  80. assert stage["observedMode"] == "unknown"
  81. def test_terminal_run_reclassifies_unfinished_retrieval_as_interrupted():
  82. events = _implementer_events()
  83. events[0].update({"status": "running", "ended_at": None})
  84. events[1].update({"status": "running", "ended_at": None})
  85. events[2].update({"status": "running", "ended_at": None})
  86. events[4].update({"status": "running", "ended_at": None})
  87. projection = RuntimeEventProjector().project(
  88. events,
  89. valid_rounds={1},
  90. valid_branches={7},
  91. run_status="failed",
  92. )
  93. stage = projection["retrievalStagesByBranch"][7]
  94. assert stage["status"] == "partial"
  95. assert stage["directToolGroups"][0]["calls"][0]["status"] == "interrupted"
  96. assert stage["agentRuns"][0]["status"] == "interrupted"
  97. assert stage["agentRuns"][0]["attempts"][0]["status"] == "interrupted"
  98. def test_running_naive_database_time_uses_asia_shanghai_before_wave_projection():
  99. events = _implementer_events()
  100. events[0].update({"status": "running", "ended_at": None})
  101. events[1].update({"status": "running", "started_at": "2026-07-14T12:00:00", "ended_at": None})
  102. events[2].update({"status": "running", "started_at": "2026-07-14T12:00:02", "ended_at": None})
  103. projection = RuntimeEventProjector().project(
  104. events,
  105. valid_rounds={1},
  106. valid_branches={7},
  107. captured_at=datetime(2026, 7, 14, 4, 0, 5, tzinfo=timezone.utc),
  108. )
  109. stage = projection["retrievalStagesByBranch"][7]
  110. assert stage["observedMode"] == "parallel"
  111. assert stage["startedAt"] == "2026-07-14T04:00:00+00:00"
  112. assert stage["endedAt"] == "2026-07-14T04:00:05+00:00"
  113. def test_query_output_error_precedes_empty_count_and_unstructured_object_is_unknown():
  114. events = _implementer_events()
  115. events[4]["output_preview"] = '{"error":"knowledge unavailable","items":[]}'
  116. events[5]["output_preview"] = '{"message":"query accepted"}'
  117. stage = RuntimeEventProjector().project(
  118. events, valid_rounds={1}, valid_branches={7}
  119. )["retrievalStagesByBranch"][7]
  120. assert [attempt["status"] for attempt in stage["agentRuns"][0]["attempts"]] == ["failure", "unknown"]
  121. assert stage["agentRuns"][0]["querySummary"] == {
  122. "total": 2, "hit": 0, "empty": 0, "failure": 1, "unknown": 1,
  123. }
  124. def test_agent_card_screening_preview_uses_business_cleaning():
  125. events = _implementer_events()
  126. events[2]["agentOutputData"] = {
  127. "summary": "## 委托任务完成\n\n### 初筛结果\n- 找到可用案例 post_id=abc123\n\n**执行统计**:\n- Tokens: 1200\n- 成本: $0.1"
  128. }
  129. stage = RuntimeEventProjector().project(
  130. events, valid_rounds={1}, valid_branches={7}
  131. )["retrievalStagesByBranch"][7]
  132. preview = stage["agentRuns"][0]["screening"]["preview"]
  133. assert "可用案例" in preview
  134. assert "post_id" not in preview
  135. assert "abc123" not in preview
  136. assert "Tokens" not in preview
  137. assert "成本" not in preview
  138. def test_creative_business_projection_translates_runtime_tool_names():
  139. events = [
  140. {
  141. "id": 10,
  142. "event_seq": 1,
  143. "event_type": "agent_invoke",
  144. "event_name": "script_implementer",
  145. "round_index": 1,
  146. "branch_id": 7,
  147. "status": "ok",
  148. "inputData": {
  149. "task": "路序号:1 action:增 target:烹饪贴士 path_type:内容 目标:补齐烹饪贴士与总结。"
  150. },
  151. },
  152. {
  153. "id": 11,
  154. "event_seq": 2,
  155. "event_type": "tool_call",
  156. "event_name": "think_and_plan",
  157. "agent_role": "script_implementer",
  158. "scope_event_id": 10,
  159. "round_index": 1,
  160. "branch_id": 7,
  161. "status": "ok",
  162. "inputData": {
  163. "thought": "缺少直接证据,先保留为待验证推导。",
  164. "action": "record_data_decision",
  165. "plan": "1. 调用 get_topic_detail。\n2. 委派 retrieve_data_knowledge。\n3. 调用 batch_update_script_paragraphs。",
  166. },
  167. },
  168. ]
  169. decision = RuntimeEventProjector().project(
  170. events, valid_rounds={1}, valid_branches={7}
  171. )["creativeByBranch"][7]
  172. assert decision["body"]["task"] == "补齐烹饪贴士与总结。"
  173. assert decision["body"]["actions"] == ["记录数据取舍"]
  174. assert decision["body"]["steps"] == [
  175. {
  176. "stepIndex": 1,
  177. "action": "记录数据取舍",
  178. "summary": None,
  179. "plan": "1. 读取选题。\n2. 委派知识 Agent取数。\n3. 更新候选脚本内容。",
  180. "reasoning": "缺少直接证据,先保留为待验证推导。",
  181. "eventRef": "event:11",
  182. }
  183. ]
  184. process_block = decision["detail"]["blocks"][0]
  185. assert process_block["type"] == "creative-process"
  186. assert process_block["steps"][0]["action"] == "记录数据取舍"
  187. business = str(decision["detail"]["blocks"])
  188. assert "record_data_decision" not in business
  189. assert "get_topic_detail" not in business
  190. assert "batch_update_script_paragraphs" not in business
  191. assert "path_type" not in business
  192. def test_creative_projection_keeps_upstream_and_actual_mutation_refs_separate():
  193. events = [
  194. {
  195. "id": 100,
  196. "event_seq": 1,
  197. "event_type": "agent_invoke",
  198. "event_name": "script_implementer",
  199. "round_index": 1,
  200. "branch_id": 7,
  201. "status": "ok",
  202. "inputData": {"task": "目标:补齐正文。"},
  203. },
  204. {
  205. "id": 101,
  206. "event_seq": 2,
  207. "event_type": "tool_call",
  208. "event_name": "get_script_snapshot",
  209. "agent_role": "script_implementer",
  210. "scope_event_id": 100,
  211. "round_index": 1,
  212. "branch_id": 7,
  213. "status": "ok",
  214. },
  215. {
  216. "id": 102,
  217. "event_seq": 3,
  218. "event_type": "tool_call",
  219. "event_name": "think_and_plan",
  220. "agent_role": "script_implementer",
  221. "scope_event_id": 100,
  222. "round_index": 1,
  223. "branch_id": 7,
  224. "status": "ok",
  225. "inputData": {"action": "更新候选正文", "thought": "先根据已有结构补写。"},
  226. },
  227. {
  228. "id": 103,
  229. "event_seq": 4,
  230. "event_type": "tool_call",
  231. "event_name": "batch_update_script_paragraphs",
  232. "agent_role": "script_implementer",
  233. "scope_event_id": 100,
  234. "round_index": 1,
  235. "branch_id": 7,
  236. "status": "ok",
  237. },
  238. ]
  239. decision = RuntimeEventProjector().project(
  240. events, valid_rounds={1}, valid_branches={7}
  241. )["creativeByBranch"][7]
  242. assert decision["eventRefs"] == ["event:101", "event:102", "event:103"]
  243. assert decision["body"]["availableUpstream"][0]["decisionUse"] == "not-recorded"
  244. assert decision["body"]["runtimeActions"][0]["label"] == "更新候选脚本内容"
  245. blocks = {item["title"]: item for item in decision["detail"]["blocks"]}
  246. upstream_item = blocks["可确认的上游信息"]["items"][0]
  247. assert "无法确认是否被本次创作采用" in upstream_item["note"]
  248. assert blocks["实际脚本改动"]["items"][0]["detailRef"] == "event:103"
  249. def test_exact_evaluator_names_create_two_different_business_stages():
  250. events = [
  251. {
  252. "id": 80, "event_seq": 1, "event_type": "agent_invoke",
  253. "event_name": "script_multipath_evaluator", "round_index": 1,
  254. "status": "ok", "inputData": {"task": "比较 branch_id=7, branch_id=8"},
  255. "agentOutputData": {"summary": "### 分支评估 branch_id=7\n**该支结论**:通过\n**目标推进**:结构完整。\n\n### 分支评估 branch_id=8\n**该支结论**:部分通过\n**目标推进**:仍需补充。\n\n### 对比与建议\n**采纳建议**:优先采用 branch_id=7。"},
  256. "started_at": "2026-07-13T10:03:00", "ended_at": "2026-07-13T10:03:10",
  257. },
  258. {
  259. "id": 81, "event_seq": 2, "event_type": "agent_invoke",
  260. "event_name": "script_evaluator", "round_index": 1,
  261. "status": "ok", "agentOutputData": {"summary": "### 整体结论:部分通过"},
  262. "started_at": "2026-07-13T10:04:00", "ended_at": "2026-07-13T10:04:10",
  263. },
  264. {
  265. "id": 82, "event_seq": 3, "event_type": "agent_invoke",
  266. "event_name": "some_future_evaluator", "round_index": 1,
  267. "status": "ok", "started_at": "2026-07-13T10:05:00", "ended_at": "2026-07-13T10:05:10",
  268. },
  269. ]
  270. projection = RuntimeEventProjector().project(events, valid_rounds={1}, valid_branches={7, 8})
  271. review = projection["multipathReviewsByRound"][1][0]
  272. assert review["branchIds"] == [7, 8]
  273. assert [item["conclusion"] for item in review["branchConclusions"]] == ["通过", "部分通过"]
  274. assert "优先采用" in review["recommendation"]
  275. assert projection["overallEvaluationsByRound"][1][0]["eventId"] == 81
  276. assert "evaluationsByBranch" not in projection
  277. assert "evaluationsByRound" not in projection
  278. def test_multipath_evaluator_parses_one_labelled_comma_separated_branch_list():
  279. event = {
  280. "id": 3687,
  281. "event_seq": 1,
  282. "event_type": "agent_invoke",
  283. "event_name": "script_multipath_evaluator",
  284. "round_index": 2,
  285. "status": "ok",
  286. "inputData": {
  287. "task": "评估对象:branch_id = 4, 5, 6, 7(分工,各填充一段)"
  288. },
  289. "agentOutputData": {
  290. "summary": "\n".join(
  291. f"### 分支评估 branch_id={branch_id}\n**该支结论**:通过"
  292. for branch_id in (4, 5, 6, 7)
  293. )
  294. },
  295. "started_at": "2026-07-14T17:17:09",
  296. "ended_at": "2026-07-14T17:18:06",
  297. }
  298. projection = RuntimeEventProjector().project(
  299. [event], valid_rounds={2}, valid_branches={4, 5, 6, 7}
  300. )
  301. assert projection["multipathReviewsByRound"][2][0]["branchIds"] == [4, 5, 6, 7]
  302. def test_facade_builds_one_index_and_merges_projector_outputs():
  303. received = []
  304. class RetrievalSpy:
  305. def project(self, index, **kwargs):
  306. received.append(index)
  307. return {
  308. "retrievalStagesByBranch": {7: {"id": "retrieval"}},
  309. "unassigned": [{"id": "retrieval-unassigned"}],
  310. }
  311. class EvaluationSpy:
  312. def project(self, index, **kwargs):
  313. received.append(index)
  314. return {
  315. "multipathReviewsByRound": {1: [{"id": "review"}]},
  316. "overallEvaluationsByRound": {1: [{"id": "evaluation"}]},
  317. "unassigned": [{"id": "evaluation-unassigned"}],
  318. }
  319. class MainAgentSpy:
  320. def project(self, index, **kwargs):
  321. received.append(index)
  322. return {
  323. "objective": {"stage": "objective"},
  324. "roundGoalsByRound": {},
  325. "implementationPlansByRound": {},
  326. }
  327. projection = RuntimeEventProjector(
  328. retrieval_projector=RetrievalSpy(),
  329. evaluation_projector=EvaluationSpy(),
  330. main_agent_projector=MainAgentSpy(),
  331. ).project(
  332. [{
  333. "id": 9,
  334. "event_seq": 1,
  335. "event_type": "tool_call",
  336. "event_name": "record_multipath_plan",
  337. "round_index": 1,
  338. "status": "ok",
  339. }],
  340. valid_rounds={1},
  341. valid_branches={7},
  342. )
  343. assert len({id(index) for index in received}) == 1
  344. assert projection["retrievalStagesByBranch"][7]["id"] == "retrieval"
  345. assert projection["multipathReviewsByRound"][1][0]["id"] == "review"
  346. assert projection["overallEvaluationsByRound"][1][0]["id"] == "evaluation"
  347. assert projection["mainAgentDecisions"]["objective"]["stage"] == "objective"
  348. assert "planRevisionsByRound" not in projection
  349. assert projection["unassigned"] == [
  350. {"id": "retrieval-unassigned"},
  351. {"id": "evaluation-unassigned"},
  352. ]