From d80f155866f20a691a250c6894250487428c4c5d Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Sat, 11 Apr 2026 01:38:03 +0800 Subject: [PATCH] fix: isolate feishu requirement plan threads --- .../ai_pmo_kernel_mixins_ingress_render.py | 2 + .../mixin_execute_query_project.py | 81 +++++++++++ .../mounted_ai_runtime_write_service.py | 31 ++++ backend/tests/test_ai_pmo_ingress_mainline.py | 134 ++++++++++++++++++ .../test_mounted_ai_runtime_write_service.py | 65 +++++++++ 5 files changed, 313 insertions(+) diff --git a/backend/app/services/ai_pmo/ai_pmo_kernel/mixins/ingress/ai_pmo_kernel_mixins_ingress_render.py b/backend/app/services/ai_pmo/ai_pmo_kernel/mixins/ingress/ai_pmo_kernel_mixins_ingress_render.py index 94c57955..f231079e 100644 --- a/backend/app/services/ai_pmo/ai_pmo_kernel/mixins/ingress/ai_pmo_kernel_mixins_ingress_render.py +++ b/backend/app/services/ai_pmo/ai_pmo_kernel/mixins/ingress/ai_pmo_kernel_mixins_ingress_render.py @@ -149,6 +149,8 @@ class KernelIngressRenderMixin: result["rag_result"] = dict(execution_result["rag_result"]) if workflow_route == ROUTE_QUERY_ANSWER and isinstance(execution_result.get("requirement_plan_context"), dict): result["requirement_plan_context"] = dict(execution_result["requirement_plan_context"]) + if isinstance(execution_result.get("draft_artifact"), dict): + result["draft_artifact"] = dict(execution_result["draft_artifact"]) if intent.name == INTENT_PROJECT_CREATE: raw_payload = execution_result.get("raw_payload") if isinstance(execution_result.get("raw_payload"), dict) else {} result["project"] = raw_payload.get("project") diff --git a/backend/app/services/ai_pmo/ai_pmo_typed_executor_execute/mixin_execute_query_project.py b/backend/app/services/ai_pmo/ai_pmo_typed_executor_execute/mixin_execute_query_project.py index 082fe7c7..d0c43cbb 100644 --- a/backend/app/services/ai_pmo/ai_pmo_typed_executor_execute/mixin_execute_query_project.py +++ b/backend/app/services/ai_pmo/ai_pmo_typed_executor_execute/mixin_execute_query_project.py @@ -3791,6 +3791,13 @@ class ExecuteQueryProjectMixin: if draft_state == "draft_pending_confirm": return True + identity_conflicts = self._requirement_plan_identity_conflict_fields( + message_text=message_text, + draft=draft, + ) + if identity_conflicts: + return True + overlap_ratio = self._requirement_seed_overlap_ratio( message_text=message_text, topic=topic, @@ -3798,6 +3805,52 @@ class ExecuteQueryProjectMixin: ) return overlap_ratio < 0.18 + @classmethod + def _requirement_plan_identity_conflict_fields( + cls, + *, + message_text: str, + draft: Optional[MountedDraftWorkspace], + ) -> list[str]: + if draft is None: + return [] + current_card = cls._sanitize_requirement_plan_card_runtime_context( + card=dict(getattr(draft, "requirement_card", None) or {}) + ) + if not current_card: + return [] + incoming_updates = cls._extract_requirement_plan_updates(message_text=message_text, card={}) + incoming_title = str(incoming_updates.get("title") or "").strip() + if not incoming_title: + derived_title = cls._derive_requirement_plan_title(message_text=message_text, card={}) + if str(derived_title or "").strip() and str(derived_title or "").strip() != "需求收敛草稿": + incoming_title = str(derived_title).strip() + + def _normalize(value: Any) -> str: + return WorkflowSemanticPlanner._normalize_text(cls._sanitize_requirement_plan_text(str(value or ""))).strip() + + conflicts: list[str] = [] + current_title = _normalize(current_card.get("title")) + normalized_incoming_title = _normalize(incoming_title) + if ( + current_title + and normalized_incoming_title + and normalized_incoming_title not in current_title + and current_title not in normalized_incoming_title + ): + conflicts.append("title") + + current_target_user = _normalize(current_card.get("target_user")) + incoming_target_user = _normalize(incoming_updates.get("target_user")) + if ( + current_target_user + and incoming_target_user + and incoming_target_user not in current_target_user + and current_target_user not in incoming_target_user + ): + conflicts.append("target_user") + return list(dict.fromkeys(conflicts)) + async def _maybe_convert_query_answer_contract_to_requirement_plan_result( self, *, @@ -3900,6 +3953,34 @@ class ExecuteQueryProjectMixin: result["requirement_plan_context"] = requirement_plan_context return result + if ( + not force_new_topic + and not self._has_requirement_plan_thread_anchor(thread_context=thread_context) + and self._requirement_plan_identity_conflict_fields(message_text=message_text, draft=draft) + ): + logger.info( + "requirement_plan_same_scope_conflict_force_new_topic", + tenant_id=self.tenant_id, + source_system=source_system, + source_message_id=source_message_id, + ) + topic, draft = await self._load_requirement_plan_workspace( + source_system=source_system, + source_message_id=source_message_id, + source_chat_id=source_chat_id, + sender=sender, + source_sender_open_id=source_sender_open_id, + thread_context=thread_context, + conversation_key=conversation_key, + force_new_topic=True, + ) + force_new_topic = True + if topic is None or draft is None: + return { + "status": "failed_safe", + "error_message": "需求收敛新 topic 初始化失败，请稍后重试。", + } + topic, draft = await self._repair_requirement_plan_workspace_from_reply_context( topic=topic, draft=draft, diff --git a/backend/app/services/mounted_ai_runtime_write_service.py b/backend/app/services/mounted_ai_runtime_write_service.py index 4908466d..490eb262 100644 --- a/backend/app/services/mounted_ai_runtime_write_service.py +++ b/backend/app/services/mounted_ai_runtime_write_service.py @@ -181,6 +181,23 @@ class MountedAIRuntimeWriteService: return True return False + @staticmethod + def _extract_result_topic_workspace_id(result: Optional[dict[str, Any]]) -> Optional[int]: + if not isinstance(result, dict): + return None + artifact = result.get("draft_artifact") if isinstance(result.get("draft_artifact"), dict) else {} + for candidate in ( + artifact.get("topic_workspace_id"), + result.get("topic_workspace_id"), + ): + try: + if candidate is None or candidate == "": + continue + return int(candidate) + except (TypeError, ValueError): + continue + return None + async def _resolve_source_activation( self, *, @@ -748,6 +765,20 @@ class MountedAIRuntimeWriteService: run_row.execution_outcome, persisted_result_fields.get("requirement_plan_context"), ) + canonical_topic_workspace_id = self._extract_result_topic_workspace_id(result) + if canonical_topic_workspace_id is not None and canonical_topic_workspace_id != int(topic_row.id): + canonical_topic = ( + await self.db.execute( + select(MountedTopicWorkspace).where( + MountedTopicWorkspace.tenant_id == int(topic_row.tenant_id), + MountedTopicWorkspace.id == int(canonical_topic_workspace_id), + ) + ) + ).scalar_one_or_none() + if canonical_topic is not None: + topic_row = canonical_topic + tracking.topic_workspace = canonical_topic + run_row.topic_workspace_id = int(canonical_topic.id) run_row.completed_at = now_at self.db.add( MountedAgentTrace( diff --git a/backend/tests/test_ai_pmo_ingress_mainline.py b/backend/tests/test_ai_pmo_ingress_mainline.py index e69d068a..a8a062fa 100644 --- a/backend/tests/test_ai_pmo_ingress_mainline.py +++ b/backend/tests/test_ai_pmo_ingress_mainline.py @@ -7298,6 +7298,140 @@ async def test_repair_requirement_plan_workspace_from_reply_context_prefers_sema assert draft.requirement_card["round_option_maps"]["target_user"]["A"] == "单一区域经理" +@pytest.mark.asyncio +async def test_build_requirement_planning_direct_result_forces_new_topic_when_same_scope_identity_conflicts( + monkeypatch, + requirement_plan_db_session, +) -> None: + class FakeExecutor(mixin_execute_query_project.ExecuteQueryProjectMixin): + def __init__(self) -> None: + self.db = requirement_plan_db_session + self.tenant_id = 3 + + async def _fake_build_requirement_plan_next_question( + self, + *, + missing_field: str, + profile: str, + message_text: str = "", + card: dict[str, object] | None = None, + latest_updates: dict[str, object] | None = None, + ) -> str: + _ = (self, profile, message_text, latest_updates) + assert isinstance(card, dict) + card["round_option_maps"] = { + **(dict(card.get("round_option_maps") or {}) if isinstance(card.get("round_option_maps"), dict) else {}), + str(missing_field): { + "A": "轻量晨报工具", + "B": "业务平台入口", + "C": "能力底座组件", + }, + } + card["last_round_options"] = { + "field": str(missing_field), + "options": [ + {"key": "A", "title": "轻量晨报工具"}, + {"key": "B", "title": "业务平台入口"}, + {"key": "C", "title": "能力底座组件"}, + ], + "recommendation": {"key": "A"}, + } + return "第一轮：项目性质\n\nA. 轻量晨报工具\nB. 业务平台入口\nC. 能力底座组件" + + monkeypatch.setattr( + mixin_execute_query_project.ExecuteQueryProjectMixin, + "_build_requirement_plan_next_question", + _fake_build_requirement_plan_next_question, + raising=False, + ) + + conversation_scope_key = build_feishu_conversation_scope_key( + source_chat_id="chat-plan-same-scope-conflict", + source_sender_open_id="ou-plan-same-scope-conflict", + sender=None, + ) + old_topic = MountedTopicWorkspace( + tenant_id=3, + topic_key="feishu:chat-plan-same-scope-conflict:message:om-old-plan-root", + source_system="feishu", + channel_ref="chat-plan-same-scope-conflict", + conversation_scope_key=conversation_scope_key, + root_message_id=None, + last_message_id="om-old-plan-root", + mode="plan", + status="active", + plan_status="collecting", + metadata_json={"plan_mode": True}, + ) + requirement_plan_db_session.add(old_topic) + await requirement_plan_db_session.flush() + old_draft = MountedDraftWorkspace( + tenant_id=3, + topic_workspace_id=int(old_topic.id), + state="collecting", + draft_version=0, + requirement_card={ + "plan_profile": "it_project", + "title": "直播复盘辅助工具", + "goal": "先把直播复盘辅助工具的目标、服务对象、第一版范围和交付方式收清楚，再决定是否正式创建。", + "project_nature": "轻量工具", + "target_user": "运营团队", + "question_round": 3, + }, + draft_payload=None, + required_confirmations=[], + ) + requirement_plan_db_session.add(old_draft) + + current_topic = MountedTopicWorkspace( + tenant_id=3, + topic_key="feishu:chat-plan-same-scope-conflict:message:msg-plan-same-scope-conflict-root", + source_system="feishu", + channel_ref="chat-plan-same-scope-conflict", + conversation_scope_key=conversation_scope_key, + root_message_id=None, + last_message_id="msg-plan-same-scope-conflict-root", + mode="chat", + status="active", + plan_status="idle", + metadata_json={}, + ) + requirement_plan_db_session.add(current_topic) + await requirement_plan_db_session.flush() + + executor = FakeExecutor() + result = await executor._build_requirement_planning_direct_result( + message_text="正式验真T9A-0410：做一个给区域经理用的门店库存异常晨报助手，只收第一版需求，不要直接创建项目。第一版先看负库存和超阈值缺货，每天自动汇总并发给区域经理。", + source_system="feishu", + source_message_id="msg-plan-same-scope-conflict-root", + source_chat_id="chat-plan-same-scope-conflict", + sender=None, + source_sender_open_id="ou-plan-same-scope-conflict", + thread_context=None, + conversation_key=conversation_scope_key, + topic_switch=None, + ) + + await requirement_plan_db_session.flush() + current_draft = ( + await requirement_plan_db_session.execute( + select(MountedDraftWorkspace).where( + MountedDraftWorkspace.topic_workspace_id == int(current_topic.id) + ) + ) + ).scalar_one_or_none() + + assert result["status"] == "succeeded" + requirement_plan_context = result.get("requirement_plan_context") + assert isinstance(requirement_plan_context, dict) + assert requirement_plan_context["title"] == "给区域经理用的门店库存异常晨报助手" + assert requirement_plan_context["target_user"] == "区域经理" + assert current_draft is not None + assert current_draft.requirement_card["title"] == "给区域经理用的门店库存异常晨报助手" + assert old_draft.requirement_card["title"] == "直播复盘辅助工具" + assert int(current_draft.topic_workspace_id) == int(current_topic.id) + + @pytest.mark.asyncio async def test_build_requirement_planning_direct_result_prefers_reply_anchor_topic_over_same_scope_sibling( monkeypatch, diff --git a/backend/tests/test_mounted_ai_runtime_write_service.py b/backend/tests/test_mounted_ai_runtime_write_service.py index 90321554..fe572fed 100644 --- a/backend/tests/test_mounted_ai_runtime_write_service.py +++ b/backend/tests/test_mounted_ai_runtime_write_service.py @@ -652,3 +652,68 @@ async def test_finalize_run_reconciles_semantic_packet_summary_requirement_card_ assert packet_card["target_user"] == "单一层级区域经理" assert packet_card["question_round"] == 4 assert packet_card["last_round_options"]["field"] == "first_scope" + + +@pytest.mark.asyncio +async def test_finalize_run_rebinds_conversation_run_to_canonical_draft_artifact_topic(db_session) -> None: + service = MountedAIRuntimeWriteService(db_session) + + tracking = await service.begin_ingress_run( + tenant_id=1, + source_system="feishu", + source_message_id="om_message_plan_canonical_rebind", + message_text="想聊个新需求。", + sender="林宇航", + channel_ref="chat_plan_canonical_rebind", + thread_context=None, + source_sender_open_id="ou_plan_canonical_rebind", + ) + assert tracking is not None + ingress_topic_id = int(tracking.topic_workspace.id) + + canonical_topic = MountedTopicWorkspace( + tenant_id=1, + topic_key="feishu:chat_plan_canonical_rebind:message:om_existing_requirement_root", + source_system="feishu", + channel_ref="chat_plan_canonical_rebind", + conversation_scope_key="feishu:chat_plan_canonical_rebind:ou_plan_canonical_rebind", + root_message_id=None, + mode="plan", + status="active", + plan_status="collecting", + metadata_json={"plan_mode": True}, + ) + db_session.add(canonical_topic) + await db_session.flush() + + await service.finalize_run( + tracking=tracking, + result={ + "handled": True, + "status": "captured", + "classification": "command", + "workflow_route": "project_create.create", + "direct_reply_text": "已整理成草稿，等待确认。", + "reply_template_key": "draft_pending_confirm", + "draft_artifact": { + "type": "requirement_plan_draft", + "topic_workspace_id": int(canonical_topic.id), + "draft_workspace_id": 36, + "draft_version": 1, + "promote_to_confirmation": True, + }, + }, + ) + await db_session.commit() + + saved_run = (await db_session.execute(select(MountedConversationRun))).scalar_one() + traces = ( + await db_session.execute( + select(MountedAgentTrace).order_by(MountedAgentTrace.id.asc()) + ) + ).scalars().all() + + assert int(saved_run.topic_workspace_id) == int(canonical_topic.id) + assert int(saved_run.topic_workspace_id) != ingress_topic_id + assert saved_run.execution_outcome["draft_artifact"]["topic_workspace_id"] == int(canonical_topic.id) + assert traces[-1].agent_name == "ReplyComposer"