from types import SimpleNamespace import pytest from pydantic import ValidationError from app.api import rag_ask_contract from app.api.router import api_router from app.api.ai_steward_api import router as ai_steward_router from app.api.rag_api import routes_query from app.core.config import settings from app.schemas.rag import RAGAskRequest from app.services.adaptive_rag_service.mixin_answer_payload import AdaptiveRAGAnswerPayloadMixin def _direct_route_plan(*, goal_kind: str, answer_contract: str) -> dict: return { "source_strategy": "direct", "goal_kind": goal_kind, "answer_contract": answer_contract, "allowed_executors": ["direct_answer"], } def test_rag_ask_request_normalizes_contract_fields() -> None: assert RAGAskRequest(question="状态怎么样").thinking_mode == "auto" req = RAGAskRequest( question="状态怎么样", asked_via="ai-pmo-process-text", message_intent="MIXED", mixed_signal_part="新增任务确认流", action_mode="read", consultant_mode_preference="ADVise", ) assert req.asked_via == "/ai-pmo/process-text" assert req.message_intent == "mixed" assert req.mixed_signal_part == "新增任务确认流" assert req.action_mode == "read" assert req.consultant_mode_preference == "advise" def test_rag_ask_request_rejects_unknown_asked_via() -> None: with pytest.raises(ValidationError): RAGAskRequest( question="hello", asked_via="email", ) def test_build_answer_question_kwargs_maps_fields() -> None: request = RAGAskRequest( question="请确认任务", asked_via="pmflow_manual", session_id=33, new_session=1, thinking_mode="deep", action_mode="read_write", sync_requirement=0, message_intent="mixed", mixed_signal_part="新增任务确认流", consultant_mode_preference="challenge", ) kwargs = rag_ask_contract.build_answer_question_kwargs( request=request, current_user=SimpleNamespace(id="77"), ) assert kwargs == { "question": "请确认任务", "asked_by": 77, "tenant_id": None, "asked_via": "pmflow_manual", "project_id": None, "source_chat_id": None, "session_id": 33, "force_new_session": True, "thinking_mode": "deep", "action_mode": "read_write", "sync_requirement": False, "message_intent": "mixed", "mixed_signal_part": "新增任务确认流", "consultant_mode_preference": "challenge", "surface_context": {}, } def test_build_kernel_ingress_kwargs_assigns_unique_source_message_id_for_rag() -> None: request = RAGAskRequest(question="现在进展怎么样") kwargs_a = rag_ask_contract.build_kernel_ingress_kwargs( request=request, current_user=SimpleNamespace(id=1, tenant_id=1, username="admin"), ) kwargs_b = rag_ask_contract.build_kernel_ingress_kwargs( request=request, current_user=SimpleNamespace(id=1, tenant_id=1, username="admin"), ) source_message_id_a = str(kwargs_a["source_message_id"] or "") source_message_id_b = str(kwargs_b["source_message_id"] or "") assert source_message_id_a.startswith("rag-ask-") assert source_message_id_b.startswith("rag-ask-") assert source_message_id_a != source_message_id_b def test_build_kernel_ingress_kwargs_preserves_explicit_anchor_fields() -> None: request = RAGAskRequest( question="这个项目怎么样", project_id=17, source_chat_id="pmflow-project-17", ) kwargs = rag_ask_contract.build_kernel_ingress_kwargs( request=request, current_user=SimpleNamespace(id=1, tenant_id=1, username="admin"), ) assert kwargs["project_hint_id"] == 17 assert kwargs["channel_ref"] == "pmflow-project-17" assert kwargs["ingress_options"]["rag_request_context"]["project_id"] == 17 assert kwargs["ingress_options"]["rag_request_context"]["source_chat_id"] == "pmflow-project-17" def test_build_kernel_ingress_kwargs_uses_conversation_context_project_binding() -> None: request = RAGAskRequest( question="把我们的项目产品和宝宝树对比下", session_id=524, ) kwargs = rag_ask_contract.build_kernel_ingress_kwargs( request=request, current_user=SimpleNamespace(id=1, tenant_id=1, username="admin"), conversation_context={ "last_entity_ids": [141], "active_workspace": { "primary_object_ref": "project:141", "primary_object_id": 141, "primary_object_title": "童析—学龄前儿童行为特质AI评估助手", }, }, ) assert kwargs["ingress_options"]["rag_request_context"]["project_id"] == 141 assert kwargs["ingress_options"]["rag_request_context"]["active_workspace"]["primary_object_ref"] == "project:141" assert kwargs["ingress_options"]["tenant_project_context"]["current_project_id"] == 141 assert kwargs["ingress_options"]["project_operating_context"]["current_project_id"] == 141 def test_build_kernel_ingress_kwargs_attaches_project_operating_context() -> None: request = RAGAskRequest( question="OpenAI Docs 这种能力适不适合我们当前项目？", project_id=17, source_chat_id="pmflow-project-17", asked_via="feishu", ) kwargs = rag_ask_contract.build_kernel_ingress_kwargs( request=request, current_user=SimpleNamespace(id=1, tenant_id=1, username="admin"), ) packet = kwargs["ingress_options"]["project_operating_context"] assert packet["current_project_id"] == 17 assert packet["current_repo_scope"] assert packet["page_anchor"]["project_id"] == 17 assert packet["fit_scope"] == "current_project_and_repo" assert packet["decision_strategy"] == "answer_then_offer_plan" def test_build_kernel_ingress_kwargs_attaches_surface_and_tenant_project_context() -> None: request = RAGAskRequest( question="我今天应该先处理什么？", surface_context={ "surface": "personal_space", "workbench": { "counts": {"nextActions": 2, "pendingConfirmations": 1, "aiProgress": 1}, "top_next_actions": [ { "title": "确认项目草案", "project": "AI 周报助手", "priority": "高", "href": "/change-review?projectId=17&changeSetId=cs-1&itemId=ci-1", "object_ref": "change_item:ci-1", "object_anchor": { "object_ref": "change_item:ci-1", "object_type": "change_item", "object_id": "ci-1", "surface": "change_review", "project_id": "17", "project_title": "AI 周报助手", "href": "/change-review?projectId=17&changeSetId=cs-1&itemId=ci-1", "parent_object_ref": "project:17", }, } ], }, "ai_readiness": {"status": "ready", "can_generate_project_draft": True, "can_manage": False}, }, ) kwargs = rag_ask_contract.build_kernel_ingress_kwargs( request=request, current_user=SimpleNamespace(id=7, tenant_id=3, username="owner"), ) options = kwargs["ingress_options"] assert options["surface_context"]["surface"] == "personal_space" assert options["tenant_project_context"]["tenant_id"] == 3 assert options["tenant_project_context"]["surface_context"]["workbench"]["counts"]["nextActions"] == 2 action = options["tenant_project_context"]["surface_context"]["workbench"]["top_next_actions"][0] assert action["object_ref"] == "change_item:ci-1" assert action["object_anchor"]["parent_object_ref"] == "project:17" assert options["rag_request_context"]["surface_context"]["workbench"]["counts"]["pendingConfirmations"] == 1 @pytest.mark.asyncio async def test_rag_ask_route_forwards_contract_fields(monkeypatch) -> None: captured: dict[str, object] = {} class FakeKernel: def __init__(self, db, tenant_id: int): _ = db captured["tenant_id"] = tenant_id async def process_ingress(self, **kwargs): captured["ingress_kwargs"] = kwargs return { "handled": True, "status": "answered", "intent": "query.read_only", "direct_reply_required": True, "direct_reply_text": "ok", "rag_result": {"answer": "ok", "session_id": 12, "processing_stage": "qa"}, } monkeypatch.setattr(routes_query, "AIPMOKernelService", FakeKernel) request = RAGAskRequest( question="现在进展怎么样", asked_via="pmflow_manual", action_mode="read", message_intent="mixed", mixed_signal_part="新增任务确认流", consultant_mode_preference="clarify", ) current_user = SimpleNamespace(id=101, tenant_id=9) response = await routes_query.rag_ask( request=request, current_user=current_user, db=object(), tenant_id=9, ) assert response["code"] == 0 data = response["data"] assert data["answer"] == "ok" assert data["session_id"] == 12 ingress_kwargs = captured["ingress_kwargs"] assert isinstance(ingress_kwargs, dict) assert ingress_kwargs["source_system"] == "pmflow_manual" assert ingress_kwargs["message_text"] == "现在进展怎么样" rag_request_context = ingress_kwargs["ingress_options"]["rag_request_context"] assert rag_request_context["asked_via"] == "pmflow_manual" assert rag_request_context["action_mode"] == "read" assert rag_request_context["message_intent"] == "mixed" assert rag_request_context["mixed_signal_part"] == "新增任务确认流" assert rag_request_context["consultant_mode_preference"] == "clarify" project_operating_context = ingress_kwargs["ingress_options"]["project_operating_context"] assert project_operating_context["current_repo_scope"] assert project_operating_context["fit_scope"] == "current_project_and_repo" assert project_operating_context["decision_strategy"] == "answer_then_offer_plan" def test_ai_steward_router_does_not_expose_legacy_rag_ask_route() -> None: route_paths = {str(getattr(route, "path", "")).strip() for route in ai_steward_router.routes} assert "/ai-steward/rag/ask" not in route_paths def test_api_router_uses_single_prefix_for_rag_and_ai_steward() -> None: route_paths = {str(getattr(route, "path", "")).strip() for route in api_router.routes} if settings.api_route_profile == "lean" and "knowledge" not in settings.api_optional_route_groups: assert "/rag/ask" not in route_paths else: assert "/rag/ask" in route_paths assert "/rag/rag/ask" not in route_paths if settings.api_route_profile == "lean" and "ai_control" not in settings.api_optional_route_groups: assert "/ai-steward/feedback-center/summary" not in route_paths else: assert "/ai-steward/feedback-center/summary" in route_paths assert "/ai-steward/ai-steward/feedback-center/summary" not in route_paths @pytest.mark.asyncio async def test_rag_ask_route_uses_shared_contract_builder(monkeypatch) -> None: captured_kwargs: list[dict[str, object]] = [] captured_tenants: list[int] = [] def fake_build_answer_kwargs(request: RAGAskRequest, current_user) -> dict[str, object]: return { "question": request.question, "asked_by": int(current_user.id), "asked_via": "shared-builder", } monkeypatch.setattr(rag_ask_contract, "build_answer_question_kwargs", fake_build_answer_kwargs) monkeypatch.setattr(rag_ask_contract, "resolve_effective_tenant_id", lambda *_args, **_kwargs: 88) class FakeKernel: def __init__(self, db, tenant_id: int): _ = db captured_tenants.append(int(tenant_id)) async def process_ingress(self, **kwargs): captured_kwargs.append(dict(kwargs)) return {"handled": True, "status": "answered", "direct_reply_text": "ok", "rag_result": {"answer": "ok"}} monkeypatch.setattr(routes_query, "AIPMOKernelService", FakeKernel) response_rag = await routes_query.rag_ask( request=RAGAskRequest(question="q1"), current_user=SimpleNamespace(id=1, tenant_id=1), db=object(), tenant_id=1, ) assert response_rag["code"] == 0 assert captured_tenants == [88] assert len(captured_kwargs) == 1 forwarded = captured_kwargs[0] assert str(forwarded["source_message_id"]).startswith("rag-ask-") assert forwarded["source_system"] == "pmflow" assert forwarded["message_text"] == "q1" assert forwarded["sender"] == "PMFlow用户" assert forwarded["channel_ref"] is None assert forwarded["source_sender_open_id"] is None assert forwarded["captured_at"] is None assert forwarded["chat_type"] == "manual" assert forwarded["project_hint_id"] is None assert forwarded["ingress_options"]["rag_request_context"] == { "question": "q1", "asked_by": 1, "asked_via": "shared-builder", } assert forwarded["ingress_options"]["project_operating_context"]["current_project_id"] is None assert forwarded["ingress_options"]["project_operating_context"]["current_repo_scope"] assert forwarded["ingress_options"]["project_operating_context"]["fit_scope"] == "current_project_and_repo" assert forwarded["ingress_options"]["project_operating_context"]["decision_strategy"] == "answer_then_offer_plan" def test_build_rag_response_payload_preserves_rag_fields() -> None: payload = rag_ask_contract.build_rag_response_payload( ingress_result={ "status": "answered", "intent": "query.read_only", "work_item_id": 321, "ai_trace_id": "ai-pmo-321", "direct_reply_required": True, "direct_reply_text": "fallback", "rag_result": { "answer": "final answer", "session_id": 18, "sources": [{"source": "snapshot"}], "answer_mode": "retrieval_deep", "grounding_status": "grounded", "policy_version": "gr-v2.1|routing-v3|prompt:feishu_direct", "citations": [{"source_type": "doc", "source_id": "doc-1"}], "sources_used": [{"source_type": "doc", "source_id": "doc-1"}], "evidence_pack": { "summary": { "conclusion": "final answer", "evidence": ["doc-1"], "uncertainties": [], }, "tool_trace": [{"tool": "vector_search"}], }, "processing_stage": "qa", }, } ) assert payload["answer"] == "final answer" assert payload["session_id"] == 18 assert payload["work_item_id"] == 321 assert payload["trace_id"] == "ai-pmo-321" assert payload["reply_required"] is True assert payload["reply_text"] == "fallback" assert payload["answer_mode"] == "retrieval_deep" assert payload["grounding_status"] == "grounded" assert payload["policy_version"] == "gr-v2.1|routing-v3|prompt:feishu_direct" assert payload["citations"] == [{"source_type": "doc", "source_id": "doc-1"}] assert payload["sources_used"] == [{"source_type": "doc", "source_id": "doc-1"}] assert payload["evidence_pack"]["summary"]["conclusion"] == "final answer" assert payload["evidence_pack"]["tool_trace"] == [{"tool": "vector_search"}] def test_answer_payload_normalizes_source_registry_fields_into_citations() -> None: class _FakeService(AdaptiveRAGAnswerPayloadMixin): @staticmethod def _to_float(value, default: float) -> float: try: return float(value) except Exception: return float(default) service = _FakeService() citations = service._build_citations_from_sources( [ { "source": "pmflow", "label": "PMFlow[task#88]", "score": 0.91, "metadata": { "record_type": "task", "record_id": 88, "project_id": 17, "target_url": "/projects/17/tasks/88", "updated_at": "2026-04-01T09:30:00+08:00", }, }, { "source": "doc", "label": "Document[架构说明]", "score": 0.73, "metadata": { "doc_id": 3, "chunk_index": 2, "file_url": "https://docs.example.com/spec", }, }, ] ) assert citations == [ { "source_type": "pmflow", "source_label": "PMFlow[task#88]", "source_id": "task:88", "chunk_id": None, "target_url": "/projects/17/tasks/88", "project_id": "17", "updated_at": "2026-04-01T09:30:00+08:00", "permission_scope": "project", "source_priority": 100, "confidence": 0.91, }, { "source_type": "doc", "source_label": "Document[架构说明]", "source_id": "3", "chunk_id": "2", "target_url": "https://docs.example.com/spec", "project_id": None, "updated_at": None, "permission_scope": "tenant", "source_priority": 70, "confidence": 0.73, }, ] assert service._build_sources_used_from_citations(citations) == [ { "source_type": "pmflow", "source_label": "PMFlow[task#88]", "source_id": "task:88", "chunk_id": None, "target_url": "/projects/17/tasks/88", "project_id": "17", "updated_at": "2026-04-01T09:30:00+08:00", "permission_scope": "project", "source_priority": 100, }, { "source_type": "doc", "source_label": "Document[架构说明]", "source_id": "3", "chunk_id": "2", "target_url": "https://docs.example.com/spec", "project_id": None, "updated_at": None, "permission_scope": "tenant", "source_priority": 70, }, ] def test_build_rag_response_payload_adds_grounding_contract_defaults() -> None: payload = rag_ask_contract.build_rag_response_payload( ingress_result={ "status": "answered", "intent": "query.read_only", "direct_reply_text": "fallback", "rag_result": { "answer": "final answer", }, } ) assert payload["answer_mode"] == "retrieval_normal" assert payload["grounding_status"] == "partially_grounded" assert payload["policy_version"] == "retrieval-v2.1|routing:unknown|prompt:unknown" assert payload["citations"] == [] assert payload["sources_used"] == [] assert payload["evidence_pack"] == { "summary": { "conclusion": "final answer", "evidence": [], "uncertainties": [], }, "tool_trace": [], } def test_build_rag_response_payload_falls_back_to_top_level_consultant_dispatch() -> None: consultant_dispatch = { "domain": "project", "mode": "challenge", "risk_level": "high", "missing_fields": [], "allow_execute": False, "next_action": "请先拆分范围，再确认是否继续推进高风险动作。", } payload = rag_ask_contract.build_rag_response_payload( ingress_result={ "status": "answered", "intent": "project.create", "direct_reply_text": "当前建议先拆分范围。", "consultant_dispatch": consultant_dispatch, "rag_result": { "answer": "当前建议先拆分范围。", "processing_stage": "qa", }, } ) assert payload["consultant_dispatch"] == consultant_dispatch def test_build_rag_response_payload_does_not_inject_fake_answer_text() -> None: payload = rag_ask_contract.build_rag_response_payload( ingress_result={ "status": "answered", "intent": "query.read_only", "direct_reply_text": "", "rag_result": { "processing_stage": "qa", }, } ) assert payload["answer"] == "" def test_customer_direct_answer_handles_greeting_without_route_clarification() -> None: payload = rag_ask_contract.build_customer_direct_answer( RAGAskRequest(question="hi"), route_plan=_direct_route_plan(goal_kind="casual_chat", answer_contract="pmflow_product_help"), ) assert payload is not None assert "外部公开事实" not in payload["reply_text"] assert payload["processing_stage"] == "planner_direct" def test_customer_direct_answer_handles_identity_question_without_internal_retrieval() -> None: payload = rag_ask_contract.build_customer_direct_answer( RAGAskRequest(question="你是谁"), route_plan=_direct_route_plan(goal_kind="pmflow_product_help", answer_contract="pmflow_product_help"), ) assert payload is not None assert payload["reason_code"] == "planner_product_capability" assert "我是 PMFlow AI" in payload["answer"] assert "系统代码" in payload["answer"] assert payload["processing_stage"] == "planner_direct" @pytest.mark.parametrize( "question", [ "这个系统能干啥", "你能干嘛", "PMFlow 能解决什么问题？", "这个平台有什么功能", "你能帮我什么", ], ) def test_customer_direct_answer_handles_product_capability_questions_without_internal_retrieval(question: str) -> None: payload = rag_ask_contract.build_customer_direct_answer( RAGAskRequest(question=question), route_plan=_direct_route_plan(goal_kind="pmflow_product_help", answer_contract="pmflow_product_help"), ) assert payload is not None assert payload["reason_code"] == "planner_product_capability" assert "我是 PMFlow AI" in payload["answer"] assert "当前租户下的项目" in payload["answer"] assert payload["processing_stage"] == "planner_direct" def test_customer_payload_sanitizes_internal_source_and_prompt_leaks() -> None: payload = rag_ask_contract.sanitize_customer_rag_payload( { "answer": "模型调用失败：name 'PROMPT_RAG_PENDING_CLARIFICATION_SELECTION_SEMANTICS' is not defined", "sources": [{"source": "system", "label": "Session State", "score": 0.8, "metadata": {"trace_id": "t1"}}], "sources_used": [{"raw_source_label": "Project database"}], "tool_trace": [{"tool": "query_route_gateway"}], }, request=RAGAskRequest(question="hi"), current_user=SimpleNamespace(id=7, tenant_id=3), ) assert payload["reply_text"] == "这次回答生成失败，系统没有拿到可展示的业务结论。你可以换一种问法，或稍后重试。" assert payload["sources"][0]["label"] == "当前空间状态" assert payload["sources_used"][0]["display_label"] == "项目数据" assert "tool_trace" not in payload assert "trace_id" not in payload assert payload["customer_answer_policy"]["visibility"] == "customer" def test_packet_ad_customer_payload_keeps_project_operating_source_label() -> None: payload = rag_ask_contract.sanitize_customer_rag_payload( { "answer": "当前项目可以继续推进。", "reply_text": "当前项目可以继续推进。", "sources": [ {"source": "project_status", "label": "PMFlow 项目执行态证据", "score": 1}, {"source": "project_object_resolver", "label": "PMFlow 项目对象解析", "score": 1}, ], "sources_used": [ {"source_type": "project_status", "display_label": "PMFlow 项目执行态证据"}, ], "customer_evidence_packet": {"answer_type": "execution_guidance", "facts": [], "source_summary": []}, }, apply_presentation_contract=False, ) assert payload["sources"][0]["label"] == "项目运行态证据" assert payload["sources"][1]["label"] == "项目数据" assert payload["sources_used"][0]["display_label"] == "项目运行态证据" assert not any(item["label"] == "PMFlow 上下文" for item in payload["sources"]) @pytest.mark.asyncio async def test_rag_ask_route_returns_customer_safe_failure_for_empty_rag_answer(monkeypatch) -> None: class FakeKernel: def __init__(self, db, tenant_id: int): _ = db _ = tenant_id async def process_ingress(self, **kwargs): _ = kwargs return { "handled": True, "status": "answered", "intent": "query.read_only", "direct_reply_text": "", "rag_result": { "processing_stage": "qa", }, } monkeypatch.setattr(routes_query, "AIPMOKernelService", FakeKernel) response = await routes_query.rag_ask( request=RAGAskRequest(question="空回答测试"), current_user=SimpleNamespace(id=1, tenant_id=1), db=object(), tenant_id=1, ) assert getattr(response, "status_code", None) is None assert response["code"] == 0 assert response["message"] == "success" assert response["data"]["answer"] == "我现在没有生成足够明确的可展示回答。你可以换一种问法，或补充项目/对象范围后我再查。" assert response["data"]["answer_mode"] == "customer_deterministic" assert response["data"]["customer_answer_policy"]["visibility"] == "customer" assert response["data"]["answer_process_packet"]["steps"] == []