from __future__ import annotations 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"