77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from app.config.ai_prompts import PMFLOW_SEMANTIC_CONTEXT_DECISION_PROMPT
|
|
from app.services.ai_pmo.semantic_context.decision_schema import (
|
|
DecisionGateResult,
|
|
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
|