from __future__ import annotations from types import SimpleNamespace import pytest from app.services.pmflow_ai_agent_runtime.agent_goal_frame_runtime import _goal_frame_safe_conversation_context from app.services.pmflow_ai_agent_runtime.agent_step_planner import _planner_safe_conversation_context from app.services.pmflow_ai_operating_runtime.mission_service import _mission_safe_conversation_context from app.services.pmflow_ai_context_pack_service import ( PMFlowAIContextPackBuilder, context_pack_memory_claim_rows, context_pack_observability_packet, context_pack_thread_claim_rows, evaluate_context_replay_case, ) from app.services.pmflow_ai_context_memory_service import PMFlowAIContextMemoryService from app.services.pmflow_ai_thread_context_service import ( PMFlowAIThreadContextService, thread_context_claim_rows, update_thread_context_summary_metadata, ) def _message(index: int, role: str, content: str) -> SimpleNamespace: return SimpleNamespace( public_message_id=f"msg_{index}", role=role, message_index=index, run_public_id=f"run_{index}", sent_at=None, content=content, ) @pytest.mark.asyncio async def test_thread_context_pack_recovers_same_thread_purchase_discussion(monkeypatch: pytest.MonkeyPatch) -> None: service = PMFlowAIThreadContextService(db=SimpleNamespace(), tenant_id=12) messages = [ _message(1, "user", "AI采购平台怎么样"), _message( 2, "assistant", "AI采购平台通常用于智能寻源与供应商匹配、合同分析与合规审查、支出分析与预测。构建者应避免全能平台，先从发票自动比对或合同审查切入 MVP。", ), _message(3, "user", "最近广州天气怎么样"), _message(4, "assistant", "广州近期有降雨，适合调整线下安排。"), ] async def _fake_load_messages(*_args, **_kwargs): return messages monkeypatch.setattr(service, "_load_messages", _fake_load_messages) pack = await service.build_context_pack( thread=SimpleNamespace(id=26, public_thread_id="thread_71ade449d4b5"), user_question="你还记得你上面的对话有采购相关的内容吗？再给我详细说说是什么内容", ) relevant_text = "\n".join(str(item.get("content") or "") for item in pack["relevant_messages"]) assert "智能寻源与供应商匹配" in relevant_text assert "合同分析与合规审查" in relevant_text claim_text = "\n".join(row["content"] for row in thread_context_claim_rows(pack)) assert "发票自动比对" in claim_text context = {"thread_context_pack": pack} assert "thread_context_pack" in _mission_safe_conversation_context(context) assert "thread_context_pack" in _goal_frame_safe_conversation_context(context, question="继续说上面内容") assert "thread_context_pack" in _planner_safe_conversation_context(context, question="继续说上面内容") def test_unified_context_pack_budgets_and_preserves_relevant_thread_history() -> None: thread_pack = { "packet_version": "pmflow_ai_thread_context_pack.v1", "thread_id": "thread_purchase", "recent_messages": [ { "message_id": f"recent_{index}", "role": "user" if index % 2 else "assistant", "message_index": index, "content": f"普通上下文 {index} " + ("产品背景 " * 80), } for index in range(1, 18) ], "relevant_messages": [ { "message_id": "relevant_procurement", "role": "assistant", "message_index": 2, "content": "前文已讨论 AI 采购平台：智能寻源、供应商匹配、合同分析、支出预测，MVP 可先从发票自动比对切入。", } ], "rolling_summary": { "durable_points": [ { "role": "assistant", "source_message_indexes": [2], "summary": "AI 采购平台的核心价值是采购流程自动化与成本/合规风险控制。", } ] }, } conversation_context = { "session_id": "qa_1", "conversation_scope": "space_ai", "thread_context_pack": thread_pack, "context_capsules": [ { "capsule_type": "pmflow_ai_object_ledger", "current_focus_object": {"object_type": "project", "object_id": "p1", "display_name": "AI采购平台"}, "recent_objects": [{"object_type": "project", "object_id": "p1"}], } ], } tenant_project_context = { "tenant_id": 1, "user_id": 2, "surface_context": { "visible_objects": [{"object_type": "project", "object_id": "p1", "display_name": "AI采购平台"}], "space_runtime_context": { "workbench": { "projects": [{"id": "p1"}], "pending_confirmations": [{"id": "c1"}], }, "space_operating_packet": {"summary": "visible " * 500}, }, }, } context_pack = PMFlowAIContextPackBuilder(budget_chars=8500).build( user_question="你还记得上面采购平台相关内容吗？", thread_context_pack=thread_pack, conversation_context=conversation_context, tenant_project_context=tenant_project_context, ) assert context_pack["packet_version"] == "pmflow_ai_context_pack.v1" assert context_pack["current_user_message"]["content"] == "你还记得上面采购平台相关内容吗？" assert context_pack["budget_report"]["max_chars"] == 8500 assert any(row["section"] == "thread_context" and row["status"] in {"included", "compressed"} for row in context_pack["budget_report"]["sections"]) relevant_text = "\n".join(str(item.get("content") or "") for item in context_pack["thread_context"]["relevant_messages"]) assert "发票自动比对" in relevant_text claim_text = "\n".join(row["content"] for row in context_pack_thread_claim_rows(context_pack)) assert "智能寻源" in claim_text safe_context = {"context_pack": context_pack, "thread_context_pack": thread_pack} assert "context_pack" in _mission_safe_conversation_context(safe_context) assert "context_pack" in _goal_frame_safe_conversation_context(safe_context, question="继续上文") assert "context_pack" in _planner_safe_conversation_context(safe_context, question="继续上文") def test_unified_context_pack_includes_layered_memory_and_replay_observability() -> None: memory_context_pack = { "packet_version": "pmflow_ai_layered_memory_pack.v1", "scope_refs": {"thread": ["thread:abc"], "user": ["user:2"], "workspace": ["workspace:space"]}, "memories": [ { "memory_id": "ctxmem_1", "memory_scope": "thread", "scope_ref": "thread:abc", "summary": "前文已经把 AI 采购平台定位为从发票自动比对切入，后续要补供应商匹配和合同分析。", "durable_facts": [{"summary": "MVP 切口是发票自动比对。"}], "open_loops": [{"summary": "还缺预算、供应商类型和审批流程信息。"}], "confidence": "high", }, { "memory_id": "ctxmem_2", "memory_scope": "user", "scope_ref": "user:2", "summary": "用户要求不要用硬编码关键词或枚举凑场景，要做终局架构。", "preferences": [{"summary": "少请示，直接查主链并推进。", "confidence": "high"}], "confidence": "high", }, ], } context_pack = PMFlowAIContextPackBuilder(budget_chars=9000).build( user_question="继续上面采购平台这个方向", thread_context_pack={}, conversation_context={"session_id": "qa_1", "conversation_scope": "space_ai"}, tenant_project_context={"tenant_id": 1, "user_id": 2, "surface_context": {}}, memory_context_pack=memory_context_pack, ) memory_text = "\n".join(row["content"] for row in context_pack_memory_claim_rows(context_pack)) assert "发票自动比对" in memory_text assert "不要用硬编码关键词" in memory_text observability = context_pack_observability_packet(context_pack) assert observability["memory"]["memory_count"] == 2 assert "thread" in observability["memory"]["scopes"] replay = evaluate_context_replay_case( context_pack=context_pack, expected_evidence=["发票自动比对", "终局架构"], ) assert replay["passed"] is True @pytest.mark.asyncio async def test_context_memory_compactor_falls_back_without_runtime_generator() -> None: service = PMFlowAIContextMemoryService(db=SimpleNamespace(), tenant_id=1, generator=SimpleNamespace()) result = await service._compact_turn( user_message="AI采购平台怎么样", assistant_answer="建议先从发票自动比对切入，再验证供应商匹配和合同分析。", previous_memory_pack={}, context_pack={}, user_message_index=1, assistant_message_index=2, ) assert result["compaction_source"] == "deterministic_fallback" assert "thread" in result["memories"] assert "发票自动比对" in result["memories"]["thread"]["summary"] def test_thread_context_summary_metadata_compacts_turns_without_scene_keywords() -> None: metadata = update_thread_context_summary_metadata( thread_metadata={}, user_message="AI采购平台怎么样", assistant_answer="AI采购平台通常用于智能寻源与供应商匹配、合同分析与合规审查。", user_message_index=1, assistant_message_index=2, ) summary = metadata["thread_context_summary"] assert summary["summary_version"] == "pmflow_ai_thread_summary.v1" assert summary["covered_message_count"] == 2 assert len(summary["durable_points"]) == 2 assert "智能寻源与供应商匹配" in summary["durable_points"][1]["summary"]