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

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