PMFlow/backend/tests/test_semantic_decision_service.py

from __future__ import annotations

from app.config.ai_prompts import PMFLOW_SEMANTIC_CONTEXT_DECISION_PROMPT
from app.services.ai_pmo.semantic_context.business_frame_schema import (
    AnswerShapeContract,
    BusinessSemanticFrame,
    OperationIntentContract,
)
from app.services.ai_pmo.semantic_context.decision_schema import (
    AnswerControlDecision,
    DecisionGateResult,
    InteractionBoundaryPacket,
    ModeCard,
    SemanticDecision,
)
from app.services.ai_pmo.semantic_context.gatekeeper import SemanticDecisionGatekeeper


def test_semantic_decision_gatekeeper_accepts_allowed_plan_patch() -> None:
    mode_card = ModeCard(
        workflow_mode="PLAN",
        mode_state="ACTIVE",
        allowed_actions=["reply.compose", "draft.patch", "plan.mark_ready_for_confirmation"],
        forbidden_actions=["project.create", "task.create", "db.write"],
        exit_conditions=["用户明确要求开始执行"],
    )
    decision = SemanticDecision(
        topic_relation="continue_current_topic",
        mode_decision="ready_confirm",
        draft_patch={"goal": "先收一版需求草稿"},
        assistant_reply="我先整理成待确认草稿。",
        requires_confirmation=True,
        requested_actions=["draft.patch", "reply.compose"],
        confidence=0.92,
    )

    result = SemanticDecisionGatekeeper.evaluate(mode_card=mode_card, decision=decision)

    assert isinstance(result, DecisionGateResult)
    assert result.accepted is True
    assert result.blocked_reasons == []
    assert result.allowed_actions == ["draft.patch", "reply.compose"]
    assert result.next_plan_status == "awaiting_execute_confirm"


def test_semantic_decision_gatekeeper_blocks_forbidden_action() -> None:
    mode_card = ModeCard(
        workflow_mode="PLAN",
        mode_state="ACTIVE",
        allowed_actions=["reply.compose", "draft.patch"],
        forbidden_actions=["project.create", "task.create", "db.write"],
        exit_conditions=["用户明确要求开始执行"],
    )
    decision = SemanticDecision(
        topic_relation="continue_current_topic",
        mode_decision="request_execute",
        draft_patch={},
        assistant_reply="我准备直接创建项目。",
        requires_confirmation=True,
        requested_actions=["project.create"],
        confidence=0.73,
    )

    result = SemanticDecisionGatekeeper.evaluate(mode_card=mode_card, decision=decision)

    assert result.accepted is False
    assert "forbidden_action:project.create" in result.blocked_reasons
    assert result.next_plan_status == "active"


def test_semantic_context_decision_prompt_avoids_business_semantic_examples() -> None:
    forbidden_fragments = (
        "信号源呢？",
        "负责人呢？",
        "技术路线呢？",
        "商业模式呢？",
        "给项目2接入",
        "GitHub webhook",
        "RTMP/WebRTC",
    )

    for fragment in forbidden_fragments:
        assert fragment not in PMFLOW_SEMANTIC_CONTEXT_DECISION_PROMPT


def test_semantic_context_decision_prompt_includes_structured_requirement_plan_protocol_contract() -> None:
    required_fragments = (
        "business_frame",
        "requirement_plan_protocol",
        "choice：A / B / C / D",
        "action：continue / clarify / reroll / draft",
        "operation_intent_contract",
    )

    for fragment in required_fragments:
        assert fragment in PMFLOW_SEMANTIC_CONTEXT_DECISION_PROMPT


def test_interaction_boundary_packet_schema_preserves_negative_boundary() -> None:
    packet = InteractionBoundaryPacket.model_validate(
        {
            "requested_level": "recommendation",
            "forbidden_levels": ["draft", "execute"],
            "allow_status": True,
            "allow_recommendation": True,
            "allow_draft": False,
            "allow_execute": False,
            "correction_targets": ["project", "level"],
            "confidence": 0.91,
            "evidence": [{"source": "semantic_planner", "note": "用户明确说先别出草稿"}],
        }
    )

    assert packet.requested_level == "recommendation"
    assert packet.forbidden_levels == ["draft", "execute"]
    assert packet.allow_draft is False
    assert packet.allow_execute is False
    assert packet.correction_targets == ["project", "level"]


def test_answer_control_decision_schema_tracks_blocked_workflows() -> None:
    decision = AnswerControlDecision.model_validate(
        {
            "effective_level": "recommendation",
            "blocked_levels": ["draft", "execute"],
            "blocked_workflows": ["signal_source_flow:onboard"],
            "allowed_workflows": ["query_flow:answer"],
            "downgraded_from": "draft",
            "requires_confirmation": False,
            "requires_self_review": True,
            "requires_correction_slot": True,
            "reason": "用户只要建议，禁止直接起草或执行。",
        }
    )

    assert decision.effective_level == "recommendation"
    assert decision.blocked_workflows == ["signal_source_flow:onboard"]
    assert decision.allowed_workflows == ["query_flow:answer"]
    assert decision.downgraded_from == "draft"


def test_semantic_decision_schema_preserves_business_frame_contracts() -> None:
    decision = SemanticDecision.model_validate(
        {
            "topic_relation": "new_topic",
            "mode_decision": "answer_only",
            "workflow_plans": [
                {
                    "workflow_type": "query_flow",
                    "workflow_action": "answer",
                    "target_scope": "project",
                    "operation_codes": ["query.answer"],
                    "confidence": 0.91,
                }
            ],
            "business_frame": {
                "goal_kind": "recommendation",
                "target_scope": "current_project",
                "resource_role": "signal_source",
                "source_kind": "external",
                "expected_answer_shape": "ranked_candidates",
                "conversation_posture": "business_operating",
                "project_binding": 6,
            },
            "answer_shape_contract": {
                "shape": "ranked_candidates",
                "confidence": 0.88,
            },
            "operation_intent_contract": {
                "intent_phase": "recommendation",
                "requires_confirmation": False,
                "project_binding": 6,
            },
            "requested_actions": ["workflow.bundle.compose"],
        }
    )

    assert isinstance(decision.business_frame, BusinessSemanticFrame)
    assert decision.business_frame.goal_kind == "recommendation"
    assert isinstance(decision.answer_shape_contract, AnswerShapeContract)
    assert decision.answer_shape_contract.shape == "ranked_candidates"
    assert isinstance(decision.operation_intent_contract, OperationIntentContract)
    assert decision.operation_intent_contract.intent_phase == "recommendation"