from __future__ import annotations from app.decision_projection import AgentDecisionProjector def test_direction_keeps_observed_reads_separate_from_explicit_basis(): decision = AgentDecisionProjector().direction( { "id": "run:objective", "subtype": "objective", "decisionItems": ["讲清柔性排产为什么能减少浪费"], "constraints": ["数据可信", "结构清楚"], "inputs": [ {"label": "选题", "relation": "direct-read"}, { "label": "账号段落模式", "relation": "direct-read", "decisionUse": "explicit-basis", }, ], "promptRef": "main:direction", } ) assert decision["semanticKind"] == "decision" assert decision["decisionType"] == "direction" assert decision["authority"] == "final" assert decision["inputs"][0] == { "label": "选题", "summary": None, "observedRelation": "read-by-actor", "decisionUse": "not-recorded", "detailRef": None, } assert decision["inputs"][1]["decisionUse"] == "explicit-basis" input_block = next( block for block in decision["detail"]["blocks"] if block["title"] == "根据以下信息做出的决策" ) assert input_block["items"][0]["note"] == "决策前直接读取；是否采用未记录" assert input_block["items"][1]["note"] == "决策前直接读取；明确作为决策依据" assert input_block["visualRole"] == "evidence" assert input_block["presentation"] == "list" assert input_block["collapsible"] is False final_block = next( block for block in decision["detail"]["blocks"] if block["title"] == "最终方向" ) assert final_block["visualRole"] == "key-result" assert len(decision["card"]["secondary"]) <= 3 assert decision["detail"]["detailKind"] == "agent-decision" def test_implementation_plan_uses_dedicated_routes_and_full_evidence(): full_goal = "修复结构完整性。" * 40 decision = AgentDecisionProjector().direction( { "id": "round:4:plan", "subtype": "implementation-plan", "decisionItems": ["方案 1 · 改：段落 2207、2208"], "explicitReasoning": "这是收敛阶段的定点修复，不需要竞争。", "inputs": [ {"label": "本轮目标", "summary": full_goal, "relation": "standing-constraint"}, ], "implementationPlan": { "mode": "单路", "routes": [{ "pathIndex": 1, "pathType": "内容", "action": "改", "target": "段落 2207、2208 的 form_elements", "method": "产生元素", "emphasis": "结构完整性", }], }, } ) route_block = decision["detail"]["blocks"][0] evidence_block = decision["detail"]["blocks"][1] assert route_block["type"] == "implementation-plan" assert route_block["routes"][0]["target"] == "段落 2207、2208 的 形式信息" assert route_block["routes"][0]["emphasis"] == "结构完整性" assert evidence_block["title"] == "根据以下信息做出的决策" assert evidence_block["visualRole"] == "evidence" assert evidence_block["collapsible"] is False assert evidence_block["items"][0]["value"] == full_goal assert evidence_block["items"][0]["note"] == "持续约束；本轮必须遵守" def test_data_tradeoff_is_implementer_scope_and_sources_are_not_auto_adopted(): decision = AgentDecisionProjector().data_tradeoff( { "id": "data-decision:12", "decision": "采用两条交叉印证的数据，暂不采用单一来源估算。", "reasoning": "两类来源结论一致。", "sources": [ {"type": "行业报告", "title": "报告 A"}, {"type": "搜索结果", "title": "结果 B"}, ], } ) assert decision["decisionType"] == "tradeoff" assert decision["subtype"] == "data-tradeoff" assert decision["actor"]["role"] == "implementer" assert decision["authority"] == "implementation-scope" assert {item["decisionUse"] for item in decision["inputs"]} == {"not-recorded"} assert decision["body"]["selected"] == [] assert len(decision["card"]["secondary"]) == 2 blocks = {block["title"]: block for block in decision["detail"]["blocks"]} assert blocks["参与判断"]["visualRole"] == "evidence" assert blocks["明确取舍"]["visualRole"] == "key-result" assert blocks["理由"]["visualRole"] == "reason" def test_data_tradeoff_uses_real_source_type_and_content_fields(): decision = AgentDecisionProjector().data_tradeoff( { "id": "data-decision:13", "decision": "组合领域信息与账号模式。", "sources": [ { "data_type": "领域信息", "data_content": "米豆腐热量约 50-60kcal/100g", } ], } ) source = decision["detail"]["blocks"][0]["items"][0] assert source["label"] == "领域信息" assert source["value"] == "米豆腐热量约 50-60kcal/100g" def test_multipath_tradeoff_is_main_final_and_does_not_render_internal_ids(): decision = AgentDecisionProjector().multipath_tradeoff( { "id": "multipath-decision:21", "branch_ids": [3, 4], "decision": "组合两个候选方案后合入主脚本。", "reasoning": "两路内容互补。", "reviewComparison": [ { "recommendation": "建议组合", "finalDecision": "组合采用", "handling": None, } ], } ) assert decision["authority"] == "final" assert decision["actor"] == {"role": "main", "label": "主 Agent"} assert decision["body"]["branchIds"] == [3, 4] business_text = str(decision["card"]) + str(decision["detail"]["blocks"]) assert "branch_id" not in business_text assert "比较 2 个候选方案" in business_text def test_evaluator_is_recommendation_with_structured_business_blocks(): decision = AgentDecisionProjector().evaluation( { "id": "multipath-review:40", "subtype": "multipath-evaluation", "subjects": ["候选方案 A", "候选方案 B"], "criteria": ["结构完整", "证据可信"], "itemConclusions": [ {"subject": "候选方案 A", "conclusion": "通过"}, {"subject": "候选方案 B", "conclusion": "部分通过"}, ], "achievements": ["核心结构已经形成"], "problems": [{"severity": "一般", "summary": "结尾仍需收束"}], "conclusion": "部分通过", "recommendation": "优先采用候选方案 A。", } ) assert decision["decisionType"] == "evaluation" assert decision["authority"] == "recommendation" assert decision["actor"]["role"] == "multipath-evaluator" assert [block["title"] for block in decision["detail"]["blocks"]] == [ "评审对象", "评审标准", "逐项结论", "已达成", "问题", "总结论", "评审建议", ] conclusion = next(block for block in decision["detail"]["blocks"] if block["title"] == "总结论") assert conclusion["visualRole"] == "key-result" def test_creative_requires_safe_think_and_plan_and_never_infers_from_artifact(): projector = AgentDecisionProjector() event = { "id": 70, "event_type": "tool_call", "event_name": "think_and_plan", "agent_role": "script_implementer", "status": "ok", "inputData": { "thought": "现有证据不足，抗性淀粉只作为待验证推导。", "action": "补充烹饪贴士段", "plan": "先写可确认内容，再标注待核验项。", }, } assert projector.creative({"safeLink": False, "event": event}) is None assert projector.creative({"safeLink": True, "artifactRef": "artifact:9"}) is None decision = projector.creative( { "safeLink": True, "event": event, "task": "补齐烹饪贴士", "output": "形成候选脚本段落", "uncertainties": ["抗性淀粉结论需要进一步核验"], "artifactRef": "artifact:9", "promptRef": "implementer:70", } ) assert decision is not None assert decision["decisionType"] == "creative" assert decision["authority"] == "implementation-scope" assert decision["artifactRef"] == "artifact:9" assert "待验证推导" in decision["body"]["reasoning"] assert decision["body"]["uncertainties"] == ["抗性淀粉结论需要进一步核验"] assert all(item["key"] != "task" for item in decision["card"]["secondary"]) assert "创作任务" not in [block["title"] for block in decision["detail"]["blocks"]] def test_missing_decision_is_explicit_without_fabricated_reasoning(): decision = AgentDecisionProjector().evaluation( {"id": "overall-review:missing", "subtype": "overall-evaluation"} ) assert decision["completeness"] == "missing" assert decision["card"]["primary"]["value"] == "决策结论未记录" assert decision["body"]["recommendation"] is None assert decision["notices"] == [ {"code": "record-missing", "message": "没有找到结构化决策结论。"} ] def test_business_copy_translates_internal_ids_without_deleting_business_numbers(): projection = AgentDecisionProjector().evaluation( { "id": "event:9", "subtype": "multipath-evaluation", "achievements": [ "branch3 完成 P2，参考 postid: 67c6c723，保留领域信息 31/32。" ], "recommendation": "建议合入 base。", } ) business = str(projection["detail"]["blocks"]) assert "方案 3" in business assert "段落 2" in business assert "账号内容样本" in business assert "主脚本" in business assert "领域信息 31/32" in business assert "branch3" not in business assert "postid" not in business