PMFlow/backend/tests/test_pmflow_ai_agent_runtime_step1.py

from __future__ import annotations

import asyncio
import inspect
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 (
    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,
    _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,
    research_coverage_gap,
)
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
from app.services.pmflow_ai_operating_runtime.agent_registry import PMFlowAIOperatingAgentRegistry
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_direct_agent_runtime() -> None:
    runtime = PMFlowAgentRuntimeService(_FakeDB(), tenant_id=12)._dynamic_agent_runtime()

    assert isinstance(runtime, PMFlowAIAgentRunService)


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_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_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"] == [
        {"id": "art_2", "title": "推进建议", "summary": "先收口最高优先级动作。", "status_label": "已完成"}
    ]
    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_duplicate_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"] == [
        {
            "id": "art_meaningful",
            "title": "公开材料结论",
            "summary": "公开材料显示医保目录管理更强调监测、基金安全和供应保障。",
            "status_label": "已完成",
        }
    ]


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"] == [
        {"id": "artifact_report", "title": "审查说明", "summary": "当前最高优先级动作需要你本人确认。", "status_label": "已完成"}
    ]
    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 payload["answer"] == "项目当前需要先收口核心验证，再准备可确认的下一步。"
    assert payload["reply_text"] == payload["answer"]
    assert payload["answer_canvas"]["direct_answer"] == payload["answer"]
    assert payload["operating_summary"] != payload["answer"]
    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"
    assert [step["key"] for step in payload["answer_process_packet"]["steps"]] == ["agent_answer_runtime"]
    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["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


@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 len(recovery_payload["source_plan"]) == 3
    assert recovery_payload["queries"]
    assert "OpportunityBrief" not in {item["object_type"] for item in payload["product_object_projections"]}
    assert payload["answer"] == "这轮先进入研究补证，不能把候选来源包装成市场结论。"
    assert payload["answer_canvas"]["direct_answer"] == payload["answer"]
    assert "还不能升级成机会简报或执行动作" not in payload["answer"]
    assert "下一步只推进前置研究补证" not 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",
        "growth_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", "growth_agent", "monetization_agent"}.issubset(blocked)


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 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 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 "&emsp;" 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 &emsp; 联系我们 订阅邮箱",
            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 "&emsp;" 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 &emsp; 订阅邮箱",
                "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 &#x27;AI events&#x27; 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_critical_semantic_nodes_do_not_foreground_timeout_before_provider_budget() -> 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=8,
    ) == 30
    assert _effective_foreground_wait_seconds(
        stage="evidence_normalizer",
        provider_timeout_seconds=60,
        configured_foreground_wait_seconds=6,
    ) == 60
    assert _effective_foreground_wait_seconds(
        stage="answer_composer",
        provider_timeout_seconds=35,
        configured_foreground_wait_seconds=10,
    ) == 10


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_uses_llm_normalizer_even_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 False


@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_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会议市场怎么样",
        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_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 "
                            "&emsp;&emsp;正文片段 导航 市场地位 商业趋势 网站投稿"
                        ),
                        "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 &emsp;&emsp; "
                            "网站投稿 媒体合作 导航 首页 报告定制 行业白皮书"
                        ),
                    }
                ],
                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 _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 _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,
            },
        }