from __future__ import annotations import asyncio from typing import Any import pytest from app.services.ai_pmo.semantic_context.decision_schema import SemanticDecision from app.services.ai_pmo.workflow_semantic_planner import WorkflowSemanticPlanner def _decision(*, workflows: list[dict[str, Any]], scene_metadata: dict[str, Any] | None = None) -> SemanticDecision: return SemanticDecision( topic_relation="new_topic", mode_decision="answer_only", workflow_plans=workflows, scene_metadata=scene_metadata or {}, requested_actions=["workflow.bundle.compose"], confidence=0.91, ) def test_workflow_semantic_planner_plan_message_uses_model_decision(monkeypatch: pytest.MonkeyPatch) -> None: async def _fake_decide(cls, *, message_text: str, source_system: str, metadata: dict[str, Any] | None = None): assert message_text == "帮我创建一个项目，并把这个需求接入信号源" assert source_system == "feishu" return _decision( workflows=[ { "workflow_type": "new_project_flow", "workflow_action": "create", "target_scope": "global", "operation_codes": ["project.create"], "confidence": 0.93, }, { "workflow_type": "signal_source_flow", "workflow_action": "onboard", "target_scope": "global", "operation_codes": ["signal_source.onboard"], "confidence": 0.92, }, ] ) monkeypatch.setattr(WorkflowSemanticPlanner, "_decide_cold_start_semantics", classmethod(_fake_decide)) bundle = WorkflowSemanticPlanner.plan_message( message_text="帮我创建一个项目，并把这个需求接入信号源", source_system="feishu", ) assert bundle.primary_workflow == "new_project_flow" assert [item.workflow_type for item in bundle.workflows] == ["new_project_flow", "signal_source_flow"] assert bundle.reply_contract.mode == "aggregate_card" @pytest.mark.asyncio async def test_workflow_semantic_planner_aplan_message_preserves_scene_metadata(monkeypatch: pytest.MonkeyPatch) -> None: async def _fake_decide(cls, *, message_text: str, source_system: str, metadata: dict[str, Any] | None = None): assert message_text == "想跟你聊个新需求" return _decision( workflows=[ { "workflow_type": "query_flow", "workflow_action": "answer", "target_scope": "global", "operation_codes": ["query.answer"], "confidence": 0.88, "metadata": {"message_intent": "requirement_plan"}, } ], scene_metadata={ "message_intent": "requirement_plan", "consultant_mode_preference": "clarify", "topic_switch": "new", }, ) monkeypatch.setattr(WorkflowSemanticPlanner, "_decide_cold_start_semantics", classmethod(_fake_decide)) bundle = await WorkflowSemanticPlanner.aplan_message( message_text="想跟你聊个新需求", source_system="feishu", ) assert bundle.primary_workflow == "query_flow" assert bundle.workflows[0].metadata["message_intent"] == "requirement_plan" assert bundle.diagnostics["topic_switch"] == "new" def test_workflow_semantic_planner_sync_wrapper_fails_inside_running_loop() -> None: async def _runner() -> None: with pytest.raises(RuntimeError): WorkflowSemanticPlanner.plan_message(message_text="test", source_system="feishu") asyncio.run(_runner()) def test_workflow_semantic_planner_no_longer_exposes_legacy_semantic_helpers() -> None: legacy_names = ( "plan_from_intent", "classify_scene", "is_action_request_capture", "parse_followup_action", "extract_project_lookup_keyword", "wants_followup_continue", "wants_followup_status", "looks_like_project_create_request", "looks_like_task_create_request", "looks_like_signal_source_request", "looks_like_external_connect_request", "looks_like_contextual_write_confirmation", "is_query_like_question", "is_obvious_question", ) for name in legacy_names: assert not hasattr(WorkflowSemanticPlanner, name), name def test_workflow_semantic_planner_recommendation_boundary_downgrades_signal_source_onboard() -> None: bundle = WorkflowSemanticPlanner._bundle_from_cold_start_decision( decision=SemanticDecision.model_validate( { "topic_relation": "new_topic", "mode_decision": "answer_only", "workflow_plans": [ { "workflow_type": "signal_source_flow", "workflow_action": "onboard", "target_scope": "project", "operation_codes": ["signal_source.onboard"], "requires_confirmation": True, "confidence": 0.94, } ], "interaction_boundary": { "requested_level": "recommendation", "forbidden_levels": ["draft", "execute"], "allow_status": True, "allow_recommendation": True, "allow_draft": False, "allow_execute": False, "self_review_required": True, "correction_allowed": True, "correction_targets": ["project", "level"], }, "requested_actions": ["workflow.bundle.compose"], "confidence": 0.94, } ), source_system="feishu", message_text="先别出草稿，只按当前项目情况推荐几个最值得接的外部信号源。", ) assert bundle.primary_workflow == "query_flow" assert [item.workflow_type for item in bundle.workflows] == ["query_flow"] assert bundle.workflows[0].operation_codes == ["query.answer"] assert bundle.diagnostics["interaction_boundary"]["requested_level"] == "recommendation" assert bundle.diagnostics["answer_control"]["effective_level"] == "recommendation" assert bundle.diagnostics["answer_control"]["blocked_workflows"] == ["signal_source_flow:onboard"] assert bundle.diagnostics["answer_control"]["downgraded_from"] == "draft" def test_workflow_semantic_planner_explicit_draft_upgrade_preserves_signal_source_onboard() -> None: bundle = WorkflowSemanticPlanner._bundle_from_cold_start_decision( decision=SemanticDecision.model_validate( { "topic_relation": "continue_current_topic", "mode_decision": "ready_confirm", "workflow_plans": [ { "workflow_type": "signal_source_flow", "workflow_action": "onboard", "target_scope": "project", "operation_codes": ["signal_source.onboard"], "requires_confirmation": True, "confidence": 0.96, } ], "interaction_boundary": { "requested_level": "draft", "forbidden_levels": ["execute"], "allow_status": True, "allow_recommendation": True, "allow_draft": True, "allow_execute": False, "self_review_required": True, "correction_allowed": True, "correction_targets": ["project", "level"], }, "requested_actions": ["workflow.bundle.compose"], "confidence": 0.96, } ), source_system="feishu", message_text="可以，把第一个整理成接入草稿给我看。", ) assert bundle.primary_workflow == "signal_source_flow" assert [item.workflow_type for item in bundle.workflows] == ["signal_source_flow"] assert bundle.diagnostics["answer_control"]["effective_level"] == "draft" assert bundle.diagnostics["answer_control"]["blocked_workflows"] == [] def test_workflow_semantic_planner_non_plan_signal_source_prepare_write_stays_execute_lane() -> None: bundle = WorkflowSemanticPlanner._bundle_from_cold_start_decision( decision=SemanticDecision.model_validate( { "topic_relation": "continue_current_topic", "mode_decision": "request_execute", "workflow_plans": [ { "workflow_type": "signal_source_flow", "workflow_action": "onboard", "target_scope": "project", "operation_codes": ["signal_source.onboard"], "requires_confirmation": True, "confidence": 0.97, } ], "interaction_boundary": { "requested_level": "execute", "forbidden_levels": ["draft"], "allow_status": True, "allow_recommendation": True, "allow_draft": False, "allow_execute": True, }, "operation_intent_contract": { "intent_phase": "prepare_write", "requires_confirmation": True, "project_binding": 6, "candidate_set_binding": "recset_invoice_01", "candidate_refs": ["cand_invoice_retry"], }, "requested_actions": ["workflow.bundle.compose"], "confidence": 0.97, } ), source_system="feishu", message_text="把刚才推荐的第一个信号源写进自动开票系统。", ) assert bundle.primary_workflow == "signal_source_flow" assert [item.workflow_type for item in bundle.workflows] == ["signal_source_flow"] assert bundle.diagnostics["answer_control"]["effective_level"] == "execute" assert bundle.diagnostics["answer_control"]["blocked_workflows"] == [] def test_workflow_semantic_planner_exposes_business_frame_diagnostics_for_recommendation() -> None: bundle = WorkflowSemanticPlanner._bundle_from_cold_start_decision( 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.92, } ], "interaction_boundary": { "requested_level": "recommendation", "forbidden_levels": ["draft", "execute"], "allow_status": True, "allow_recommendation": True, "allow_draft": False, "allow_execute": False, }, "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.89, }, "operation_intent_contract": { "intent_phase": "recommendation", "requires_confirmation": False, "project_binding": 6, }, "requested_actions": ["workflow.bundle.compose"], "confidence": 0.92, } ), source_system="feishu", message_text="自动开票系统这个项目，你有推荐的外部信号源吗？", ) assert bundle.diagnostics["business_frame"]["goal_kind"] == "recommendation" assert bundle.diagnostics["business_frame"]["resource_role"] == "signal_source" assert bundle.diagnostics["answer_shape_contract"]["shape"] == "ranked_candidates" assert bundle.diagnostics["operation_intent_contract"]["intent_phase"] == "recommendation" def test_workflow_semantic_planner_preserves_reply_anchor_candidate_set_in_bundle_diagnostics() -> None: bundle = WorkflowSemanticPlanner._bundle_from_cold_start_decision( decision=SemanticDecision.model_validate( { "topic_relation": "continue_current_topic", "mode_decision": "request_execute", "workflow_plans": [ { "workflow_type": "signal_source_flow", "workflow_action": "onboard", "target_scope": "project", "operation_codes": ["signal_source.onboard"], "requires_confirmation": True, "confidence": 0.93, } ], "interaction_boundary": { "requested_level": "execute", "forbidden_levels": ["draft"], "allow_status": True, "allow_recommendation": True, "allow_draft": False, "allow_execute": True, }, "operation_intent_contract": { "intent_phase": "execute_write", "requires_confirmation": True, "project_binding": 6, "candidate_set_binding": "recset_invoice_01", "candidate_refs": ["cand_invoice_retry"], }, "requested_actions": ["workflow.bundle.compose"], "confidence": 0.93, } ), source_system="feishu", message_text="把刚才那组外部信号源加入自动开票系统。", metadata={ "project_id": 6, "reply_to_message_id": "om-reply-1", "reply_anchor_qa_context": { "project_id": 6, "signal_source_recommendation": { "candidate_set_id": "recset_invoice_01", "project_binding": 6, "candidates": [ { "candidate_ref": "cand_invoice_retry", "display_name": "开票失败或重试回执", "materialization_payload": { "display_name": "开票失败或重试回执", "business_value": "最早看到票开不出去", }, } ], }, }, }, ) assert bundle.diagnostics["reply_anchor_qa_context"]["project_id"] == 6 assert ( bundle.diagnostics["reply_anchor_qa_context"]["signal_source_recommendation"]["candidate_set_id"] == "recset_invoice_01" ) def test_workflow_semantic_planner_keeps_live_review_tool_non_plan_write_in_prepare_write_lane() -> None: bundle = WorkflowSemanticPlanner._bundle_from_cold_start_decision( decision=SemanticDecision.model_validate( { "topic_relation": "continue_current_topic", "mode_decision": "request_execute", "workflow_plans": [ { "workflow_type": "signal_source_flow", "workflow_action": "onboard", "target_scope": "project", "operation_codes": ["signal_source.onboard"], "requires_confirmation": True, "confidence": 0.95, } ], "interaction_boundary": { "requested_level": "execute", "forbidden_levels": ["draft"], "allow_status": True, "allow_recommendation": True, "allow_draft": False, "allow_execute": True, }, "business_frame": { "goal_kind": "prepare_write", "target_scope": "explicit_project", "resource_role": "signal_source", "source_kind": "external", "expected_answer_shape": "write_confirmation", "conversation_posture": "business_operating", "project_binding": 88, }, "answer_shape_contract": { "shape": "write_confirmation", "confidence": 0.91, }, "operation_intent_contract": { "intent_phase": "prepare_write", "requires_confirmation": True, "project_binding": 88, "candidate_set_binding": "recset_live_review_01", "candidate_refs": ["cand_comment_risk", "cand_platform_notice"], }, "requested_actions": ["workflow.bundle.compose"], "confidence": 0.95, } ), source_system="feishu", message_text="把刚才推荐给直播复盘辅助工具（MVP1）的两个外部信号源直接加进去，不要出草稿，先回显准备写什么。", ) assert bundle.primary_workflow == "signal_source_flow" assert bundle.diagnostics["answer_control"]["effective_level"] == "execute" assert bundle.diagnostics["business_frame"]["goal_kind"] == "prepare_write" assert bundle.diagnostics["answer_shape_contract"]["shape"] == "write_confirmation" assert bundle.diagnostics["operation_intent_contract"]["candidate_set_binding"] == "recset_live_review_01"