import pytest from app.services.integrations.feishu_integration_service.mixin_card_action import FeishuCardActionMixin from app.services.integrations.feishu_integration_service.mixin_receive_routing import FeishuReceiveRoutingMixin from app.services.integrations.feishu_integration_service.mixin_reply_templates import FeishuReplyTemplatesMixin from app.services.integrations.feishu_integration_service.service import FeishuIntegrationService def test_extract_card_action_payload_parses_dict_value() -> None: payload = FeishuCardActionMixin._extract_card_action_payload( { "open_message_id": "om_123", "open_chat_id": "oc_123", "operator": {"open_id": "ou_abc", "name": "测试用户"}, "action": { "value": { "action": "confirm_draft", "work_item_id": 88, "draft_version": 3, "revision_note": "保持现有范围", } }, } ) assert payload["action"] == "confirm" assert payload["work_item_id"] == 88 assert payload["draft_version"] == 3 assert payload["source_message_id"] == "om_123" assert payload["chat_id"] == "oc_123" assert payload["sender_open_id"] == "ou_abc" def test_extract_card_action_payload_parses_json_string_value_and_aliases() -> None: payload = FeishuCardActionMixin._extract_card_action_payload( { "message_id": "msg_1", "chat_id": "chat_9", "operator": {"user_id": "ou_user", "display_name": "张三"}, "action": {"value": '{"draft_action":"修改","workItemId":"100","version":"4","note":"改成下周上线"}'}, } ) assert payload["action"] == "revise" assert payload["work_item_id"] == 100 assert payload["draft_version"] == 4 assert payload["revision_note"] == "改成下周上线" assert payload["source_message_id"] == "msg_1" assert payload["chat_id"] == "chat_9" def test_extract_card_action_payload_parses_plan_synthetic_text() -> None: payload = FeishuCardActionMixin._extract_card_action_payload( { "message_id": "msg_plan_1", "chat_id": "chat_plan_1", "operator": {"open_id": "ou_plan", "name": "计划用户"}, "action": {"value": {"action": "plan_choice", "choice": "C", "synthetic_text": "C"}}, } ) assert payload["action"] == "plan_choice" assert payload["choice"] == "C" assert payload["synthetic_text"] == "C" assert payload["source_message_id"] == "msg_plan_1" assert payload["chat_id"] == "chat_plan_1" @pytest.mark.asyncio async def test_handle_message_event_dispatches_card_action(monkeypatch) -> None: service = FeishuIntegrationService(db=object()) called: dict[str, object] = {} async def _fake_card_handler(event): called["event_type"] = event.get("type") return {"code": 0, "msg": "success", "data": {"handled": True}} monkeypatch.setattr(service, "_handle_card_action_trigger", _fake_card_handler) result = await service.handle_message_event({"type": "card.action.trigger", "event": {}}) assert result["code"] == 0 assert called["event_type"] == "card.action.trigger" def test_build_action_card_for_draft_pending_confirm() -> None: card = FeishuCardActionMixin._build_action_card_from_capture_result( capture_result={ "reply_template_key": "draft_pending_confirm", "work_item_id": 321, "draft_version": 5, "direct_reply_text": "📌 AI直播复盘助手\n面向跨境 DTC 团队的 AI 直播复盘工具\n\n✅ 当前判断\n可行性：待讨论\n\n⚠️ 仍需确认\n· 验收口径\n· 触发时机", }, mention_open_id="ou_xxx", mention_name="测试同学", ) assert isinstance(card, dict) assert card["header"]["title"]["content"] == "AI PMO 草稿待确认" div_blocks = [item for item in card["elements"] if item.get("tag") == "div"] assert len(div_blocks) >= 3 action_row = next((item for item in card["elements"] if item.get("tag") == "action"), {}) actions = action_row.get("actions") if isinstance(action_row.get("actions"), list) else [] action_names = {((btn.get("value") or {}).get("action")) for btn in actions if isinstance(btn, dict)} assert {"confirm", "revise", "cancel"} <= action_names def test_build_action_card_for_pending_manual_continue() -> None: card = FeishuCardActionMixin._build_action_card_from_capture_result( capture_result={ "reply_template_key": "pending_manual", "results": [{"work_item_id": 99}], "direct_reply_text": "需要继续执行。", }, mention_open_id=None, mention_name=None, ) assert isinstance(card, dict) assert card["header"]["title"]["content"] == "AI PMO 待继续执行" action_row = next((item for item in card["elements"] if item.get("tag") == "action"), {}) actions = action_row.get("actions") if isinstance(action_row.get("actions"), list) else [] assert len(actions) == 1 assert (actions[0].get("value") or {}).get("action") == "continue" assert (actions[0].get("value") or {}).get("work_item_id") == 99 def test_build_action_card_for_requirement_plan_followup() -> None: card = FeishuCardActionMixin._build_action_card_from_capture_result( capture_result={ "reply_template_key": "requirement_plan_followup", "work_item_id": 77, "direct_reply_text": "A. 规则引擎驱动\nB. 轻量模型辅助\nC. 混合增强路径\n\n⭐ 当前我更推荐：C", }, mention_open_id="ou_plan_xxx", mention_name="测试计划用户", ) assert isinstance(card, dict) assert card["header"]["title"]["content"] == "AI PMO 需求收敛" action_rows = [item for item in card["elements"] if item.get("tag") == "action"] assert len(action_rows) == 2 row_1_actions = action_rows[0].get("actions") if isinstance(action_rows[0].get("actions"), list) else [] row_2_actions = action_rows[1].get("actions") if isinstance(action_rows[1].get("actions"), list) else [] assert [((btn.get("value") or {}).get("synthetic_text")) for btn in row_1_actions] == ["A", "B", "C", "D"] assert [((btn.get("value") or {}).get("synthetic_text")) for btn in row_2_actions] == ["继续出草稿", "重出一版"] @pytest.mark.parametrize( ("external_recommendation", "expected_fragment"), [ ( { "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, "目前更像背景相关", ), ( { "retrieval_status": "verifier_blocked", "impact_status": "uncertain", }, "校验未通过", ), ( { "retrieval_status": "no_relevant_results", "impact_status": "no_direct_match", }, "暂未找到与当前项目直接相关的公开证据", ), ( { "retrieval_status": "retrieval_failed", "impact_status": "uncertain", }, "这次外部检索失败", ), ], ) def test_build_feishu_external_summary_line_preserves_distinct_states( external_recommendation: dict[str, str], expected_fragment: str, ) -> None: line = FeishuReceiveRoutingMixin._build_feishu_external_summary_line(external_recommendation) assert line.startswith("外部判断：") assert expected_fragment in line def test_should_accept_feishu_qa_natural_reply_rejects_fixed_opener_without_project_entities() -> None: accepted = FeishuReceiveRoutingMixin._should_accept_feishu_qa_natural_reply( "这事可以继续看，但现在还不能直接说它适合这个项目。", selected_entity_candidates=[ "采购经理每日实时库存上报服务", "WMS库存快照自动拉取与清洗", "每日固定时段生成并推送结构化摘要卡片", ], blocked_entity_candidates=[], ) assert accepted is False def test_should_accept_feishu_qa_natural_reply_accepts_project_entity_analysis() -> None: accepted = FeishuReceiveRoutingMixin._should_accept_feishu_qa_natural_reply( ( "Responses API 不适合放在 WMS库存快照自动拉取或者 TOP10缺货排序这两层，" "更像放在每日固定时段生成并推送结构化摘要卡片之后的解释和追问层。" ), selected_entity_candidates=[ "采购经理每日实时库存上报服务", "WMS库存快照自动拉取与清洗", "TOP10商品缺货排序逻辑实现（基于当前库存与安全库存差值）", "每日固定时段生成并推送结构化摘要卡片", ], blocked_entity_candidates=[], ) assert accepted is True def test_should_accept_feishu_qa_natural_reply_rejects_unplanned_project_entities() -> None: accepted = FeishuReceiveRoutingMixin._should_accept_feishu_qa_natural_reply( ( "Responses API 更像放在每日固定时段生成并推送结构化摘要卡片之后的解释层，" "不适合直接放进 WMS库存快照自动拉取与清洗 主链。" "但它和外部信号源 - WMS实时库存接口 也许也能直接打通。" ), selected_entity_candidates=[ "WMS库存快照自动拉取与清洗", "每日固定时段生成并推送结构化摘要卡片", ], blocked_entity_candidates=[ "外部信号源 - WMS实时库存接口", "TOP10商品缺货排序逻辑实现（基于当前库存与安全库存差值）", ], ) assert accepted is False def test_build_feishu_qa_analysis_packet_reply_text_deduplicates_gap_and_next_step() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "external_impact_analysis", "conclusion": "这条外部变化现在还不能直接挂到 Alpha 的项目结论上。", "retrieval_status": "verifier_blocked", "impact_status": "no_direct_impact", "reason_chain": [ "外部信号源 - WMS实时库存接口这条线索还没通过校验，当前不能直接挂到项目结论上。", "每日固定时段生成并推送结构化摘要卡片这一步如果要消费这条变化，还缺稳定的字段映射和证据。", ], "blocked_reason_or_gap": "先修正证据质量，再决定是否继续外部推荐。", "next_step": "先修正证据质量，再决定是否继续外部推荐。", } ) assert rendered.count("先修正证据质量，再决定是否继续外部推荐。") == 1 def test_build_feishu_qa_analysis_packet_reply_text_for_verifier_blocked_uses_human_next_step() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "external_impact_analysis", "conclusion": "这条外部变化现在还不能直接挂到 Alpha 的项目结论上。", "retrieval_status": "verifier_blocked", "impact_status": "no_direct_impact", "reason_chain": [ "现在卡在外部信号源 - WMS实时库存接口这条线索上：它指向提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源，但还没通过校验。", "如果后面要让每日固定时段生成并推送结构化摘要卡片消费这条变化，还得先把稳定的字段映射和证据补齐。", ], "blocked_reason_or_gap": "先把外部信号源 - WMS实时库存接口这条线索补齐到可校验，再看要不要挂到每日固定时段生成并推送结构化摘要卡片这一步。", "next_step": "先把外部信号源 - WMS实时库存接口这条线索补齐到可校验，再看要不要挂到每日固定时段生成并推送结构化摘要卡片这一步。", } ) expected = ( "这条外部变化现在还不能直接挂到 Alpha 的项目结论上。\n\n" "现在卡在外部信号源 - WMS实时库存接口这条线索上：它指向提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源，但还没通过校验。" "如果后面要让每日固定时段生成并推送结构化摘要卡片消费这条变化，还得先把稳定的字段映射和证据补齐。\n\n" "先把外部信号源 - WMS实时库存接口这条线索补齐到可校验，再看要不要挂到每日固定时段生成并推送结构化摘要卡片这一步。" ) assert rendered == expected assert "外部推荐" not in rendered assert "证据质量" not in rendered def test_build_feishu_qa_analysis_packet_reply_text_for_project_fit_compacts_reason_chain() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "project_fit_analysis", "conclusion": "放到采购经理每日实时库存上报服务里，它更像接在每日固定时段生成并推送结构化摘要卡片后面的解释层，不像WMS库存快照自动拉取与清洗这种主链能力。", "fit_verdict": "fit_later", "focus_objects": [ { "name": "每日固定时段生成并推送结构化摘要卡片", "kind": "task", "stance": "better_fit", "reason": "这一步更像结果出来之后的解释和追问层。", }, { "name": "WMS库存快照自动拉取与清洗", "kind": "task", "stance": "not_fit", "reason": "这一步属于确定性主链，不适合直接塞进生成能力。", }, ], "reason_chain": [ "WMS库存快照自动拉取与清洗这条主链负责确定性取数和计算，结果要稳定、可复算。", "相比之下，每日固定时段生成并推送结构化摘要卡片更适合承接结果出来之后的解释和追问。", ], "missing_mapping": "还缺它在每日固定时段生成并推送结构化摘要卡片这一步到底承担解释、问答还是编排角色的明确映射。", "next_step": "先确认它在当前项目里具体承接哪个节点。", } ) expected = ( "放到采购经理每日实时库存上报服务里，它更像接在每日固定时段生成并推送结构化摘要卡片后面的解释层，不像WMS库存快照自动拉取与清洗这种主链能力。\n\n" "WMS库存快照自动拉取与清洗这条主链负责确定性取数和计算，结果要稳定、可复算。相比之下，每日固定时段生成并推送结构化摘要卡片更适合承接结果出来之后的解释和追问。\n\n" "还缺它在每日固定时段生成并推送结构化摘要卡片这一步到底承担解释、问答还是编排角色的明确映射。先确认它在当前项目里具体承接哪个节点。" ) assert rendered == expected def test_build_feishu_qa_analysis_packet_reply_text_for_direct_match_leads_with_business_task() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "external_impact_analysis", "conclusion": "这条外部变化已经直接打到采购经理每日实时库存上报服务里的WMS库存快照自动拉取与清洗这一步。", "retrieval_status": "evidence_found", "impact_status": "direct_match", "reason_chain": [ "WMS库存快照自动拉取与清洗负责从WMS系统每小时拉取库存快照，按可用/在途/冻结三态聚合，清洗异常值后写入中台临时表，会最先接住这条变化，可以当成后续收口和计划讨论的正式输入。", "对应的正式锚点是外部信号源 - WMS实时库存接口，它指向提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源，已经被外部证据直接打到，不再只是背景相关。", ], "focus_objects": [ { "name": "WMS库存快照自动拉取与清洗", "kind": "task", "stance": "dependent", "reason": "这是当前最直接受影响的项目对象。", }, { "name": "外部信号源 - WMS实时库存接口", "kind": "signal", "stance": "anchor", "reason": "它已经和当前项目的正式对象建立了直接映射。", }, ], "blocked_reason_or_gap": "直接看它会影响哪条下游任务，再决定要不要进 plan。", "next_step": "直接看它会影响哪条下游任务，再决定要不要进 plan。", } ) expected = ( "这条外部变化已经直接打到采购经理每日实时库存上报服务里的WMS库存快照自动拉取与清洗这一步。\n\n" "WMS库存快照自动拉取与清洗负责从WMS系统每小时拉取库存快照，按可用/在途/冻结三态聚合，清洗异常值后写入中台临时表，会最先接住这条变化，可以当成后续收口和计划讨论的正式输入。对应的正式锚点是外部信号源 - WMS实时库存接口，它指向提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源，已经被外部证据直接打到，不再只是背景相关。\n\n" "直接看它会影响哪条下游任务，再决定要不要进 plan。" ) assert rendered == expected def test_build_feishu_qa_analysis_packet_reply_text_for_signal_only_project_fit_uses_real_signal_names() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "project_fit_analysis", "conclusion": "放到直播复盘辅助工具（速记版）里，它更像落在输出规范 - 三栏速记清单格式这一层，不像输入适配 - 双源时间对齐能力这种前置约束或主链对象。", "fit_verdict": "fit_later", "focus_objects": [ { "name": "输出规范 - 三栏速记清单格式", "kind": "signal", "stance": "better_fit", "reason": "这个对象更像结果出来之后的解释和追问层。", }, { "name": "输入适配 - 双源时间对齐能力", "kind": "signal", "stance": "not_fit", "reason": "这个对象更像当前要先跑通的确定性主链或前置约束。", }, ], "reason_chain": [ "输入适配 - 双源时间对齐能力当前代表的是支持手动校准录屏与评论的时间偏移，并在此基础上完成全链路切片分析，更像要先校验、对齐或跑通的核心约束。", "输出规范 - 三栏速记清单格式当前代表的是输出纯文本文件，含高光、掉点、执行偏差三栏，更接近最终输出、用户感知或后续解释层。", ], "missing_mapping": "还缺它和输出规范 - 三栏速记清单格式这一层之间的明确映射。", "next_step": "先把输出规范 - 三栏速记清单格式和输入适配 - 双源时间对齐能力的边界卡清楚，再决定要不要继续推进。", } ) assert "结果出来之后的解释层后面的解释层" not in rendered assert "当前主链这种主链能力" not in rendered assert "输出规范 - 三栏速记清单格式" in rendered assert "输入适配 - 双源时间对齐能力" in rendered def test_build_feishu_qa_analysis_packet_reply_text_for_sparse_project_fit_uses_plain_language() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "project_fit_analysis", "conclusion": "光看自动开票系统现在这些项目信息，我还不想直接说它该放哪一步。", "fit_verdict": "fit_later", "current_subject": "自动开票系统", "focus_objects": [], "reason_chain": [ "现在项目里还没有足够具体的任务或信号源，硬把它挂到某个位置上，结论很容易说偏。" ], "missing_mapping": "至少还缺一个能对上的业务步骤或信号源。", "next_step": "先把这个项目最关键的业务步骤补清楚，我再判断它更适合放哪里。", } ) expected = ( "自动开票系统这个项目，我现在还不想硬说它该放在哪一步。\n\n" "现在关于自动开票系统，能看到的信息还比较粗，项目里还没有能对上的具体业务步骤或外部信号。" "要是现在硬把它挂到某个位置上，结论很容易说偏。\n\n" "先把这个项目最关键的业务步骤补清楚，我再判断它更适合放哪里。" ) assert rendered == expected def test_build_feishu_qa_analysis_packet_reply_text_for_sparse_project_fit_with_description_stays_business_like() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "project_fit_analysis", "conclusion": "光看自动开票系统现在这些项目信息，我还不想直接说它该放哪一步。", "fit_verdict": "fit_later", "current_subject": "自动开票系统", "focus_objects": [], "reason_chain": [ "我现在只看得到这个项目的大致目标：开发一套自动开票系统，实现开票接口到达后自动开票并发送发票图片的功能。", "但还看不到这条链路里谁来触发、要接哪些外部输入、以及哪个结果算真的跑通。", ], "missing_mapping": "还缺能对上的实际流程、关键结果和外部输入。", "next_step": "至少先把实际流程拆出来：谁发起、系统接什么输入、最后产出什么结果。", } ) expected = ( "自动开票系统这个项目，我现在还不想硬说它该放在哪一步。\n\n" "我现在只看得到这个项目的大致目标：开发一套自动开票系统，实现开票接口到达后自动开票并发送发票图片的功能。" "但还看不到这条链路里谁来触发、要接哪些外部输入、以及哪个结果算真的跑通。\n\n" "至少先把实际流程拆出来：谁发起、系统接什么输入、最后产出什么结果。" ) assert rendered == expected def test_build_feishu_qa_analysis_packet_reply_text_for_verifier_blocked_signal_only_project_avoids_generic_current_action() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "external_impact_analysis", "conclusion": "这条外部变化现在还不能直接挂到直播复盘辅助工具（速记版）的项目结论上。", "retrieval_status": "verifier_blocked", "impact_status": "no_direct_impact", "reason_chain": [ "现在卡在外部信号源 - TikTok Shop 直播开放API（评论流这条线索上：它指向获取带毫秒级时间戳的原始评论序列，用于与录屏帧对齐，但还没通过校验。", "如果后面要让输出规范 - 三栏速记清单格式消费这条变化，还得先把稳定的字段映射和证据补齐。", ], "blocked_reason_or_gap": "先把外部信号源 - TikTok Shop 直播开放API（评论流这条线索补齐到可校验，再看要不要挂到输出规范 - 三栏速记清单格式这一步。", "next_step": "先把外部信号源 - TikTok Shop 直播开放API（评论流这条线索补齐到可校验，再看要不要挂到输出规范 - 三栏速记清单格式这一步。", } ) assert "当前项目动作" not in rendered assert "输出规范 - 三栏速记清单格式" in rendered def test_build_feishu_qa_analysis_packet_reply_text_for_internal_explain_avoids_contract_jargon() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "internal_explain_analysis", "conclusion": "当前活动引擎是 Qwen 低阶引擎。", "answer_contract": "status_summary", "current_subject": "模型路由配置", "resolved_target": "模型路由配置", "reason_chain": [ "你这轮问的是模型路由配置的当前状态，所以这里只回答现状，不往规划建议上展开。", ], } ) expected = ( "当前活动引擎是 Qwen 低阶引擎。\n\n" "你这轮问的是模型路由配置的当前状态，所以这里只回答现状，不往规划建议上展开。" ) assert rendered == expected assert "status_summary" not in rendered def test_build_feishu_qa_analysis_packet_reply_text_for_internal_explain_uncertainty_deduplicates_gap() -> None: rendered = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_packet_reply_text( { "packet_type": "internal_explain_analysis", "conclusion": "当前活动引擎是 Qwen 低阶引擎。", "answer_contract": "status_summary", "current_subject": "模型路由配置", "resolved_target": "模型路由配置", "reason_chain": [ "你这轮问的是模型路由配置的当前状态，所以这里只回答现状，不往规划建议上展开。", ], "evidence_gap_or_uncertainty": "尚未确认是否已经切到最新灰度配置。", } ) expected = ( "当前活动引擎是 Qwen 低阶引擎。\n\n" "你这轮问的是模型路由配置的当前状态，所以这里只回答现状，不往规划建议上展开。\n\n" "尚未确认是否已经切到最新灰度配置。" ) assert rendered == expected assert rendered.count("尚未确认是否已经切到最新灰度配置") == 1 def test_should_accept_feishu_qa_natural_reply_rejects_internal_packet_jargon() -> None: accepted = FeishuReceiveRoutingMixin._should_accept_feishu_qa_natural_reply( ( "自动开票系统这个项目，现在还不能确定它该放在哪个阶段。\n\n" "分析包里明确写了，当前项目里至少还缺一个能对上的业务步骤或信号源。" "光看系统里现有的数据，没有足够具体的任务或外部信号支撑判断。\n\n" "analysis_packet 没提供 next_step，missing_mapping 也还缺字段。" ), selected_entity_candidates=["自动开票系统"], blocked_entity_candidates=["app"], ) assert accepted is False def test_normalize_feishu_qa_analysis_plan_keeps_only_allowed_focus_entities() -> None: plan = FeishuReceiveRoutingMixin._normalize_feishu_qa_analysis_plan( payload={ "conclusion": "Responses API 更像摘要卡片之后的解释层。", "focus_entities": [ { "name": "每日固定时段生成并推送结构化摘要卡片", "stance": "better_fit", "why": "这里更适合承接解释和追问。", }, { "name": "WMS库存快照自动拉取与清洗", "stance": "not_fit", "why": "这是确定性取数链路，不适合塞进生成能力。", }, { "name": "外部并不存在的对象", "stance": "not_fit", "why": "这条应该被过滤掉。", }, ], "missing_mapping": "当前没有证据说明它已接入现有信号源或主计算链。", "next_step": "先确认它到底承担哪个节点的角色。", }, entity_catalog=[ { "name": "每日固定时段生成并推送结构化摘要卡片", "kind": "task", }, { "name": "WMS库存快照自动拉取与清洗", "kind": "task", }, { "name": "外部信号源 - WMS实时库存接口", "kind": "signal", }, ], fallback_next_step="继续确认它到底落在这个项目的哪一层，再决定要不要进入 plan。", ) assert plan["conclusion"] == "Responses API 更像摘要卡片之后的解释层。" assert [item["name"] for item in plan["focus_entities"]] == [ "每日固定时段生成并推送结构化摘要卡片", "WMS库存快照自动拉取与清洗", ] assert [item["stance"] for item in plan["focus_entities"]] == ["better_fit", "not_fit"] assert all(item["name"] != "外部并不存在的对象" for item in plan["focus_entities"]) assert plan["missing_mapping"] == "当前没有证据说明它已接入现有信号源或主计算链。" def test_build_feishu_qa_analysis_fallback_plan_uses_deterministic_catalog_order_for_indirect_only_without_structured_focus() -> None: plan = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_fallback_plan( result={ "rag_result": { "external_recommendation": { "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, "recommended_next_step": "继续确认它到底落在这个项目的哪一层，再决定要不要进入 plan。", }, }, entity_catalog=[ {"name": "WMS库存快照自动拉取与清洗", "kind": "task"}, {"name": "TOP10商品缺货排序逻辑实现（基于当前库存与安全库存差值）", "kind": "task"}, {"name": "每日固定时段生成并推送结构化摘要卡片", "kind": "task"}, {"name": "试点小组成员接收通道配置与权限开通", "kind": "task"}, ], ) assert plan["conclusion"] == "目前还不能直接说它已经适合当前项目。" assert [item["name"] for item in plan["focus_entities"]] == [ "WMS库存快照自动拉取与清洗", "TOP10商品缺货排序逻辑实现（基于当前库存与安全库存差值）", ] assert [item["stance"] for item in plan["focus_entities"]] == ["better_fit", "not_fit"] @pytest.mark.asyncio async def test_render_feishu_qa_direct_reply_text_uses_shared_analysis_packet_family() -> None: captured: dict[str, object] = {} class FakeRouting(FeishuReceiveRoutingMixin): async def _build_feishu_qa_packet_context(self, *, result, message_content): # noqa: ANN001 _ = (result, message_content) return { "analysis_packet": { "packet_version": "v1", "packet_type": "project_fit_analysis", "conclusion": "放到采购经理每日实时库存上报服务里，Responses API 更像接在摘要卡片后面的解释层，不像主计算链。", "focus_objects": [ { "name": "每日固定时段生成并推送结构化摘要卡片", "kind": "task", "stance": "better_fit", "reason": "这里更像结果出来之后的解释和追问层。", }, { "name": "WMS库存快照自动拉取与清洗", "kind": "task", "stance": "not_fit", "reason": "这是稳定取数的主链，不适合直接塞进生成能力。", }, ], "fit_verdict": "fit_later", "current_subject": "采购经理每日实时库存上报服务", "candidate_position": "每日固定时段生成并推送结构化摘要卡片", "rejected_position": "WMS库存快照自动拉取与清洗", "preferred_layer": "每日固定时段生成并推送结构化摘要卡片", "rejected_layer": "WMS库存快照自动拉取与清洗", "reason_chain": [ "WMS库存快照自动拉取与清洗这一步属于确定性取数或计算主链，结果要稳定、可复算。", "每日固定时段生成并推送结构化摘要卡片更接近结果出来之后的解释和追问层，更适合承接这类能力。", ], "missing_mapping": "还缺它在每日固定时段生成并推送结构化摘要卡片这一步到底承担解释、问答还是编排角色的明确映射。", "next_step": "先确认它是不是只用在摘要卡片之后。", }, "selected_entity_candidates": [ "每日固定时段生成并推送结构化摘要卡片", "WMS库存快照自动拉取与清洗", ], "blocked_entity_candidates": [ "TOP10商品缺货排序逻辑实现（基于当前库存与安全库存差值）", ], } async def _generate_feishu_qa_natural_reply( self, *, message_content, analysis_packet, selected_entity_candidates, blocked_entity_candidates, ): # noqa: ANN001 captured["message_content"] = message_content captured["analysis_packet"] = analysis_packet captured["selected_entity_candidates"] = selected_entity_candidates captured["blocked_entity_candidates"] = blocked_entity_candidates return "" routing = FakeRouting() reply = await routing._render_feishu_qa_direct_reply_text( result={ "direct_reply_text": "外部实时查询：目前证据不足。", "rag_result": { "fit_brief": { "project_summary": "Alpha 是一个 AI 项目管理工作台。", }, "external_recommendation": { "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, }, }, message_content="这个适合吗", ) assert captured["message_content"] == "这个适合吗" assert captured["analysis_packet"]["packet_type"] == "project_fit_analysis" assert captured["selected_entity_candidates"] == [ "每日固定时段生成并推送结构化摘要卡片", "WMS库存快照自动拉取与清洗", ] expected = ( "放到采购经理每日实时库存上报服务里，Responses API 更像接在摘要卡片后面的解释层，不像主计算链。\n\n" "WMS库存快照自动拉取与清洗这一步属于确定性取数或计算主链，结果要稳定、可复算。每日固定时段生成并推送结构化摘要卡片更接近结果出来之后的解释和追问层，更适合承接这类能力。\n\n" "还缺它在每日固定时段生成并推送结构化摘要卡片这一步到底承担解释、问答还是编排角色的明确映射。先确认它是不是只用在摘要卡片之后。" ) assert reply == expected @pytest.mark.asyncio async def test_render_feishu_qa_direct_reply_text_internal_signal_source_state_strips_raw_internal_terms() -> None: class FakeRouting(FeishuReceiveRoutingMixin): async def _build_feishu_qa_packet_context(self, *, result, message_content): # noqa: ANN001 _ = (result, message_content) return { "analysis_packet": { "packet_version": "v1", "packet_type": "internal_explain_analysis", "conclusion": "当前还没看到已经记录、但还没接入的新信号源候选。", "focus_objects": [ { "name": "测试产能警告", "kind": "project", "stance": "relevant", "reason": "这轮问题就在问这个项目当前有没有新信号源候选。", } ], "answer_contract": "status_summary", "current_subject": "测试产能警告", "reason_chain": [ "这轮只回答当前项目里已经记录到的现状，不往方案设计上展开。" ], "resolved_target": "测试产能警告", }, "selected_entity_candidates": ["测试产能警告"], "blocked_entity_candidates": [], } async def _generate_feishu_qa_natural_reply( self, *, message_content, analysis_packet, selected_entity_candidates, blocked_entity_candidates, ): # noqa: ANN001 _ = (message_content, analysis_packet, selected_entity_candidates, blocked_entity_candidates) return "" routing = FakeRouting() reply = await routing._render_feishu_qa_direct_reply_text( result={ "direct_reply_text": ( "内部知识库查询：没有。项目“测试产能警告”（ID: 81）在内部知识库中未提及任何信号源。\n\n" "当前盘面：signal_total=0，pending_signal_count=0。\n" "建议动作：继续补充项目线索。" ), "rag_result": { "route_decision": { "route_mode": "internal", "effective_source_policy": "internal", }, "analysis_packet": { "packet_version": "v1", "packet_type": "internal_explain_analysis", "conclusion": "当前还没看到已经记录、但还没接入的新信号源候选。", }, }, }, message_content="测试产能警告这个项目当前有没有已经提到、但还没接入的新信号源？按现状回答。", ) assert reply == ( "当前还没看到已经记录、但还没接入的新信号源候选。\n\n" "这轮只回答当前项目里已经记录到的现状，不往方案设计上展开。\n\n" "来源：项目库" ) assert "Project database" not in reply assert "signal_total" not in reply assert "当前盘面" not in reply assert "建议动作" not in reply @pytest.mark.asyncio async def test_render_feishu_qa_direct_reply_text_prefers_recommendation_answer_for_external_signal_recommendation() -> None: class FakeRouting(FeishuReceiveRoutingMixin): async def _build_feishu_qa_packet_context(self, *, result, message_content): # noqa: ANN001 raise AssertionError("signal source recommendation should use direct recommendation answer before packet fallback") routing = FakeRouting() reply = await routing._render_feishu_qa_direct_reply_text( result={ "direct_reply_text": ( "外部实时查询：如果只是先给建议，我会先看这 3 类外部信号：\n" "1. 测试执行失败和阻塞记录，因为它最早反映测试队列是不是已经开始堆积。\n" "2. 构建或发布变更记录，因为它能帮你判断产能波动是不是由最近的版本改动带出来的。\n" "3. 工单或异常反馈，因为它能看出问题有没有开始影响到实际交付。\n" "这些先算建议，不代表已经接入或验证完成。" ), "rag_result": { "route_decision": { "route_mode": "external", "effective_source_policy": "external", }, "external_query_plan": { "interaction_level": "recommendation", }, "external_recommendation": { "retrieval_status": "verifier_blocked", "impact_status": "no_direct_impact", }, }, }, message_content="测试产能警告这个项目，如果只考虑外部信号源，你会优先推荐哪 3 类？直接给推荐结果和理由，不要复述当前现状，不要起草需求。", ) assert reply == ( "如果只是先给建议，我会先看这 3 类外部信号：\n" "1. 测试执行失败和阻塞记录，因为它最早反映测试队列是不是已经开始堆积。\n" "2. 构建或发布变更记录，因为它能帮你判断产能波动是不是由最近的版本改动带出来的。\n" "3. 工单或异常反馈，因为它能看出问题有没有开始影响到实际交付。\n" "这些先算建议，不代表已经接入或验证完成。\n\n" "来源：项目库 + 外部公开资料（当前仍在校验）" ) assert "External Freshness Guard" not in reply assert "这一层" not in reply assert "主链" not in reply def test_build_feishu_qa_analysis_plan_reply_text_avoids_stepwise_checklist_tone() -> None: reply = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_plan_reply_text( { "conclusion": "目前还不能直接说它已经适合当前项目。", "focus_entities": [ { "name": "每日固定时段生成并推送结构化摘要卡片", "stance": "better_fit", "why": "这里更像结果出来之后的解释和追问层。", }, { "name": "WMS库存快照自动拉取与清洗", "stance": "not_fit", "why": "这是稳定取数的主链，不适合直接塞进生成能力。", }, ], "missing_mapping": "还缺它在当前项目里到底承接哪个节点的明确映射。", "next_step": "先确认它是不是只用在摘要卡片之后。", } ) expected = ( "目前还不能直接说它已经适合当前项目。\n\n" "如果真要放进这个项目，我会先放在每日固定时段生成并推送结构化摘要卡片这一步，" "不太会直接改WMS库存快照自动拉取与清洗这条主流程。\n\n" "还缺它在当前项目里到底承接哪个节点的明确映射。\n\n" "先确认它是不是只用在摘要卡片之后。" ) assert reply == expected def test_build_feishu_qa_analysis_plan_reply_text_avoids_layer_jargon_for_fit() -> None: reply = FeishuReceiveRoutingMixin._build_feishu_qa_analysis_plan_reply_text( { "conclusion": "现在还不能直接说它已经适合当前项目。", "focus_entities": [ { "name": "需求澄清与范围确认", "stance": "better_fit", "why": "这一步更适合先接住解释、补说明或继续追问。", }, { "name": "核心方案落地", "stance": "not_fit", "why": "这一步要先稳定跑通，不适合先拿来试这种能力。", }, ], "missing_mapping": "现在还说不准它会先帮到需求澄清与范围确认，还是会直接改到核心方案落地。", "next_step": "先定清楚它是给需求澄清与范围确认补说明，还是会直接改到核心方案落地。", } ) assert "这一层" not in reply assert "主链能力" not in reply assert "边界卡清楚" not in reply assert reply == ( "现在还不能直接说它已经适合当前项目。\n\n" "如果真要放进这个项目，我会先放在需求澄清与范围确认这一步，不太会直接改核心方案落地这条主流程。\n\n" "现在还说不准它会先帮到需求澄清与范围确认，还是会直接改到核心方案落地。\n\n" "先定清楚它是给需求澄清与范围确认补说明，还是会直接改到核心方案落地。" ) def test_choose_feishu_qa_analysis_plan_prefers_fallback_when_indirect_only_plan_lacks_contrast() -> None: chosen = FeishuReceiveRoutingMixin._choose_feishu_qa_analysis_plan( normalized_plan={ "conclusion": "目前还不能直接说它已经适合当前项目。", "focus_entities": [ { "name": "WMS库存快照自动拉取与清洗", "stance": "relevant", "why": "该任务聚焦内部WMS数据接入。", }, { "name": "外部信号源 - WMS实时库存接口", "stance": "relevant", "why": "同为外部API调用场景。", }, ], "missing_mapping": "缺少具体作用层映射。", "next_step": "确认它是不是用于摘要生成环节。", }, fallback_plan={ "conclusion": "目前还不能直接说它已经适合当前项目。", "focus_entities": [ { "name": "每日固定时段生成并推送结构化摘要卡片", "stance": "better_fit", "why": "这一步更像结果出来之后的解释和追问层。", }, { "name": "WMS库存快照自动拉取与清洗", "stance": "not_fit", "why": "这一步属于确定性取数或计算主链，不适合直接塞进生成能力。", }, ], "missing_mapping": "还缺它在当前项目里到底承接哪个节点的明确映射。", "next_step": "继续确认它到底落在这个项目的哪一层，再决定要不要进入 plan。", }, retrieval_status="evidence_found", impact_status="indirect_only", ) assert [item["name"] for item in chosen["focus_entities"]] == [ "每日固定时段生成并推送结构化摘要卡片", "WMS库存快照自动拉取与清洗", ] def test_build_execution_receipt_reply_prefers_compact_multiline_summary() -> None: reply = FeishuReplyTemplatesMixin._build_execution_receipt_reply( message_content="请处理这条信号", capture_result={ "status": "captured", "signal_id": 18, "linked_project_ids": [9], }, ) assert reply == "✅ 已执行完成\n信号#18 已入库\n项目#9 已关联" def test_build_action_card_for_queue_pending_confirm_returns_none() -> None: card = FeishuCardActionMixin._build_action_card_from_capture_result( capture_result={ "reply_template_key": "queue_pending_confirm", "results": [{"item_id": 201, "route_status": "queued_for_decision"}], "direct_reply_text": "需要你确认后继续执行。", }, mention_open_id=None, mention_name=None, ) assert card is None @pytest.mark.asyncio async def test_handle_card_action_trigger_continue_dispatches_resume(monkeypatch) -> None: called: dict[str, object] = {} class DummyDB: async def commit(self) -> None: called["committed"] = True async def rollback(self) -> None: called["rolled_back"] = True service = FeishuIntegrationService(db=DummyDB()) service.tenant_id = 7 async def _fake_reply_text(**kwargs): called["reply_text"] = kwargs.get("text") called["reply_chat"] = kwargs.get("chat_id") monkeypatch.setattr(service, "_safe_reply_text", _fake_reply_text) class _FakeKernel: def __init__(self, db, tenant_id): called["kernel_tenant_id"] = tenant_id async def process_ingress(self, **kwargs): called["ingress_kwargs"] = kwargs return { "handled": True, "status": "captured", "direct_reply_text": "已继续执行。", } monkeypatch.setattr( "app.services.ai_pmo.ai_pmo_kernel.ai_pmo_kernel_runtime.AIPMOKernelService", _FakeKernel, ) result = await service._handle_card_action_trigger( { "open_message_id": "om_continue_1", "open_chat_id": "oc_continue_1", "operator": {"open_id": "ou_continue", "name": "继续用户"}, "action": {"value": {"action": "continue", "work_item_id": 55}}, } ) assert result["code"] == 0 data = result.get("data") if isinstance(result.get("data"), dict) else {} assert data.get("handled") is True assert data.get("target_work_item_id") == 55 ingress_kwargs = called["ingress_kwargs"] assert ingress_kwargs["message_text"] == "继续处理 work_item #55" assert ingress_kwargs["source_system"] == "ai_pmo_ui" assert called["reply_text"] == "已继续执行。" assert called["reply_chat"] == "oc_continue_1" @pytest.mark.asyncio async def test_handle_card_action_trigger_plan_choice_dispatches_synthetic_ingress(monkeypatch) -> None: called: dict[str, object] = {} class DummyDB: async def commit(self) -> None: called["committed"] = True async def rollback(self) -> None: called["rolled_back"] = True service = FeishuIntegrationService(db=DummyDB()) service.tenant_id = 7 async def _fake_reply_card(**kwargs): called["reply_card"] = kwargs.get("card_content") called["reply_chat"] = kwargs.get("chat_id") monkeypatch.setattr(service, "_safe_reply_card", _fake_reply_card) class _FakeKernel: def __init__(self, db, tenant_id): called["kernel_tenant_id"] = tenant_id async def process_ingress(self, **kwargs): called["ingress_kwargs"] = kwargs return { "handled": True, "status": "captured", "reply_template_key": "requirement_plan_followup", "direct_reply_text": "C. 混合增强路径\n\n⭐ 当前我更推荐：C", "results": [{"work_item_id": 88}], "work_item_id": 88, } monkeypatch.setattr( "app.services.ai_pmo.ai_pmo_kernel.ai_pmo_kernel_runtime.AIPMOKernelService", _FakeKernel, ) result = await service._handle_card_action_trigger( { "open_message_id": "om_plan_1", "open_chat_id": "oc_plan_1", "operator": {"open_id": "ou_plan", "name": "计划用户"}, "action": {"value": {"action": "plan_choice", "choice": "C", "synthetic_text": "C", "work_item_id": 88}}, } ) assert result["code"] == 0 data = result.get("data") if isinstance(result.get("data"), dict) else {} assert data.get("handled") is True assert data.get("target_work_item_id") == 88 ingress_kwargs = called["ingress_kwargs"] assert ingress_kwargs["message_text"] == "C" assert ingress_kwargs["source_system"] == "feishu" assert str(ingress_kwargs["source_message_id"]).startswith("feishu-card-plan-88-plan_choice-") ingress_options = ingress_kwargs.get("ingress_options") if isinstance(ingress_kwargs.get("ingress_options"), dict) else {} thread_context = ingress_options.get("thread_context") if isinstance(ingress_options.get("thread_context"), dict) else {} assert thread_context["reply_to_message_id"] == "om_plan_1" assert thread_context["root_message_id"] == "om_plan_1" assert called["reply_chat"] == "oc_plan_1" assert isinstance(called["reply_card"], dict) assert called["reply_card"]["header"]["title"]["content"] == "AI PMO 需求收敛" @pytest.mark.asyncio async def test_handle_card_action_trigger_queue_continue_is_ignored(monkeypatch) -> None: called: dict[str, object] = {} class DummyDB: async def commit(self) -> None: called["committed"] = True async def rollback(self) -> None: called["rolled_back"] = True service = FeishuIntegrationService(db=DummyDB()) service.tenant_id = 7 async def _fake_reply_text(**kwargs): called["reply_text"] = kwargs.get("text") monkeypatch.setattr(service, "_safe_reply_text", _fake_reply_text) result = await service._handle_card_action_trigger( { "open_message_id": "om_queue_1", "open_chat_id": "oc_queue_1", "operator": {"open_id": "ou_queue", "name": "队列用户"}, "action": {"value": {"action": "continue", "queue_item_id": 201}}, } ) assert result["code"] == 0 assert result["msg"] == "card_action_ignored" data = result.get("data") if isinstance(result.get("data"), dict) else {} assert data.get("reason") == "missing_work_item_id" assert "reply_text" not in called @pytest.mark.asyncio async def test_route_scene_reply_sends_requirement_plan_followup_as_card(monkeypatch) -> None: calls: dict[str, object] = {} class FakeRouting(FeishuReceiveRoutingMixin): def _build_capability_reply(self, message_content: str): # noqa: ANN001 _ = message_content return None def _build_action_card_from_capture_result(self, *, capture_result, mention_open_id, mention_name): # noqa: ANN001 _ = (mention_open_id, mention_name) if str(capture_result.get("reply_template_key") or "").strip().lower() == "requirement_plan_followup": return { "header": {"title": {"tag": "plain_text", "content": "AI PMO 需求收敛"}}, "elements": [], } return None async def _safe_reply_card(self, **kwargs): # noqa: ANN001 calls["reply_card"] = kwargs async def _safe_reply_text(self, **kwargs): # noqa: ANN001 calls["reply_text"] = kwargs def _emit_state_transition(self, **kwargs): # noqa: ANN001 calls["state_transition"] = kwargs def _should_allow_group_qa_reply(self, **kwargs): # noqa: ANN001 _ = kwargs return True routing = FakeRouting() result = await routing._route_scene_reply( scene="qa", result={ "reply_template_key": "requirement_plan_followup", "work_item_id": 88, "direct_reply_text": "C. 混合增强路径\n\n⭐ 当前我更推荐：C", }, message_content="C", chat_id="chat-plan-card", chat_type="group", sender_open_id="ou-plan", sender_name="计划用户", mentioned_bot_names=[], thread_context=None, trace_id="trace-plan-card", message_id="msg-plan-card", perf_t0=None, ) assert result == "qa_action_card" assert "reply_card" in calls assert "reply_text" not in calls card_content = calls["reply_card"]["card_content"] if isinstance(calls["reply_card"], dict) else {} assert card_content["header"]["title"]["content"] == "AI PMO 需求收敛" @pytest.mark.asyncio async def test_route_scene_reply_sends_rich_qa_direct_text_for_handoff_metadata() -> None: calls: dict[str, object] = {} class FakeRouting(FeishuReceiveRoutingMixin): def _build_capability_reply(self, message_content: str): # noqa: ANN001 _ = message_content return None def _build_action_card_from_capture_result(self, *, capture_result, mention_open_id, mention_name): # noqa: ANN001 _ = (capture_result, mention_open_id, mention_name) return None async def _safe_reply_text(self, **kwargs): # noqa: ANN001 calls["reply_text"] = kwargs def _emit_state_transition(self, **kwargs): # noqa: ANN001 calls["state_transition"] = kwargs def _should_allow_group_qa_reply(self, **kwargs): # noqa: ANN001 _ = kwargs return True async def _render_feishu_qa_direct_reply_text(self, *, result, message_content): # noqa: ANN001 _ = (result, message_content) return ( "Responses API 不适合放在 WMS库存快照自动拉取或者 TOP10缺货排序这两层，" "更像放在每日结构化摘要卡片之后的解释和追问层。\n\n" "这个项目当前在做的是先把 WMS库存快照拉进来，再完成 TOP10缺货排序，" "最后按固定时段推送结构化摘要卡片。前两层都属于取数和计算主链，应该保持确定、可复算，" "不适合把 Responses API 放进去做主判断。\n\n" "它真正更可能有价值的地方，是采购经理收到摘要卡片之后继续追问的时候。" ) routing = FakeRouting() result = await routing._route_scene_reply( scene="qa", result={ "direct_reply_required": True, "direct_reply_text": "外部实时查询：目前证据不足，建议先明确要查融资还是招聘动态。", "rag_result": { "fit_brief": { "project_summary": "Alpha 是一个 AI 项目管理工作台。", "repo_summary": "当前仓库负责统一项目上下文与 AI handoff。", }, "external_recommendation": { "retrieval_status": "evidence_found", "impact_status": "indirect_only", "external_impact": "目前更像背景相关，还不能直接映射到当前正式对象。", }, "recommended_next_step": "如果你要，我可以基于当前结论进入 plan，继续把需求范围和下一步收清楚。", }, }, message_content="这个适合吗", chat_id="chat-rich-qa", chat_type="group", sender_open_id="ou-rich-qa", sender_name="测试用户", mentioned_bot_names=[], thread_context={"reply_to_message_id": "om-rich-qa"}, trace_id="trace-rich-qa", message_id="msg-rich-qa", perf_t0=None, ) assert result == "qa_direct" reply_text = calls["reply_text"]["text"] if isinstance(calls.get("reply_text"), dict) else "" expected = ( "Responses API 不适合放在 WMS库存快照自动拉取或者 TOP10缺货排序这两层，" "更像放在每日结构化摘要卡片之后的解释和追问层。\n\n" "这个项目当前在做的是先把 WMS库存快照拉进来，再完成 TOP10缺货排序，" "最后按固定时段推送结构化摘要卡片。前两层都属于取数和计算主链，应该保持确定、可复算，" "不适合把 Responses API 放进去做主判断。\n\n" "它真正更可能有价值的地方，是采购经理收到摘要卡片之后继续追问的时候。" ) assert reply_text == expected assert "这事可以继续看" not in reply_text assert "先说我的判断" not in reply_text