from __future__ import annotations import asyncio import inspect import json from pathlib import Path from types import SimpleNamespace import pytest from app.api import space_api from app.api.rag_ask_contract import sanitize_customer_rag_payload from app.schemas.pmflow_ai_agent_runtime import ( PMFlowAIAgentActionCandidateLedgerEntry, PMFlowAIAgentAnswerDraft, PMFlowAIAgentAnswerIntentPlan, PMFlowAIAgentBoundary, PMFlowAIAgentClaim, PMFlowAIAgentCriticResult, PMFlowAIAgentEvidenceNormalization, PMFlowAIAgentFactLedgerEntry, PMFlowAIAgentGapLedgerEntry, PMFlowAIAgentGoalFrame, PMFlowAIAgentObservation, PMFlowAIAgentRunBudget, PMFlowAIAgentSourceNormalization, PMFlowAIAgentStepProposal, PMFlowAIAgentTerminalAnswerState, PMFlowAIAgentWorkspace, ) from app.schemas.pmflow_ai_operating_runtime import ( PMFlowAIAgentAssignment, PMFlowAIAgentWorkProduct, PMFlowAIAutonomyDecision, PMFlowAIEvidenceLedgerEntry, PMFlowAIOperatingMission, PMFlowAIOperatingPlan, PMFlowAIProjectWorldFacet, PMFlowAIProjectWorldStatePacket, ) from app.schemas.rag import RAGAskRequest from app.services.pmflow_agent_runtime import ( PMFlowAgentEnvironmentService, PMFlowAgentRuntimeService, PMFlowAgentSpecialistRuntimeService, ) from app.services.pmflow_agent_runtime.service import _prepared_actions_from_agent_payload from app.services.pmflow_ai_agent_runtime import PMFlowAIAgentAnswerRuntime, PMFlowAIAgentRunService from app.services.pmflow_ai_agent_runtime import agent_answer_runtime as agent_answer_runtime_module from app.services.pmflow_ai_agent_runtime import agent_evidence_normalizer as agent_evidence_normalizer_module from app.services.pmflow_ai_agent_runtime.agent_answer_runtime import ( build_agent_terminal_answer_state, critique_agent_customer_answer, repair_agent_customer_answer, render_agent_terminal_state_fallback, ) from app.services.pmflow_ai_agent_runtime.agent_run_service import ( _candidate_source_urls_for_fetch, _context_first_proposal_gate, _proposal_with_goal_frame_runtime_hints, _rank_source_candidates, _recovery_proposal_from_critic, ) from app.services.pmflow_ai_agent_runtime.agent_evidence_normalizer import ( PMFlowAIAgentEvidenceNormalizer, _should_use_deterministic_external_normalizer, apply_evidence_normalization, fallback_normalize_observation, ) from app.services.pmflow_ai_agent_runtime.agent_goal_frame_runtime import ( _fallback_goal_frame, _goal_frame_safe_conversation_context, _repair_goal_frame_for_open_semantics, ) from app.services.pmflow_ai_agent_runtime.agent_llm_budget import ( _effective_foreground_wait_seconds, resolve_agent_llm_runtime_profile, ) from app.services.pmflow_ai_agent_runtime.agent_research_contract import ( enrich_external_search_proposal, goal_frame_has_explicit_external_probe_contract, goal_frame_requires_external_evidence, research_coverage_gap, ) from app.services.pmflow_ai_agent_runtime.agent_payload_gate import build_agent_answer_canvas from app.services.pmflow_ai_agent_runtime.agent_state_critic import PMFlowAIAgentStateCritic, _enforce_critic_contract from app.services.pmflow_ai_agent_runtime.agent_step_planner import PMFlowAIAgentStepPlanner from app.services.pmflow_ai_agent_runtime.agent_workspace import ( agent_workspace_from_context_capsules, upsert_agent_workspace_capsule, workspace_from_conversation_context, ) from app.services.pmflow_ai_agent_runtime.agent_workspace_ledger import record_critic_ledgers from app.services.pmflow_ai_agent_runtime.agent_tool_registry import ( AGENT_TOOL_CONTRACTS, PMFlowAIAgentToolRegistry, _AgentRuntimeExternalPlanner, _rank_external_evidence, ) from app.services.external_query_service.schemas import ExternalEvidenceItem from app.services.pmflow_ai_operating_runtime import PMFlowAIOperatingRunService, PMFlowAISpecialistAgentService from app.services.pmflow_ai_operating_runtime.agent_registry import PMFlowAIOperatingAgentRegistry from app.services.pmflow_ai_operating_runtime import specialist_agents as operating_specialist_agents_module from app.services.pmflow_ai_operating_runtime.run_service import ( _claim_texts, _compose_research_operating_answer, _external_source_rows, ) from app.services.prompt_runtime_service import PromptRuntimeRenderResult @pytest.mark.asyncio async def test_agent_runtime_step1_builds_run_workspace_and_process_packet() -> None: events = [] async def sink(event: dict) -> None: events.append(event) payload = await PMFlowAIAgentRunService( object(), tenant_id=12, planner=_FakePlanner("inspect_space_collections"), tool_registry=_FakeToolRegistry(), critic=_FakeCritic(ready=True), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="帮我分析下用AI来协助采购监督的系统怎么样"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, started_at=None, event_sink=sink, ) assert payload["processing_stage"] == "agent_runtime" assert payload["answer_mode"] == "agent_runtime" assert payload["agent_run_packet"]["packet_version"] == "pmflow_ai_agent_run.v1" assert payload["agent_workspace"]["workspace_version"] == "pmflow_ai_agent_workspace.v1" assert payload["agent_workspace"]["objective_ledger"] assert payload["agent_workspace"]["fact_ledger"] assert payload["agent_workspace"]["goal_frame"]["packet_version"] == "pmflow_ai_agent_goal_frame.v1" assert payload["agent_workspace"]["claim_ledger"] assert payload["agent_workspace"]["evidence_normalizations"] assert payload["agent_workspace"]["evidence_normalizations"][0]["packet_version"] == "pmflow_ai_agent_evidence_normalization.v1" assert payload["agent_workspace"]["attempt_ledger"] assert payload["agent_workspace"]["action_ledger"] assert payload["agent_workspace"]["deliverable_state"]["status"] assert payload["deliverable_packet"]["packet_version"] == "pmflow_ai_agent_deliverable_packet.v1" assert payload["deliverable_packet"]["deliverable_kind"] == "status_brief" assert payload["agent_workspace"]["deliverable_state"]["deliverable_kind"] == payload["deliverable_packet"]["deliverable_kind"] assert payload["answer_canvas"]["deliverable_kind"] == payload["deliverable_packet"]["deliverable_kind"] assert all(isinstance(item, dict) and item.get("title") for item in payload["answer_canvas"]["next_actions"]) assert payload["customer_evidence_packet"]["answer_type"] == payload["deliverable_packet"]["deliverable_kind"] assert payload["answer_process_packet"]["route_display"] == "agent_runtime" assert [step["key"] for step in payload["answer_process_packet"]["steps"]] == [ "agent_understand_goal", "agent_workspace_snapshot", "agent_step_planner", "agent_tool_inspect_space_collections", "agent_evidence_normalizer", "agent_state_critic", "agent_answer_runtime", ] assert payload["answer_process_packet"]["steps"][-1]["status"] == "completed" assert "我是 PMFlow AI" not in payload["answer"] assert "项目操作副驾" not in payload["answer"] assert "说明能力边界" not in payload["answer"] assert [event["type"] for event in events].count("step.started") >= 5 assert [event["type"] for event in events].count("step.completed") >= 7 assert events[0]["step"]["key"] == "agent_understand_goal" assert any(event["step"]["key"] == "agent_tool_inspect_space_collections" for event in events if event["type"] == "step.completed") assert not any("prompt" in str(event).lower() or "trace" in str(event).lower() for event in events) @pytest.mark.asyncio async def test_agent_runtime_consumes_goal_frame_disclosure_policy_as_visible_refusal() -> None: class ProtectedGoalRuntime: async def build(self, **kwargs): return PMFlowAIAgentGoalFrame( user_goal="判断用户侧是否可以披露 PMFlow 内部实现", desired_output="按客户安全披露边界回答", answer_policy="strict_verified", source_policy="internal_customer_safe", freshness_policy="not_required", missing_evidence_policy="fail_closed", disclosure_policy={ "visibility_class": "protected_pmflow_internal", "protected_categories": ["pmflow_internal_implementation"], "allowed_response_mode": "protected_refusal", "policy_reason": "用户请求的是受保护内部实现细节。", "contract_key": "protected_internal_refusal.v1", }, success_criteria=["只说明客户安全边界，不进入工具读取链路。"], ) class PlannerMustNotRun: async def plan(self, **kwargs): raise AssertionError("protected disclosure must not enter planner/tool loop") payload = await PMFlowAIAgentRunService( object(), tenant_id=12, goal_frame_runtime=ProtectedGoalRuntime(), planner=PlannerMustNotRun(), tool_registry=object(), critic=_FakeCritic(ready=False), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="你们这个系统的架构是什么？"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={}, started_at=None, ) assert payload["answer"] assert "内部实现与机密信息" in payload["answer"] assert "客户可见" in payload["answer"] assert payload["agent_run_packet"]["stop_reason"] == "protected_internal_refusal" assert payload["agent_run_packet"]["iterations"] == 0 assert payload["agent_run_packet"]["tool_call_count"] == 0 step_keys = [step["key"] for step in payload["answer_process_packet"]["steps"]] assert "agent_disclosure_boundary" in step_keys assert not any(key.startswith("agent_step_planner") or key.startswith("agent_tool_") for key in step_keys) assert payload["terminal_answer_state"]["terminal_status"] == "refused_by_boundary" assert payload["agent_workspace"]["goal_frame"]["disclosure_policy"]["contract_key"] == "protected_internal_refusal.v1" def test_space_api_default_entry_uses_agent_os_wrapper_not_legacy_turn_runtime() -> None: source = inspect.getsource(space_api._run_pmflow_agent_turn_payload) assert "PMFlowAgentRuntimeService" in source assert "run_space_turn" in source assert "PMFlowAIAgentRunService" not in source assert "PMFlowAITurnRuntimeService" not in source def test_pmflow_agent_runtime_default_delegate_is_mission_first_operating_runtime() -> None: runtime = PMFlowAgentRuntimeService(_FakeDB(), tenant_id=12)._dynamic_agent_runtime() assert isinstance(runtime, PMFlowAIOperatingRunService) def test_agent_runtime_tool_environment_has_no_scenario_specific_opportunity_tool() -> None: assert "run_opportunity_discovery_agents" not in AGENT_TOOL_CONTRACTS assert "discover_project_opportunities" not in str(AGENT_TOOL_CONTRACTS) assert "run_market_opportunity_review" not in str(AGENT_TOOL_CONTRACTS) assert "inspect_pending_confirmations" not in AGENT_TOOL_CONTRACTS assert "inspect_pending_confirmations" not in str(AGENT_TOOL_CONTRACTS) assert "inspect_attention_queue" in AGENT_TOOL_CONTRACTS assert "inspect_project_execution_package" in AGENT_TOOL_CONTRACTS def test_context_first_gate_does_not_let_empty_saved_ideas_override_prior_context() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="继续之前方案") workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="context_assistant_prior_proposal", claim_text="上一轮已经形成一个可验证的最小闭环方案。", source_tool="context_evidence_absorber", allowed_to_state=True, ) ] proposal = PMFlowAIAgentStepProposal( action="tool", tool_name="inspect_saved_ideas", rationale="读取保存草稿。", ) gated = _context_first_proposal_gate( proposal=proposal, workspace=workspace, context_evidence_packet={ "continuation_obligations": [{"claim_id": "ctxclaim_1"}], "current_user_message": "继续之前方案", }, ) assert gated.action == "answer" assert gated.tool_name is None assert gated.tool_input["context_first_gate"] == "answer_from_context_evidence" def test_space_thread_persistence_prefers_canonical_customer_answer_over_context_scaffold() -> None: answer = space_api._canonical_customer_answer_from_payload( { "answer": "我会把它收敛成产品蓝图来推进：同一对话历史第 2 条 assistant 消息提到：上一版方案；列出功能模块：列出功能模块：", "deliverable_packet": { "answer_text": "这版应继续沿用上一轮已确认的三个最小闭环，并输出一版能直接验证的新版方案。\n\n- 先验证数据质量\n- 再验证智能建议\n- 最后验证流程嵌入" }, "answer_canvas": { "direct_answer": "这版应继续沿用上一轮已确认的三个最小闭环，并输出一版能直接验证的新版方案。" }, } ) assert "同一对话历史第" not in answer assert "三个最小闭环" in answer def test_goal_frame_repair_adds_dynamic_objective_contract_without_business_enum() -> None: frame = _repair_goal_frame_for_open_semantics( frame=PMFlowAIAgentGoalFrame( user_goal="找到低门槛、容易变现的 AI 应用方向", desired_output="给出可比较、可验证的方向建议", evidence_requirements=["确认哪些判断需要公开证据支撑"], success_criteria=["答案要能支撑用户选择优先做哪个方向"], ), question="有哪些容易变现的AI应用可以做呢？", ) assert frame.answer_objective["user_success"] assert frame.dynamic_rubric assert frame.evidence_lanes assert frame.latency_policy["max_tool_calls"] assert frame.deliverable_shape["primary_shape"] serialized = str(frame.model_dump(mode="json", exclude_none=True)) assert "客服" not in serialized assert "招聘" not in serialized assert "财务" not in serialized def test_agent_goal_frame_fallback_prefers_customer_safe_context() -> None: frame = _fallback_goal_frame(question="现在有哪些项目了") assert frame.source_policy == "internal_customer_safe" assert frame.freshness_policy == "not_required" assert not any("公开来源" in item for item in frame.evidence_requirements[:1]) def test_agent_step_planner_fallback_does_not_turn_evidence_requirements_into_external_search() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="现在有哪些项目了") workspace.goal_frame = _fallback_goal_frame(question="现在有哪些项目了") proposal = PMFlowAIAgentStepPlanner._fallback_plan( question="现在有哪些项目了", workspace=workspace, reason="planner_generation_failed", ) assert proposal.tool_name == "inspect_space_collections" assert "external_search" not in str(proposal.model_dump(mode="json", exclude_none=True)) def test_goal_frame_keeps_public_source_policy_when_probe_contract_exists() -> None: frame = PMFlowAIAgentGoalFrame( user_goal="推进一个开放产品判断", desired_output="给出带公开证据的判断。", answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", interpretation_branches=[ { "branch_id": "public_probe", "meaning": "需要用公开来源验证的解释分支", "evidence_probe_queries": ["representative public products official features"], } ], ) repaired = _repair_goal_frame_for_open_semantics(frame=frame, question="推进一个开放产品判断") assert repaired.source_policy == "credible_public_sources" assert goal_frame_has_explicit_external_probe_contract(repaired) is True assert goal_frame_requires_external_evidence(repaired) is True def test_goal_frame_respects_public_source_policy_without_probe_contract() -> None: frame = PMFlowAIAgentGoalFrame( user_goal="推进一个开放产品想法", desired_output="先给方向性判断。", answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", evidence_lanes=[ { "lane_id": "reasoning_lane", "lane": "产品价值假设", "purpose": "整理用户想法中的可验证方向。", "minimum_useful_signal": "可形成受控假设。", } ], ) repaired = _repair_goal_frame_for_open_semantics(frame=frame, question="推进一个开放产品想法") assert repaired.source_policy == "credible_public_sources" assert goal_frame_requires_external_evidence(repaired) is True @pytest.mark.asyncio async def test_agent_step_planner_open_reasoning_probe_contract_must_search() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="推进一个开放产品判断") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="推进一个开放产品判断", desired_output="给出带公开证据的判断。", answer_policy="directional_answer", source_policy="open_reasoning", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", interpretation_branches=[ { "branch_id": "public_probe", "meaning": "需要用公开来源验证的解释分支", "evidence_probe_queries": ["representative public products official features"], } ], ) proposal = await PMFlowAIAgentStepPlanner(db=object(), tenant_id=12, generator=object()).plan( question="推进一个开放产品判断", workspace=workspace, conversation_context={}, tenant_project_context={}, ) assert proposal.action == "tool" assert proposal.tool_name == "external_search" assert "representative public products official features" in proposal.tool_input["search_queries"] @pytest.mark.asyncio async def test_agent_state_critic_continues_when_public_obligation_unattempted() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="推进一个开放产品判断") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="推进一个开放产品判断", desired_output="给出带公开证据的判断。", answer_policy="directional_answer", source_policy="open_reasoning", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", interpretation_branches=[ { "branch_id": "public_probe", "meaning": "需要用公开来源验证的解释分支", "evidence_probe_queries": ["representative public products official features"], } ], ) result = await PMFlowAIAgentStateCritic(db=object(), tenant_id=12, generator=object()).review( question="推进一个开放产品判断", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="answer", tool_input={"source_policy": "open_reasoning"}), observation=None, ) assert result.ready_to_answer is False assert result.should_continue is True assert result.retry_would_change_state is True assert result.next_step_hint == "select_next_read_tool" assert any("公开来源证据" in item for item in result.missing_state) def test_space_context_normalizer_preserves_complete_collection_inventory() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="现在有哪些项目了") normalization = fallback_normalize_observation( question="现在有哪些项目了", workspace=workspace, observation=PMFlowAIAgentObservation( source="inspect_space_collections", observation_type="space_context", summary="已读取客户安全的空间集合摘要。", facts=[ { "type": "project_collection_summary", "count": 3, "titles": ["项目 A", "项目 B", "项目 C"], } ], confidence="high", relevance_to_goal="high", ), ) claim = normalization.usable_claims[0] terminal_state = PMFlowAIAgentTerminalAnswerState( user_goal="现在有哪些项目了", terminal_status="answerable", claims=[claim], ) assert "当前可见项目共 3 个" in claim.claim_text assert all(title in claim.claim_text for title in ["项目 A", "项目 B", "项目 C"]) assert "missing_collection_inventory_items" in critique_agent_customer_answer( terminal_state=terminal_state, answer="当前可见项目共 3 个：\n- 项目 A\n- 项目 B", ) assert "项目 C" in repair_agent_customer_answer(terminal_state=terminal_state) @pytest.mark.asyncio async def test_agent_runtime_repairs_external_tool_when_source_contract_is_internal() -> None: payload = await PMFlowAIAgentRunService( object(), tenant_id=12, planner=_FakePlanner("external_search"), tool_registry=_FakeToolRegistry(), critic=_FakeCritic(ready=True), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="当前空间状态"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={ "surface_context": {"visible_objects": [{"object_ref": "project:141"}]}, "query_route": { "route_mode": "internal", "effective_source_policy": "internal", "query_profile": {"needs_private_context": True, "needs_freshness": False}, }, }, started_at=None, ) step_keys = [step["key"] for step in payload["answer_process_packet"]["steps"]] assert "agent_tool_inspect_space_collections" in step_keys assert "agent_tool_external_search" not in step_keys assert payload["agent_workspace"]["action_ledger"][-1]["tool_name"] == "inspect_space_collections" def test_agent_tool_registry_dispatches_through_registered_handlers() -> None: registry = PMFlowAIAgentToolRegistry(db=object(), tenant_id=12) source = inspect.getsource(PMFlowAIAgentToolRegistry.execute) assert set(registry._handlers) == set(AGENT_TOOL_CONTRACTS) assert "if tool_name ==" not in source assert "handler = self._handlers.get(tool_name)" in source def test_pmflow_ai_thread_message_projection_hides_raw_run_internals() -> None: payload = { "answer": "这是用户应该看到的回答。", "agent_os_run_id": "run_demo", "answer_process_packet": { "steps": [ { "key": "workspace_binding", "label": "绑定工作区", "status": "completed", "display_summary": "已读取当前空间。", "duration_ms": 18, }, { "key": "final_answer", "label": "step_plan_abc123", "status": "passed", "display_summary": "target_ref final_answer passed", }, ] }, "agent_os_artifacts": [ {"artifact_id": "art_1", "kind": "final_answer", "title": "PMFlow AI 回答", "summary": "这是用户应该看到的回答。"}, {"artifact_id": "art_2", "kind": "research_note", "title": "推进建议", "summary": "先收口最高优先级动作。"}, ], "agent_os_verifications": [{"verification_id": "verification_1", "target_ref": "step_plan_abc123", "result": "passed"}], } message_payload = space_api._pmflow_ai_assistant_message_payload(payload) assert message_payload["work_log"][0]["label"] == "绑定工作区" assert message_payload["work_log"][0]["status_label"] == "已完成" assert len(message_payload["work_log"]) == 1 assert message_payload["artifacts"] == [] assert "verifications" not in message_payload assert "step_plan_abc123" not in str(message_payload) assert "passed" not in str(message_payload) def test_pmflow_ai_thread_message_projection_returns_visible_and_total_work_log_counts() -> None: payload = { "answer": "已经整理完成。", "answer_process_packet": { "steps": [ { "key": f"step_{index}", "label": f"读取材料 {index}", "status": "completed", "display_summary": f"已完成第 {index} 步。", } for index in range(1, 20) ] }, } message_payload = space_api._pmflow_ai_assistant_message_payload(payload) assert len(message_payload["work_log"]) == 12 assert len(message_payload["work_log_full"]) == 19 assert message_payload["work_log_meta"]["visible_step_count"] == 12 assert message_payload["work_log_meta"]["total_step_count"] == 19 assert message_payload["work_log_meta"]["truncated"] is True def test_pmflow_ai_message_view_backfills_full_work_log_from_run_steps() -> None: row = SimpleNamespace( public_message_id="msg_1", role="assistant", content="已经完成本轮处理。", run_public_id="run_1", qa_record_id=None, sent_at=None, message_payload={ "work_log": [ { "id": f"stored_{index}", "label": f"已存步骤 {index}", "summary": "历史投影里只有前 12 条。", "status": "done", } for index in range(1, 13) ], }, ) run_steps = [ { "id": f"step_{index}", "label": f"真实运行步骤 {index}", "summary": f"已执行第 {index} 步。", "status": "completed", } for index in range(1, 20) ] view = space_api._pmflow_ai_message_view(row, run_steps=run_steps) assert view["work_log_meta"]["visible_step_count"] == 12 assert view["work_log_meta"]["total_step_count"] == 19 assert view["work_log_meta"]["truncated"] is True assert len(view["work_log"]) == 12 assert len(view["work_log_full"]) == 19 def test_pmflow_ai_message_view_keeps_public_source_url_with_underscore_slug() -> None: answer = ( "基于当前可核验的公开材料，近半年国家医保政策在药品目录管理上呈现出更强的监测与保供导向。" "来源包括[国家医保局政策材料](https://www.nhsa.gov.cn/art/2026/5/14/art_14_18920.html?content_7050533)。" ) row = SimpleNamespace( public_message_id="msg_1", role="assistant", content=answer, run_public_id="run_1", qa_record_id=None, sent_at=None, message_payload={"answer": "", "reply_text": "", "work_log": []}, ) view = space_api._pmflow_ai_message_view(row, run_steps=[]) assert view["content"] == answer assert "content_7050533" in view["content"] message_payload = space_api._pmflow_ai_assistant_message_payload({"answer": answer}) assert message_payload["answer"] == answer def test_pmflow_ai_message_view_keeps_public_prompt_ai_business_language() -> None: answer = ( "AI漫剧行业在2026年初已进入爆发式增长阶段，核心特征是生产范式从传统拍摄转向" "“Prompt-AI生成-合成”的自动化流程。这一转变显著降低了制作门槛与成本。" ) row = SimpleNamespace( public_message_id="msg_1", role="assistant", content=answer, run_public_id="run_1", qa_record_id=None, sent_at=None, message_payload={"answer": "", "reply_text": "", "work_log": []}, ) view = space_api._pmflow_ai_message_view(row, run_steps=[]) assert view["content"] == answer message_payload = space_api._pmflow_ai_assistant_message_payload({"answer": answer}) assert message_payload["answer"] == answer def test_pmflow_ai_message_view_hides_inline_artifact_cards() -> None: row = SimpleNamespace( public_message_id="msg_1", role="assistant", content="已经给出政策动态分析。", run_public_id="run_1", qa_record_id=None, sent_at=None, message_payload={ "answer": "已经给出政策动态分析。", "artifacts": [ { "id": "art_duplicate", "title": "分析下近半年国家医保的政策动态", "summary": "分析下近半年国家医保的政策动态", "status_label": "已完成", }, { "id": "art_meaningful", "title": "公开材料结论", "summary": "公开材料显示医保目录管理更强调监测、基金安全和供应保障。", "status_label": "已完成", }, ], }, ) view = space_api._pmflow_ai_message_view(row, run_steps=[]) assert view["artifacts"] == [] def test_pmflow_ai_thread_message_projection_hides_deliverable_artifacts_and_unsafe_actions() -> None: payload = { "answer": "你当前最该处理的是确认变更 2，进入变更审查页后确认或退回。", "answer_canvas": { "direct_answer": "你当前最该处理的是确认变更 2，进入变更审查页后确认或退回。", "action_candidates": [ { "title": "确认变更 2", "summary": "打开变更审查页处理这条待确认项。", "href": "/space/change-review?projectId=141", } ], }, "agent_os_artifacts": [ { "artifact_id": "artifact_bad", "kind": "agent_deliverable", "title": "明确儿童项目下一步", "summary": "GoalFrame 明确限定 project:141 和 approval:21。", }, { "artifact_id": "artifact_report", "kind": "customer_report", "title": "审查说明", "summary": "当前最高优先级动作需要你本人确认。", }, ], "agent_os_prepared_action_packets": [ { "action_id": "action_good", "source": "deliverable_action_candidate", "title": "确认变更 2", "summary": "打开变更审查页处理这条待确认项。", "href": "/space/change-review?projectId=141", }, { "action_id": "action_no_href", "source": "deliverable_action_candidate", "title": "修改后再审 3", "summary": "没有可跳转目标的动作不应展示成按钮。", }, { "action_id": "action_internal", "source": "deliverable_action_candidate", "title": "确认变更 2；对象：重析；依据：GoalFrame top_next_actions[0]", "href": "/space/change-review?projectId=141", }, ], } message_payload = space_api._pmflow_ai_assistant_message_payload(payload) assert message_payload["artifacts"] == [] assert message_payload["actions"] == [ { "id": "action_good", "label": "确认变更 2", "summary": "打开变更审查页处理这条待确认项。", "href": "/space/change-review?projectId=141", } ] assert "GoalFrame" not in str(message_payload) assert "approval:21" not in str(message_payload) def test_pmflow_ai_thread_message_projection_preserves_operating_answer_and_confirmation_action_type() -> None: payload = { "answer": "这轮先继续补公开研究来源，确认前不会写项目记忆或发起执行。", "reply_text": "这轮先继续补公开研究来源，确认前不会写项目记忆或发起执行。", "processing_stage": "multi_agent_operating_runtime", "answer_canvas": { "direct_answer": "这轮结果里混入了内部运行信息，我先不展示那段内容。", "action_candidates": [], }, "agent_os_confirmation_requests": [ { "confirmation_id": "creq_test", "title": "确认准备动作：继续补研究", "summary": "确认后只继续做只读研究补证。", "action_label": "继续补研究", "action_type": "continue_research", "can_execute_after_confirmation": True, "action_payload": {"research_recovery": {"queries": ["2026 market report"]}}, } ], } message_payload = space_api._pmflow_ai_assistant_message_payload(payload) assert message_payload["answer_canvas"]["direct_answer"] == payload["answer"] assert message_payload["answer"] == payload["answer"] assert message_payload["processing_stage"] == "multi_agent_operating_runtime" assert message_payload["confirmations"][0]["action_type"] == "continue_research" assert message_payload["confirmations"][0]["can_execute"] is True def test_agent_runtime_message_projection_uses_single_customer_truth_for_blocked_research() -> None: raw_answer = ( "这轮还不能负责任地下市场结论。真实进展是：已整理 7 个候选来源、0 条可用判断；" "主要缺口：搜索结果只作为候选来源，未读正文前不能形成 ResearchNote claim；" "claim_ledger 仍为空，gap_ledger 中仍存在 blocking gaps。" ) payload = { "answer": raw_answer, "reply_text": raw_answer, "processing_stage": "agent_runtime", "answer_canvas": { "direct_answer": "这轮结果里混入了内部运行信息，我先不展示那段内容。", "action_candidates": [], }, "deliverable_packet": { "deliverable_kind": "blocked_state", "status": "blocked", "answer_text": raw_answer, "summary": raw_answer, }, "customer_evidence_packet": { "answer_type": "blocked_state", "fallback_summary": raw_answer, }, } sanitized = sanitize_customer_rag_payload(payload, apply_presentation_contract=False) message_payload = space_api._pmflow_ai_assistant_message_payload(sanitized) assert sanitized["answer"] assert message_payload["answer"] == sanitized["answer"] assert message_payload["reply_text"] == sanitized["answer"] assert message_payload["answer_canvas"]["direct_answer"] == sanitized["answer"] assert "混入了内部运行信息" not in message_payload["answer_canvas"]["direct_answer"] assert "ResearchNote claim" not in str(message_payload) assert "claim_ledger" not in str(message_payload) assert "gap_ledger" not in str(message_payload) def test_research_answer_projection_blocks_when_research_note_claims_are_not_gated() -> None: answer = _compose_research_operating_answer( mission=PMFlowAIOperatingMission(original_user_input="最近AI会议市场怎么样"), project_world_state=PMFlowAIProjectWorldStatePacket( primary_project_ref="space:external_research", status="partial", remaining_gaps=["external_public_sources_need_agent_reading"], ), research_product=PMFlowAIAgentWorkProduct( product_type="ResearchNote", structured_output={ "evidence_state": {"source_count": 5, "usable_claim_count": 2, "ready_for_opportunity_brief": True}, "market_claims": [ "Forbes Councils Membership Learn more about our services and contact us.", "公开来源候选目前只能作为候选线索，尚未抽取能直接支撑业务判断的具体事实。", ], "research_gaps": ["external_public_sources_need_agent_reading"], }, ), operating_plan=PMFlowAIOperatingPlan(highest_leverage_move="继续补三类研究来源"), autonomy_gate=PMFlowAIAutonomyDecision( confirmation_requests=[ { "action_type": "continue_research", "action_label": "继续补研究", } ] ), operating_cycle=SimpleNamespace(consumed_learning_ledger=SimpleNamespace(learning_items=[{"summary": "上一轮来源不可用"}])), ) assert answer.startswith("现在还不能直接下市场结论。") assert "初步市场判断" not in answer def test_research_answer_projection_uses_gated_partial_claims_before_opportunity_threshold() -> None: answer = _compose_research_operating_answer( mission=PMFlowAIOperatingMission(original_user_input="最近智能会议市场怎么样"), project_world_state=PMFlowAIProjectWorldStatePacket( primary_project_ref="space:external_research", status="partial", remaining_gaps=["还缺少第二个可核验公开来源"], ), research_product=PMFlowAIAgentWorkProduct( product_type="ResearchNote", structured_output={ "evidence_state": {"source_count": 1, "usable_claim_count": 1, "ready_for_opportunity_brief": False}, "market_claims": [ "2025年智能会议系统市场规模达到186亿元，具备AI能力的产品占比超过45%。", ], "research_gaps": ["还缺少第二个可核验公开来源"], }, ), operating_plan=PMFlowAIOperatingPlan(highest_leverage_move="继续补研究来源"), autonomy_gate=PMFlowAIAutonomyDecision(confirmation_requests=[]), operating_cycle=SimpleNamespace(consumed_learning_ledger=SimpleNamespace(learning_items=[])), ) assert answer.startswith("现在可以先给一个有边界的初判：") assert "186亿元" in answer assert "还不能升级成机会简报或执行动作" in answer assert "Forbes Councils" not in answer assert "公开来源候选" not in answer assert "external_public_sources_need_agent_reading" not in answer assert "Project Memory" not in answer assert "handoff" not in answer assert space_api._customer_safe_display_text(answer, fallback="fallback", limit=4000) == answer def test_agent_payload_prepared_actions_only_use_complete_action_candidates() -> None: payload = { "answer_canvas": { "action_candidates": [ { "title": "确认变更 2", "summary": "打开变更审查页处理这条待确认项。", "href": "/space/change-review?projectId=141", "target_ref": "approval:21", }, { "title": "确认变更 2；对象：重析；依据：GoalFrame top_next_actions[0]", "href": "/space/change-review?projectId=141", }, { "title": "修改后再审 3", "summary": "缺少 href 的候选动作不能投影成前端按钮。", "target_ref": "approval:20", }, ], "next_actions": [ { "title": "GoalFrame 明确限定仅绑定 project:141 和 approval:21，其他待办项虽存在但不满足目标约束条件。", } ], } } actions = _prepared_actions_from_agent_payload(payload) assert actions == [ { "action_id": "agent_action_1", "title": "确认变更 2", "summary": "打开变更审查页处理这条待确认项。", "href": "/space/change-review?projectId=141", "target_ref": "approval:21", "status": "suggested", "confirmation_required": False, "source": "deliverable_action_candidate", } ] def test_agent_answer_quality_rejects_internal_goalframe_and_repairs_from_clean_actions() -> None: terminal_state = PMFlowAIAgentTerminalAnswerState( user_goal="儿童那个项目下一步我要做什么呢", terminal_status="answerable", claims=[ PMFlowAIAgentClaim( claim_id="claim_internal", claim_type="project_next_action", claim_text="GoalFrame 明确限定仅绑定 project:141 和 approval:21，必须满足 top_next_actions[0]。", allowed_to_state=True, ) ], action_candidates=[ PMFlowAIAgentActionCandidateLedgerEntry( title="确认变更 2", target_title="童析—学龄前儿童行为特质AI评估助手", reason="这是当前工作台里优先级最高的待确认项。", href="/space/change-review?projectId=141", requires_confirmation=True, ) ], answer_intent_plan=PMFlowAIAgentAnswerIntentPlan(lead_claim_id="claim_internal"), ) bad_answer = "GoalFrame 明确限定 project:141 和 approval:21，且不满足 top_next_actions[0] 唯一性约束。" assert "internal_runtime_terms" in critique_agent_customer_answer(terminal_state=terminal_state, answer=bad_answer) repaired = render_agent_terminal_state_fallback(terminal_state) assert "确认变更 2" in repaired assert "童析" in repaired assert "GoalFrame" not in repaired assert "top_next_actions" not in repaired assert "approval:21" not in repaired assert "project:141" not in repaired @pytest.mark.asyncio async def test_pmflow_agent_runtime_step1_persists_run_and_keeps_customer_answer() -> None: db = _FakeDB() payload = await PMFlowAgentRuntimeService( db, tenant_id=12, delivery_runtime=_FakeDeliveryRuntime(), ).run_space_turn( request=RAGAskRequest(question="你是谁"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert payload["processing_stage"] == "pmflow_agent_runtime" assert payload["answer_mode"] == "pmflow_agent_runtime" assert payload["pmflow_agent_run"]["packet_version"] == "pmflow_agent_run.v1" assert payload["pmflow_agent_run"]["run_id"] == payload["agent_os_run_id"] assert payload["agent_os_environment"]["packet_version"] == "pmflow_agent_environment.v1" assert payload["agent_os_steps"][0]["kind"] == "agent_answer_runtime" assert not any(step.get("tool_name") == "read_capability_contract" for step in payload["agent_os_steps"]) assert not any(step.get("tool_name") == "read_space_state" for step in payload["agent_os_steps"]) assert payload["agent_os_steps"] assert payload["agent_os_observations"] assert not any(observation.get("source_kind") == "inspect_project_operating_state" for observation in payload["agent_os_observations"]) assert payload["agent_os_artifacts"] assert payload["agent_os_verifications"] assert payload["answer"] == "项目当前需要先收口核心验证，再准备可确认的下一步。" assert "project_object_required_for_world_state" not in payload["answer"] assert any(item.__class__.__name__ == "PMFlowAgentRun" for item in db.items) assert any(item.__class__.__name__ == "PMFlowAgentStep" for item in db.items) assert any(item.__class__.__name__ == "PMFlowAgentObservation" for item in db.items) assert any(item.__class__.__name__ == "PMFlowAgentArtifact" for item in db.items) assert any(item.__class__.__name__ == "PMFlowAgentVerification" for item in db.items) sanitized = sanitize_customer_rag_payload(payload, apply_presentation_contract=False) assert sanitized["processing_stage"] == "pmflow_agent_runtime" assert sanitized["pmflow_agent_run"]["packet_version"] == "pmflow_agent_run.v1" assert sanitized["agent_os_environment"]["packet_version"] == "pmflow_agent_environment.v1" assert "answer_runtime_packet" not in sanitized assert "agent_workspace" not in sanitized @pytest.mark.asyncio async def test_pmflow_agent_environment_step2_reads_project_world_state_only_with_structured_anchor() -> None: visible_only_registry = _RecordingAgentOSToolRegistry() visible_only = await PMFlowAgentEnvironmentService( db=object(), tenant_id=12, tool_registry=visible_only_registry, ).observe_initial( request=RAGAskRequest(question="现在有哪些项目"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert [item["tool_name"] for item in visible_only_registry.calls] == ["inspect_space_collections"] assert not any(step.tool_name == "inspect_project_operating_state" for step in visible_only.steps) assert any(gap.get("type") == "object_anchor_gap" for observation in visible_only.observations for gap in observation.gaps) anchored_registry = _RecordingAgentOSToolRegistry() anchored = await PMFlowAgentEnvironmentService( db=object(), tenant_id=12, tool_registry=anchored_registry, ).observe_initial( request=RAGAskRequest(question="这个项目怎么推进", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert [item["tool_name"] for item in anchored_registry.calls] == [ "inspect_space_collections", "inspect_project_operating_state", ] assert anchored_registry.calls[-1]["tool_input"] == {"project_ref": "project:141"} assert any(step.tool_name == "inspect_project_operating_state" for step in anchored.steps) assert any( fact.get("type") == "project_world_state" for observation in anchored.observations for fact in observation.facts ) @pytest.mark.asyncio async def test_pmflow_agent_runtime_step3_uses_dynamic_agent_runtime_without_specialist_override() -> None: db = _FakeDB() payload = await PMFlowAgentRuntimeService( db, tenant_id=12, delivery_runtime=_FakeDeliveryRuntime(), environment_service=PMFlowAgentEnvironmentService( db=object(), tenant_id=12, tool_registry=_RecordingAgentOSToolRegistry(), ), specialist_runtime=PMFlowAgentSpecialistRuntimeService( db=object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), ), ).run_space_turn( request=RAGAskRequest(question="这个项目怎么推进好呢", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert payload["processing_stage"] == "pmflow_agent_runtime" assert payload["policy_version"] == "pmflow-agent-os-v2" assert payload["agent_os_specialists"] == {} assert payload["agent_os_specialist_assignments"] == [] assert payload["agent_os_work_products"] == [] assert payload["pmflow_agent_run"]["work_products"] == payload["agent_os_work_products"] assert payload["agent_os_operating_cycle"] == {} assert payload["agent_os_learning_ledger"] == {} assert payload["agent_os_next_cycle_triggers"] == [] assert any(item["kind"] == "agent_deliverable" for item in payload["agent_os_artifacts"]) assert any(item["target_ref"] == "final_answer" for item in payload["agent_os_verifications"]) assert payload["answer"] == "项目当前需要先收口核心验证，再准备可确认的下一步。" assert "project_object_required_for_world_state" not in payload["answer"] assert [step["key"] for step in payload["answer_process_packet"]["steps"][:1]] == ["agent_answer_runtime"] sanitized = sanitize_customer_rag_payload(payload, apply_presentation_contract=False) assert sanitized["agent_os_specialists"] == {} assert sanitized["agent_os_autonomy_gate"] == {} assert "answer_runtime_packet" not in sanitized assert "agent_workspace" not in sanitized @pytest.mark.asyncio async def test_pmflow_agent_runtime_step4_prepares_actions_but_blocks_writes_and_handoffs() -> None: payload = await PMFlowAgentRuntimeService( _FakeDB(), tenant_id=12, delivery_runtime=_FakeDeliveryRuntime(), environment_service=PMFlowAgentEnvironmentService( db=object(), tenant_id=12, tool_registry=_RecordingAgentOSToolRegistry(), ), specialist_runtime=PMFlowAgentSpecialistRuntimeService( db=object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), ), ).run_space_turn( request=RAGAskRequest(question="这个项目怎么推进好呢", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert payload["agent_os_prepared_action_packets"] assert payload["agent_os_confirmation_requests"] == [] assert payload["agent_os_execution_preparations"] == [] assert payload["prepared_action_packets"] == payload["agent_os_prepared_action_packets"] assert all(item["confirmation_required"] is False for item in payload["agent_os_prepared_action_packets"]) assert all(item["status"] == "suggested" for item in payload["agent_os_prepared_action_packets"]) assert payload["agent_os_autonomy_gate"] == {} assert "项目当前需要先收口核心验证" in payload["answer"] @pytest.mark.asyncio async def test_pmflow_agent_runtime_step5_consumes_updates_and_persists_learning_ledger() -> None: previous_ledger = { "packet_version": "pmflow_ai_learning_ledger.v1", "ledger_id": "lgr_previous", "scope_ref": "project:141", "learning_items": [ { "learning_key": "single_priority_order", "summary": "上一轮确认：用户希望只看一个优先级顺序。", "outcome_type": "user_preference", "outcome_status": "observed", } ], "decision_outcomes": [], "experiment_results": [], "execution_outcomes": [], "user_preferences": [], "consumed_memory_refs": [], } payload = await PMFlowAgentRuntimeService( _FakeDB(), tenant_id=12, delivery_runtime=_FakeDeliveryRuntime(), environment_service=PMFlowAgentEnvironmentService( db=object(), tenant_id=12, tool_registry=_RecordingAgentOSToolRegistry(), ), specialist_runtime=PMFlowAgentSpecialistRuntimeService( db=object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), ), ).run_space_turn( request=RAGAskRequest(question="继续推进这个项目", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={ "context_capsules": [ { "capsule_type": "operating_learning_ledger", "scope_ref": "project:141", "learning_ledger": previous_ledger, } ], "operating_outcomes": [ { "outcome_type": "recommendation_decision", "outcome_status": "accepted", "source_ref": "creq_previous", "target_ref": "project:141", "summary": "用户接受了上一轮 MVP 收口建议。", "learning_signals": {"decision_state": "accepted"}, } ], }, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert payload["policy_version"] == "pmflow-agent-os-v2" assert payload["agent_os_consumed_learning_ledger"] == {} assert payload["agent_os_learning_ledger"] == {} assert payload["learning_ledger"] == {} assert payload["agent_os_next_cycle_triggers"] == [] assert any(step["key"] == "agent_answer_runtime" for step in payload["answer_process_packet"]["steps"]) assert any(item.__class__.__name__ == "PMFlowAgentMemory" for item in payload["pmflow_agent_run"].get("memories", [])) is False assert "项目当前需要先收口核心验证" in payload["answer"] sanitized = sanitize_customer_rag_payload(payload, apply_presentation_contract=False) assert sanitized["agent_os_operating_cycle"] == {} assert sanitized["agent_os_learning_ledger"] == {} @pytest.mark.asyncio async def test_pmflow_agent_specialist_runtime_step3_does_not_bind_visible_project_list() -> None: result = await PMFlowAgentSpecialistRuntimeService( db=object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), ).run( request=RAGAskRequest(question="你是谁"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert result.skipped_reason == "本轮没有强项目锚点，未运行项目专业 agent。" assert result.assignments == [] assert result.work_products == [] assert result.answer is None @pytest.mark.asyncio async def test_pmflow_agent_runtime_step3_keeps_direct_dialogue_answer_even_with_project_anchor() -> None: payload = await PMFlowAgentRuntimeService( _FakeDB(), tenant_id=12, delivery_runtime=_FakeDirectDeliveryRuntime(), environment_service=PMFlowAgentEnvironmentService( db=object(), tenant_id=12, tool_registry=_RecordingAgentOSToolRegistry(), ), specialist_runtime=PMFlowAgentSpecialistRuntimeService( db=object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), ), ).run_space_turn( request=RAGAskRequest(question="你是谁", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert payload["answer"].startswith("我是 PMFlow AI") assert "可以帮你查看空间、项目和下一步" in payload["answer"] assert payload["agent_os_work_products"] == [] assert "最应该先推进" not in payload["answer"] @pytest.mark.asyncio async def test_operating_runtime_step1_wraps_answer_agent_with_mission_and_work_products() -> None: payload = await PMFlowAIOperatingRunService( object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), delivery_runtime=_FakeDeliveryRuntime(), ).run_turn( request=RAGAskRequest(question="这个项目怎么推进好呢", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert payload["processing_stage"] == "multi_agent_operating_runtime" assert payload["answer_mode"] == "operating_runtime" assert payload["operating_mission"]["packet_version"] == "pmflow_ai_operating_mission.v1" assert payload["operating_mission"]["project_scope"]["object_refs"] == ["project:141"] assert payload["project_world_state_packet"]["packet_version"] == "pmflow_ai_project_world_state.v1" assert payload["project_world_state_packet"]["status"] == "ready" assert payload["project_world_state_packet"]["coverage"]["covered_facets"] assert payload["evidence_ledger"] assert payload["operating_run_packet"]["packet_version"] == "pmflow_ai_operating_run.v1" assert payload["operating_run_packet"]["project_world_state"]["packet_version"] == "pmflow_ai_project_world_state.v1" assert payload["operating_run_packet"]["agent_council_review"]["packet_version"] == "pmflow_ai_agent_council_review.v1" assert payload["operating_run_packet"]["status"] == "waiting_for_confirmation" assert len(payload["agent_assignments"]) >= 6 assert {item["agent_id"] for item in payload["agent_assignments"]} >= { "product_strategy_agent", "mvp_scope_agent", "pm_breakdown_agent", "tech_architect_agent", "qa_risk_agent", "learning_agent", } assert all("project_world_state_packet" in item["input_refs"] for item in payload["agent_assignments"]) assert all("learning_ledger" in item["input_refs"] for item in payload["agent_assignments"]) assert payload["agent_work_products"] assert all(item["claims"] for item in payload["agent_work_products"]) assert {item["product_type"] for item in payload["agent_work_products"]} >= { "OpportunityBrief", "MVPOption", "TaskDraft", "TechnicalArchitectureAssessment", "QAPlan", "LearningUpdate", } assert all(item["structured_output"] for item in payload["agent_work_products"]) assert all(item["verification"]["role_contract"] == item["product_type"] for item in payload["agent_work_products"]) assert any("evd_test" in item["evidence_refs"] for item in payload["agent_work_products"]) assert payload["product_object_projections"] assert payload["product_object_projections"][0]["packet_version"] == "pmflow_ai_product_object_projection.v1" assert payload["product_object_projections"] == payload["command_payload"]["product_object_projections"] assert "最高杠杆动作" in payload["answer"] or "准备 MVP 范围和需求候选变更" in payload["answer"] assert "确认前只展示候选" in payload["answer"] assert payload["answer"] != "项目当前需要先收口核心验证，再准备可确认的下一步。" assert payload["reply_text"] == payload["answer"] assert payload["answer_canvas"]["direct_answer"] == payload["answer"] assert payload["operating_summary"] assert "最高杠杆" in payload["operating_summary"] or "最该推进" in payload["operating_summary"] assert payload["command_payload"]["packet_version"] == "pmflow_ai_operating_command_payload.v1" assert payload["command_payload"]["project_world_state"]["status"] == "ready" assert payload["command_payload"]["evidence_ledger"] assert payload["agent_council_review"]["priority_order"] assert payload["command_payload"]["priority_order"] == payload["agent_council_review"]["priority_order"] assert payload["operating_plan"]["priority_order"] == payload["agent_council_review"]["priority_order"] assert payload["operating_plan"]["conflict_ledger"] == payload["conflict_ledger"] assert payload["command_payload"]["prepared_actions"] assert payload["autonomy_gate"]["packet_version"] == "pmflow_ai_autonomy_gate.v1" assert payload["autonomy_gate"]["status"] == "prepared_waiting_confirmation" assert payload["prepared_action_packets"] == payload["command_payload"]["prepared_actions"] assert payload["confirmation_requests"] == payload["command_payload"]["confirmation_requests"] assert payload["execution_preparations"] == payload["command_payload"]["execution_preparations"] assert payload["operating_cycle"]["packet_version"] == "pmflow_ai_operating_cycle_state.v1" assert payload["operating_cycle"]["trigger"]["trigger_type"] == "user_turn" assert {item["trigger_type"] for item in payload["operating_cycle"]["supported_triggers"]} >= { "user_turn", "signal", "project_creation", "task_update", "execution_result", "daily_review", "weekly_strategy_review", } assert payload["learning_ledger"]["packet_version"] == "pmflow_ai_learning_ledger.v1" assert payload["learning_updates"] == payload["command_payload"]["learning_updates"] assert payload["next_cycle_triggers"] == payload["command_payload"]["next_cycle"]["triggers"] assert any(item.get("source_agent") in {"mvp_scope_agent", "pm_breakdown_agent", "product_strategy_agent"} for item in payload["prepared_action_packets"]) assert payload["answer_process_packet"]["route_display"] == "agent_runtime" process_keys = [step["key"] for step in payload["answer_process_packet"]["steps"]] assert process_keys[:8] == [ "operating_mission", "operating_mission_critic", "specialist_registry", "project_world_state", "customer_safe_claim_ledger", "agent_council", "autonomy_gate", "learning_loop", ] assert "agent_answer_runtime" in process_keys sanitized = sanitize_customer_rag_payload(payload, apply_presentation_contract=False) assert sanitized["processing_stage"] == "multi_agent_operating_runtime" assert sanitized["project_world_state_packet"]["packet_version"] == "pmflow_ai_project_world_state.v1" assert sanitized["customer_safe_claim_ledger"]["packet_version"] == "pmflow_ai_customer_safe_claim_ledger.v1" assert sanitized["agent_council_review"]["packet_version"] == "pmflow_ai_agent_council_review.v1" assert sanitized["autonomy_gate"]["packet_version"] == "pmflow_ai_autonomy_gate.v1" assert sanitized["conflict_ledger"] assert sanitized["prepared_action_packets"] assert sanitized["confirmation_requests"] assert sanitized["execution_preparations"] assert sanitized["product_object_projections"] assert sanitized["operating_cycle"]["packet_version"] == "pmflow_ai_operating_cycle_state.v1" assert sanitized["learning_ledger"]["packet_version"] == "pmflow_ai_learning_ledger.v1" assert sanitized["outcome_observations"] assert sanitized["learning_updates"] assert sanitized["next_cycle_triggers"] assert sanitized["evidence_ledger"] assert sanitized["command_payload"]["packet_version"] == "pmflow_ai_operating_command_payload.v1" assert "operating_summary" not in sanitized assert "agent_workspace" not in sanitized assert "answer_runtime_packet" not in sanitized class _FakeSpecialistRuntimeProfile: temperature = 0.17 response_mode = "structured_json" max_tokens = 2400 def routing_meta(self) -> dict[str, object]: return {"profile_source": "unit_test", "profile_version": "specialist-test-v1"} class _FakeSpecialistPromptDB: def add(self, item) -> None: pass async def flush(self) -> None: pass def execute(self, *args, **kwargs): # pragma: no cover - only used as a callable capability marker. raise AssertionError("Prompt/profile access should be monkeypatched in this unit test.") class _RecordingSpecialistLLMGenerator: def __init__(self) -> None: self.calls: list[dict[str, object]] = [] async def generate(self, prompt: str, **kwargs) -> str: routing_meta = dict(kwargs.get("routing_meta") or {}) self.calls.append({"prompt": prompt, "routing_meta": routing_meta, "kwargs": kwargs}) agent_id = str(routing_meta.get("specialist_agent_id") or "") if agent_id == "mvp_scope_agent": payload = { "product_type": "MVPOption", "status": "ready", "summary": "先把 AI 漫剧工具收敛为角色一致性与分镜复用的首版闭环。", "structured_output": { "artifact_title": "AI 漫剧工具首版 MVP", "mvp_boundary": ["角色设定库", "分镜脚本生成", "关键帧一致性校验"], }, "claims": [{"claim_text": "首版应围绕可验证的创作闭环，而不是一次做完整生产平台。"}], "recommendations": [{"summary": "先验证 3 分钟短剧从脚本到关键帧的稳定产出。"}], "prepared_actions": [{"title": "生成 MVP 草案待确认", "requires_user_confirm": True}], "blocked_actions": [{"summary": "未确认前不直接创建项目或任务。"}], "evidence_refs": ["evd_open_research_seed"], } else: payload = { "product_type": "RevenueSignal", "status": "ready", "summary": "变现验证应优先面向短剧创作者和小团队，测试模板包或按剧集计费。", "structured_output": { "artifact_title": "AI 漫剧工具变现假设", "pricing_hypotheses": ["模板订阅", "按剧集导出计费"], }, "claims": [{"claim_text": "小团队愿意为降低返工和稳定角色一致性付费。"}], "recommendations": [{"summary": "用 5 个创作者访谈验证付费意愿。"}], "prepared_actions": [{"title": "准备变现访谈清单", "requires_user_confirm": True}], "blocked_actions": [], "evidence_refs": ["evd_open_research_seed"], } return json.dumps(payload, ensure_ascii=False) @pytest.mark.asyncio async def test_specialist_registry_uses_mission_selected_agents_and_llm_work_products(monkeypatch) -> None: async def fake_render(self, **kwargs): return PromptRuntimeRenderResult( prompt="specialist work product prompt", prompt_key=kwargs["prompt_key"], source="unit_test", release_id=9001, runtime_tenant_id=12, ) async def fake_resolve_profile(self, **kwargs): assert kwargs["function_name"] == "pmflow_ai_specialist_work_product" assert kwargs["scene_key"] == "space_pmflow_ai_operating_runtime" assert kwargs["prompt_key"] == "PROMPT_PMFLOW_AI_SPECIALIST_WORK_PRODUCT" return _FakeSpecialistRuntimeProfile() monkeypatch.setattr(operating_specialist_agents_module.PromptRuntimeService, "render", fake_render) monkeypatch.setattr(operating_specialist_agents_module.AIFunctionRuntimeProfileService, "resolve_profile", fake_resolve_profile) generator = _RecordingSpecialistLLMGenerator() assignments = [ PMFlowAIAgentAssignment( agent_id="mvp_scope_agent", role_goal="把创意转成可确认的首版 MVP 边界。", expected_work_product="MVPOption", ), PMFlowAIAgentAssignment( agent_id="monetization_agent", role_goal="形成最小变现验证方案。", expected_work_product="RevenueSignal", ), ] products = await PMFlowAISpecialistAgentService( db=_FakeSpecialistPromptDB(), tenant_id=12, generator=generator, agent_registry=PMFlowAIOperatingAgentRegistry(), ).produce_work_products_async( mission=PMFlowAIOperatingMission( mission_goal="把 AI 漫剧工具创意推进到可确认的产品草案。", stage="mvp", desired_business_outcome="形成可验证 MVP 与变现假设。", original_user_input="我想做个AI漫剧的工具", required_agents=["mvp_scope_agent", "monetization_agent"], project_scope={"scope_type": "new_idea", "object_refs": []}, ), assignments=assignments, project_world_state=PMFlowAIProjectWorldStatePacket( primary_project_ref="", project_refs=[], status="partial", evidence_ledger=[ PMFlowAIEvidenceLedgerEntry( evidence_id="evd_open_research_seed", project_ref="space:idea", facet_key="idea_seed", source_ref="user_question", summary="用户提出 AI 漫剧工具创意。", confidence="medium", ) ], remaining_gaps=["需要补充目标用户和制作链路证据。"], ), learning_ledger={"accepted_patterns": ["把创意落到可确认草案，而不是泛聊。"]}, ) assert [item.agent_id for item in products] == ["mvp_scope_agent", "monetization_agent"] assert [call["routing_meta"]["specialist_agent_id"] for call in generator.calls] == ["mvp_scope_agent", "monetization_agent"] assert all(call["routing_meta"]["prompt_key"] == "PROMPT_PMFLOW_AI_SPECIALIST_WORK_PRODUCT" for call in generator.calls) assert all(call["routing_meta"]["profile_source"] == "unit_test" for call in generator.calls) assert all(call["routing_meta"]["response_contract"] == "json" for call in generator.calls) assert all(call["routing_meta"]["response_format"] == "json_object" for call in generator.calls) assert products[0].summary == "先把 AI 漫剧工具收敛为角色一致性与分镜复用的首版闭环。" assert products[1].summary == "变现验证应优先面向短剧创作者和小团队，测试模板包或按剧集计费。" assert products[0].structured_output["mvp_boundary"] == ["角色设定库", "分镜脚本生成", "关键帧一致性校验"] assert products[1].structured_output["pricing_hypotheses"] == ["模板订阅", "按剧集导出计费"] assert all(item.verification["llm_generated"] is True for item in products) assert all("specialist_llm_fallback" not in item.verification.get("quality_findings", []) for item in products) @pytest.mark.asyncio async def test_operating_runtime_accepts_stream_event_sink_and_delegates_it() -> None: delivery_runtime = _RecordingEventSinkDeliveryRuntime() events: list[dict] = [] async def sink(event: dict) -> None: events.append(event) payload = await PMFlowAIOperatingRunService( object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), delivery_runtime=delivery_runtime, ).run_turn( request=RAGAskRequest(question="这个项目怎么推进好呢", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, event_sink=sink, ) assert payload["processing_stage"] == "multi_agent_operating_runtime" assert delivery_runtime.saw_event_sink is True assert events == [{"type": "step.completed", "step": {"key": "agent_answer_runtime", "label": "生成回答", "status": "completed"}}] @pytest.mark.asyncio async def test_operating_runtime_open_research_projects_partial_evidence_to_research_recovery() -> None: payload = await PMFlowAIOperatingRunService( object(), tenant_id=12, project_world_state_service=_FakeOpenResearchProjectWorldStateService(), delivery_runtime=_FakeOpenResearchDeliveryRuntime(), ).run_turn( request=RAGAskRequest(question="最近AI会议市场怎么样"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": []}}, ) world_state = payload["project_world_state_packet"] research_facet = next(item for item in world_state["facets"] if item["facet_key"] == "external_research_evidence") projection = payload["product_object_projections"][0] prepared = payload["prepared_action_packets"] confirmations = payload["confirmation_requests"] assert world_state["project_refs"] == [] assert research_facet["status"] == "partial" assert projection["object_type"] == "ResearchNote" assert projection["status"] == "draft" assert any(item["action_family"] == "research_recovery" for item in prepared) assert any(item["source_action_type"] == "continue_research" for item in prepared) assert any(item["action_type"] == "continue_research" and item["action_label"] == "继续补研究" for item in confirmations) assert all(item["can_execute_after_confirmation"] is True for item in confirmations if item["action_type"] == "continue_research") recovery_payload = next(item for item in prepared if item["action_family"] == "research_recovery")["payload"]["research_recovery"] assert recovery_payload["source_plan"] == [] assert recovery_payload["queries"] assert "OpportunityBrief" not in {item["object_type"] for item in payload["product_object_projections"]} assert "继续补研究" in payload["answer"] assert "继续补研究" in payload["answer"] assert payload["answer"] != "这轮先进入研究补证，不能把候选来源包装成市场结论。" assert payload["answer_canvas"]["direct_answer"] == payload["answer"] assert "候选来源" in payload["answer"] or "候选" in payload["answer"] assert "还不能升级成机会简报或执行动作" in payload["operating_summary"] @pytest.mark.asyncio async def test_operating_runtime_promotes_nested_agent_run_observations_before_specialists() -> None: payload = await PMFlowAIOperatingRunService( object(), tenant_id=12, project_world_state_service=_FakeOpenResearchProjectWorldStateService(), delivery_runtime=_FakeNestedResearchDeliveryRuntime(), ).run_turn( request=RAGAskRequest(question="最近AI会议市场怎么样"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": []}}, ) world_state = payload["project_world_state_packet"] research_facet = next(item for item in world_state["facets"] if item["facet_key"] == "external_research_evidence") facts = {item["fact_key"]: item["value"] for item in research_facet["facts"]} confirmations = payload["confirmation_requests"] assert len(facts["source_candidates"]) == 2 assert len(facts["usable_claims"]) == 1 assert {item["agent_id"] for item in payload["agent_work_products"]} == { "idea_incubation_agent", "market_research_agent", "competitor_agent", "product_strategy_agent", "monetization_agent", "learning_agent", } assert {item["action_family"] for item in payload["prepared_action_packets"]} == {"research_recovery"} assert all(item["action_type"] == "continue_research" for item in confirmations) assert all(item["can_execute_after_confirmation"] is True for item in confirmations) assert payload["answer_canvas"]["direct_answer"] == payload["answer"] blocked = {item["agent_id"]: item for item in payload["agent_work_products"] if item["status"] == "blocked"} assert {"competitor_agent", "monetization_agent"}.issubset(blocked) assert "growth_agent" not in {item["agent_id"] for item in payload["agent_work_products"]} @pytest.mark.asyncio async def test_operating_runtime_replans_after_evidence_and_preserves_terminal_claims() -> None: payload = await PMFlowAIOperatingRunService( object(), tenant_id=12, mission_service=_FakeShallowIdeaMissionService(), project_world_state_service=_FakeOpenResearchProjectWorldStateService(), delivery_runtime=_FakeProductBlueprintDeliveryRuntime(), ).run_turn( request=RAGAskRequest(question="做个AI音乐的自动生成"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": []}}, ) agent_ids = {item["agent_id"] for item in payload["agent_assignments"]} assert "idea_incubation_agent" in agent_ids assert {"product_strategy_agent", "mvp_scope_agent", "tech_architect_agent", "qa_risk_agent", "monetization_agent"}.issubset(agent_ids) assert payload["operating_mission_critic"]["needs_replan"] is True critic_added_roles = set() for review in payload["operating_mission_critic_reviews"]: critic_added_roles.update(review["missing_capability_roles"]) assert critic_added_roles >= { "product_strategy_agent", "mvp_scope_agent", "tech_architect_agent", } world_state = payload["project_world_state_packet"] research_facet = next(item for item in world_state["facets"] if item["facet_key"] == "external_research_evidence") facts = {item["fact_key"]: item["value"] for item in research_facet["facts"]} assert len(facts["source_candidates"]) == 2 assert len(facts["usable_claims"]) >= 3 claim_ledger = payload["customer_safe_claim_ledger"] assert claim_ledger["packet_version"] == "pmflow_ai_customer_safe_claim_ledger.v1" assert claim_ledger["status"] == "ready" assert claim_ledger["claim_count"] >= 3 assert [item["claim_text"] for item in claim_ledger["claims"][:3]] == facts["usable_claims"][:3] assert payload["command_payload"]["customer_safe_claim_ledger"]["claim_count"] == claim_ledger["claim_count"] assert payload["project_world_state_packet"]["customer_safe_claim_ledger"]["claim_count"] == claim_ledger["claim_count"] assert "0 个候选来源" not in payload["answer"] assert "0 条可用判断" not in payload["answer"] assert "AI 音乐工具的核心用户价值" in payload["answer"] assert "产品蓝图" in payload["answer"] assert any(item.get("artifact_kind") == "product_blueprint" for item in payload["answer_canvas"]["artifacts"]) assert {item["product_type"] for item in payload["agent_work_products"]} >= { "OpportunityBrief", "MVPOption", "TechnicalArchitectureAssessment", "QAPlan", "RevenueSignal", } def test_operating_registry_keeps_requested_specialists_and_leaves_dependency_gate_to_roles() -> None: mission = PMFlowAIOperatingMission( stage="research", original_user_input="最近AI会议市场怎么样", required_agents=["growth_agent", "monetization_agent", "market_research_agent"], ) assignments = PMFlowAIOperatingAgentRegistry().assignments_for_mission(mission) assert [item.agent_id for item in assignments] == ["growth_agent", "monetization_agent", "market_research_agent"] @pytest.mark.asyncio async def test_operating_runtime_step6_consumes_and_updates_learning_ledger() -> None: previous_ledger = { "packet_version": "pmflow_ai_learning_ledger.v1", "ledger_id": "lgr_previous", "scope_ref": "project:141", "learning_items": [ { "learning_key": "single_priority_order", "summary": "上一轮确认：用户希望只看一个优先级顺序。", "outcome_type": "user_preference", "outcome_status": "observed", } ], "decision_outcomes": [], "experiment_results": [], "execution_outcomes": [], "user_preferences": [], "consumed_memory_refs": [], } payload = await PMFlowAIOperatingRunService( object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), delivery_runtime=_FakeDeliveryRuntime(), ).run_turn( request=RAGAskRequest(question="继续推进这个项目", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={ "context_capsules": [ { "capsule_type": "operating_learning_ledger", "scope_ref": "project:141", "learning_ledger": previous_ledger, } ], "operating_outcomes": [ { "outcome_type": "recommendation_decision", "outcome_status": "accepted", "source_ref": "creq_previous", "target_ref": "project:141", "summary": "用户接受了上一轮 MVP 收口建议。", "learning_signals": {"decision_state": "accepted"}, } ], }, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) assert payload["consumed_learning_ledger"]["ledger_id"] == "lgr_previous" assert payload["consumed_learning_ledger"]["learning_items"][0]["learning_key"] == "single_priority_order" assert payload["learning_ledger"]["ledger_id"] == "lgr_previous" assert any(item["outcome_status"] == "accepted" for item in payload["learning_ledger"]["decision_outcomes"]) assert any(item["trigger_type"] == "execution_result" for item in payload["next_cycle_triggers"]) assert "读取了上一轮" in payload["operating_summary"] assert "最高杠杆动作" in payload["answer"] or "准备 MVP 范围和需求候选变更" in payload["answer"] assert "确认前只展示候选" in payload["answer"] capsules = space_api._upsert_operating_learning_ledger_capsule(capsules=[], learning_ledger=payload["learning_ledger"]) assert capsules[0]["capsule_type"] == "operating_learning_ledger" assert capsules[0]["learning_ledger"]["ledger_id"] == "lgr_previous" @pytest.mark.asyncio async def test_operating_runtime_step5_prepares_actions_but_blocks_writes_and_handoffs() -> None: payload = await PMFlowAIOperatingRunService( object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService(), delivery_runtime=_FakeDeliveryRuntime(), ).run_turn( request=RAGAskRequest(question="这个项目怎么推进好呢", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) families = {item["action_family"] for item in payload["prepared_action_packets"]} required_before = { required for request in payload["confirmation_requests"] for required in request.get("required_before", []) } prohibited = set(payload["autonomy_gate"]["prohibited_side_effects"]) assert {"project_memory_candidate", "product_object_projection"} <= families assert payload["execution_preparations"] assert payload["execution_preparations"][0]["packet_version"] == "pmflow_ai_execution_preparation.v1" assert payload["execution_preparations"][0]["readiness"] == "prepared_waiting_confirmation" assert payload["execution_preparations"][0]["handoff_package"]["packet_version"] == "execution_bridge_handoff_preparation.v1" assert payload["execution_preparations"][0]["handoff_package"]["result_writeback_contract"]["protocol_version"] == "execution_result_protocol_v1" assert all(item["confirmation_required"] is True for item in payload["prepared_action_packets"]) assert all(item["status"] == "prepared_waiting_confirmation" for item in payload["prepared_action_packets"]) assert all(request["status"] == "waiting_for_user_confirmation" for request in payload["confirmation_requests"]) assert "write_project_memory" in prohibited assert "execute_handoff" in prohibited assert "write_project_memory" in required_before assert "execute_handoff" in required_before executable_requests = [request for request in payload["confirmation_requests"] if request.get("can_execute_after_confirmation")] assert executable_requests assert {request.get("action_type") for request in executable_requests} >= {"save_candidate", "start_execution_handoff"} assert any( request.get("action_payload", {}).get("payload", {}).get("handoff_package", {}).get("result_writeback_contract", {}).get("protocol_version") == "execution_result_protocol_v1" for request in executable_requests ) assert "确认前只展示候选，不会写入项目资料、改动任务或交给执行环境" in payload["operating_summary"] assert "最高杠杆动作" in payload["answer"] or "准备 MVP 范围和需求候选变更" in payload["answer"] assert "确认前只展示候选" in payload["answer"] def test_space_confirmation_continue_research_builds_server_side_rerun_prompt() -> None: prompt = space_api._research_recovery_prompt_from_confirmation( confirmation={ "title": "确认准备动作：准备 ResearchNote", "action_payload": { "payload": { "research_recovery": { "research_question": "最近 AI 会议市场怎么样", "queries": ["AI conference market 2026 sponsorship", "AI event platform market"], "research_gaps": ["缺少参会增长", "缺少商业化证据"], } } }, }, request=space_api.PMFlowAIConfirmationExecutionRequest(), ) assert "只读研究补证" in prompt assert "重新经过 specialist/council" in prompt assert "AI conference market 2026 sponsorship" in prompt assert "不要写 Project Memory" in prompt def test_space_confirmation_execution_handoff_parses_formal_task_contract() -> None: result = space_api._execution_handoff_request_from_confirmation( action_payload={ "target_ref": "project:141", "payload": { "handoff_package": { "task_candidates": [{"target_ref": "task:9001", "title": "实现执行桥回写"}], "repo_binding_id": "8", "runtime_target": "codex", } }, }, supplement_text="先跑最小测试。", ) assert result["project_id"] == 141 assert result["task_id"] == 9001 assert result["repo_binding_id"] == "8" assert result["runtime_target"] == "codex" assert result["note"] == "先跑最小测试。" def test_space_confirmation_execution_handoff_blocks_without_formal_task() -> None: result = space_api._execution_handoff_request_from_confirmation( action_payload={"target_ref": "project:141", "payload": {"handoff_package": {"task_candidates": []}}}, supplement_text="", ) assert result["blocked"] is True assert "正式任务" in result["missing"] assert "不能创建执行交接" in result["customer_message"] @pytest.mark.asyncio async def test_operating_runtime_step4_council_records_risk_veto_and_single_priority() -> None: payload = await PMFlowAIOperatingRunService( object(), tenant_id=12, project_world_state_service=_FakeProjectWorldStateService( blocked_actions=[ { "display_label": "先处理当前任务阻塞", "reason": "项目世界状态显示该动作仍受阻，需要先解除阻塞。", } ] ), delivery_runtime=_FakeDeliveryRuntime(), ).run_turn( request=RAGAskRequest(question="这个项目怎么推进好呢", project_id=141), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) conflict_types = {item["conflict_type"] for item in payload["conflict_ledger"]} blocked_titles = {item["title"] for item in payload["operating_plan"]["blocked_actions"]} priority_titles = [item["title"] for item in payload["operating_plan"]["priority_order"]] assert payload["agent_council_review"]["status"] == "blocked" assert "risk_veto_conflict" in conflict_types assert "先处理当前任务阻塞" in blocked_titles assert "先处理当前任务阻塞" not in priority_titles assert priority_titles == [item["title"] for item in payload["command_payload"]["priority_order"]] assert len(priority_titles) == len(set(priority_titles)) def test_space_api_frame_type_ignores_legacy_turn_state() -> None: frame_type = space_api._space_frame_type_from_agent_payload( { "pmflow_ai_turn_state": {"answer_state": {"answer_shape": "legacy_status_answer"}}, "terminal_answer_state": {"terminal_status": "answerable"}, "agent_run_packet": {"status": "answered"}, } ) assert frame_type == "answerable" @pytest.mark.asyncio async def test_agent_runtime_step2_lets_planner_choose_external_search_without_identity_template() -> None: payload = await PMFlowAIAgentRunService( object(), tenant_id=12, planner=_FakePlanner("external_search"), tool_registry=_FakeToolRegistry(), critic=_FakeCritic(ready=True), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="帮我分析下用AI来协助采购监督的系统怎么样"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, started_at=None, ) step_keys = [step["key"] for step in payload["answer_process_packet"]["steps"]] assert "agent_step_planner" in step_keys assert "agent_tool_external_search" in step_keys assert "agent_tool_inspect_source_detail" in step_keys assert "agent_state_critic" in step_keys assert payload["agent_workspace"]["observations"][-1]["observation_type"] == "external_source_document" assert payload["sources"] == [{"label": "公开来源", "url": "https://example.com/source"}] assert "我是 PMFlow AI" not in payload["answer"] assert "身份" not in payload["answer"] assert "能力边界" not in payload["answer"] assert payload["terminal_answer_state"]["packet_version"] == "pmflow_ai_agent_terminal_answer_state.v1" assert payload["answer_runtime_packet"]["packet_version"] == "pmflow_ai_agent_answer_runtime_packet.v1" assert payload["deliverable_packet"]["source_observation_ids"] assert payload["deliverable_packet"]["sections"] assert "判断依据：" not in payload["answer"] assert "Claim Ledger" not in payload["answer"] sanitized = sanitize_customer_rag_payload(payload, apply_presentation_contract=False) assert "answer_runtime_packet" not in sanitized assert "agent_workspace" not in sanitized assert "pmflow_ai_turn_state" not in sanitized assert "pmflow_ai_route_plan" not in sanitized assert sanitized["deliverable_packet"]["packet_version"] == "pmflow_ai_agent_deliverable_packet.v1" assert sanitized["terminal_answer_state"]["packet_version"] == "pmflow_ai_agent_terminal_answer_state.v1" def test_agent_runtime_step2_keeps_legacy_goal_assessor_out_of_new_service() -> None: from app.services.pmflow_ai_agent_runtime import agent_run_service source = inspect.getsource(agent_run_service) assert "PROMPT_PMFLOW_AI_TURN_GOAL_STATE_ASSESSOR" not in source assert "PMFlowAITurnRuntimeService" not in source assert "looks_like_" not in source assert "operation_kind" not in source assert "external_search" in AGENT_TOOL_CONTRACTS def test_pmflow_ai_runtime_prompt_renderers_do_not_use_code_fallback() -> None: service_roots = [ Path("backend/app/services/pmflow_ai_agent_runtime"), Path("backend/app/services/pmflow_ai_operating_runtime"), ] source = "\n".join( path.read_text(encoding="utf-8") for root in service_roots for path in root.glob("*.py") ) assert "code_fallback" not in source assert "fallback_template=" not in source def test_agent_tool_environment_exposes_typed_contracts_not_name_whitelist() -> None: contract = AGENT_TOOL_CONTRACTS["inspect_project_operating_state"] assert contract.read_or_write == "read" assert contract.risk_level == "low" assert "inspect_project_status" in contract.aliases assert contract.input_schema assert contract.output_schema assert contract.permission_boundary assert contract.success_signal assert contract.failure_signal assert contract.side_effects == [] @pytest.mark.asyncio async def test_agent_runtime_v2_replans_when_critic_requests_more_evidence() -> None: payload = await PMFlowAIAgentRunService( object(), tenant_id=12, planner=_SequencePlanner(["inspect_space_collections", "external_search"]), tool_registry=_FakeToolRegistry(), critic=_SequenceCritic( [ PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, state_delta={"remaining_gap": "needs_external_source"}, reason="还需要继续读取外部证据。", ), PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, state_delta={"evidence_ready": True}, reason="证据已足够。", ), ] ), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="帮我分析下用AI来协助采购监督的系统怎么样"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, started_at=None, ) step_keys = [step["key"] for step in payload["answer_process_packet"]["steps"]] assert "agent_step_planner_2" in step_keys assert "agent_tool_external_search_2" in step_keys assert "agent_state_critic_2" in step_keys assert payload["agent_run_packet"]["stop_reason"] == "ready_to_answer" assert payload["agent_run_packet"]["iterations"] == 2 assert payload["agent_run_packet"]["tool_call_count"] == 2 assert [event["event_type"] for event in payload["agent_run_packet"]["event_log"]].count("plan") == 2 assert len(payload["agent_workspace"]["action_ledger"]) >= 2 assert any(item.get("tool_name") == "external_search" for item in payload["agent_workspace"]["attempt_ledger"]) assert payload["agent_workspace"]["deliverable_state"]["stop_reason"] == "ready_to_answer" @pytest.mark.asyncio async def test_agent_runtime_recovers_from_failed_source_detail_by_reading_untried_ranked_sources() -> None: tool_registry = _RecoveryExternalToolRegistry() payload = await PMFlowAIAgentRunService( object(), tenant_id=12, planner=_SequencePlanner(["external_search"]), tool_registry=tool_registry, critic=_SequenceCritic( [ PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=False, retry_would_change_state=False, state_delta={"source_quality_blocked_claims": True}, reason="第一个来源不可读，旧链路会直接停住。", ), PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, retry_would_change_state=False, state_delta={"recovered_with_second_source": True}, reason="补读未尝试来源后可以回答。", ), ] ), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="最近AI会议市场怎么样"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": []}}, started_at=None, ) step_keys = [step["key"] for step in payload["answer_process_packet"]["steps"]] assert "agent_recovery_planner_2" in step_keys assert "agent_tool_inspect_source_detail_2" in step_keys assert payload["agent_run_packet"]["stop_reason"] == "ready_to_answer" assert payload["agent_run_packet"]["tool_call_count"] == 3 failed_first_url = tool_registry.inspect_calls[0][0] assert failed_first_url not in tool_registry.inspect_calls[1] assert "https://example.com/ai-conference-attendance-2026" in tool_registry.inspect_calls[1] @pytest.mark.asyncio async def test_agent_runtime_recovers_when_only_source_candidate_claims_exist() -> None: payload = await PMFlowAIAgentRunService( object(), tenant_id=12, planner=_SequencePlanner(["external_search"]), tool_registry=_CandidateClaimRecoveryToolRegistry(), evidence_normalizer=_CandidateOnlyNormalizer(), critic=_SequenceCritic( [ PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, retry_would_change_state=True, state_delta={"remaining_gaps": ["needs_conference_specific_sources"]}, reason="当前只有候选来源，没有可陈述业务事实，继续检索会改变状态。", ), PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, state_delta={"recovered": True}, reason="第二轮已补充观察。", ), ] ), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="最近AI会议市场怎么样"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": []}}, started_at=None, ) step_keys = [step["key"] for step in payload["answer_process_packet"]["steps"]] assert "agent_recovery_planner_2" in step_keys assert "agent_tool_external_search_2" in step_keys assert payload["agent_run_packet"]["stop_reason"] == "ready_to_answer" assert payload["agent_run_packet"]["iterations"] == 2 @pytest.mark.asyncio async def test_pmflow_agent_runtime_uses_external_search_and_ledger_for_open_market_goal() -> None: dynamic_runtime = PMFlowAIAgentRunService( object(), tenant_id=12, planner=_SequencePlanner(["external_search"]), tool_registry=_FakeToolRegistry(), critic=_SequenceCritic( [ PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, state_delta={"external_claim_ready": True}, reason="公开材料已进入 claim ledger。", ), ] ), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ) payload = await PMFlowAgentRuntimeService( _FakeDB(), tenant_id=12, delivery_runtime=dynamic_runtime, ).run_space_turn( request=RAGAskRequest(question="帮我看看市场上有没有什么新的好项目"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, ) process_keys = [step["key"] for step in payload["answer_process_packet"]["steps"]] assert "agent_tool_external_search" in process_keys assert "agent_evidence_normalizer_fetch" in process_keys assert any(step.get("tool_name") == "external_search" for step in payload["agent_os_steps"]) assert any(observation.get("source_kind") == "external_search" for observation in payload["agent_os_observations"]) assert not any(step.get("tool_name") == "run_opportunity_discovery_agents" for step in payload["agent_os_steps"]) assert "我是 PMFlow AI" not in payload["answer"] assert payload["customer_evidence_packet"]["answer_type"] assert payload["agent_os_verifications"][-1]["target_ref"] == "final_answer" @pytest.mark.asyncio async def test_agent_runtime_blocks_raw_external_source_noise_from_customer_answer() -> None: payload = await PMFlowAIAgentRunService( object(), tenant_id=12, planner=_SequencePlanner(["external_search", "inspect_source_detail"]), tool_registry=_NoisyExternalToolRegistry(), critic=PMFlowAIAgentStateCritic(db=object(), tenant_id=12, generator=object()), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="最近AI漫剧行业怎么样"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": []}}, started_at=None, ) answer = payload["answer"] assert payload["deliverable_packet"]["status"] == "blocked" assert payload["agent_workspace"]["claim_ledger"] == [] assert payload["agent_workspace"]["evidence_normalizations"][0]["source_quality"] == "noisy" assert "news@iimedia.cn" not in answer assert "020-8922-4700" not in answer assert " " not in answer assert "联系我们 订阅邮箱" not in answer assert "网页噪音" in answer or "证据" in answer assert all("news@iimedia.cn" not in str(item) for item in payload["deliverable_packet"]["sections"]) def test_agent_answer_runtime_never_promotes_fact_ledger_to_customer_claims() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="检查外部来源") workspace.fact_ledger = [ PMFlowAIAgentFactLedgerEntry( fact_type="external_evidence", fact_text="原始网页：news@example.com 020-1234-5678   联系我们 订阅邮箱", source_tool="external_search", ) ] terminal_state = build_agent_terminal_answer_state( question="检查外部来源", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="stop_blocked", rationale="停止"), observation=None, critic_result=PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=False, retry_would_change_state=False, missing_state=["缺少可入账 claim"], reason="claim ledger 为空。", ), ) answer = render_agent_terminal_state_fallback(terminal_state) assert terminal_state.claims == [] assert "news@example.com" not in answer assert "020-1234-5678" not in answer assert " " not in answer assert "还没有足够可靠的材料支撑结论" in answer def test_agent_answer_draft_accepts_streaming_composer_sources() -> None: draft = PMFlowAIAgentAnswerDraft(answer_text="已生成回答。", source="llm_composer_stream") repaired = PMFlowAIAgentAnswerDraft(answer_text="已修复回答。", source="llm_composer_stream_readability_repair") assert draft.source == "llm_composer_stream" assert repaired.source == "llm_composer_stream_readability_repair" def test_non_usable_normalization_cannot_enter_claim_ledger() -> None: observation = PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源。", facts=[ { "type": "external_evidence", "title": "公开页面", "summary": "联系我们 020-1234-5678 news@example.com   订阅邮箱", "url": "https://example.com/noisy", } ], sources=[{"label": "公开页面", "url": "https://example.com/noisy"}], confidence="high", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="检查外部来源") normalization = fallback_normalize_observation(workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "noisy" assert normalization.usable_claims == [] assert workspace.claim_ledger == [] assert workspace.boundary_ledger def test_source_candidate_normalization_does_not_enter_claim_ledger() -> None: observation = PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源。", facts=[ { "type": "external_evidence", "title": "AI Meeting Assistants 2026", "summary": "This page may contain related market material.", "url": "https://example.com/ai-meeting-assistants", } ], sources=[{"label": "AI Meeting Assistants 2026", "url": "https://example.com/ai-meeting-assistants"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议行业怎么样") normalization = fallback_normalize_observation(workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "noisy" assert normalization.usable_claims == [] assert workspace.claim_ledger == [] assert workspace.boundary_ledger assert any("候选来源" in boundary.customer_safe_reason for boundary in workspace.boundary_ledger) def test_external_search_candidate_summary_never_enters_claim_ledger_before_reading_source() -> None: observation = PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源。", facts=[ { "type": "external_evidence", "title": "Industry report candidate", "summary": "A 2026 public report says adoption increased by 42% and sponsors expanded budgets.", "url": "https://example.com/report-candidate", } ], sources=[{"label": "Industry report candidate", "url": "https://example.com/report-candidate"}], confidence="high", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="最近行业市场怎么样") normalization = fallback_normalize_observation(workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "noisy" assert normalization.usable_claims == [] assert workspace.claim_ledger == [] assert normalization.source_normalizations[0].rejected_reason == "search_candidate_requires_source_reading" def test_external_source_document_can_use_event_market_signals_after_source_reading() -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "RAISE Summit 2026", "url": "https://example.com/raise-summit", "text": ( "European AI conferences are shifting toward applied enterprise AI, buyer networking, " "and sponsor-backed communities. Upcoming events like RAISE Summit 2026 report over " "9,000 attendees and dedicated enterprise AI tracks." ), } ], sources=[{"label": "RAISE Summit 2026", "url": "https://example.com/raise-summit"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace( current_goal="最近AI会议市场怎么样", goal_frame=PMFlowAIAgentGoalFrame( user_goal="判断AI会议活动市场的近期态势", desired_output="给出有边界的市场研究判断", evidence_requirements=["AI conference attendance", "conference market activity"], interpretation_branches=[ { "branch_id": "b1", "meaning": "AI会议和活动市场", "user_language_prior": "high", "anchor_terms": ["AI conference", "conference", "event", "attendance", "attendees", "summit"], }, { "branch_id": "b2", "meaning": "AI技术软件市场", "user_language_prior": "medium", "anchor_terms": ["software", "model", "agentic AI"], }, ], ), ) normalization = fallback_normalize_observation( question="最近AI会议市场怎么样", workspace=workspace, observation=observation, ) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "usable" assert workspace.claim_ledger assert any("9,000 attendees" in claim.claim_text for claim in workspace.claim_ledger) assert all("###" not in claim.claim_text for claim in workspace.claim_ledger) def test_external_source_page_shell_does_not_enter_claim_ledger() -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "Forbes Councils Article", "url": "https://example.com/forbes-councils", "text": ( "Forbes Councils Paid Program Subscribe Sign In Newsletter Share This Article " "Contact Us Privacy Terms Copyright 'AI events' Read More Premium " "About Us Site Map Follow Us Advertisement" ), } ], sources=[{"label": "Forbes Councils Article", "url": "https://example.com/forbes-councils"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="做一轮公开市场研究") normalization = fallback_normalize_observation(workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "noisy" assert normalization.usable_claims == [] assert workspace.claim_ledger == [] assert normalization.source_normalizations[0].rejected_reason == "raw_source_noise" def test_external_source_ai_infra_ad_fragment_does_not_enter_claim_ledger() -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "2025年AI漫剧产业深度研究报告", "url": "https://example.com/ai-comic-drama-report", "text": ( "Qwen-Image 定制镜像 | RTX4090D CUDA12.4 大模型推理专用 一键部署。" "适配 24GB 显存，预装通义千问视觉语言模型依赖库、推理脚本与常用工具，开箱即用，" "支持大模型快速加载、图像理解、多模态推理等任务。" ), } ], sources=[{"label": "2025年AI漫剧产业深度研究报告", "url": "https://example.com/ai-comic-drama-report"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="最近AI漫剧行业怎么样") normalization = fallback_normalize_observation( question="最近AI漫剧行业怎么样", workspace=workspace, observation=observation, ) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "noisy" assert normalization.usable_claims == [] assert workspace.claim_ledger == [] assert normalization.source_normalizations[0].rejected_reason == "raw_source_noise" @pytest.mark.asyncio async def test_agent_evidence_normalizer_does_not_let_llm_rescue_deterministic_source_noise( monkeypatch: pytest.MonkeyPatch, ) -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "AI漫剧报告候选页", "url": "https://example.com/ai-comic-drama-report", "text": ( "Qwen-Image 定制镜像 | RTX4090D CUDA12.4 大模型推理专用 一键部署。" "预装视觉语言模型依赖库和推理脚本，开箱即用，支持图像理解、多模态推理。" ), } ], sources=[{"label": "AI漫剧报告候选页", "url": "https://example.com/ai-comic-drama-report"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="最近AI漫剧行业怎么样") monkeypatch.setattr(PMFlowAIAgentEvidenceNormalizer, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_EVIDENCE_NORMALIZER")) generator = _SequenceGenerator( [ """ { "source_quality": "usable", "usable_claims": [ { "claim_type": "market_growth", "claim": "AI漫剧行业正在爆发增长，2025年市场规模将突破200亿元。", "customer_value": "说明AI漫剧是近期内容产业的新增长方向。", "confidence": "medium", "relevance_to_user_goal": "direct", "claim_kind": "substantive_fact" } ] } """ ] ) normalization = await PMFlowAIAgentEvidenceNormalizer( db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator, ).normalize(question="最近AI漫剧行业怎么样", workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "noisy" assert normalization.usable_claims == [] assert workspace.claim_ledger == [] assert normalization.source_normalizations[0].rejected_reason == "raw_source_noise" def test_operating_research_evidence_ignores_raw_and_generic_claim_texts() -> None: workspace = { "claim_ledger": [ { "claim_type": "external_source_evidence", "claim_text": "公开来源《Example》提供了与本轮目标相关的材料：Subscribe Sign In Newsletter Privacy Terms Advertisement", }, { "claim_type": "external_source_evidence", "claim_text": "A 2026 industry survey reported 42% revenue growth among vendors, and enterprise buyers increased budgets for specialized solutions.", }, { "claim_type": "external_source_evidence", "claim_text": "3万人 | Example | ### 关键增长驱动因素 - 预计2026年形成$330-480亿美元市场 --- ## 区域市场格局", }, ] } deliverable = { "sections": [ { "section_kind": "claims", "items": [ {"summary": "候选来源需要结合其它来源交叉验证。"}, {"summary": "The report shows demand increased in 2026 as budgets shifted toward practical adoption programs."}, ], } ] } claims = _claim_texts(workspace, deliverable) assert len(claims) == 2 assert all("Subscribe" not in item for item in claims) assert all("候选来源" not in item for item in claims) assert all("###" not in item for item in claims) def test_external_source_document_fallback_extracts_goal_relevant_claims_without_llm() -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "AI Conference Outlook 2026", "url": "https://example.com/ai-conference-outlook", "text": ( "AI Conference Outlook 2026 reports that enterprise buyers are increasing budgets for specialized " "AI conferences and event programs. The same report says organizers are packaging sponsorship, " "training and community formats around practical AI adoption, with revenue growth expected in 2026." ), } ], sources=[{"label": "AI Conference Outlook 2026", "url": "https://example.com/ai-conference-outlook"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") normalization = fallback_normalize_observation(workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "usable" assert normalization.usable_claims assert workspace.claim_ledger assert workspace.claim_ledger[0].claim_kind == "substantive_fact" assert workspace.claim_ledger[0].claim_type == "external_source_evidence" assert "提供了与本轮目标相关的材料" not in workspace.claim_ledger[0].claim_text def test_source_relevance_gate_blocks_raw_ai_market_page_as_research_claim() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="获取近期人工智能相关会议市场的整体态势", desired_output="简要综述 AI 会议市场现状、活动活跃度及主要趋势。", interpretation_branches=[ { "branch_id": "b1", "meaning": "行业活动生态：关注AI领域会议、峰会、论坛等活动的近期动态", "semantic_decomposition": "AI领域 + 会议活动 + 市场活跃度 + 状态评估", "user_language_prior": "high", "confidence": "high", "anchor_terms": ["AI", "会议", "市场"], } ], evidence_requirements=[ "近期主要AI会议活动及其关注热点", "AI会议活动的整体趋势描述", "公开媒体或行业分析对会议活跃度的评价", ], ) observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "中国CIO峰会", "url": "https://example.com/cio-summit", "text": ( "中国CIO峰会 上海站 | 2026年10月23日 深圳站：5月15日 立即注册 " "当前，中国人工智能市场发展迅猛，IDC数据显示，2024年中国生成式AI占AI市场投资总规模的18.2%。" "随着生成式AI技术发展，2029年生成式AI投资占比将达到41.1%，投资规模超450亿美元。" "企业员工可借助生成式AI生成结构化业务洞察，开发人员能够进行代码重构。" ), } ], confidence="medium", relevance_to_goal="high", ) normalization = fallback_normalize_observation( question="最近AI会议市场怎么样", workspace=workspace, observation=observation, ) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality != "usable" assert workspace.claim_ledger == [] def test_research_note_threshold_blocks_single_dimension_external_claims() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="获取近期人工智能相关会议市场的整体态势", desired_output="简要综述 AI 会议市场现状、活动活跃度及主要趋势。", evidence_requirements=[ "近期主要AI会议活动及其关注热点", "AI会议活动的整体趋势描述", "公开媒体或行业分析对会议活跃度的评价", ], interpretation_branches=[ { "branch_id": "b1", "meaning": "行业活动生态：关注AI领域会议、峰会、论坛等活动的近期动态", "user_language_prior": "high", "confidence": "high", } ], ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="external_source_evidence", claim_text="2026年某公开来源显示多个AI主题会议正在聚焦智能体和产业落地热点。", support_refs=["https://source-a.example/report"], interpretation_branch_id="b1", relevance_to_user_goal="partial", claim_kind="substantive_fact", confidence="medium", ), PMFlowAIAgentClaim( claim_type="external_source_evidence", claim_text="另一公开来源显示AI主题会议议程正在聚焦智能体和产业落地热点。", support_refs=["https://source-b.example/report"], interpretation_branch_id="b1", relevance_to_user_goal="partial", claim_kind="substantive_fact", confidence="medium", ), ] result = _enforce_critic_contract( PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, retry_would_change_state=False, recover_or_continue="compose_answer", next_step_hint="compose_answer", ), workspace=workspace, observation=PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", confidence="medium", relevance_to_goal="high", ), ) assert result.ready_to_answer is False assert result.should_continue is True assert any("研究记录" in item for item in result.missing_state) def test_external_source_document_extracts_clean_fact_from_noisy_page_shell() -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "Enterprise Collaboration Summit", "url": "https://example.com/summit", "text": ( "Subscribe Sign In Newsletter Share Privacy Terms Advertisement Contact Us " "Enterprise Collaboration Summit 2026 reports 1,200 enterprise attendees and says sponsors " "increased budgets by 38% as workflow automation moved into mainstream conference programs. " "Read More Premium Site Map Copyright" ), } ], sources=[{"label": "Enterprise Collaboration Summit", "url": "https://example.com/summit"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace( current_goal="最近企业协作峰会市场怎么样", goal_frame=PMFlowAIAgentGoalFrame( user_goal="判断企业协作峰会市场最近怎么样", interpretation_branches=[ { "branch_id": "b1", "meaning": "企业协作峰会和活动市场趋势", "user_language_prior": "high", "anchor_terms": ["enterprise collaboration summit", "conference", "sponsor"], }, { "branch_id": "b2", "meaning": "企业协作软件市场趋势", "user_language_prior": "medium", "anchor_terms": ["collaboration software"], }, ], ), ) normalization = fallback_normalize_observation( question="最近企业协作峰会市场怎么样", workspace=workspace, observation=observation, ) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "usable" assert workspace.claim_ledger claim = workspace.claim_ledger[0] assert claim.interpretation_branch_id == "b1" assert "1,200 enterprise attendees" in claim.claim_text assert "Subscribe Sign In Newsletter" not in claim.claim_text assert "Contact Us" not in claim.claim_text def test_operating_source_rows_use_clean_gated_fragments_instead_of_page_shell() -> None: observations = [ { "source_kind": "inspect_source_detail", "observation_type": "external_source_document", "facts": [ { "type": "external_source_text", "title": "Enterprise Collaboration Summit", "url": "https://example.com/summit", "text": ( "Subscribe Sign In Newsletter Privacy Terms Advertisement " "The 2026 enterprise collaboration summit drew 1,200 attendees and reported 38% sponsor budget growth." ), } ], } ] rows = _external_source_rows(observations) assert len(rows) == 1 assert "1,200 attendees" in rows[0]["summary"] assert "Subscribe Sign In" not in rows[0]["summary"] def test_agent_critic_state_delta_new_facts_enter_claim_ledger() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") critic = PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, state_delta={ "new_facts": [ { "claim_text": "公开材料显示，AI会议辅助工具市场在2022年约15亿美元，预计2030年达到62亿美元。", "source_label": "Grand View Research", "confidence": "medium", } ] }, reason="已有可入账事实。", ) record_critic_ledgers(workspace, critic_result=critic) assert workspace.claim_ledger assert "62亿美元" in workspace.claim_ledger[0].claim_text assert workspace.fact_ledger assert "Grand View Research" in workspace.fact_ledger[0].support_refs @pytest.mark.asyncio async def test_agent_evidence_normalizer_timeboxes_slow_llm_and_falls_back(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("PMFLOW_AI_AGENT_EVIDENCE_NORMALIZER_TIMEOUT_SECONDS", "0.1") class SlowGenerator: async def generate(self, *_args, **_kwargs): await asyncio.sleep(1) return "{}" observation = PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源。", facts=[ { "type": "external_evidence", "title": "AI Meeting Assistants 2026", "summary": "This page may contain related market material.", "url": "https://example.com/ai-meeting-assistants", } ], sources=[{"label": "AI Meeting Assistants 2026", "url": "https://example.com/ai-meeting-assistants"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议行业怎么样") normalization = await PMFlowAIAgentEvidenceNormalizer( db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=SlowGenerator(), ).normalize(question="最近AI会议行业怎么样", workspace=workspace, observation=observation) assert normalization.source_quality == "noisy" assert normalization.usable_claims == [] assert normalization.boundaries @pytest.mark.asyncio async def test_agent_node_runtime_profile_has_platform_contract_fallback() -> None: profile = await resolve_agent_llm_runtime_profile(object(), stage="answer_composer") assert profile.function_name == "pmflow_ai_agent_answer_composer" assert profile.prompt_key == "PROMPT_PMFLOW_AI_AGENT_ANSWER_COMPOSER" assert profile.scene_key == "space_pmflow_ai_agent_runtime" assert profile.routing_meta["function_runtime_profile_source"] == "code_default" def test_agent_semantic_nodes_do_not_cut_off_provider_wait() -> None: assert _effective_foreground_wait_seconds( stage="goal_frame", provider_timeout_seconds=30, configured_foreground_wait_seconds=8, ) == 30 assert _effective_foreground_wait_seconds( stage="step_planner", provider_timeout_seconds=30, configured_foreground_wait_seconds=25, ) == 30 assert _effective_foreground_wait_seconds( stage="evidence_normalizer", provider_timeout_seconds=60, configured_foreground_wait_seconds=16, ) == 60 assert _effective_foreground_wait_seconds( stage="answer_composer", provider_timeout_seconds=35, configured_foreground_wait_seconds=10, ) == 35 def test_external_search_goal_frame_latency_hint_does_not_shrink_provider_timeout() -> None: proposal = PMFlowAIAgentStepProposal( action="tool", tool_name="external_search", tool_input={"query": "国家医保局 近半年 政策动态"}, rationale="需要外部公开材料。", expected_observation="公开来源候选。", ) goal_frame = SimpleNamespace( latency_policy={ "mode": "standard", "external_search_timeout_seconds": "12", } ) repaired = _proposal_with_goal_frame_runtime_hints(proposal=proposal, goal_frame=goal_frame) assert "provider_timeout_seconds" not in repaired.tool_input assert repaired.tool_input["result_limit"] == 6 def test_external_source_candidate_ranking_prefers_authoritative_public_domains() -> None: proposal = PMFlowAIAgentStepProposal( action="tool", tool_name="external_search", tool_input={"query": "国家医保局 近半年 政策动态"}, rationale="需要外部公开材料。", expected_observation="公开来源候选。", ) ranked = _rank_source_candidates( rows=[ { "url": "https://news.example.com/2026-policy", "title": "国家医保局 2026 政策动态解读", "summary": "转载解读。", }, { "url": "https://www.nhsa.gov.cn/art/2025/12/7/art_14_18972.html", "title": "国家医保局政策动态", "summary": "官方发布。", }, ], workspace=None, proposal=proposal, ) assert ranked[0]["url"].startswith("https://www.nhsa.gov.cn/") def test_external_evidence_ranking_filters_explicitly_stale_items_when_freshness_required() -> None: proposal = PMFlowAIAgentStepProposal( action="tool", tool_name="external_search", tool_input={"query": "国家医保局 近半年 政策动态", "freshness_required": True}, rationale="需要外部公开材料。", expected_observation="公开来源候选。", ) ranked = _rank_external_evidence( evidence=[ ExternalEvidenceItem( source_type="url", title="国家医保局 2024 年 DRG/DIP 政策解读", target_url="https://www.nhsa.gov.cn/art/2024/7/23/art_105_13316.html", snippet="2024 年发布。", raw_metadata={"published_at": "2024-07-23"}, ), ExternalEvidenceItem( source_type="url", title="国家医保局 2026 年政策动态", target_url="https://www.nhsa.gov.cn/art/2026/3/1/example.html", snippet="2026 年发布。", raw_metadata={"published_at": "2026-03-01"}, ), ], proposal=proposal, question="国家医保局 近半年 政策动态", ) assert [item.target_url for item in ranked] == ["https://www.nhsa.gov.cn/art/2026/3/1/example.html"] def test_external_source_document_can_use_deterministic_normalizer_in_standard_mode() -> None: workspace = PMFlowAIAgentWorkspace( current_goal="分析下近半年国家医保的政策动态", goal_frame=PMFlowAIAgentGoalFrame( user_goal="分析下近半年国家医保的政策动态", answer_objective="提取公开来源里的干净政策判断。", source_policy="credible_public_sources", latency_policy={"mode": "standard"}, ), ) observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "url": "https://www.nhsa.gov.cn/art/2026/5/9/art_105_20455.html", "text": "国家医疗保障局 政策解读 2026年目录调整目标任务...", } ], ) assert _should_use_deterministic_external_normalizer(workspace=workspace, observation=observation) is True @pytest.mark.asyncio async def test_agent_evidence_normalizer_runtime_mode_bypass_skips_llm(monkeypatch: pytest.MonkeyPatch) -> None: async def fake_resolve(*_args, **_kwargs): return SimpleNamespace(runtime_mode="bypass") async def fail_render(*_args, **_kwargs): raise AssertionError("prompt should not render when evidence normalizer is bypassed") class FailGenerator: async def generate(self, *_args, **_kwargs): raise AssertionError("llm should not be called when evidence normalizer is bypassed") monkeypatch.setattr(agent_evidence_normalizer_module, "resolve_agent_llm_runtime_profile", fake_resolve) monkeypatch.setattr(PMFlowAIAgentEvidenceNormalizer, "_render_prompt", fail_render) observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "AI 会议市场", "text": "2026 年 AI 会议工具需求继续增长，企业正在把会议纪要转向协作工作流。", } ], confidence="medium", relevance_to_goal="high", ) normalization = await PMFlowAIAgentEvidenceNormalizer( db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=FailGenerator(), ).normalize(question="最近AI会议市场怎么样", workspace=PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样"), observation=observation) assert normalization.rejected_reason == "evidence_normalizer_runtime_mode_bypass" @pytest.mark.asyncio async def test_agent_answer_quality_gate_mode_reads_runtime_profile(monkeypatch: pytest.MonkeyPatch) -> None: async def fake_resolve(*_args, **_kwargs): return SimpleNamespace(quality_gate_mode="off") monkeypatch.setattr(agent_answer_runtime_module, "resolve_agent_llm_runtime_profile", fake_resolve) assert await PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object())._quality_gate_mode() == "off" @pytest.mark.asyncio async def test_agent_answer_runtime_keeps_provider_tokens_private(monkeypatch: pytest.MonkeyPatch) -> None: async def fake_runtime_profile(_db, *, stage: str): return SimpleNamespace( function_name=f"pmflow_ai_agent_{stage}", prompt_key="PROMPT_PMFLOW_AI_AGENT_ANSWER_COMPOSER", scene_key="space_pmflow_ai_agent_runtime", temperature=0.2, response_mode="fast", max_tokens=2600, timeout_seconds=10, runtime_mode="active", quality_gate_mode="off", routing_meta={}, ) events: list[dict[str, object]] = [] generator = _SequenceGenerator(["最终可见的 Markdown 答案。"]) async def collect_event(event: dict[str, object]) -> None: events.append(event) monkeypatch.setattr(agent_answer_runtime_module, "resolve_agent_llm_runtime_profile", fake_runtime_profile) monkeypatch.setattr(PMFlowAIAgentAnswerRuntime, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_ANSWER_COMPOSER")) result = await PMFlowAIAgentAnswerRuntime( db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator, ).run( question="帮我判断一下", workspace=PMFlowAIAgentWorkspace(current_goal="帮我判断一下"), proposal=PMFlowAIAgentStepProposal(action="answer", rationale="材料已足够"), observation=None, critic_result=PMFlowAIAgentCriticResult(ready_to_answer=True, reason="ready"), event_sink=collect_event, ) assert result.stream_mode == "final_answer_snapshot" assert result.answer == "最终可见的 Markdown 答案。" assert [event.get("type") for event in events] == ["answer.finalizing"] assert not any(event.get("type") == "answer.delta" for event in events) @pytest.mark.asyncio async def test_agent_evidence_normalizer_uses_deterministic_extraction_when_llm_only_returns_gaps( monkeypatch: pytest.MonkeyPatch, ) -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "智能协作会议系统市场观察", "url": "https://example.com/collaboration-meeting-market", "text": ( "2025年智能会议系统市场规模达到186亿元，具备AI能力的产品占比超过45%。" "报告预计2027年AI会议系统市占率将突破75%，会议场景正在从单点纪要转向协作工作流。" ), } ], sources=[{"label": "智能协作会议系统市场观察", "url": "https://example.com/collaboration-meeting-market"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="最近智能会议市场怎么样") monkeypatch.setattr(PMFlowAIAgentEvidenceNormalizer, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_EVIDENCE_NORMALIZER")) generator = _SequenceGenerator( [ """ { "source_quality": "noisy", "gaps": ["模型没有抽出可用事实"] } """ ] ) normalization = await PMFlowAIAgentEvidenceNormalizer( db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator, ).normalize(question="最近智能会议市场怎么样", workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "usable" assert workspace.claim_ledger assert "186亿元" in workspace.claim_ledger[0].claim_text assert "模型没有抽出可用事实" in normalization.gaps @pytest.mark.asyncio async def test_agent_evidence_normalizer_prefers_llm_high_resolution_claims( monkeypatch: pytest.MonkeyPatch, ) -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "公开市场研究", "url": "https://example.com/meeting-ai-market", "text": "Report shell with sparse snippets.", } ], sources=[{"label": "公开市场研究", "url": "https://example.com/meeting-ai-market"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="最近智能会议工具市场怎么样") monkeypatch.setattr(PMFlowAIAgentEvidenceNormalizer, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_EVIDENCE_NORMALIZER")) generator = _SequenceGenerator( [ """ { "source_quality": "usable", "source_normalizations": [ { "source_url": "https://example.com/meeting-ai-market", "source_title": "公开市场研究", "source_quality": "usable", "usable_claims": [ { "claim_type": "market_growth", "claim": "公开材料显示，智能会议助手市场正在从会议纪要自动化扩展到行动项追踪、销售对话分析和协作工作流，需求侧由企业提升会议效率驱动。", "customer_value": "智能会议助手市场的近期机会不只在自动生成纪要，而是在把会议内容接入后续协作和业务流程。", "confidence": "medium", "support_refs": ["https://example.com/meeting-ai-market"], "interpretation_branch_id": "b1", "relevance_to_user_goal": "direct", "claim_kind": "directional_judgement" } ], "boundaries": [ { "boundary_type": "coverage_boundary", "blocked_claim": "不能据此断言全行业份额排名。", "customer_safe_reason": "当前材料还没有覆盖完整竞争份额。", "allowed_alternative": "继续补充竞品和融资来源。" } ] } ] } """ ] ) normalization = await PMFlowAIAgentEvidenceNormalizer( db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator, ).normalize(question="最近智能会议工具市场怎么样", workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "usable" assert workspace.claim_ledger assert "行动项追踪" in workspace.claim_ledger[0].claim_text assert workspace.claim_ledger[0].customer_value.startswith("智能会议助手市场的近期机会") assert "把判断建立在已读取正文" not in workspace.claim_ledger[0].customer_value def test_attention_queue_normalization_keeps_action_candidates() -> None: observation = PMFlowAIAgentObservation( source="inspect_attention_queue", observation_type="attention_queue", summary="已读取 1 条客户安全注意力队列候选。", facts=[ { "type": "action_candidate", "title": "审查任务 194 的 PR 结果", "summary": "任务 194 当前 PR 已提交，需要确认验收状态。", "target_ref": "project:26", "target_title": "儿童项目", "href": "/execution-bridge?projectId=26&taskId=194", "reason": "执行状态 pr_open，下一步是人工审查。", "completion_criteria": "PR 审查完成，验收状态更新。", "safety": "prepare_only", "requires_confirmation": True, }, { "type": "object_affordance", "object_ref": "project:26", "title": "儿童项目", "action_label": "审查任务 194 的 PR 结果", "href": "/execution-bridge?projectId=26&taskId=194", }, ], sources=[{"label": "PMFlow 注意力队列"}], confidence="high", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="儿童那个项目下一步我要做什么呢") normalization = fallback_normalize_observation(workspace=workspace, observation=observation) apply_evidence_normalization(workspace, normalization=normalization) assert normalization.source_quality == "usable" assert workspace.action_candidate_ledger candidate = workspace.action_candidate_ledger[0] assert candidate.title == "审查任务 194 的 PR 结果" assert candidate.href == "/execution-bridge?projectId=26&taskId=194" assert "PR 审查完成" in candidate.completion_criteria assert workspace.affordance_ledger[0].object_ref == "project:26" def test_terminal_answer_uses_action_candidate_without_generic_industry_fallback() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="儿童那个项目下一步我要做什么呢") workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="project_execution_package", claim_text="儿童项目当前选中任务 194，执行状态是 PR 已提交。", customer_value="支撑项目下一步判断。", confidence="high", ) ] workspace.action_candidate_ledger = [ PMFlowAIAgentActionCandidateLedgerEntry( title="审查任务 194 的 PR 结果", target_ref="project:26", target_title="儿童项目", href="/execution-bridge?projectId=26&taskId=194", reason="执行状态 pr_open，下一步需要确认交付与验收。", completion_criteria="PR 审查完成，验收状态更新。", safety="prepare_only", confidence="high", ) ] terminal_state = build_agent_terminal_answer_state( question="儿童那个项目下一步我要做什么呢", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="answer"), observation=None, critic_result=PMFlowAIAgentCriticResult(ready_to_answer=True, should_continue=False), ) answer = render_agent_terminal_state_fallback(terminal_state) assert "审查任务 194 的 PR 结果" in answer assert "PR 审查完成" in answer assert "行业信号" not in answer assert "受控初判" not in answer @pytest.mark.asyncio async def test_project_execution_package_preserves_selected_task_action_detail() -> None: registry = PMFlowAIAgentToolRegistry(db=object(), tenant_id=12) package = await registry._build_project_execution_package_payload( project_id=26, project_ref="project:26", project_title="儿童项目", execution_payload={ "execution": {"taskId": 194, "status": "pr_open", "testSummary": {"status": "passed"}}, "activeDevTasks": [{"id": 194, "title": "修复儿童项目首页登录链路", "status": "in_review"}], "acceptance": {"status": "pending_review"}, "releaseGate": {"status": "blocked"}, "qualityContract": {"summary": {"releaseBlockingMissing": 1}}, }, truth_payload={"selectedTaskId": 194}, ) # Use the public observation projection shape that the normalizer consumes. from app.services.pmflow_ai_agent_runtime.agent_tool_registry import _execution_package_facts projected = _execution_package_facts(package) selected = next(item for item in projected if item.get("type") == "selected_task") action = next(item for item in projected if item.get("type") == "action_candidate") assert selected["task_id"] == "194" assert selected["title"] == "修复儿童项目首页登录链路" assert action["target_ref"] == "project:26" assert action["href"] == "/execution-bridge?projectId=26&taskId=194" assert "发布阻塞级缺失证据" in action["completion_criteria"] assert "补齐 1 个发布阻塞级证据" in action["title"] def test_external_source_detail_rejects_404_and_navigation_noise() -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "404 页面不存在", "url": "https://example.com/missing", "text": "404 页面不存在 登录 注册 隐私政策 用户协议 Copyright 网站地图", } ], sources=[{"label": "404 页面不存在", "url": "https://example.com/missing"}], confidence="medium", relevance_to_goal="high", ) workspace = PMFlowAIAgentWorkspace(current_goal="我这个项目和宝宝树对比怎么样") normalization = fallback_normalize_observation(workspace=workspace, observation=observation) assert normalization.source_quality == "noisy" assert normalization.usable_claims == [] assert normalization.boundaries @pytest.mark.asyncio async def test_agent_answer_runtime_returns_chinese_partial_answer_when_safe_claims_exist() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI行业怎么样") workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="market_investment_trend", claim_text="Global AI investment is projected to reach $300 billion by 2026.", support_refs=["Morgan Stanley"], confidence="medium", ), PMFlowAIAgentClaim( claim_type="model_performance_trend", claim_text="Open-weight models are narrowing the performance gap with proprietary models.", support_refs=["LLM Stats"], confidence="medium", ), ] workspace.boundary_ledger = [ PMFlowAIAgentBoundary( boundary_type="freshness_gap", blocked_claim="缺少 2026 年 5 月之后的实时材料。", customer_safe_reason="这轮公开材料还没有覆盖 2026 年 5 月之后的变化。", allowed_alternative="继续补充最新公开来源后，再把判断更新成完整结论。", ) ] result = await PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()).run( question="最近AI行业怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="stop_blocked", rationale="停止"), observation=None, critic_result=PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=False, retry_would_change_state=False, reason="已有部分 claim，但缺少实时材料。", ), ) assert result.terminal_state.terminal_status == "partially_answerable" assert result.deliverable_packet.status == "partial" assert result.deliverable_packet.deliverable_kind == "decision_packet" assert result.quality_findings == [] assert "Global AI investment is projected to reach $300 billion by 2026" in result.answer assert "Open-weight models are narrowing the performance gap" in result.answer assert any("\u4e00" <= ch <= "\u9fff" for ch in result.answer) assert result.deliverable_packet.sections def test_agent_answer_critic_allows_longer_customer_safe_blocked_answer() -> None: terminal_state = PMFlowAIAgentTerminalAnswerState( user_goal="最近AI会议行业怎么样", terminal_status="waiting_for_evidence", boundaries=[ PMFlowAIAgentBoundary( boundary_type="source_gap", blocked_claim="缺少会议行业垂直公开来源。", customer_safe_reason="这轮查到的材料偏向泛 AI 行业，不能直接说明 AI 会议行业最近的发展。", allowed_alternative="继续换用会议技术、活动平台和标准组织来源补证据。", ) ], answer_intent_plan=PMFlowAIAgentAnswerIntentPlan(lead_boundary_id="boundary_missing", detail_level="brief"), ) answer = ( "目前还不能可靠判断最近 AI 会议行业怎么样。已经查到的材料更多是在讲泛 AI 趋势、语音平台或宏观产业变化，" "没有足够覆盖会议与活动技术场景，例如虚拟/混合会议平台的新功能、实时翻译在会场中的使用、智能议程生成、" "线下和线上融合的落地进展，也缺少最近三个月内可追溯的垂直来源。" "所以这轮更稳妥的结论是：不能把现有材料当作行业判断依据。下一步应该换查 PCMA、MPI、Zoom、Teams、Webex、" "Airmeet、Run The World、IEEE 或 ISO/IEC 这类会议技术、平台公告和标准组织来源，再用发布时间和具体事件交叉确认。" ) assert critique_agent_customer_answer(terminal_state=terminal_state, answer=answer) == [] @pytest.mark.asyncio async def test_agent_state_critic_respects_recovery_veto_when_more_reading_would_change_state() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议行业怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议行业怎么样", desired_output="给出方向性判断并说明边界。", answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="vendor_product_update", claim_text="公开来源显示会议平台正在把 AI 会议能力从摘要扩展到工作流和后续执行。", customer_value="支撑方向性行业判断。", support_refs=["公开来源"], confidence="medium", ) ] workspace.gap_ledger = [ # The remaining gap should become a boundary, not a hard blocker. PMFlowAIAgentGapLedgerEntry( gap_type="market_size_gap", description="缺少可用于判断精确市场规模和份额的独立报告。", blocking=True, user_actionable=False, ) ] generator = _SequenceGenerator( [ """ { "ready_to_answer": false, "should_continue": true, "needs_user_input": false, "retry_would_change_state": true, "state_delta": {"remaining_gaps": ["缺少完整报告"]}, "findings": ["还没覆盖全部维度"], "missing_state": ["完整行业报告"], "recover_or_continue": "continue_with_tool", "next_step_hint": "select_next_read_tool", "reason": "仍缺少完整证据。" } """ ] ) result = await PMFlowAIAgentStateCritic(db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator).review( question="最近AI会议行业怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="tool", tool_name="external_search"), observation=PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源。", confidence="medium", relevance_to_goal="high", ), ) assert result.ready_to_answer is False assert result.should_continue is True assert result.retry_would_change_state is True assert result.recover_or_continue == "continue_with_tool" assert result.next_step_hint == "select_next_read_tool" assert any("缺少可用于判断精确市场规模和份额的独立报告" in item for item in result.state_delta["remaining_gaps"]) @pytest.mark.asyncio async def test_agent_state_critic_does_not_answer_from_source_candidate_only(monkeypatch: pytest.MonkeyPatch) -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议行业怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议行业怎么样", desired_output="给出方向性判断并说明边界。", answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="external_source_candidate", claim_text="公开来源《AI Meeting Assistants 2026》提供了与本轮目标相关的外部材料，但仍需要结合其它来源交叉验证后再下强结论。", customer_value="确认本轮有可追溯外部材料可进入后续抽取和交叉验证。", support_refs=["https://example.com/ai-meeting-assistants"], confidence="medium", ) ] generator = _SequenceGenerator( [ """ { "ready_to_answer": false, "should_continue": true, "needs_user_input": false, "retry_would_change_state": true, "state_delta": {"remaining_gaps": ["还没有业务事实 claim"]}, "findings": ["只有候选来源"], "missing_state": ["需要读取正文或换来源抽取业务事实"], "recover_or_continue": "continue_with_tool", "next_step_hint": "select_next_read_tool", "reason": "候选来源不能支撑最终回答。" } """ ] ) monkeypatch.setattr(PMFlowAIAgentStateCritic, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_STATE_CRITIC")) result = await PMFlowAIAgentStateCritic(db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator).review( question="最近AI会议行业怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="tool", tool_name="external_search"), observation=PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源。", confidence="medium", relevance_to_goal="high", ), ) assert result.ready_to_answer is False assert result.should_continue is True assert result.recover_or_continue == "continue_with_tool" assert result.next_step_hint == "select_next_read_tool" assert any("业务事实" in item or "读取正文" in item for item in result.missing_state) @pytest.mark.asyncio async def test_agent_state_critic_vetoes_raw_page_fragment_claim(monkeypatch: pytest.MonkeyPatch) -> None: workspace = PMFlowAIAgentWorkspace(current_goal="做一轮公开市场研究") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="做一轮公开市场研究", desired_output="给出方向性判断并说明边界。", answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="external_source_evidence", claim_text="Forbes Councils Paid Program Subscribe Sign In Newsletter Share This Article Contact Us Privacy Terms Copyright Advertisement", customer_value="支撑公开研究。", support_refs=["https://example.com/noisy"], confidence="medium", ) ] generator = _SequenceGenerator( [ """ { "ready_to_answer": true, "should_continue": false, "needs_user_input": false, "retry_would_change_state": false, "state_delta": {}, "findings": ["已有材料"], "missing_state": [], "recover_or_continue": "compose_answer", "next_step_hint": "compose_answer", "reason": "模型误判为可回答。" } """ ] ) monkeypatch.setattr(PMFlowAIAgentStateCritic, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_STATE_CRITIC")) result = await PMFlowAIAgentStateCritic(db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator).review( question="做一轮公开市场研究", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="tool", tool_name="inspect_source_detail"), observation=PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", confidence="medium", relevance_to_goal="high", ), ) assert result.ready_to_answer is False assert result.should_continue is False assert result.recover_or_continue == "stop_blocked" assert any("Claim Ledger" in item for item in result.missing_state) def test_goal_frame_context_sanitizer_drops_prior_blocked_research_hypotheses() -> None: safe = _goal_frame_safe_conversation_context( { "session_id": 654, "last_subject": "最近AI会议行业怎么样", "context_capsules": [ { "capsule_type": "pmflow_agent_run", "status": "blocked", "research_goal": "理解最近三个月内AI会议行业的整体发展状况", "agent_workspace": {"current_goal": "围绕 NeurIPS/ICML 搜索"}, }, { "capsule_type": "external_research", "topic": "AI conferences 2026", "research_goal": "学术会议行业", }, { "capsule_type": "pmflow_ai_object_ledger", "current_focus_object": { "object_id": "project:1", "object_type": "project", "display_name": "童析一学龄前儿童行为评估助手", "binding_strength": "session_object_ledger", }, }, ], }, question="最近AI会议市场怎么样", ) rendered = str(safe) assert safe["current_user_question"] == "最近AI会议市场怎么样" assert "research_goal" not in rendered assert "NeurIPS" not in rendered assert "external_research" not in rendered assert "pmflow_agent_run" not in rendered assert safe["context_capsules"][0]["current_focus_object"]["display_name"] == "童析一学龄前儿童行为评估助手" def test_goal_frame_does_not_keyword_repair_compact_modifier_phrase() -> None: frame = PMFlowAIAgentGoalFrame( user_goal="理解最近XR教育市场怎么样", desired_output="给出方向性判断。", interpretation_branches=[ { "branch_id": "b_academic", "meaning": "围绕 XR 主题举办的教育活动和内容组织生态", "confidence": "high", "user_language_prior": "high", }, { "branch_id": "b_event", "meaning": "商业 XR 教育论坛和展会的活动市场", "confidence": "medium", "user_language_prior": "medium", }, ], selected_interpretation="围绕 XR 主题举办的教育活动和内容组织生态", ) repaired = _repair_goal_frame_for_open_semantics(frame=frame, question="最近XR教育市场怎么样") branch_ids = [item["branch_id"] for item in repaired.interpretation_branches] assert "b_compound_application_context" not in branch_ids assert [item["branch_id"] for item in repaired.interpretation_branches[:2]] == ["b_academic", "b_event"] assert repaired.selected_interpretation == "围绕 XR 主题举办的教育活动和内容组织生态" def test_goal_frame_repairs_selected_interpretation_branch_id_to_meaning() -> None: frame = PMFlowAIAgentGoalFrame( user_goal="判断AI会议市场", desired_output="给出方向性判断。", selected_interpretation="b1", interpretation_branches=[ { "branch_id": "b1", "meaning": "AI 会议软件/工具/服务市场", "confidence": "high", "user_language_prior": "high", }, { "branch_id": "b2", "meaning": "AI 主题会议/大会/活动生态", "confidence": "medium", "user_language_prior": "medium", }, ], ) repaired = _repair_goal_frame_for_open_semantics(frame=frame, question="最近AI会议市场怎么样") assert repaired.selected_interpretation == repaired.interpretation_branches[0]["meaning"] assert repaired.interpretation_branches[0]["branch_id"] == "b1" def test_research_contract_expands_external_search_from_dynamic_lanes() -> None: frame = PMFlowAIAgentGoalFrame( user_goal="分析一个近期政策方向", desired_output="给出覆盖充分的方向性判断。", evidence_lanes=[ {"lane_id": "lane_a", "lane": "制度目标变化", "purpose": "判断政策意图", "minimum_useful_signal": "官方解释或一手文件"}, {"lane_id": "lane_b", "lane": "执行机制变化", "purpose": "判断落地影响", "minimum_useful_signal": "实施方案或执行通知"}, {"lane_id": "lane_c", "lane": "监管约束变化", "purpose": "判断风险边界", "minimum_useful_signal": "监管行动或检查计划"}, ], ) proposal = PMFlowAIAgentStepProposal( action="tool", tool_name="external_search", tool_input={"query": "分析一个近期政策方向", "search_queries": ["分析一个近期政策方向"]}, ) enriched = enrich_external_search_proposal(proposal=proposal, goal_frame=frame, question="分析一个近期政策方向") branch_searches = enriched.tool_input["branch_searches"] assert {item["branch_id"] for item in branch_searches} >= {"lane_a", "lane_b", "lane_c"} assert enriched.tool_input["requires_coverage"] is True assert len(enriched.tool_input["search_queries"]) >= 4 def test_answer_design_plan_uses_structure_primitives_not_scene_templates() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="做一个开放方向判断") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="做一个开放方向判断", answer_objective={"user_success": "用户需要快速比较多个外部对象并知道下一步怎么选。"}, ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="capability_signal", claim_text="对象 A 在公开材料中强调检索和引用。", support_refs=["https://example.com/a"], claim_kind="substantive_fact", confidence="medium", ), PMFlowAIAgentClaim( claim_type="workflow_signal", claim_text="对象 B 在公开材料中强调工作流集成。", support_refs=["https://example.com/b"], claim_kind="substantive_fact", confidence="medium", ), PMFlowAIAgentClaim( claim_type="risk_signal", claim_text="对象 C 的公开材料显示跨源可信度仍是边界。", support_refs=["https://example.com/c"], claim_kind="substantive_fact", confidence="low", ), ] workspace.boundary_ledger = [ PMFlowAIAgentBoundary( boundary_type="source_boundary", blocked_claim="仍缺少用户评价来源。", customer_safe_reason="用户评价还没有覆盖，不能把接受度说成确定结论。", ) ] terminal_state = build_agent_terminal_answer_state( question="做一个开放方向判断", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="answer"), observation=PMFlowAIAgentObservation(source="inspect_source_detail", observation_type="external_source_document"), critic_result=PMFlowAIAgentCriticResult(ready_to_answer=True, should_continue=False), ) plan = terminal_state.answer_intent_plan assert "multi_dimension_comparison" in plan.content_relationships assert "comparison_matrix" in plan.block_order assert "source_digest" in plan.block_order assert plan.structure_preference == "lead_then_comparison_table" assert plan.avoid_sections is False assert "competitor" not in json.dumps(plan.model_dump(mode="json"), ensure_ascii=False).lower() def test_answer_canvas_projects_structured_answer_blocks_from_ledgers() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="做一个开放方向判断") workspace.goal_frame = PMFlowAIAgentGoalFrame(user_goal="做一个开放方向判断") workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="capability_signal", claim_text="对象 A 在公开材料中强调检索和引用。", support_refs=["https://example.com/a"], claim_kind="substantive_fact", confidence="medium", ), PMFlowAIAgentClaim( claim_type="workflow_signal", claim_text="对象 B 在公开材料中强调工作流集成。", support_refs=["https://example.com/b"], claim_kind="substantive_fact", confidence="medium", ), PMFlowAIAgentClaim( claim_type="risk_signal", claim_text="对象 C 的公开材料显示跨源可信度仍是边界。", support_refs=["https://example.com/c"], claim_kind="substantive_fact", confidence="low", ), ] terminal_state = build_agent_terminal_answer_state( question="做一个开放方向判断", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="answer"), observation=PMFlowAIAgentObservation(source="inspect_source_detail", observation_type="external_source_document"), critic_result=PMFlowAIAgentCriticResult(ready_to_answer=True, should_continue=False), ) canvas = build_agent_answer_canvas( answer="这轮可以先比较三个对象，再选择切入口。", workspace=workspace, terminal_state=terminal_state, deliverable_packet=None, ) block_types = [item["type"] for item in canvas["answer_blocks"]] assert "comparison_matrix" in block_types assert "evidence_backed_claims" in block_types assert "source_digest" in block_types matrix = next(item for item in canvas["answer_blocks"] if item["type"] == "comparison_matrix") assert len(matrix["rows"]) == 3 assert matrix["columns"] == ["对象/维度", "判断", "依据", "置信度"] def test_research_contract_does_not_turn_branch_explanations_into_search_queries() -> None: frame = PMFlowAIAgentGoalFrame( user_goal="推进一个新产品想法", desired_output="给出可验证的方向判断。", interpretation_branches=[ { "branch_id": "b_context_noise", "meaning": "现有项目功能扩展：在“童析”或其他当前项目中加入该能力。这个解释来自可见工作区噪声。", "semantic_decomposition": "把当前空间项目和新想法做了过强绑定。", } ], evidence_lanes=[ { "lane_id": "lane_value", "lane": "用户为什么需要这个新能力", "purpose": "判断是否有清晰价值闭环，不能把工作区项目噪声带入检索。", "minimum_useful_signal": "真实用户问题或公开材料。", } ], ) proposal = PMFlowAIAgentStepProposal( action="tool", tool_name="external_search", tool_input={"query": "做个AI简报助手"}, ) enriched = enrich_external_search_proposal(proposal=proposal, goal_frame=frame, question="做个AI简报助手") query_text = " ".join(enriched.tool_input["search_queries"]) assert "童析" not in query_text assert "可见工作区噪声" not in query_text assert all(len(item) <= 180 for item in enriched.tool_input["search_queries"]) def test_research_coverage_gap_requires_dynamic_lane_breadth() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="分析一个近期政策方向") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="分析一个近期政策方向", latency_policy={"mode": "standard"}, evidence_lanes=[ {"lane_id": "lane_a", "lane": "制度目标变化", "purpose": "判断政策意图"}, {"lane_id": "lane_b", "lane": "执行机制变化", "purpose": "判断落地影响"}, {"lane_id": "lane_c", "lane": "监管约束变化", "purpose": "判断风险边界"}, ], ) claims = [ PMFlowAIAgentClaim( claim_type="policy_update", claim_text="一手文件说明制度目标正在调整。", evidence_lane="lane_a", claim_kind="substantive_fact", relevance_to_user_goal="direct", confidence="high", ) ] gap = research_coverage_gap(workspace=workspace, claims=claims) assert "1/3" in gap assert "执行机制变化" in gap def test_research_coverage_gap_requires_public_claim_for_public_source_goal() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="把当前项目和外部公开信息结合起来判断") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="把当前项目和外部公开信息结合起来判断", source_policy="credible_public_sources", latency_policy={"mode": "standard"}, evidence_lanes=[ { "lane_id": "internal_anchor", "lane": "项目内部定位", "source_policy": "internal_customer_safe", "candidate_tools": ["inspect_project_execution_package"], }, { "lane_id": "public_reference", "lane": "外部公开参照", "source_policy": "credible_public_sources", "candidate_tools": ["external_search"], }, ], ) claims = [ PMFlowAIAgentClaim( claim_type="project_positioning_fact", claim_text="内部执行包说明项目已有一个明确定位。", evidence_lane="internal_anchor", source_tool="inspect_project_execution_package", source_kind="project_execution_package", claim_kind="substantive_fact", relevance_to_user_goal="direct", confidence="high", ) ] gap = research_coverage_gap(workspace=workspace, claims=claims) assert "1/2" in gap assert "外部公开参照" in gap @pytest.mark.asyncio async def test_agent_state_critic_does_not_preflight_answer_before_public_lane_is_read() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="把当前项目和外部公开信息结合起来判断") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="把当前项目和外部公开信息结合起来判断", source_policy="credible_public_sources", missing_evidence_policy="answer_with_boundaries", latency_policy={"mode": "standard"}, evidence_lanes=[ { "lane_id": "internal_anchor", "lane": "项目内部定位", "source_policy": "internal_customer_safe", "candidate_tools": ["inspect_project_execution_package"], }, { "lane_id": "public_reference", "lane": "外部公开参照", "source_policy": "credible_public_sources", "candidate_tools": ["external_search"], }, ], ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="project_positioning_fact", claim_text="内部执行包说明项目已有一个明确定位。", evidence_lane="internal_anchor", source_tool="inspect_project_execution_package", source_kind="project_execution_package", claim_kind="substantive_fact", relevance_to_user_goal="direct", confidence="high", ) ] result = await PMFlowAIAgentStateCritic(db=object(), tenant_id=12, generator=object()).review( question="把当前项目和外部公开信息结合起来判断", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="tool", tool_name="inspect_project_execution_package"), observation=PMFlowAIAgentObservation( source="inspect_project_execution_package", observation_type="project_execution_package", summary="已读取内部执行包。", confidence="high", ), ) assert result.ready_to_answer is False assert result.should_continue is True assert result.next_step_hint == "select_next_read_tool" assert any("外部公开参照" in item for item in result.missing_state) def test_recovery_proposal_uses_goal_contract_to_continue_with_external_search_after_internal_read() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="把当前项目和外部公开信息结合起来判断") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="把当前项目和外部公开信息结合起来判断", source_policy="credible_public_sources", latency_policy={"mode": "standard"}, evidence_lanes=[ { "lane_id": "internal_anchor", "lane": "项目内部定位", "source_policy": "internal_customer_safe", "candidate_tools": ["inspect_project_execution_package"], }, { "lane_id": "public_reference", "lane": "外部公开参照", "source_policy": "credible_public_sources", "candidate_tools": ["external_search"], }, ], ) critic_result = PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, retry_would_change_state=True, state_delta={"remaining_gaps": ["还缺外部公开参照。"]}, recover_or_continue="continue_with_tool", next_step_hint="select_next_read_tool", ) proposal = _recovery_proposal_from_critic( question="把当前项目和外部公开信息结合起来判断", workspace=workspace, latest_observation=PMFlowAIAgentObservation( source="inspect_project_execution_package", observation_type="project_execution_package", summary="已读取内部执行包。", ), critic_result=critic_result, tool_call_count=1, run_budget=PMFlowAIAgentRunBudget(max_tool_calls=6), ) assert proposal is not None assert proposal.tool_name == "external_search" assert proposal.tool_input["requires_coverage"] is True assert any(item["branch_id"] == "public_reference" for item in proposal.tool_input["branch_searches"]) def test_evidence_normalizer_preserves_research_metadata_from_claim_payload() -> None: observation = PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="读取公开来源正文。", sources=[{"label": "官方页面", "url": "https://example.gov/policy"}], ) normalization = agent_evidence_normalizer_module._normalization_from_payload( { "source_quality": "usable", "usable_claims": [ { "claim_type": "policy_update", "claim": "官方页面说明执行机制发生调整。", "support_refs": ["https://example.gov/policy"], "evidence_lane": "lane_b", "coverage_contribution": "补足执行机制变化通道。", "time_match": "recent", "target_match": "direct", "authority_level": "official", "relevance_to_user_goal": "direct", "claim_kind": "substantive_fact", "confidence": "high", } ], }, observation=observation, ) claim = normalization.usable_claims[0] assert claim.evidence_lane == "lane_b" assert claim.coverage_contribution == "补足执行机制变化通道。" assert claim.time_match == "recent" assert claim.target_match == "direct" assert claim.authority_level == "official" @pytest.mark.asyncio async def test_agent_step_planner_expands_external_search_across_interpretation_branches(monkeypatch: pytest.MonkeyPatch) -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议行业怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议行业怎么样", desired_output="给出方向性判断并说明边界。", interpretation_branches=[ { "branch_id": "b_academic", "meaning": "围绕 AI 主题举办的学术会议和投稿组织生态", "confidence": "medium", "user_language_prior": "medium", }, { "branch_id": "b_product", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "confidence": "medium", "user_language_prior": "high", "evidence_probe_queries": ["AI meeting assistant market adoption 2026"], }, ], selected_interpretation="围绕 AI 主题举办的学术会议和投稿组织生态", answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) generator = _SequenceGenerator( [ """ { "action": "tool", "tool_name": "external_search", "tool_input": { "search_queries": ["NeurIPS 2026 conference updates submission"], "freshness_required": false }, "rationale": "读取学术会议证据。", "expected_observation": "公开来源。", "stop_condition": "有证据。", "risks_if_answer_now": ["证据不足"], "confidence": "high" } """ ] ) monkeypatch.setattr(PMFlowAIAgentStepPlanner, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_STEP_PLANNER")) proposal = await PMFlowAIAgentStepPlanner(db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator).plan( question="最近AI会议行业怎么样", workspace=workspace, conversation_context={}, tenant_project_context={}, ) queries = proposal.tool_input["search_queries"] assert "NeurIPS 2026 conference updates submission" in queries assert "AI meeting assistant market adoption 2026" in queries assert any(item.get("branch_id") == "b_product" for item in proposal.tool_input["branch_searches"]) @pytest.mark.asyncio async def test_agent_step_planner_promotes_goal_frame_branch_queries(monkeypatch: pytest.MonkeyPatch) -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议市场怎么样", source_policy="credible_public_sources", interpretation_branches=[ { "branch_id": "b_product", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "confidence": "medium", "user_language_prior": "high", "evidence_probe_queries": ["AI meeting scenario application market trends"], }, { "branch_id": "b_macro", "meaning": "AI主题会议活动的宏观会展市场", "confidence": "medium", "user_language_prior": "high", }, ], ) generator = _SequenceGenerator( [ """ { "action": "tool", "tool_name": "external_search", "tool_input": { "search_queries": ["AI conference market trends overview"], "branch_searches": [ { "branch_id": "b_product", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "query": "AI meeting assistant adoption", "queries": ["AI meeting assistant adoption", "AI meeting scenario application market trends"], "purpose": "验证 GoalFrame 解释分支" } ], "freshness_required": false }, "rationale": "检索。", "expected_observation": "公开来源。", "stop_condition": "有证据。", "risks_if_answer_now": ["证据不足"], "confidence": "medium" } """ ] ) monkeypatch.setattr(PMFlowAIAgentStepPlanner, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_STEP_PLANNER")) proposal = await PMFlowAIAgentStepPlanner(db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator).plan( question="最近AI会议市场怎么样", workspace=workspace, conversation_context={}, tenant_project_context={}, ) assert proposal.tool_input["search_queries"][0] == "最近AI会议市场怎么样" assert "AI meeting assistant adoption" in proposal.tool_input["search_queries"] assert "AI conference market trends overview" in proposal.tool_input["search_queries"] @pytest.mark.asyncio async def test_agent_state_critic_answers_with_evidence_backed_interpretation_branch(monkeypatch: pytest.MonkeyPatch) -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议行业怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议行业怎么样", desired_output="给出方向性判断并说明边界。", interpretation_branches=[ { "branch_id": "b_academic", "meaning": "围绕 AI 主题举办的学术会议和投稿组织生态", "confidence": "medium", "user_language_prior": "medium", }, { "branch_id": "b_product", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "confidence": "medium", "user_language_prior": "high", "answer_assumption_instruction": "我先按 AI 进入会议场景后的应用生态来回答。", }, ], answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="conference_scope_expansion", claim_text="NeurIPS 2026 征稿范围覆盖多个跨学科方向。", customer_value="支撑学术会议解释分支。", support_refs=["https://neurips.cc/Conferences/2026/CallForPapers"], interpretation_branch_id="b_academic", relevance_to_user_goal="partial", claim_kind="substantive_fact", confidence="high", ) ] generator = _SequenceGenerator( [ """ { "ready_to_answer": true, "should_continue": false, "needs_user_input": false, "retry_would_change_state": false, "state_delta": {"remaining_gaps": []}, "findings": ["已有 NeurIPS claim"], "missing_state": [], "recover_or_continue": "compose_answer", "next_step_hint": "compose_answer", "reason": "已有局部证据。" } """ ] ) monkeypatch.setattr(PMFlowAIAgentStateCritic, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_STATE_CRITIC")) result = await PMFlowAIAgentStateCritic(db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator).review( question="最近AI会议行业怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="tool", tool_name="external_search"), observation=PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源。", confidence="medium", relevance_to_goal="high", ), ) assert result.ready_to_answer is True assert result.should_continue is False assert result.recover_or_continue == "compose_answer" assert result.next_step_hint == "compose_answer" assert result.retry_would_change_state is False assert result.state_delta["interpretation_decision"]["winning_branch_id"] == "b_academic" assert result.state_delta["interpretation_decision"]["uncovered_branches"][0]["branch_id"] == "b_product" def test_agent_state_critic_prefers_evidence_backed_branch_without_keyword_repair() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议市场怎么样", desired_output="给出方向性判断。", interpretation_branches=[ { "branch_id": "b_product", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "confidence": "medium", "user_language_prior": "high", "answer_assumption_instruction": "我先按 AI 进入会议场景后的应用生态来回答。", "missing_anchors": ["用户没有明确限定解释。"], }, { "branch_id": "b_macro", "meaning": "AI主题会议活动的宏观会展市场", "confidence": "medium", "user_language_prior": "high", "missing_anchors": ["用户没有明确限定地域。"], }, ], answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="application_trend", claim_text="协作平台正在把 AI 用于会议纪要、参会体验和会后工作流。", interpretation_branch_id="b_product", relevance_to_user_goal="direct", claim_kind="directional_judgement", confidence="high", support_refs=["https://example.com/ai-meeting"], ), PMFlowAIAgentClaim( claim_type="event_market", claim_text="大型AI主题峰会仍保持举办。", interpretation_branch_id="b_macro", relevance_to_user_goal="partial", claim_kind="substantive_fact", confidence="medium", support_refs=["https://example.com/ai-event"], ), PMFlowAIAgentClaim( claim_type="event_market", claim_text="部分会议活动围绕企业AI治理展开。", interpretation_branch_id="b_macro", relevance_to_user_goal="partial", claim_kind="substantive_fact", confidence="medium", support_refs=["https://example.com/ai-governance-event"], ), ] result = _enforce_critic_contract( PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, retry_would_change_state=False, recover_or_continue="compose_answer", next_step_hint="compose_answer", ), workspace=workspace, observation=PMFlowAIAgentObservation(source="external_search", observation_type="external_public_evidence"), ) assert result.ready_to_answer is True assert result.state_delta["interpretation_decision"]["winning_branch_id"] == "b_product" assert result.state_delta["interpretation_decision"]["decision_basis"] == "evidence_backed_branch" def test_agent_state_critic_does_not_loop_when_priority_branch_has_evidence() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议市场怎么样", desired_output="给出方向性判断。", interpretation_branches=[ { "branch_id": "b_product", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "confidence": "medium", "user_language_prior": "high", "answer_assumption_instruction": "我先按 AI 进入会议场景后的应用生态来回答。", "missing_anchors": ["用户没有明确限定解释。"], }, { "branch_id": "b_macro", "meaning": "AI主题会议活动的宏观会展市场", "confidence": "medium", "user_language_prior": "high", "missing_anchors": ["用户没有明确限定地域。"], }, ], answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="application_market", claim_text="AI meeting assistants 市场正在增长，公开材料覆盖纪要、搜索和自动化工作流。", interpretation_branch_id="b_product", relevance_to_user_goal="direct", claim_kind="substantive_fact", confidence="medium", support_refs=["https://example.com/ai-meeting-assistants"], ) ] result = _enforce_critic_contract( PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, retry_would_change_state=True, recover_or_continue="continue_with_tool", next_step_hint="select_next_read_tool", reason="继续补最高语言先验分支。", ), workspace=workspace, observation=PMFlowAIAgentObservation(source="external_search", observation_type="external_public_evidence"), ) assert result.ready_to_answer is True assert result.should_continue is False assert result.retry_would_change_state is False assert result.recover_or_continue == "compose_answer" assert result.state_delta["interpretation_decision"]["winning_branch_id"] == "b_product" assert result.state_delta["interpretation_decision"]["decision_basis"] == "evidence_backed_branch" def test_agent_state_critic_accepts_structured_delta_when_claim_ledger_is_empty() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议市场怎么样", desired_output="给出方向性判断。", interpretation_branches=[ { "branch_id": "b_product", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "confidence": "medium", "user_language_prior": "high", } ], answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) workspace.claim_ledger = [] result = _enforce_critic_contract( PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, retry_would_change_state=True, recover_or_continue="continue_with_tool", next_step_hint="select_next_read_tool", state_delta={ "new_facts": ["AI会议助手市场受企业会议效率自动化需求推动，公开来源显示该细分市场处于增长阶段。"], "interpretation_decision": { "winning_branch_id": "b_product", "winning_meaning": "AI作为能力进入会议场景后的应用生态", "decision_basis": "critic_evidence_delta", }, }, ), workspace=workspace, observation=PMFlowAIAgentObservation(source="external_search", observation_type="external_public_evidence"), ) assert result.ready_to_answer is True assert result.should_continue is False assert result.retry_would_change_state is False assert result.recover_or_continue == "compose_answer" def test_agent_state_critic_uses_source_claims_when_ledger_is_empty() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="了解最近AI会议市场怎么样", desired_output="给出方向性判断。", interpretation_branches=[ { "branch_id": "b_product", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "confidence": "medium", "user_language_prior": "high", "answer_assumption_instruction": "我先按 AI 进入会议场景后的应用生态来回答。", "missing_anchors": ["用户没有明确限定解释。"], }, { "branch_id": "b1", "meaning": "AI行业会议活动的整体生态与趋势", "confidence": "high", "user_language_prior": "high", "missing_anchors": ["活动市场证据尚未覆盖。"], }, ], answer_policy="directional_answer", source_policy="credible_public_sources", freshness_policy="prefer_recent_not_strict", missing_evidence_policy="answer_with_boundaries", ) workspace.evidence_normalizations = [ PMFlowAIAgentEvidenceNormalization( source_quality="usable", source_normalizations=[ PMFlowAIAgentSourceNormalization( source_ref="https://example.com/ai-meeting-assistants", source_title="AI Meeting Assistants Market", source_quality="usable", usable_claims=[ PMFlowAIAgentClaim( claim_type="market_trend", claim_text="AI会议助手市场受企业会议效率自动化需求推动，处于增长阶段。", customer_value="支撑 AI 进入会议场景后的应用生态判断。", interpretation_branch_id="b_product", relevance_to_user_goal="partial", claim_kind="directional_judgement", confidence="medium", ) ], ) ], ) ] result = _enforce_critic_contract( PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, retry_would_change_state=True, recover_or_continue="continue_with_tool", next_step_hint="select_next_read_tool", state_delta={ "new_facts": ["当前来源均聚焦 AI 会议助手市场。"], "interpretation_decision": { "winning_branch_id": "b1", "winning_meaning": "AI行业会议活动的整体生态与趋势", "decision_basis": "evidence_backed_branch", }, }, ), workspace=workspace, observation=PMFlowAIAgentObservation(source="inspect_source_detail", observation_type="external_source_document"), ) assert result.ready_to_answer is True assert result.state_delta["interpretation_decision"]["winning_branch_id"] == "b_product" assert result.state_delta["interpretation_decision"]["decision_basis"] == "evidence_backed_branch" def test_agent_runtime_skips_source_detail_when_search_already_yields_claims() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="market_trend", claim_text="AI会议助手市场呈现增长态势。", customer_value="支撑方向性判断。", support_refs=["https://example.com/ai-meeting-assistants"], interpretation_branch_id="b_product", relevance_to_user_goal="partial", claim_kind="directional_judgement", confidence="medium", ) ] urls = _candidate_source_urls_for_fetch( proposal=PMFlowAIAgentStepProposal(action="tool", tool_name="external_search", tool_input={"query": "AI meeting"}), observation=PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", sources=[{"label": "公开来源", "url": "https://example.com/ai-meeting-assistants"}], ), workspace=workspace, tool_call_count=1, run_budget=PMFlowAIAgentRunBudget(max_tool_calls=4), ) assert urls == [] def test_agent_runtime_fetches_source_detail_when_public_obligation_has_only_reasoning_claims() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="做一个需要公开证据的对比判断") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="做一个需要公开证据的对比判断", source_policy="credible_public_sources", answer_policy="directional_answer", missing_evidence_policy="answer_with_boundaries", ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="reasoned_hypothesis", claim_text="当前可以先形成一个产品逻辑假设。", customer_value="先形成一个产品逻辑假设。", claim_kind="directional_judgement", confidence="medium", ) ] urls = _candidate_source_urls_for_fetch( proposal=PMFlowAIAgentStepProposal(action="tool", tool_name="external_search", tool_input={"query": "AI briefing"}), observation=PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", sources=[{"label": "公开来源", "url": "https://example.com/ai-briefing"}], ), workspace=workspace, tool_call_count=1, run_budget=PMFlowAIAgentRunBudget(max_tool_calls=4), ) assert urls == ["https://example.com/ai-briefing"] def test_answer_readability_repairs_single_line_gfm_table() -> None: raw = ( "这里先给结论。 " "| 竞品名称 | 核心定位 | 关键优势 | 主要劣势/待验证点 ||:---|:---|:---|:---|| " "NewsBang | 深度媒体对比 | 有观点对立 | 需要验证留存 |" ) repaired = agent_answer_runtime_module._repair_answer_readability(raw) assert "| 竞品名称 | 核心定位 | 关键优势 | 主要劣势/待验证点 |" in repaired assert "\n| :--- | :--- | :--- | :--- |" in repaired assert "\n| NewsBang | 深度媒体对比 | 有观点对立 | 需要验证留存 |" in repaired def test_customer_markdown_normalizer_recovers_malformed_pipe_continuation() -> None: from app.api.rag_ask_contract import _normalize_customer_answer_markdown raw = ( "### 核心价值主张 - **独特优势**：相比现有竞品，新助手应强调可信过滤。\n\n" "| 功能模块 | 具体描述 | 优先级 |\n| :--- | :--- | :--- |\n| 核心数据源接入 | 支持主流新闻源。 | P0 |\n\n" "| P0 | 差异化功能：多源交叉验证 | 对关键事实进行多来源比对。 | " "P1 | 差异化功能：交互式追问 | 允许用户深度追问。 |\n\n" "| P1 | 多模态输出 | 支持图表输出。 | P2 | ### 竞争壁垒构建思路 – 数据网络效应：持续优化。" ) repaired = _normalize_customer_answer_markdown(raw) assert "###" not in repaired assert "****" not in repaired assert "**核心价值主张**" in repaired assert "**独特优势**：相比现有竞品，新助手应强调可信过滤。" in repaired assert "| 优先级 | 项目 | 说明 |" in repaired assert "| P0 | 差异化功能：多源交叉验证 | 对关键事实进行多来源比对。 |" in repaired assert "| P1 | 差异化功能：交互式追问 | 允许用户深度追问。 |" in repaired assert "**竞争壁垒构建思路" in repaired assert "数据网络效应：持续优化。" in repaired def test_customer_markdown_normalizer_repairs_fragmented_gfm_table_rows() -> None: from app.api.rag_ask_contract import _normalize_customer_answer_markdown raw = ( "先给结论。\n\n" "| 对比维度 | 工具 A | 工具 B | 工具 C |\n" "|:---|:---|:---|:---|\n" "| **核心定位** | A 的定位。 | B 的定位。\n\n" "| C 的定位。| | **最佳适用团队** | A 适合的团队。\n\n" "| B 适合的团队。| C 适合的团队。\n\n" "| | **核心强项** | A 强项一；
A 强项二。| B 强项。\n\n" "| C 强项。|\n\n" "| \n\n" "**后续判断**\n\n" "继续说明。" ) repaired = _normalize_customer_answer_markdown(raw) assert "| 对比维度 | 工具 A | 工具 B | 工具 C |" in repaired assert "| **核心定位** | A 的定位。 | B 的定位。 | C 的定位。 |" in repaired assert "| **最佳适用团队** | A 适合的团队。 | B 适合的团队。 | C 适合的团队。 |" in repaired assert "| **核心强项** | A 强项一；A 强项二。 | B 强项。 | C 强项。 |" in repaired assert "\n| C 的定位。| | **最佳适用团队**" not in repaired assert "\n| \n" not in repaired assert "**后续判断**" in repaired assert "继续说明。" in repaired def test_customer_markdown_normalizer_repairs_numbered_table_continuations() -> None: from app.api.rag_ask_contract import _normalize_customer_answer_markdown raw = ( "这版方案聚焦在**数据接入质量**与**AI助手能力**这两个切口。\n\n" "| 模块名称 | 关键功能描述 | 验证目标 |\n" "| :--- | :--- | :--- |\n" "| 1. 智能数据清洗与录入 | 支持上传 PDF/Excel 报价单，AI 自动提取关键字段。 | 验证解析准确率。 |\n\n" "2. **风险与异常预警助手** 基于提取的数据，自动比对历史采购价或行业基准价。\n\n" "若发现价格偏离超过阈值，立即生成红色预警报告。 | 验证风控价值。\n\n" "|| 3. 自然语言询价建议 | 用户可通过对话界面询问供应商对比。\n\n" "| 验证智能决策辅助的交互体验。\n\n" "**后续安排**\n\n" "继续说明。" ) repaired = _normalize_customer_answer_markdown(raw) assert "在 **数据接入质量** 与 **AI助手能力** 这两个切口" in repaired assert "| 模块名称 | 关键功能描述 | 验证目标 |" in repaired assert "| 1. 智能数据清洗与录入 | 支持上传 PDF/Excel 报价单，AI 自动提取关键字段。 | 验证解析准确率。 |" in repaired assert "| 2. **风险与异常预警助手** | 基于提取的数据，自动比对历史采购价或行业基准价。 若发现价格偏离超过阈值，立即生成红色预警报告。 | 验证风控价值。 |" in repaired assert "| 3. 自然语言询价建议 | 用户可通过对话界面询问供应商对比。 | 验证智能决策辅助的交互体验。 |" in repaired assert "|| 3." not in repaired assert "**后续安排**" in repaired assert "继续说明。" in repaired def test_agent_answer_critic_rejects_mechanical_fallback_answer() -> None: terminal_state = PMFlowAIAgentTerminalAnswerState( user_goal="最近AI会议行业怎么样", terminal_status="partially_answerable", claims=[ PMFlowAIAgentClaim( claim_type="trend", claim_text="会议 AI 产品正在从纪要转向执行工作流。", customer_value="支撑方向性判断。", confidence="medium", ) ], ) findings = critique_agent_customer_answer( terminal_state=terminal_state, answer="这轮已有材料能支撑一部分判断：可展示事实已有公开材料支撑，置信度较高；来源包括neurips.cc、openreview.net。。但还有缺口。", ) assert "mechanical_fallback_answer" in findings def test_agent_answer_critic_rejects_customer_visible_process_leak() -> None: terminal_state = PMFlowAIAgentTerminalAnswerState( user_goal="分析一个近期政策方向", terminal_status="partially_answerable", claims=[ PMFlowAIAgentClaim( claim_type="policy_update", claim_text="公开材料显示该政策方向有阶段性调整。", customer_value="支撑一个有边界的方向判断。", confidence="medium", ) ], ) findings = critique_agent_customer_answer( terminal_state=terminal_state, answer="我先按这个解释来回答：这个判断还不完整，后面需要继续补齐缺口。", ) assert "process_leak_phrase" in findings def test_agent_terminal_state_carries_interpretation_decision_from_critic() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="判断最近AI会议市场", interpretation_branches=[ { "branch_id": "b_application", "meaning": "把 AI 作为能力进入会议场景后的应用生态", "answer_assumption_instruction": "我先按 AI 进入会议场景后的应用生态来回答。", } ], ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="market_trend", claim_text="AI 会议助手正在从记录转向会后执行工作流。", interpretation_branch_id="b_application", relevance_to_user_goal="direct", claim_kind="directional_judgement", confidence="medium", support_refs=["https://example.com/meeting-ai"], ) ] terminal_state = build_agent_terminal_answer_state( question="最近AI会议市场怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="answer"), observation=None, critic_result=PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, state_delta={ "interpretation_decision": { "winning_branch_id": "b_application", "winning_meaning": "把 AI 作为能力进入会议场景后的应用生态", "decision_basis": "evidence_backed_branch", "answer_assumption_instruction": "我先按 AI 进入会议场景后的应用生态来回答。", } }, ), ) fallback = render_agent_terminal_state_fallback(terminal_state) assert terminal_state.interpretation_decision["winning_branch_id"] == "b_application" assert "本轮解释口径是" in fallback assert "我先按" not in fallback def test_agent_terminal_state_orders_claims_by_winning_interpretation_branch() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="market_size", claim_text="会议行业整体规模正在恢复。", interpretation_branch_id="b_macro", relevance_to_user_goal="direct", claim_kind="substantive_fact", confidence="medium", ), PMFlowAIAgentClaim( claim_type="application_trend", claim_text="协作平台正在把 AI 用于会议纪要、参会体验和会后工作流。", interpretation_branch_id="b_product", relevance_to_user_goal="partial", claim_kind="directional_judgement", confidence="medium", ), ] terminal_state = build_agent_terminal_answer_state( question="最近AI会议市场怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="answer"), observation=None, critic_result=PMFlowAIAgentCriticResult( ready_to_answer=True, should_continue=False, state_delta={ "interpretation_decision": { "winning_branch_id": "b_product", "winning_meaning": "把 AI 作为能力进入会议场景后的应用生态", "answer_assumption_instruction": "我先按 AI 进入会议场景后的应用生态来回答。", } }, ), ) assert terminal_state.claims[0].interpretation_branch_id == "b_product" def test_agent_terminal_state_waits_for_evidence_when_critic_vetoes_even_with_claims() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="external_source_evidence", claim_text="某 HR webinar 提到了 AI 管理趋势。", customer_value="只能说明一个偏窄材料。", support_refs=["https://example.com/hr-webinar"], interpretation_branch_id="b_hr", relevance_to_user_goal="partial", claim_kind="substantive_fact", confidence="medium", ) ] workspace.boundary_ledger = [ PMFlowAIAgentBoundary( boundary_type="source_gap", blocked_claim="缺少 AI 会议活动市场主分支证据。", customer_safe_reason="现有材料没有覆盖用户真正问的 AI 会议市场。", allowed_alternative="继续读取更贴近会议活动市场的公开来源。", ) ] terminal_state = build_agent_terminal_answer_state( question="最近AI会议市场怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="tool", tool_name="external_search"), observation=None, critic_result=PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, retry_would_change_state=True, state_delta={"remaining_gaps": ["主分支证据不足"]}, ), ) assert terminal_state.terminal_status == "waiting_for_evidence" assert terminal_state.answer_intent_plan.lead_claim_id == "" assert terminal_state.answer_intent_plan.lead_boundary_id def test_recovery_query_diversifies_when_existing_claim_is_off_branch() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议市场怎么样") workspace.goal_frame = PMFlowAIAgentGoalFrame( user_goal="最近AI会议市场怎么样", desired_output="判断 AI 会议活动市场趋势。", evidence_requirements=["参会规模", "赞助需求", "会议活动供给"], interpretation_branches=[ { "branch_id": "b_activity", "meaning": "AI 会议活动市场和峰会生态", "user_language_prior": "high", "anchor_terms": ["AI conference", "summit", "attendance", "sponsorship"], }, { "branch_id": "b_hr", "meaning": "HR 领域 AI webinar", "user_language_prior": "low", "anchor_terms": ["HR", "talent acquisition"], }, ], ) workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="external_source_evidence", claim_text="HR webinar 讨论了 AI 对人才招聘的影响。", interpretation_branch_id="b_hr", relevance_to_user_goal="partial", claim_kind="substantive_fact", confidence="medium", ) ] proposal = _recovery_proposal_from_critic( question="最近AI会议市场怎么样", workspace=workspace, latest_observation=PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取偏题来源正文。", confidence="medium", relevance_to_goal="medium", ), critic_result=PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, retry_would_change_state=True, state_delta={"remaining_gaps": ["主分支 AI 会议活动市场缺少参会和赞助证据"]}, missing_state=["AI 会议活动市场主分支未覆盖"], ), tool_call_count=2, run_budget=SimpleNamespace(max_tool_calls=8), ) assert proposal is not None assert proposal.tool_name == "external_search" queries = proposal.tool_input["search_queries"] assert len(queries) >= 2 assert any("AI 会议活动市场和峰会生态" in query for query in queries) assert any("参会规模" in query or "赞助需求" in query for query in queries) assert proposal.tool_input["result_limit"] >= 8 def test_operating_claim_projection_ignores_claims_when_terminal_state_waits_for_evidence() -> None: workspace = { "claim_ledger": [ { "claim_type": "external_source_evidence", "claim_text": "HR webinar 讨论了 AI 对人才招聘的影响。", "confidence": "medium", } ] } deliverable = { "sections": [ { "section_kind": "claims", "items": [{"summary": "HR webinar 讨论了 AI 对人才招聘的影响。"}], } ] } claims = _claim_texts(workspace, deliverable, terminal_state={"terminal_status": "waiting_for_evidence"}) assert claims == [] @pytest.mark.asyncio async def test_agent_external_planner_defaults_to_non_strict_freshness_gate() -> None: planner = _AgentRuntimeExternalPlanner( query="最近AI会议行业怎么样", tool_input={"search_queries": ["AI meeting assistant industry trends"]}, ) plan = await planner.build_plan(question="最近AI会议行业怎么样") assert plan.freshness_required is False assert plan.search_queries == ["AI meeting assistant industry trends"] @pytest.mark.asyncio async def test_agent_external_planner_can_request_broader_bounded_result_limit() -> None: planner = _AgentRuntimeExternalPlanner( query="开放行业研究", tool_input={ "search_queries": [f"query {index}" for index in range(8)], "result_limit": 10, }, ) plan = await planner.build_plan(question="开放行业研究") assert len(plan.search_queries) == 8 assert plan.result_limit == 10 @pytest.mark.asyncio async def test_agent_answer_runtime_revises_bad_draft_before_terminal_fallback(monkeypatch: pytest.MonkeyPatch) -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI会议行业怎么样") workspace.boundary_ledger = [ PMFlowAIAgentBoundary( boundary_type="source_gap", blocked_claim="缺少会议行业垂直公开来源。", customer_safe_reason="这轮查到的材料偏向泛 AI 行业，不能直接说明 AI 会议行业最近的发展。", allowed_alternative="继续换用会议技术、活动平台和标准组织来源补证据。", ) ] generator = _SequenceGenerator( [ "Refine search using terms like 'AI conference platform update' or 'hybrid meeting AI feature launch'.", "目前还不能可靠判断最近 AI 会议行业怎么样。已查到的材料偏向泛 AI 行业，不能直接说明会议与活动技术场景的最近变化；下一步需要换用会议技术、活动平台和标准组织来源补证据。", ] ) async def fake_runtime_profile(_db, *, stage: str): return SimpleNamespace( function_name=f"pmflow_ai_agent_{stage}", prompt_key="PROMPT_PMFLOW_AI_AGENT_ANSWER_REVISER" if stage == "answer_reviser" else "PROMPT_PMFLOW_AI_AGENT_ANSWER_COMPOSER", scene_key="space_pmflow_ai_agent_runtime", temperature=0.1, response_mode="fast", max_tokens=2600, timeout_seconds=10, runtime_mode="active", quality_gate_mode="strict", routing_meta={}, ) monkeypatch.setattr(agent_answer_runtime_module, "resolve_agent_llm_runtime_profile", fake_runtime_profile) monkeypatch.setattr(PMFlowAIAgentAnswerRuntime, "_render_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_ANSWER_COMPOSER")) monkeypatch.setattr(PMFlowAIAgentAnswerRuntime, "_render_reviser_prompt", _fake_agent_prompt_renderer("PROMPT_PMFLOW_AI_AGENT_ANSWER_REVISER")) result = await PMFlowAIAgentAnswerRuntime(db=SimpleNamespace(execute=lambda *_args, **_kwargs: None), tenant_id=12, generator=generator).run( question="最近AI会议行业怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal(action="stop_blocked", rationale="停止"), observation=None, critic_result=PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, retry_would_change_state=False, reason="claim_ledger 为空，需要换会议行业垂直来源。", ), ) assert generator.calls == 2 assert result.answer_runtime_packet.draft.source == "llm_reviser" assert result.quality_findings == [] assert "目前还不能可靠判断最近 AI 会议行业怎么样" in result.answer assert "Refine search" not in result.answer def test_agent_deliverable_actions_do_not_expose_planner_rationale() -> None: workspace = PMFlowAIAgentWorkspace(current_goal="最近AI行业怎么样") workspace.claim_ledger = [ PMFlowAIAgentClaim( claim_type="market_trend", claim_text="公开材料显示市场投入仍在增加。", support_refs=["公开来源"], confidence="medium", ) ] terminal_state = build_agent_terminal_answer_state( question="最近AI行业怎么样", workspace=workspace, proposal=PMFlowAIAgentStepProposal( action="tool", tool_name="external_search", rationale="User goal requires recent, authoritative, external public information on AI industry status.", ), observation=None, critic_result=PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=False, retry_would_change_state=False, reason="已有部分 claim。", ), ) result_text = render_agent_terminal_state_fallback(terminal_state) assert terminal_state.terminal_status == "partially_answerable" assert "User goal requires" not in result_text assert "external_search" not in result_text assert all("User goal requires" not in item for item in terminal_state.next_possible_actions) @pytest.mark.asyncio async def test_agent_runtime_v2_stops_repeated_tool_without_state_delta() -> None: payload = await PMFlowAIAgentRunService( object(), tenant_id=12, planner=_SequencePlanner(["inspect_space_collections", "inspect_space_collections"]), tool_registry=_FakeToolRegistry(), critic=_SequenceCritic( [ PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, state_delta={}, reason="还要继续。", ), PMFlowAIAgentCriticResult( ready_to_answer=False, should_continue=True, state_delta={}, reason="不应该执行到第二次 critic。", ), ] ), answer_runtime=PMFlowAIAgentAnswerRuntime(db=object(), tenant_id=12, generator=object()), ).run_turn( request=RAGAskRequest(question="现在有什么项目"), current_user=SimpleNamespace(id=7, tenant_id=12), conversation_context={}, tenant_project_context={"surface_context": {"visible_objects": [{"object_ref": "project:141"}]}}, started_at=None, ) assert payload["agent_run_packet"]["stop_reason"] == "repeated_action_without_state_delta" assert payload["deliverable_packet"]["deliverable_kind"] == "blocked_state" assert payload["deliverable_packet"]["status"] == "blocked" assert payload["agent_workspace"]["deliverable_state"]["deliverable_kind"] == payload["deliverable_packet"]["deliverable_kind"] assert payload["agent_run_packet"]["iterations"] == 2 assert payload["agent_run_packet"]["tool_call_count"] == 1 assert payload["agent_run_packet"]["event_log"][-2]["stop_reason"] == "repeated_action_without_state_delta" assert [step["key"] for step in payload["answer_process_packet"]["steps"]].count("agent_tool_inspect_space_collections") == 1 @pytest.mark.asyncio async def test_agent_runtime_project_status_without_anchor_returns_world_state_scope_gap() -> None: observation = await PMFlowAIAgentToolRegistry(db=object(), tenant_id=12).execute( proposal=PMFlowAIAgentStepProposal( action="tool", tool_name="inspect_project_status", tool_input={"query": "list_all"}, ), request=RAGAskRequest(question="现在有什么项目"), workspace=workspace_from_conversation_context(question="现在有什么项目", conversation_context={}), current_user=SimpleNamespace(id=38, tenant_id=12), conversation_context={}, tenant_project_context={ "surface_context": { "workbench": { "recent_projects": [ { "id": 141, "title": "童析—学龄前儿童行为特质AI评估助手", "status": "active", } ] } } }, ) assert observation.observation_type == "project_world_state_missing_scope" assert "项目数量" in observation.summary or "不能读取项目世界状态" in observation.summary assert any(fact.get("type") == "project_item" and fact.get("project_ref") == "project:141" for fact in observation.facts) assert not any(fact.get("type") == "project_world_state" for fact in observation.facts) @pytest.mark.asyncio async def test_agent_runtime_project_status_consumes_planner_project_ref() -> None: observation = await PMFlowAIAgentToolRegistry(db=object(), tenant_id=12).execute( proposal=PMFlowAIAgentStepProposal( action="tool", tool_name="inspect_project_status", tool_input={"query": "project:141"}, ), request=RAGAskRequest(question="儿童的那个项目你知道吗"), workspace=workspace_from_conversation_context(question="儿童的那个项目你知道吗", conversation_context={}), current_user=SimpleNamespace(id=38, tenant_id=12), conversation_context={}, tenant_project_context={ "surface_context": { "workbench": { "recent_projects": [ { "id": 141, "title": "童析—学龄前儿童行为特质AI评估助手", "status": "active", } ] } } }, ) assert observation.observation_type == "project_world_state" assert "童析—学龄前儿童行为特质AI评估助手" in observation.summary assert any(fact.get("type") == "project_anchor" and fact.get("project_ref") == "project:141" for fact in observation.facts) assert any(fact.get("type") == "project_world_state" and fact.get("project_refs") == ["project:141"] for fact in observation.facts) assert any(fact.get("type") == "project_world_facet" for fact in observation.facts) assert any(fact.get("type") == "agent_tool_contract" and fact.get("tool_name") == "inspect_project_operating_state" for fact in observation.facts) @pytest.mark.asyncio async def test_agent_runtime_project_status_does_not_use_visible_collection_as_operating_state() -> None: observation = await PMFlowAIAgentToolRegistry(db=object(), tenant_id=12).execute( proposal=PMFlowAIAgentStepProposal( action="tool", tool_name="inspect_project_status", tool_input={}, ), request=RAGAskRequest(question="现在有哪些项目"), workspace=workspace_from_conversation_context(question="现在有哪些项目", conversation_context={}), current_user=SimpleNamespace(id=38, tenant_id=12), conversation_context={}, tenant_project_context={ "surface_context": { "workbench": { "recent_projects": [ { "id": 141, "title": "童析—学龄前儿童行为特质AI评估助手", "status": "active", } ] } } }, ) assert observation.observation_type == "project_world_state_missing_scope" assert any(fact.get("type") == "project_world_state_gap" for fact in observation.facts) assert not any(fact.get("type") == "project_world_state" for fact in observation.facts) @pytest.mark.asyncio async def test_agent_tool_environment_resolves_explicit_customer_object() -> None: observation = await PMFlowAIAgentToolRegistry(db=object(), tenant_id=12).execute( proposal=PMFlowAIAgentStepProposal( action="tool", tool_name="resolve_customer_object", tool_input={"object_ref": "project:141"}, ), request=RAGAskRequest(question="儿童的那个项目你知道吗"), workspace=workspace_from_conversation_context(question="儿童的那个项目你知道吗", conversation_context={}), current_user=SimpleNamespace(id=38, tenant_id=12), conversation_context={}, tenant_project_context={ "surface_context": { "workbench": { "recent_projects": [ { "id": 141, "title": "童析—学龄前儿童行为特质AI评估助手", "status": "active", } ] } } }, ) assert observation.observation_type == "object_resolution" assert any(fact.get("type") == "resolved_object" and fact.get("object_ref") == "project:141" for fact in observation.facts) @pytest.mark.asyncio async def test_agent_tool_environment_returns_candidates_without_auto_binding() -> None: observation = await PMFlowAIAgentToolRegistry(db=object(), tenant_id=12).execute( proposal=PMFlowAIAgentStepProposal( action="tool", tool_name="resolve_customer_object", tool_input={}, ), request=RAGAskRequest(question="它现在怎么样"), workspace=workspace_from_conversation_context(question="它现在怎么样", conversation_context={}), current_user=SimpleNamespace(id=38, tenant_id=12), conversation_context={}, tenant_project_context={ "surface_context": { "workbench": { "recent_projects": [ { "id": 141, "title": "童析—学龄前儿童行为特质AI评估助手", "status": "active", } ] } } }, ) assert observation.observation_type == "object_candidates" assert "尚未自动绑定单一对象" in observation.summary assert any(fact.get("type") == "object_candidate" and fact.get("binding_strength") == "candidate" for fact in observation.facts) def test_agent_runtime_step4_restores_workspace_from_capsule_and_updates_current_goal() -> None: first = workspace_from_conversation_context( question="先分析 AI 采购监督系统", conversation_context={}, ) first.known_facts = [{"type": "external_evidence", "title": "公开材料"}] capsules = upsert_agent_workspace_capsule(capsules=[], workspace=first) restored = agent_workspace_from_context_capsules(capsules) assert restored["workspace_id"] == first.workspace_id assert restored["known_facts"][0]["title"] == "公开材料" second = workspace_from_conversation_context( question="继续看落地风险", conversation_context={"context_capsules": capsules}, ) assert second.workspace_id == first.workspace_id assert second.current_goal == "继续看落地风险" assert "先分析 AI 采购监督系统" in second.goal_history assert second.fact_ledger[0].fact_text == "公开材料" next_capsules = upsert_agent_workspace_capsule(capsules=capsules, workspace=second) next_workspace_capsule = agent_workspace_from_context_capsules(next_capsules) assert next_workspace_capsule["fact_ledger"][0]["fact_text"] == "公开材料" class _FakePlanner: def __init__(self, tool_name: str) -> None: self.tool_name = tool_name async def plan(self, **kwargs): return PMFlowAIAgentStepProposal( action="tool", tool_name=self.tool_name, tool_input={"query": kwargs.get("question", ""), "freshness_required": True}, rationale="读取下一步所需证据。", expected_observation="客户安全观察。", confidence="medium", ) class _SequencePlanner: def __init__(self, tool_names: list[str]) -> None: self.tool_names = list(tool_names) self.index = 0 async def plan(self, **kwargs): tool_name = self.tool_names[min(self.index, len(self.tool_names) - 1)] self.index += 1 return PMFlowAIAgentStepProposal( action="tool", tool_name=tool_name, tool_input={"query": kwargs.get("question", ""), "freshness_required": True}, rationale=f"执行 {tool_name} 读取证据。", expected_observation="客户安全观察。", confidence="medium", ) class _FakeToolRegistry: async def execute(self, *, proposal, **kwargs): if proposal.tool_name == "external_search": return PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源，得到 1 条可用证据。", facts=[{"type": "external_evidence", "title": "公开资料"}], sources=[{"label": "公开来源", "url": "https://example.com/source"}], confidence="high", relevance_to_goal="high", ) if proposal.tool_name == "inspect_source_detail": return PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "公开来源", "url": "https://example.com/source", "text": "公开来源正文：该方向已有公开材料支撑，市场需求和应用探索正在增加。", } ], sources=[{"label": "公开来源", "url": "https://example.com/source"}], confidence="medium", relevance_to_goal="high", ) return PMFlowAIAgentObservation( source="inspect_space_collections", observation_type="space_context", summary="已读取客户安全的空间上下文摘要。", facts=[{"type": "surface_collection_count", "collection": "projects", "count": 1}], confidence="medium", relevance_to_goal="medium", ) class _NoisyExternalToolRegistry: async def execute(self, *, proposal, **kwargs): if proposal.tool_name == "external_search": return PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源，得到 1 条候选材料。", facts=[ { "type": "external_evidence", "title": "行业报告页面", "summary": ( "行业报告页面 联系我们 订阅邮箱 020-8922-4700 news@iimedia.cn " "  正文片段 导航 市场地位 商业趋势 网站投稿" ), "url": "https://example.com/noisy", } ], sources=[{"label": "行业报告页面", "url": "https://example.com/noisy"}], confidence="high", relevance_to_goal="high", ) if proposal.tool_name == "inspect_source_detail": return PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="已读取公开来源正文。", facts=[ { "type": "external_source_text", "title": "行业报告页面", "url": "https://example.com/noisy", "text": ( "联系我们 订阅邮箱 020-8922-4700 news@iimedia.cn    " "网站投稿 媒体合作 导航 首页 报告定制 行业白皮书" ), } ], sources=[{"label": "行业报告页面", "url": "https://example.com/noisy"}], confidence="medium", relevance_to_goal="medium", ) return PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="source_detail_missing_url", summary="外部来源读取缺少可执行 URL，本轮未读取网页正文。", facts=[{"type": "missing_external_source_url", "user_actionable": False}], confidence="medium", relevance_to_goal="medium", ) class _RecoveryExternalToolRegistry: def __init__(self) -> None: self.inspect_calls: list[list[str]] = [] async def execute(self, *, proposal, **kwargs): if proposal.tool_name == "external_search": return PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源，得到 3 条候选材料。", facts=[ { "type": "external_evidence", "title": "AI market report", "summary": "Broad artificial intelligence market size report.", "url": "https://example.com/broad-ai-market", }, { "type": "external_evidence", "title": "AI conference attendance and sponsorship 2026", "summary": "Conference attendance, sponsorship and event market signals.", "url": "https://example.com/ai-conference-attendance-2026", }, { "type": "external_evidence", "title": "AI event technology alternatives", "summary": "Event tech, community events and conference software alternatives.", "url": "https://example.com/ai-event-technology", }, ], sources=[ {"label": "AI market report", "url": "https://example.com/broad-ai-market"}, {"label": "AI conference attendance and sponsorship 2026", "url": "https://example.com/ai-conference-attendance-2026"}, {"label": "AI event technology alternatives", "url": "https://example.com/ai-event-technology"}, ], confidence="high", relevance_to_goal="high", ) if proposal.tool_name == "inspect_source_detail": urls = [str(item) for item in list(proposal.tool_input.get("urls") or [])] self.inspect_calls.append(urls) if len(self.inspect_calls) == 1: return PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="source_detail_failed", summary="第一个来源不可读。", facts=[ { "type": "external_source_unreadable", "url": urls[0] if urls else "https://example.com/broad-ai-market", "summary": "403 Forbidden", } ], confidence="unknown", relevance_to_goal="medium", ) url = urls[0] if urls else "https://example.com/ai-conference-attendance-2026" return PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="补读了未尝试来源正文。", facts=[ { "type": "external_source_text", "title": "AI conference attendance and sponsorship 2026", "url": url, "text": "AI conferences are showing stronger attendance, sponsorship demand and event tooling needs.", } ], sources=[{"label": "AI conference attendance and sponsorship 2026", "url": url}], confidence="medium", relevance_to_goal="high", ) return PMFlowAIAgentObservation( source="inspect_space_collections", observation_type="space_context", summary="已读取空间上下文。", facts=[], ) class _CandidateClaimRecoveryToolRegistry: async def execute(self, *, proposal, **kwargs): if proposal.tool_name == "external_search": return PMFlowAIAgentObservation( source="external_search", observation_type="external_public_evidence", summary="已查询外部公开来源，得到候选材料。", facts=[ { "type": "external_evidence", "title": "AI market report", "summary": "Broad artificial intelligence market size report.", "url": "https://example.com/broad-ai-market", } ], sources=[{"label": "AI market report", "url": "https://example.com/broad-ai-market"}], confidence="high", relevance_to_goal="medium", ) if proposal.tool_name == "inspect_source_detail": return PMFlowAIAgentObservation( source="inspect_source_detail", observation_type="external_source_document", summary="读取到候选正文，但仍没有可入账事实。", facts=[ { "type": "external_source_text", "title": "AI market report", "url": "https://example.com/broad-ai-market", "text": "Broad artificial intelligence market size report.", } ], sources=[{"label": "AI market report", "url": "https://example.com/broad-ai-market"}], confidence="medium", relevance_to_goal="medium", ) return PMFlowAIAgentObservation( source="inspect_space_collections", observation_type="space_context", summary="已读取空间上下文。", facts=[], ) class _FakeCritic: def __init__(self, *, ready: bool) -> None: self.ready = ready async def review(self, **kwargs): return PMFlowAIAgentCriticResult( ready_to_answer=self.ready, should_continue=not self.ready, state_delta={"test": True}, findings=["状态已检查。"], next_step_hint="compose_answer" if self.ready else "select_next_read_tool", reason="状态检查完成。", ) class _SequenceCritic: def __init__(self, results: list[PMFlowAIAgentCriticResult]) -> None: self.results = list(results) self.index = 0 async def review(self, **kwargs): result = self.results[min(self.index, len(self.results) - 1)] self.index += 1 return result class _SequenceGenerator: def __init__(self, responses: list[str]) -> None: self.responses = list(responses) self.calls = 0 async def generate(self, *args, **kwargs) -> str: index = min(self.calls, len(self.responses) - 1) self.calls += 1 return self.responses[index] class _CandidateOnlyNormalizer: async def normalize(self, **kwargs): observation = kwargs.get("observation") return PMFlowAIAgentEvidenceNormalization( source_observation_id=getattr(observation, "observation_id", None), source_quality="noisy", usable_claims=[ PMFlowAIAgentClaim( claim_type="external_source_candidate", claim_kind="source_candidate", claim_text="候选外部材料可继续读取，但还不能作为业务事实陈述。", allowed_to_state=True, ) ], gaps=["needs_conference_specific_sources"], rejected_reason="source_candidate_without_substantive_claim", ) def _fake_agent_prompt_renderer(prompt_key: str): async def _render(self, **kwargs): return PromptRuntimeRenderResult( prompt=f"{prompt_key} test prompt", prompt_key=prompt_key, source="unit_test", runtime_tenant_id=12, ) return _render class _FakeProjectWorldStateService: def __init__(self, *, blocked_actions: list[dict[str, object]] | None = None) -> None: self.blocked_actions = list(blocked_actions or []) async def build_packet(self, **kwargs): return PMFlowAIProjectWorldStatePacket( primary_project_ref="project:141", project_refs=["project:141"], status="ready", facets=[ PMFlowAIProjectWorldFacet( facet_key="project_identity", display_label="项目身份与目标", status="ready", summary="项目身份与目标 已读取：project_title: 童析；loop_stage: command_idle", facts=[{"type": "world_state_fact", "key": "project_title", "value": "童析"}], evidence_refs=["project:141:project_identity"], ) ], project_operating_evidence_packets=[ { "packet_version": "project_operating_evidence.v1", "project_ref": "project:141", "project_title": "童析", "safe_next_actions": [ { "display_label": "先处理当前任务阻塞", "target_ref": "project:141", } ], "blocked_actions": self.blocked_actions, } ], evidence_ledger=[ PMFlowAIEvidenceLedgerEntry( evidence_id="evd_test", project_ref="project:141", facet_key="project_identity", source_ref="project:141:project_identity", summary="项目身份与目标 已读取。", confidence="high", ) ], coverage={"coverage_score": 1.0, "covered_facets": ["project_identity"], "partial_facets": [], "missing_facets": []}, ) class _FakeOpenResearchProjectWorldStateService: async def build_packet(self, **kwargs): return PMFlowAIProjectWorldStatePacket( primary_project_ref="", project_refs=[], status="partial", facets=[], evidence_ledger=[ PMFlowAIEvidenceLedgerEntry( evidence_id="evd_open_research_seed", project_ref="space:research", facet_key="research_scope", source_ref="user_question", summary="用户提出开放市场研究问题。", confidence="medium", ) ], coverage={ "coverage_score": 0.1, "covered_facets": [], "partial_facets": [], "missing_facets": ["external_research_evidence"], }, remaining_gaps=["external_public_sources_need_agent_reading"], ) class _FakeShallowIdeaMissionService: async def build_mission(self, **kwargs): request = kwargs["request"] return PMFlowAIOperatingMission( mission_goal="把用户的新想法整理成可继续推进的 IdeaSeed。", stage="idea", desired_outcome="形成初步创意判断，并判断是否具备商业潜力与 MVP 推进价值。", desired_business_outcome="不要只停留在创意描述，需要支撑是否继续做、怎么做首版、成本风险和变现假设。", original_user_input=str(request.question or ""), project_scope={"scope_type": "new_idea", "object_refs": []}, deliverable_type="idea_seed", action_ladder_level="prepare_object", required_agents=["idea_incubation_agent"], success_criteria=["形成可推进的产品判断，而不是只给创意摘要。"], autonomy_level="prepare_changes", ) class _FakeDB: def __init__(self) -> None: self.items = [] self._next_id = 1 def add(self, item) -> None: self.items.append(item) async def flush(self) -> None: for item in self.items: if getattr(item, "id", None) is None: item.id = self._next_id self._next_id += 1 class _RecordingAgentOSToolRegistry: def __init__(self) -> None: self.calls: list[dict[str, object]] = [] async def execute(self, *, proposal, **kwargs): self.calls.append({"tool_name": proposal.tool_name, "tool_input": dict(proposal.tool_input or {})}) if proposal.tool_name == "inspect_project_operating_state": return PMFlowAIAgentObservation( source="inspect_project_operating_state", observation_type="project_world_state", summary="已读取项目世界状态。", facts=[{"type": "project_world_state", "project_refs": ["project:141"], "status": "ready"}], confidence="high", relevance_to_goal="high", ) return PMFlowAIAgentObservation( source="inspect_space_collections", observation_type="space_context", summary="已读取当前空间集合。", facts=[{"type": "surface_collection_count", "collection": "projects", "count": 1}], confidence="medium", relevance_to_goal="medium", ) class _FakeDeliveryRuntime: async def run_turn(self, **kwargs): return { "answer": "项目当前需要先收口核心验证，再准备可确认的下一步。", "reply_text": "项目当前需要先收口核心验证，再准备可确认的下一步。", "processing_stage": "agent_runtime", "answer_mode": "agent_runtime", "grounding_status": "grounded", "answer_canvas": { "action_candidates": [ { "title": "准备本周项目推进计划", "summary": "进入项目工作台准备本周推进计划。", "href": "/pmflow-os?projectId=141", "action_type": "suggest", "requires_user_confirm": False, } ], "next_actions": [], }, "deliverable_packet": { "summary": "项目当前需要先收口核心验证，再准备可确认的下一步。", "blocked_reasons": ["真实写入项目记忆前需要用户确认。"], }, "terminal_answer_state": { "claims": [ { "claim_text": "项目当前需要先收口核心验证。", "confidence": "medium", } ] }, "customer_interaction_context_packet": { "current_focus_object": { "object_ref": "project:141", "object_type": "project", "object_id": 141, "title": "童析—学龄前儿童行为特质AI评估助手", } }, "agent_run_packet": { "run_id": "arun_test", "observations": [{"observation_id": "obs_test"}], }, "answer_process_packet": { "packet_version": "customer_answer_process.v1", "steps": [{"key": "agent_answer_runtime", "label": "生成回答", "status": "completed"}], "total_duration_ms": 10, }, } class _FakeOpenResearchDeliveryRuntime: async def run_turn(self, **kwargs): answer = "这轮先进入研究补证，不能把候选来源包装成市场结论。" return { "answer": answer, "reply_text": answer, "processing_stage": "agent_runtime", "answer_mode": "agent_runtime", "grounding_status": "partial", "agent_os_observations": [ { "observation_id": "obs_external_search_1", "source_kind": "external_search", "facts": [ { "type": "external_evidence", "title": "AI events market candidate", "url": "https://example.com/ai-events", "summary": "公开来源候选显示 AI 会议活动和赞助需求需要继续交叉验证。", }, { "type": "external_evidence", "title": "AI event tools candidate", "url": "https://example.com/ai-event-tools", "summary": "公开来源候选显示 AI 会务工具和活动运营工具正在出现。", }, ], } ], "agent_workspace": { "claim_ledger": [ { "claim_text": "AI 会议市场判断需要同时覆盖活动需求、会务工具和商业化信号。", "confidence": "medium", } ], "gap_ledger": [ { "summary": "还缺少近期参会规模、赞助收入和工具竞品的交叉来源。", } ], }, "deliverable_packet": { "deliverable_kind": "research_note", "summary": answer, "sections": [ { "section_kind": "gaps", "items": [{"summary": "需要继续补三类来源后再形成机会结论。"}], } ], }, "answer_process_packet": { "packet_version": "customer_answer_process.v1", "steps": [{"key": "agent_answer_runtime", "label": "生成回答", "status": "completed"}], "total_duration_ms": 10, }, } class _FakeNestedResearchDeliveryRuntime: async def run_turn(self, **kwargs): answer = "内部 Agent OS 已读取公开来源，但还没形成足够市场结论。" observations = [ { "observation_id": "obs_nested_search_1", "source": "external_search", "observation_type": "external_public_evidence", "facts": [ { "type": "external_evidence", "title": "AI conference attendance candidate", "url": "https://example.com/ai-conference-attendance", "summary": "AI conference attendance and sponsor demand candidate.", }, { "type": "external_evidence", "title": "AI event tools candidate", "url": "https://example.com/ai-event-tools", "summary": "AI event technology and meeting software candidate.", }, ], } ] return { "answer": answer, "reply_text": answer, "processing_stage": "agent_runtime", "answer_mode": "agent_runtime", "grounding_status": "partial", "answer_canvas": {"direct_answer": "旧 inner answer 不应该覆盖 operating answer。"}, "agent_run_packet": { "run_id": "arun_nested", "observations": observations, }, "agent_workspace": { "claim_ledger": [ { "claim_text": "AI 会议市场判断需要先分开活动需求、会务工具和商业化信号。", "confidence": "medium", } ], "gap_ledger": [{"summary": "还缺少赞助、参会规模和工具竞品的交叉来源。"}], }, "deliverable_packet": { "deliverable_kind": "research_note", "sections": [{"section_kind": "gaps", "items": [{"summary": "需要继续补三类来源。"}]}], }, "answer_process_packet": { "packet_version": "customer_answer_process.v1", "steps": [{"key": "agent_answer_runtime", "label": "生成回答", "status": "completed"}], "total_duration_ms": 10, }, } class _FakeProductBlueprintDeliveryRuntime: async def run_turn(self, **kwargs): bad_answer = "现在还不能直接下市场结论。已整理 0 个候选来源和 0 条可用判断，接下来最值得补的是：准备IdeaSeed草案对象。" return { "answer": bad_answer, "reply_text": bad_answer, "processing_stage": "agent_runtime", "answer_mode": "agent_runtime", "grounding_status": "partial", "answer_canvas": { "direct_answer": bad_answer, "artifacts": [ { "artifact_id": "product_blueprint_ai_music", "artifact_kind": "product_blueprint", "title": "AI 音乐自动生成产品蓝图", "summary": "把 AI 音乐生成工具收敛为可验证的创作闭环、MVP 范围、技术路径和商业假设。", "slots": [ { "slot_key": "value_proposition", "title": "价值主张", "summary": "核心价值是把数小时的作曲、编曲、母带或分轨处理压缩到分钟级。", }, { "slot_key": "mvp_boundary", "title": "MVP 边界", "summary": "首版优先覆盖文本到音乐、局部修复、分轨和母带四个可验收能力。", }, { "slot_key": "data_ai_pipeline", "title": "技术路径", "summary": "需要评估音频生成模型、分轨模型、版权过滤和推理成本。", }, { "slot_key": "risk_compliance", "title": "风险合规", "summary": "需要把版权、训练素材、生成内容归属和商用授权作为首轮门禁。", }, { "slot_key": "revenue_model", "title": "变现假设", "summary": "先验证创作者订阅、按导出计费和模板包售卖三类收入路径。", }, ], } ], "action_candidates": [ { "title": "生成 AI 音乐工具项目草案", "action_type": "create_project_draft", "requires_user_confirm": True, } ], }, "agent_os_observations": [ { "observation_id": "obs_ai_music_search", "source_kind": "external_search", "facts": [ { "type": "external_evidence", "title": "AI music generation market source", "url": "https://example.com/ai-music-generation", "summary": "公开来源显示 AI 音乐生成工具正在用于缩短创作者从旋律构思到可发布音频的制作周期。", }, { "type": "external_evidence", "title": "AI audio stem separation source", "url": "https://example.com/ai-audio-stems", "summary": "公开来源显示分轨、母带和音频修复能力正在成为 AI 音频工具的常见工作流。", }, ], } ], "terminal_answer_state": { "terminal_status": "partially_answerable", "claims": [ { "claim_text": "AI 音乐工具的核心用户价值在于大幅缩短创作时间，并降低母带、分轨和音频修复的专业门槛。", "confidence": "medium", }, { "claim_text": "首版 MVP 不应直接追求完整音乐工作站，而应先验证文本生成音乐、局部修复、分轨和母带处理的闭环。", "confidence": "medium", }, { "claim_text": "商业化判断必须同时看推理成本、版权合规和创作者付费意愿，不能只凭技术可行性拍板。", "confidence": "medium", }, ], }, "deliverable_packet": { "deliverable_kind": "product_blueprint", "summary": "AI 音乐自动生成工具可形成产品蓝图，但需要保留证据边界。", "sections": [ { "section_kind": "claims", "items": [ {"summary": "AI 音乐工具有明确的创作效率价值。"}, {"summary": "MVP 应围绕可验收的音频生成与处理闭环。"}, ], } ], }, "answer_process_packet": { "packet_version": "customer_answer_process.v1", "steps": [{"key": "agent_answer_runtime", "label": "生成回答", "status": "completed"}], "total_duration_ms": 10, }, "agent_run_packet": { "run_id": "arun_ai_music", "observations": [], }, } class _RecordingEventSinkDeliveryRuntime(_FakeDeliveryRuntime): def __init__(self) -> None: self.saw_event_sink = False async def run_turn(self, **kwargs): event_sink = kwargs.get("event_sink") self.saw_event_sink = callable(event_sink) if callable(event_sink): result = event_sink({"type": "step.completed", "step": {"key": "agent_answer_runtime", "label": "生成回答", "status": "completed"}}) if inspect.isawaitable(result): await result return await super().run_turn(**kwargs) class _FakeDirectDeliveryRuntime: async def run_turn(self, **kwargs): answer = "我是 PMFlow AI，可以帮你查看空间、项目和下一步，但不会替你直接写入或执行。" return { "answer": answer, "reply_text": answer, "processing_stage": "agent_runtime", "answer_mode": "agent_runtime", "grounding_status": "grounded", "answer_canvas": {"direct_answer": answer, "next_actions": []}, "deliverable_packet": { "deliverable_kind": "direct_answer", "summary": answer, "next_actions": [], }, "customer_evidence_packet": {"answer_type": "direct_answer"}, "terminal_answer_state": { "terminal_status": "answerable", "claims": [{"claim_text": answer, "confidence": "medium"}], }, "agent_run_packet": { "run_id": "arun_direct", "steps": [{"key": "agent_answer_runtime", "label": "生成回答", "status": "completed"}], "observations": [], }, "answer_process_packet": { "packet_version": "customer_answer_process.v1", "steps": [{"key": "agent_answer_runtime", "label": "生成回答", "status": "completed"}], "total_duration_ms": 10, }, }