| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687 |
- from copy import deepcopy
- from app.execution_builder import ExecutionViewBuilder
- from app.inspector_projection import project_activity_detail, project_event_detail, project_round_detail
- def test_v8_contract_separates_multipath_review_main_decision_and_overall_review(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- assert view["schemaVersion"] == "8"
- assert view["dataShape"] == "current"
- first = view["rounds"][0]
- assert len(first["convergenceBatches"]) == 2
- assert [batch["branchIds"] for batch in first["convergenceBatches"]] == [[1, 2], [3]]
- assert [batch["review"]["eventId"] for batch in first["convergenceBatches"]] == [39, 40]
- assert [batch["decision"]["recordId"] for batch in first["convergenceBatches"]] == [21, 22]
- assert first["convergenceBatches"][0]["decision"]["branchOutcomes"] == [
- {"branchId": 1, "status": "merged", "label": "已采用", "reasoning": "方案 1 的当前处理理由"},
- {"branchId": 2, "status": "discarded", "label": "未采用", "reasoning": "方案 2 的当前处理理由"},
- ]
- assert first["overallEvaluation"]["detailRef"] == "event:41"
- assert first["branches"][0]["dataDecisions"][0]["decision"] == "采用案例中的三段结构"
- assert all(decision["recordId"] > 0 for branch in first["branches"] for decision in branch["dataDecisions"])
- assert "dataDecisions" not in first["convergenceBatches"][0]
- assert "evaluation" not in first["branches"][0]
- assert "disposition" not in first["branches"][0]
- retrieval = first["branches"][0]["retrievalStage"]
- assert retrieval["directToolGroups"][0]["calls"][0]["businessLabel"] == "当前主脚本"
- assert retrieval["agentRuns"][0]["businessLabel"] == "解构 Case Agent"
- assert retrieval["agentRuns"][0]["querySummary"]["empty"] == 1
- review_projection = first["convergenceBatches"][0]["review"]["decisionProjection"]
- main_projection = first["convergenceBatches"][0]["decision"]["decisionProjection"]
- data_projection = first["branches"][0]["dataDecisions"][0]["decisionProjection"]
- overall_projection = first["overallEvaluation"]["decision"]
- assert (review_projection["decisionType"], review_projection["authority"]) == (
- "evaluation",
- "recommendation",
- )
- assert (main_projection["decisionType"], main_projection["authority"]) == (
- "tradeoff",
- "final",
- )
- assert (data_projection["decisionType"], data_projection["authority"]) == (
- "tradeoff",
- "implementation-scope",
- )
- assert (overall_projection["decisionType"], overall_projection["authority"]) == (
- "evaluation",
- "recommendation",
- )
- def test_multipath_review_keeps_grouped_branches_and_comparison_matrix(current_bundle):
- review_event = next(item for item in current_bundle["events"] if item["id"] == 39)
- review_event["agentOutputData"]["summary"] = """### 分支评估 branch_id=1(领域信息路)
- **该支结论**:部分通过
- **目标推进**:
- - 事实链完整。
- **未通过/存疑项**:
- - 元素尚未全部落库。
- **有据扎实度**:来源可指位。
- ### 分支评估 branch_id=2(内容路)
- **该支结论**:部分通过
- **目标推进**:
- - 骨架已形成。
- **未通过/部分通过项**:
- - 数量红线违规。
- **有据扎实度**:与快照一致。
- ### 对比与建议
- | 维度 | branch1 | branch2 | 差异依据 |
- |------|---------|---------|----------|
- | 目标契合度 | 高 | 高 | 各自完成主任务 |
- **采纳建议**:组合两个方案。"""
- first = ExecutionViewBuilder().from_bundle(current_bundle)["rounds"][0]
- projection = first["convergenceBatches"][0]["review"]["decisionProjection"]
- blocks = projection["detail"]["blocks"]
- assert [block["type"] for block in blocks[:2]] == [
- "evaluation-branches",
- "evaluation-matrix",
- ]
- branches = blocks[0]["branches"]
- assert [item["subject"] for item in branches] == ["方案 1", "方案 2"]
- assert branches[1]["achievements"] == ["骨架已形成。"]
- assert branches[1]["problems"] == ["数量红线违规。"]
- assert blocks[1]["rows"][0]["criterion"] == "目标契合度"
- def test_same_round_multipath_decisions_remain_separate_and_ordered(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- batches = [item["decision"] for item in view["rounds"][0]["convergenceBatches"]]
- assert [item["recordId"] for item in batches] == [21, 22]
- assert batches[0]["decision"] != batches[1]["decision"]
- assert batches[0]["node"]["detailRef"] == "multipath-decision:21"
- def test_comma_separated_review_scope_joins_matching_main_decision(current_bundle):
- review_event = next(item for item in current_bundle["events"] if item["id"] == 39)
- review_event["inputData"]["task"] = "评估对象:branch_id = 1, 2(分工)"
- first = ExecutionViewBuilder().from_bundle(current_bundle)["rounds"][0]
- batch = next(
- item
- for item in first["convergenceBatches"]
- if item.get("decision") and item["decision"]["recordId"] == 21
- )
- assert batch["association"] == "exact-branch-set-and-time"
- assert batch["review"]["eventId"] == 39
- assert batch["branchIds"] == [1, 2]
- assert batch["missingReview"] is None
- assert batch["missingDecision"] is None
- def test_review_without_main_decision_remains_a_review_only_batch(current_bundle):
- current_bundle["multipathDecisions"] = []
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- batches = view["rounds"][0]["convergenceBatches"]
- assert [item["association"] for item in batches] == ["review-only", "review-only"]
- assert [item["review"]["eventId"] for item in batches] == [39, 40]
- assert all(item["decision"] is None for item in batches)
- assert all(item["missingDecision"] for item in batches)
- def test_unscoped_same_round_review_is_reported_as_unlinked_not_missing(current_bundle):
- current_bundle["events"] = [
- item
- for item in current_bundle["events"]
- if item.get("event_name") != "script_multipath_evaluator"
- ]
- current_bundle["events"].append(
- {
- "id": 4995,
- "event_seq": 99,
- "event_type": "agent_invoke",
- "event_name": "script_multipath_evaluator",
- "round_index": 1,
- "status": "ok",
- "inputData": {"task": "请评审本轮候选方案"},
- "agentOutputData": {"summary": "评审内容完整,但任务未保存方案范围。"},
- }
- )
- first = ExecutionViewBuilder().from_bundle(current_bundle)["rounds"][0]
- missing = first["convergenceBatches"][0]["missingReview"]
- assert missing["card"]["primary"]["value"] == (
- "发现同轮评审,但无法安全关联到具体候选"
- )
- assert missing["sourceNotice"] == "association-incomplete"
- assert missing["unscopedReviewRefs"] == ["event:4995"]
- def test_main_canvas_translates_agent_terms_without_changing_recorded_decision(current_bundle):
- current_bundle["multipathDecisions"][0]["decision"] = "选择 Branch 1 合入 Base。"
- first = ExecutionViewBuilder().from_bundle(current_bundle)["rounds"][0]
- batch = first["convergenceBatches"][0]["decision"]
- assert batch["decision"] == "选择 Branch 1 合入 Base。"
- assert batch["node"]["card"]["primary"]["value"] == "选择方案 1 合入主脚本。"
- assert "方案 1 合入主脚本" in first["summary"]["decision"]["summary"]
- def test_round_summary_is_structured_and_keeps_multiple_decision_batches(current_bundle):
- current_bundle["rounds"][0]["goal"] = (
- "将三段骨架扩展到可模拟完整观感的程度\n"
- "① 为各段补充具体形式和脚本元素\n"
- "② 将详细执行描述改成可执行画面\n"
- "③ 拆分核心呈现段的二级子段\n"
- "④ 保持三个方案互不依赖"
- )
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- summary = view["rounds"][0]["summary"]
- assert summary["goal"]["headline"].startswith("将三段骨架扩展到可模拟完整观感的程度")
- assert summary["goal"]["itemCount"] == 0
- assert summary["goal"]["previewItems"] == []
- assert summary["goal"]["truncated"] is False
- assert summary["approach"]["mode"] == "分工"
- assert summary["approach"]["candidateCount"] == 3
- assert summary["decision"]["batchCount"] == 2
- assert summary["decision"]["branchIds"] == [1, 2, 3]
- assert summary["output"]["mergedContentCount"] == 1
- assert summary["output"]["domainFactCount"] == 1
- assert view["rounds"][0]["result"]["artifactRef"] == "artifact:round:1"
- def test_round_detail_keeps_complete_goal_out_of_compact_summary(current_bundle):
- full_goal = "核心目标\n① 第一项很长的具体任务\n② 第二项很长的具体任务\n③ 第三项很长的具体任务\n④ 第四项只在详情中完整保留"
- current_bundle["rounds"][0]["goal"] = full_goal
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- first = view["rounds"][0]
- detail = project_round_detail(88001, first, current_bundle)
- assert first["summary"]["goal"]["previewItems"] == []
- assert detail["businessSections"][0]["content"] == full_goal
- def test_branch_task_detail_uses_complete_dispatch_task_instead_of_card_preview(current_bundle):
- full_task = "目标:" + "完整创作任务。" * 80
- current_bundle["events"].append({
- "id": 99,
- "event_seq": 99,
- "event_type": "agent_invoke",
- "event_name": "script_implementer",
- "agent_role": "script_implementer",
- "round_index": 1,
- "branch_id": 1,
- "status": "ok",
- "inputData": {"task": full_task},
- })
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- task_card = view["rounds"][0]["branches"][0]["task"]["card"]["primary"]
- detail = project_activity_detail(
- 88001,
- "round:1:branch:1:task",
- view,
- current_bundle,
- )
- assert detail["businessSections"][0]["title"] == "目标"
- assert detail["businessSections"][0]["content"] == "完整创作任务。" * 80
- assert len(detail["businessSections"][0]["content"]) > 140
- assert task_card["label"] == "目标"
- assert task_card["value"] == "完整创作任务。" * 80
- assert "…" not in task_card["value"]
- def test_final_result_has_independent_detail_and_current_artifact_entry(current_bundle):
- current_bundle["record"]["status"] = "success"
- current_bundle["record"]["summary"] = "已形成可交付的主脚本。"
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- final = view["finalResult"]
- detail = project_activity_detail(
- 88001, "run:final-result", view, current_bundle
- )
- assert final["detailRef"] == "run:final-result"
- assert final["artifactRef"] == "artifact:base:current"
- assert final["card"]["primary"]["value"] == "已形成可交付的主脚本。"
- assert [item["title"] for item in detail["businessSections"]] == [
- "构建结论",
- "构建摘要或失败原因",
- "轮次与方案统计",
- "最终主脚本规模",
- "版本说明",
- ]
- assert detail["changes"]["artifactRef"] == "artifact:base:current"
- assert detail["changes"]["exactness"] == "current-only"
- def test_round_result_uses_dedicated_aggregate_detail(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- detail = project_activity_detail(
- 88001, "round:1:result", view, current_bundle
- )
- assert [item["title"] for item in detail["businessSections"]] == [
- "本轮目标完成情况",
- "方案处置",
- "领域事实",
- "整体评审",
- "主脚本变化",
- "下一步",
- ]
- assert "方案 1:已采用" in detail["businessSections"][1]["content"]
- assert "柔性排产可降低" in detail["businessSections"][2]["content"]
- assert detail["changes"]["artifactRef"] == "artifact:base:current"
- def test_retrieval_agent_detail_prefers_lazy_loaded_complete_event(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- full_task = "查找并对比可说明因果结构的解构 Case,保留完整限制条件。"
- full_screening = "### 初筛整理\n- 两个案例可用。\n- 一个案例因结构不符而舍弃。"
- event_detail = {
- "id": 52,
- "event_name": "retrieve_data_decode_case",
- "input": {"content": {"task": full_task}},
- "output": {"content": {"summary": full_screening}},
- }
- detail = project_activity_detail(
- 88001,
- "retrieval-agent:52",
- view,
- current_bundle,
- event_detail=event_detail,
- )
- sections = {item["title"]: item["content"] for item in detail["businessSections"]}
- assert sections["完整取数任务"] == full_task
- assert "一个案例因结构不符而舍弃" in sections["初筛整理"]
- assert "成功为空 1 次" in sections["原始命中"]
- def test_first_three_main_agent_cards_and_details_use_real_decision_context(current_bundle):
- direction = """## 选题价值主张
- 用真实工厂解释柔性排产的价值。
- ## 账号价值主张
- 保持专业、可核验的表达。
- ## 创作总目标
- ### 目标 1
- - 目标语句:让读者看懂柔性排产如何减少浪费。
- ### 目标 2
- - 目标语句:用可核验的行业事实支撑结论。
- ## 领域评估维度
- - 维度名:数据可信度
- """
- current_bundle["record"]["script_direction"] = direction
- first_round = current_bundle["rounds"][0]
- first_round["goal"] = "完成主结构。\n① 补齐机制拆解\n② 核实行业数据"
- paths = first_round["multipath_plan"]
- current_bundle["events"] = [
- {
- "id": 100,
- "event_seq": 0,
- "event_type": "agent_scope",
- "event_name": "main",
- "agent_role": "main",
- "agent_depth": 0,
- "scope_event_id": 100,
- "status": "ok",
- },
- *[
- {
- "id": event_id,
- "event_seq": 1,
- "event_type": "tool_call",
- "event_name": name,
- "agent_role": "main",
- "agent_depth": 0,
- "scope_event_id": 100,
- "parent_event_id": 100,
- "status": "ok",
- "ended_at": "2026-07-13T09:59:01",
- }
- for event_id, name in (
- (101, "get_topic_detail"),
- (102, "get_account_script_section_patterns"),
- (103, "get_account_script_persona_points"),
- )
- ],
- {
- "id": 108,
- "event_seq": 2,
- "event_type": "tool_call",
- "event_name": "think_and_plan",
- "agent_role": "main",
- "agent_depth": 0,
- "scope_event_id": 100,
- "parent_event_id": 100,
- "status": "ok",
- "inputData": {
- "thought_number": 1,
- "thought_summary": "先锁定创作目标,再竞争首轮结构。",
- "thought": "完整分析账号结构、选题价值与领域下限。",
- "plan": "保存创作目标,然后开启首轮。",
- "action": "写入创作总目标",
- },
- "ended_at": "2026-07-13T09:59:02",
- },
- {
- "id": 104,
- "event_seq": 3,
- "event_type": "tool_call",
- "event_name": "save_script_direction",
- "agent_role": "main",
- "agent_depth": 0,
- "scope_event_id": 100,
- "parent_event_id": 100,
- "status": "ok",
- "inputData": {"script_direction": direction},
- "outputData": {"success": True},
- "ended_at": "2026-07-13T09:59:02",
- },
- {
- "id": 105,
- "event_seq": 4,
- "event_type": "tool_call",
- "event_name": "begin_round",
- "agent_role": "main",
- "agent_depth": 0,
- "scope_event_id": 100,
- "parent_event_id": 100,
- "round_index": 0,
- "status": "ok",
- "outputData": {"success": True, "round_index": 1},
- "ended_at": "2026-07-13T09:59:03",
- },
- {
- "id": 106,
- "event_seq": 5,
- "event_type": "tool_call",
- "event_name": "get_script_snapshot",
- "agent_role": "main",
- "agent_depth": 0,
- "scope_event_id": 100,
- "parent_event_id": 100,
- "round_index": 1,
- "status": "ok",
- "outputData": {"count": 0},
- "ended_at": "2026-07-13T09:59:04",
- },
- {
- "id": 107,
- "event_seq": 6,
- "event_type": "tool_call",
- "event_name": "record_multipath_plan",
- "agent_role": "main",
- "agent_depth": 0,
- "scope_event_id": 100,
- "parent_event_id": 100,
- "round_index": 1,
- "status": "ok",
- "inputData": {
- "paths": paths,
- "mode": "分工",
- "note": first_round["plan_note"],
- },
- "outputData": {"success": True, "round_index": 1},
- "ended_at": "2026-07-13T09:59:05",
- },
- *current_bundle["events"],
- ]
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- planning = view["planning"]
- objective = view["objective"]
- first = view["rounds"][0]
- assert planning["title"] == "创作规划解析"
- assert planning["detailRef"] == "event:108"
- assert planning["card"]["primary"]["value"] == "先锁定创作目标,再竞争首轮结构。"
- planning_event = next(item for item in current_bundle["events"] if item["id"] == 108)
- planning_detail = project_event_detail(88001, planning_event)
- assert [section["title"] for section in planning_detail["businessSections"]] == [
- "规划概要",
- "完整思考",
- "执行计划",
- "下一步",
- ]
- assert planning_detail["businessSections"][1]["content"] == "完整分析账号结构、选题价值与领域下限。"
- assert objective["card"]["primary"]["value"] == "让读者看懂柔性排产如何减少浪费。"
- assert [line["label"] for line in objective["card"]["secondary"]] == [
- "决策前读取",
- "目标结构",
- ]
- assert objective["card"]["secondary"][0]["value"] == "选题 · 账号段落模式 · 账号人设"
- assert "2 个创作目标" in objective["card"]["secondary"][1]["value"]
- assert "完成主结构" in first["goal"]["card"]["primary"]["value"]
- assert "补齐机制拆解" in first["goal"]["card"]["primary"]["value"]
- assert first["goal"]["card"]["secondary"][0]["label"] == "形成前可见"
- assert first["plan"]["node"]["card"]["secondary"][0]["label"] == "各路任务"
- assert first["plan"]["runtimeRevisionRefs"] == ["event:107"]
- assert "decisionContext" not in str(view)
- objective_detail = project_activity_detail(88001, "run:objective", view, current_bundle)
- goal_detail = project_activity_detail(88001, "round:1:goal", view, current_bundle)
- plan_detail = project_activity_detail(88001, "round:1:plan", view, current_bundle)
- objective_titles = [block["title"] for block in objective_detail["blocks"]]
- assert objective_titles == ["当前保存的创作方向", "根据以下信息做出的决策"]
- assert objective_detail["blocks"][0]["value"] == direction.strip()
- assert objective_detail["blocks"][0]["presentation"] == "document"
- assert "关键约束" not in objective_titles
- assert "构建总结" not in str(objective_detail["blocks"])
- goal_titles = [block["title"] for block in goal_detail["blocks"]]
- assert goal_titles[0] == "本轮目标构成"
- assert goal_detail["blocks"][0]["presentation"] == "document"
- assert goal_detail["blocks"][0]["value"] == current_bundle["rounds"][0]["goal"]
- assert "核心目标" not in str(goal_detail["blocks"][0])
- plan_block = plan_detail["blocks"][0]
- assert plan_block["type"] == "implementation-plan"
- assert plan_block["title"] == "本轮实施路线"
- assert plan_block["routes"][0]["target"] == "文章结构"
- assert any(block["title"] == "根据以下信息做出的决策" for block in plan_detail["blocks"])
- assert "人设详细原文" not in str(view)
- def test_running_round_waits_for_main_agent_without_claiming_record_loss(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- second = view["rounds"][1]
- batch = second["convergenceBatches"][0]
- assert batch["missingDecision"]["kind"] == "execution"
- assert batch["missingDecision"]["card"]["primary"]["value"] == "等待主 Agent 比较候选方案"
- assert "sourceNotice" not in batch["missingDecision"]
- assert second["branches"][0]["outcome"]["label"] == "等待主 Agent 处置"
- assert not any(item["id"] == "missing-multipath-decisions" for item in view["warnings"])
- def test_stopped_round_is_process_incomplete_not_record_missing(current_bundle):
- stopped = deepcopy(current_bundle)
- stopped["record"]["status"] = "stopped"
- stopped["multipathDecisions"] = []
- stopped["events"] = []
- view = ExecutionViewBuilder().from_bundle(stopped)
- first = view["rounds"][0]
- batch = first["convergenceBatches"][0]
- assert batch["missingDecision"]["card"]["primary"]["value"] == "构建停止前尚未形成多路决策"
- assert batch["missingDecision"]["sourceNotice"] == "process-incomplete"
- assert first["overallEvaluation"]["card"]["primary"]["value"] == "构建停止前尚未进行整体评审"
- assert first["overallEvaluation"]["sourceNotice"] == "process-incomplete"
- open_branch = view["rounds"][1]["branches"][0]
- assert open_branch["outcome"]["label"] == "构建停止时尚未处置"
- assert open_branch["outcome"]["sourceNotice"] == "process-incomplete"
- assert not any(item["id"] == "missing-multipath-decisions" for item in view["warnings"])
- def test_completed_round_without_main_decision_is_a_real_record_gap(current_bundle):
- completed = deepcopy(current_bundle)
- completed["record"]["status"] = "success"
- completed["multipathDecisions"] = []
- completed["events"] = []
- view = ExecutionViewBuilder().from_bundle(completed)
- first = view["rounds"][0]
- batch = first["convergenceBatches"][0]
- assert batch["missingDecision"]["card"]["primary"]["value"] == "构建已经结束,但未找到主 Agent 多路决策记录"
- assert batch["missingDecision"]["sourceNotice"] == "record-missing"
- assert first["overallEvaluation"]["card"]["primary"]["value"] == "构建已经结束,但未找到明确的整体评审结论"
- assert first["overallEvaluation"]["sourceNotice"] == "record-missing"
- assert first["overallEvaluation"]["status"] == "not-evaluated"
- assert first["state"] == "not-evaluated"
- assert first["result"]["status"] == "not-evaluated"
- assert any(item["id"] == "missing-multipath-decisions" for item in view["warnings"])
- def test_domain_information_is_an_independent_output(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- branch = next(item for item in view["rounds"][0]["branches"] if item["branchId"] == 2)
- assert branch["pathType"] == "领域信息"
- assert branch["output"]["type"] == "domain-info"
- assert branch["output"]["factCount"] == 1
- assert branch["output"]["summary"] == "新增 1 条已核实领域事实"
- assert branch["status"] == "discarded"
- def test_runtime_event_only_supplies_explicit_evaluation_and_keeps_orphan_agent_unassigned(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- first = view["rounds"][0]
- assert first["overallEvaluation"]["sourceNotice"] == "runtime-associated"
- assert first["state"] == "partially-passed"
- assert view["rounds"][1]["state"] == "running"
- assert [item["eventId"] for item in view["unassigned"]["runtimeEvents"]] == [42]
- def test_branch_zero_rows_cannot_enter_new_projection(current_bundle):
- current_bundle["dataDecisions"].append({
- "id": 999,
- "round_index": 1,
- "branch_id": 0,
- "decision": "危险旧决策内容",
- "sources": [],
- })
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- serialized = str(view)
- assert "危险旧决策内容" not in serialized
- assert [item["decision"]["recordId"] for item in view["rounds"][0]["convergenceBatches"]] == [21, 22]
- def test_old_record_is_visible_but_not_reinterpreted(current_bundle):
- legacy = deepcopy(current_bundle)
- legacy["events"] = []
- legacy["multipathDecisions"] = []
- legacy["domainInfo"] = []
- for branch in legacy["branches"]:
- branch["path_type"] = None
- view = ExecutionViewBuilder().from_bundle(legacy)
- assert view["dataShape"] == "legacy-incomplete"
- assert all(not any(batch.get("decision") for batch in round_["convergenceBatches"]) for round_ in view["rounds"])
- assert any(item["id"] == "legacy-shape" for item in view["warnings"])
- def test_main_payload_does_not_embed_event_bodies_or_candidate_snapshot(current_bundle):
- current_bundle["events"][0]["output_body"] = "secret complete report"
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- serialized = str(view)
- assert "secret complete report" not in serialized
- assert "candidate_snapshot" not in serialized
- assert "script_build_event_body" not in serialized
- def test_inspector_exposes_business_decisions_and_only_real_changes(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- first = view["rounds"][0]
- round_detail = project_round_detail(88001, first, current_bundle)
- data_detail = project_activity_detail(88001, "data-decision:11", view, current_bundle)
- main_detail = project_activity_detail(88001, "multipath-decision:21", view, current_bundle)
- goal_detail = project_activity_detail(88001, "round:1:goal", view, current_bundle)
- assert round_detail["changes"]["exactness"] == "current-only"
- assert data_detail["detailKind"] == "agent-decision"
- assert [block["title"] for block in data_detail["blocks"]] == [
- "参与判断",
- "明确取舍",
- "理由",
- ]
- assert main_detail["detailKind"] == "agent-decision"
- assert "最终决定" in [block["title"] for block in main_detail["blocks"]]
- assert "评审建议与最终决定" in [
- block["title"] for block in main_detail["blocks"]
- ]
- assert "changes" not in goal_detail
- def test_explicit_next_round_translation_is_shown_as_recommendation_handling(current_bundle):
- changed = deepcopy(current_bundle)
- changed["multipathDecisions"][0]["reasoning"] = (
- "方案 1 直接采用。方案 2 的数据优势很有价值,"
- "但不直接合并,决定在下一轮转译为增强目标。"
- )
- view = ExecutionViewBuilder().from_bundle(changed)
- rows = view["rounds"][0]["convergenceBatches"][0]["decision"][
- "decisionProjection"
- ]["body"]["reviewComparison"]
- assert "handling" in rows[1]
- assert "下一轮转译" in rows[1]["handling"]
- def test_every_projected_business_section_is_readable_text(current_bundle):
- view = ExecutionViewBuilder().from_bundle(current_bundle)
- refs = [
- "run:objective",
- "data-decision:11",
- "multipath-decision:21",
- "domain-info:31",
- "round:1:goal",
- "round:1:plan",
- "round:1:result",
- "round:1:branch:1",
- "round:1:branch:1:task",
- "round:1:branch:1:retrieval",
- "round:1:branch:1:output",
- ]
- details = [project_round_detail(88001, view["rounds"][0], current_bundle)]
- details.extend(project_activity_detail(88001, ref, view, current_bundle) for ref in refs)
- for detail in details:
- if detail.get("detailKind") == "agent-decision":
- assert detail["blocks"]
- _assert_business_blocks_hide_technical_fields(detail["blocks"])
- else:
- assert detail["businessSections"]
- assert all(isinstance(section["content"], str) for section in detail["businessSections"])
- assert all(
- isinstance(section["content"], str)
- for section in (detail.get("changes") or {}).get("sections", [])
- )
- def _assert_business_blocks_hide_technical_fields(blocks):
- serialized = str(blocks)
- for forbidden in (
- "script_build_id",
- "branch_id",
- "paragraph_id",
- "post_id",
- "event ID",
- "Tool 名",
- "path_type",
- "Base",
- "patch",
- ):
- assert forbidden not in serialized
|