PMFlow/backend/app/services/customer_answer_presentation_service.py

from __future__ import annotations

import json
import re
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from ..models import User
from ..schemas.customer_answer import (
    CUSTOMER_EVIDENCE_PACKET_VERSION,
    CustomerActiveWorkspaceSummary,
    CustomerAnswerProcessPacket,
    CustomerAnswerProcessStep,
    CustomerAnswerVerification,
    CustomerEvidenceFact,
    CustomerEvidencePacket,
    CustomerInteractionContextPacket,
    CustomerVisibleObject,
    PMFlowAIAnswerCanvas,
    PMFlowAIAnswerCanvasEvidenceBlock,
    PMFlowAIAnswerCanvasNextAction,
    PMFlowAIAnswerCanvasObjectRef,
)
from ..schemas.context_workspace import normalize_research_workspace_payload
from ..schemas.rag import RAGAskRequest
from .ai_function_runtime_profile_service import AIFunctionRuntimeProfileService
from .llm_service import LLMService
from .prompt_runtime_service import PromptRuntimeService


PROMPT_CUSTOMER_ANSWER_COMPOSER_KEY = "PROMPT_CUSTOMER_ANSWER_COMPOSER"
PROMPT_CUSTOMER_ANSWER_VERIFIER_KEY = "PROMPT_CUSTOMER_ANSWER_VERIFIER"
PROMPT_SPACE_AI_INTERACTION_OPTION_PLANNER_KEY = "PROMPT_SPACE_AI_INTERACTION_OPTION_PLANNER"
CUSTOMER_ANSWER_COMPOSER_MAX_TOKENS = 10000
CUSTOMER_ANSWER_VERIFIER_MAX_TOKENS = 10000
PMFLOW_AI_ANSWER_PRESENTATION_SCENE_KEY = "pmflow_ai_answer_presentation"

CUSTOMER_ANSWER_PACKET_VERSION = CUSTOMER_EVIDENCE_PACKET_VERSION
_RUNTIME_ANSWER_SHAPES = {
    "strategic_judgement",
    "status_answer",
    "clarification",
    "execution_guidance",
    "fact_list",
    "composite_answer",
    "general_answer",
}
_PROTECTED_ANSWER_TYPE = "protected_scope"

_INTERNAL_TERM_PATTERNS = (
    r"System Snapshot",
    r"User Snapshot",
    r"Session State",
    r"Project database",
    r"Knowledge Graph",
    r"Feishu history",
    r"AICallLog",
    r"SemanticDecision",
    r"AdaptiveRAG",
    r"PROMPT_[A-Z0-9_]+",
    r"\bprojects\b",
    r"\bteam_members\b",
    r"\bassigned_projects\b",
    r"\btenant_id\b",
    r"\broute_decision\b",
    r"\bsource_strategy\b",
    r"\bprompt_key\b",
    r"\btrace_id\b",
    r"backend/",
    r"frontend/",
    r"app/services",
)
_INTERNAL_TERM_RE = re.compile("|".join(f"(?:{pattern})" for pattern in _INTERNAL_TERM_PATTERNS), re.IGNORECASE)
_INTERNAL_EVIDENCE_TONE_RE = re.compile(r"(字段|数组|快照|上下文明确|一致指向|出现在用户上下文|系统快照)", re.IGNORECASE)
_INTERNAL_ANCHOR_RE = re.compile(
    r"("
    r"\b(?:approval|change_set|task|evidence|project):\d+\b|"
    r"\b(?:projectId|changeSetId|taskId|evidenceId|approvalId)=\d+\b|"
    r"/(?:change-review|pmflow-os|execution-bridge|release-growth|project-launch)\?[^)\s，。；;]+"
    r")",
    re.IGNORECASE,
)
_PROJECT_LINE_RE = re.compile(  # semantic-lint: ignore
    r"(?:\*\*)?(?P<name>[^*\n（(]{2,120})(?:\*\*)?[（(]\s*(?:ID|项目编号)\s*[:：]\s*(?P<id>\d+)\s*[，,]\s*状态\s*[:：]\s*(?P<status>[^）)]+)[）)]"
)


class CustomerAnswerPresentationService:
    """Build customer-safe answer surfaces from internal/external evidence."""

    def __init__(
        self,
        db: AsyncSession,
        *,
        tenant_id: int,
        llm_service: LLMService | None = None,
    ) -> None:
        self.db = db
        self.tenant_id = int(tenant_id or 1)
        self._llm_service_supplied = llm_service is not None
        self.llm = llm_service or LLMService()

    async def present(
        self,
        *,
        request: RAGAskRequest,
        current_user: User,
        payload: dict[str, Any],
    ) -> dict[str, Any]:
        working = dict(payload or {})
        existing_packet = working.get("customer_evidence_packet") if isinstance(working.get("customer_evidence_packet"), dict) else None
        preserve_packet = bool(
            existing_packet
            and (
                working.get("preserve_customer_evidence_packet")
                or working.get("pmflow_ai_turn_state")
            )
        )
        packet = existing_packet if preserve_packet else self.build_customer_evidence_packet(
            request=request,
            current_user=current_user,
            payload=working,
        )
        answer, verification, composer_used = await self.compose_answer_from_packet(
            question=str(request.question or "").strip(),
            packet=packet,
        )

        working["answer"] = answer
        working["reply_text"] = answer
        working["customer_evidence_packet"] = packet
        working["customer_answer_verification"] = verification
        working["customer_interaction_context_packet"] = self.build_customer_interaction_context_packet(
            packet,
            research_workspace=working.get("research_workspace"),
        )
        working["answer_process_packet"] = self.build_answer_process_packet(packet=packet, verification=verification)
        working["answer_canvas"] = self.build_answer_canvas(
            packet=packet,
            answer=answer,
            research_workspace=working.get("research_workspace"),
        )
        working["answer_mode"] = "customer_composed" if composer_used else "customer_deterministic"
        working["policy_version"] = _append_policy_version(
            working.get("policy_version"),
            "customer-evidence-packet-v1",
        )
        return working

    def build_customer_evidence_packet(
        self,
        *,
        request: RAGAskRequest,
        current_user: User,
        payload: dict[str, Any],
    ) -> dict[str, Any]:
        route_plan = _normalize_dict(payload.get("pmflow_ai_route_plan"))
        research_contract = _normalize_dict(payload.get("research_contract"))
        external_fit = _normalize_dict(payload.get("external_project_fit_packet"))
        sources = _customer_source_summaries(payload)
        facts = self._build_facts(request=request, payload=payload, route_plan=route_plan)
        answer_type = self._resolve_answer_type(
            request=request,
            payload=payload,
            route_plan=route_plan,
            facts=facts,
        )
        fallback_summary = _strip_internal_answer_text(payload.get("reply_text") or payload.get("answer"))
        limitations = _build_limitations(payload)
        if external_fit:
            facts.append(_external_project_fit_fact(external_fit))
        return _packet_to_dict(
            CustomerEvidencePacket(
                tenant_id=_safe_int(getattr(current_user, "tenant_id", None)),
                user_id=_safe_int(getattr(current_user, "id", None)),
                question=_clean_text(request.question, limit=500),
                answer_type=answer_type,
                route_display=_route_display(route_plan, research_contract),
                facts=[_normalize_fact_model(item) for item in facts[:20]],
                source_summary=sources,
                confidence=_confidence_label(payload),
                limitations=limitations,
                suggested_actions=_suggested_actions_from_surface(request.surface_context),
                fallback_summary=fallback_summary,
            )
        )

    def build_customer_interaction_context_packet(
        self,
        packet: dict[str, Any] | CustomerEvidencePacket,
        *,
        research_workspace: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        packet_dict = _packet_to_dict(packet)
        workspace = normalize_research_workspace_payload(research_workspace)
        objects = _visible_objects_from_packet(packet_dict, research_workspace=workspace)
        current_focus = objects[0] if len(objects) == 1 else None
        if current_focus is None:
            current_focus = _workspace_primary_focus(workspace)
        return CustomerInteractionContextPacket(
            current_focus_object=current_focus,
            visible_objects=objects,
            last_shown_objects=objects,
            active_workspace=_active_workspace_summary(workspace),
        ).model_dump(mode="json", exclude_none=True)

    def build_answer_process_packet(
        self,
        *,
        packet: dict[str, Any] | CustomerEvidencePacket,
        verification: dict[str, Any] | CustomerAnswerVerification | None = None,
    ) -> dict[str, Any]:
        raw_packet = dict(packet) if isinstance(packet, dict) else {}
        packet_dict = _packet_to_dict(packet)
        answer_type = _canonical_runtime_answer_type(packet_dict.get("answer_type"))
        route_display = _clean_text(packet_dict.get("route_display"), limit=120) or None
        _ = verification
        steps = _explicit_process_steps_from_packet(raw_packet) or _explicit_process_steps_from_packet(packet_dict)
        return CustomerAnswerProcessPacket(
            answer_type=_canonical_runtime_answer_type(answer_type),
            route_display=route_display,
            steps=steps[:8],
        ).model_dump(mode="json", exclude_none=True)

    def build_answer_canvas(
        self,
        *,
        packet: dict[str, Any] | CustomerEvidencePacket,
        answer: str,
        research_workspace: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        packet_dict = _packet_to_dict(packet)
        answer_type = _canonical_runtime_answer_type(packet_dict.get("answer_type"))
        facts = [item for item in list(packet_dict.get("facts") or []) if isinstance(item, dict)]
        canvas_facts = _canvas_relevant_facts(answer_type=answer_type, facts=facts)
        interaction = self.build_customer_interaction_context_packet(
            {**packet_dict, "facts": canvas_facts},
            research_workspace=research_workspace,
        )
        visible_objects = [
            item
            for item in list(interaction.get("visible_objects") or [])
            if isinstance(item, dict) and str(item.get("object_ref") or "").strip()
        ]
        suggested_actions = [
            item for item in list(packet_dict.get("suggested_actions") or []) if isinstance(item, dict)
        ]
        suggested_actions = _canvas_relevant_actions(answer_type=answer_type, actions=suggested_actions)
        evidence_blocks: list[PMFlowAIAnswerCanvasEvidenceBlock] = []
        for fact in canvas_facts[:8]:
            label = _clean_text(fact.get("display_label") or "依据", limit=160)
            summary = _clean_text(fact.get("display_summary"), limit=1000)
            if not label or not summary:
                continue
            evidence_blocks.append(
                PMFlowAIAnswerCanvasEvidenceBlock(
                    label=label,
                    summary=summary,
                    source_type=_canvas_source_type(fact),
                    confidence=_confidence_value(fact.get("confidence")),
                )
            )
        object_refs: list[PMFlowAIAnswerCanvasObjectRef] = []
        for obj in visible_objects[:8]:
            object_ref = _clean_text(obj.get("object_ref"), limit=120)
            title = _clean_text(obj.get("title"), limit=160)
            object_type = _clean_text(obj.get("object_type") or obj.get("type"), limit=64)
            if not object_ref or not title or not object_type:
                continue
            object_refs.append(
                PMFlowAIAnswerCanvasObjectRef(
                    object_ref=object_ref,
                    title=title,
                    type=object_type,
                    href=_clean_text(obj.get("href"), limit=240) or None,
                    status=_clean_text(obj.get("status"), limit=80) or None,
                    summary=_clean_text(obj.get("display_summary") or obj.get("summary"), limit=500) or None,
                )
            )
        next_actions: list[PMFlowAIAnswerCanvasNextAction] = []
        for action in suggested_actions[:5]:
            title = _clean_text(action.get("title"), limit=160)
            if not title:
                continue
            href = _clean_text(action.get("href"), limit=240) or None
            next_actions.append(
                PMFlowAIAnswerCanvasNextAction(
                    title=title,
                    reason=_clean_text(action.get("project") or action.get("priority"), limit=500),
                    action_type=_canvas_action_type_from_href(href),
                    href=href,
                    object_ref=_clean_text(action.get("object_ref"), limit=120) or None,
                    object_anchor=_clean_object_anchor(action.get("object_anchor") or action.get("objectAnchor")),
                    requires_user_confirm=True,
                )
            )
        if not next_actions and object_refs:
            first = object_refs[0]
            if first.href:
                next_actions.append(
                    PMFlowAIAnswerCanvasNextAction(
                        title=f"打开{first.title}",
                        reason="继续查看这个对象的主流程状态和可处理事项。",
                        action_type="navigate",
                        href=first.href,
                        object_ref=first.object_ref,
                        object_anchor={
                            "object_ref": first.object_ref,
                            "object_type": first.type,
                            "object_id": str(first.object_id) if getattr(first, "object_id", None) is not None else "",
                            "surface": "",
                            "project_id": str(first.object_id) if getattr(first, "object_id", None) is not None else "",
                            "project_title": first.title,
                            "href": first.href,
                            "parent_object_ref": None,
                        },
                        requires_user_confirm=True,
                    )
                )
        direct_answer = _standardize_canvas_direct_answer(answer) or _standardize_canvas_direct_answer(
            packet_dict.get("fallback_summary")
        )
        limitations = [
            _clean_text(item, limit=160)
            for item in list(packet_dict.get("limitations") or [])
            if _clean_text(item, limit=160)
        ][:6]
        canvas = PMFlowAIAnswerCanvas(
            headline=_answer_canvas_headline(answer_type=answer_type, answer=direct_answer),
            direct_answer=direct_answer,
            evidence_blocks=evidence_blocks[:8],
            object_refs=object_refs[:8],
            next_actions=next_actions[:5],
            limitations=limitations,
        )
        return canvas.model_dump(mode="json", exclude_none=True)

    def build_customer_evidence_packet_from_runtime(
        self,
        *,
        question: str,
        user_id: int | None = None,
        surface_context: dict[str, Any] | None = None,
        payload: dict[str, Any] | None = None,
        sources: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        """Build a customer-safe packet before the model composes the answer."""

        working = dict(payload or {})
        if sources is not None:
            working["sources"] = list(sources)
            working["sources_used"] = list(sources)
        request = RAGAskRequest(
            question=str(question or "").strip() or "PMFlow 问答",
            surface_context=dict(surface_context or {}),
        )
        facts = self._build_facts(
            request=request,
            payload=working,
            route_plan=_normalize_dict(working.get("pmflow_ai_route_plan")),
        )
        answer_type = self._resolve_answer_type(
            request=request,
            payload=working,
            route_plan=_normalize_dict(working.get("pmflow_ai_route_plan")),
            facts=facts,
        )
        if str(working.get("grounding_status") or "").strip().lower() in {
            "insufficient_evidence",
            "blocked",
            "needs_clarification",
        }:
            answer_type = "insufficient_evidence"
        return _packet_to_dict(
            CustomerEvidencePacket(
                tenant_id=self.tenant_id,
                user_id=_safe_int(user_id),
                question=_clean_text(question, limit=500),
                answer_type=answer_type,
                route_display=_route_display(
                    _normalize_dict(working.get("pmflow_ai_route_plan")),
                    _normalize_dict(working.get("research_contract")),
                ),
                facts=[_normalize_fact_model(item) for item in facts[:20]],
                source_summary=_customer_source_summaries(working),
                confidence=_confidence_label(working),
                limitations=_build_limitations(working),
                suggested_actions=_suggested_actions_from_surface(surface_context),
                fallback_summary=_strip_internal_answer_text(working.get("answer") or working.get("reply_text")),
            )
        )

    async def compose_answer_from_packet(
        self,
        *,
        question: str,
        packet: dict[str, Any] | CustomerEvidencePacket,
    ) -> tuple[str, dict[str, Any], bool]:
        packet_dict = _packet_to_dict(packet)
        answer = self._render_deterministic_answer(packet_dict)
        composer_used = False
        if not answer and self._should_use_llm_composer(packet_dict) and self._can_use_llm_composer():
            answer = await self._compose_answer(question=question, packet=packet_dict)
            composer_used = bool(answer)
        if not answer:
            answer = self._render_safe_fallback(packet_dict)
        verification = await self._verify_answer(packet=packet_dict, answer=answer, use_llm=composer_used)
        if not verification.get("ok"):
            repaired = _repair_customer_answer_text(answer)
            repaired_verification = self._verify_answer_deterministic(repaired)
            if repaired_verification.get("ok"):
                answer = repaired
                verification = {
                    **repaired_verification,
                    "origin": "deterministic_soft_rewrite",
                    "rewrite_required": False,
                    "reason": "soft_display_repaired",
                }
            else:
                answer = self._render_safe_fallback(packet_dict)
                verification = self._verify_answer_deterministic(answer)
        return answer, verification, composer_used

    def _build_facts(
        self,
        *,
        request: RAGAskRequest,
        payload: dict[str, Any],
        route_plan: dict[str, Any],
    ) -> list[dict[str, Any]]:
        facts: list[dict[str, Any]] = []
        quarantine_surface = _should_quarantine_surface_context(
            request=request,
            payload=payload,
            route_plan=route_plan,
        )
        if not quarantine_surface:
            facts.extend(
                _merge_project_facts(
                    _project_facts_from_surface(request.surface_context),
                    _project_facts_from_answer(payload.get("answer") or payload.get("reply_text")),
                )
            )
            facts.extend(_saved_idea_facts_from_surface(request.surface_context))
            facts.extend(_space_status_facts_from_surface(request.surface_context))
            facts.extend(_ai_readiness_facts_from_surface(request.surface_context))
            facts.extend(_release_blocker_facts_from_surface(request.surface_context))
        facts.extend(_evidence_findings_from_payload(payload))
        facts.extend(_external_facts_from_payload(payload))
        facts.extend(_retrieved_summary_facts_from_payload(payload))
        return _dedupe_facts(facts)

    def _resolve_answer_type(
        self,
        *,
        request: RAGAskRequest,
        payload: dict[str, Any],
        route_plan: dict[str, Any],
        facts: list[dict[str, Any]],
    ) -> str:
        customer_packet = _normalize_dict(payload.get("customer_evidence_packet"))
        turn_shape = _answer_shape_from_runtime_payload(payload)
        if turn_shape:
            return turn_shape
        packet_answer_type = _canonical_runtime_answer_type(customer_packet.get("answer_type"))
        if packet_answer_type:
            return packet_answer_type
        _ = request, route_plan, facts
        return "general_answer"

    def _render_deterministic_answer(self, packet: dict[str, Any]) -> str:
        answer_type = _canonical_runtime_answer_type(packet.get("answer_type"))
        if answer_type in _RUNTIME_ANSWER_SHAPES:
            return _render_state_shape_answer(packet)
        if answer_type == _PROTECTED_ANSWER_TYPE:
            return "这个问题不在用户侧 PMFlow AI 的可回答范围内。你可以问项目状态、下一步、空间事项或带来源的外部公开信息。"
        return ""

    def _should_use_llm_composer(self, packet: dict[str, Any]) -> bool:
        answer_type = _canonical_runtime_answer_type(packet.get("answer_type"))
        if answer_type in _RUNTIME_ANSWER_SHAPES or answer_type == _PROTECTED_ANSWER_TYPE:
            return False
        return bool(packet.get("facts") or packet.get("fallback_summary"))

    def _can_use_llm_composer(self) -> bool:
        if not hasattr(self.llm, "generate"):
            return False
        if self._llm_service_supplied:
            return True
        return hasattr(self.db, "execute")

    async def _compose_answer(self, *, question: str, packet: dict[str, Any]) -> str:
        prompt_key = PROMPT_CUSTOMER_ANSWER_COMPOSER_KEY
        function_name = "customer_answer_composer"
        try:
            prompt_result = await PromptRuntimeService(self.db).render(
                tenant_id=self.tenant_id,
                prompt_key=prompt_key,
                question=question[:1000],
                customer_evidence_packet=json.dumps(_packet_for_model_prompt(packet), ensure_ascii=False)[:7000],
            )
            prompt = prompt_result.prompt
            routing_meta = prompt_result.routing_meta()
            routing_meta["scene_key"] = "pmflow_ai_answer_presentation"
            routing_meta["scene_contract"] = "customer_answer_presentation"
        except Exception:
            return ""
        try:
            runtime_profile = await self._resolve_runtime_profile(
                function_name=function_name,
                prompt_key=prompt_key,
                fallback_max_tokens=CUSTOMER_ANSWER_COMPOSER_MAX_TOKENS,
                fallback_temperature=0.2,
            )
            routing_meta.update(runtime_profile["routing_meta"])
            text = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=self.tenant_id,
                function_name=function_name,
                temperature=runtime_profile["temperature"],
                max_tokens=runtime_profile["max_tokens"],
                use_prompt_profile=False,
                routing_meta=routing_meta,
            )
        except Exception:
            return ""
        return _strip_internal_answer_text(text)

    async def compose_interaction_request(
        self,
        *,
        question: str,
        packet: dict[str, Any],
        answer: str,
        fallback_interaction_request: dict[str, Any],
    ) -> dict[str, Any]:
        """Generate customer-specific option candidates for a follow-up input request."""

        fallback = dict(fallback_interaction_request or {})
        if not fallback or not self._can_use_llm_composer():
            return fallback
        try:
            prompt_result = await PromptRuntimeService(self.db).render(
                tenant_id=self.tenant_id,
                prompt_key=PROMPT_SPACE_AI_INTERACTION_OPTION_PLANNER_KEY,
                question=str(question or "")[:1000],
                customer_evidence_packet=json.dumps(_packet_for_model_prompt(packet), ensure_ascii=False)[:7000],
                answer=str(answer or "")[:3000],
                fallback_interaction_request=json.dumps(fallback, ensure_ascii=False)[:4000],
            )
            prompt = prompt_result.prompt
            routing_meta = prompt_result.routing_meta()
            routing_meta["scene_key"] = "pmflow_ai_answer_presentation"
            routing_meta["scene_contract"] = "space_ai_interaction_option_planner"
        except Exception:
            return fallback
        try:
            runtime_profile = await self._resolve_runtime_profile(
                function_name="space_ai_interaction_option_planner",
                prompt_key=PROMPT_SPACE_AI_INTERACTION_OPTION_PLANNER_KEY,
                fallback_max_tokens=4000,
                fallback_temperature=0.1,
            )
            routing_meta.update(runtime_profile["routing_meta"])
            raw = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=self.tenant_id,
                function_name="space_ai_interaction_option_planner",
                temperature=runtime_profile["temperature"],
                max_tokens=runtime_profile["max_tokens"],
                use_prompt_profile=False,
                routing_meta=routing_meta,
            )
            parsed = _parse_json_object(raw)
        except Exception:
            return fallback
        planned = _normalize_interaction_request_from_planner(parsed, fallback=fallback)
        return planned or fallback

    async def _verify_answer(self, *, packet: dict[str, Any], answer: str, use_llm: bool) -> dict[str, Any]:
        deterministic = self._verify_answer_deterministic(answer)
        relevance = _verify_answer_relevance(packet=packet, answer=answer)
        if deterministic.get("ok") and not relevance.get("ok"):
            return relevance
        if not deterministic.get("ok") or not use_llm:
            return deterministic
        try:
            prompt_result = await PromptRuntimeService(self.db).render(
                tenant_id=self.tenant_id,
                prompt_key=PROMPT_CUSTOMER_ANSWER_VERIFIER_KEY,
                customer_evidence_packet=json.dumps(_packet_for_model_prompt(packet), ensure_ascii=False)[:6000],
                answer=str(answer or "")[:2000],
            )
            prompt = prompt_result.prompt
            routing_meta = prompt_result.routing_meta()
            routing_meta["scene_key"] = "pmflow_ai_answer_presentation"
            routing_meta["scene_contract"] = "customer_answer_presentation"
        except Exception:
            return deterministic
        try:
            runtime_profile = await self._resolve_runtime_profile(
                function_name="customer_answer_verifier",
                prompt_key=PROMPT_CUSTOMER_ANSWER_VERIFIER_KEY,
                fallback_max_tokens=CUSTOMER_ANSWER_VERIFIER_MAX_TOKENS,
                fallback_temperature=0.0,
            )
            routing_meta.update(runtime_profile["routing_meta"])
            raw = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=self.tenant_id,
                function_name="customer_answer_verifier",
                temperature=runtime_profile["temperature"],
                max_tokens=runtime_profile["max_tokens"],
                use_prompt_profile=False,
                routing_meta=routing_meta,
            )
            parsed = _parse_json_object(raw)
        except Exception:
            return deterministic
        parsed_ok = bool(parsed.get("ok")) and not bool(parsed.get("leak_found")) and not bool(parsed.get("rewrite_required"))
        if not parsed_ok and deterministic.get("ok") and not _has_hard_internal_anchor(answer) and not _needs_soft_display_repair(answer):
            return {
                "ok": True,
                "origin": "llm_verifier_soft_allowed",
                "leak_found": False,
                "untranslated_terms": [],
                "unanswered": False,
                "overclaim": False,
                "rewrite_required": False,
                "reason": "llm_verifier_flagged_allowed_product_terms",
            }
        return {
            "ok": parsed_ok,
            "origin": "llm_verifier",
            "leak_found": bool(parsed.get("leak_found")),
            "untranslated_terms": [
                _clean_text(item, limit=80)
                for item in list(parsed.get("untranslated_terms") or [])[:8]
                if _clean_text(item, limit=80)
            ],
            "unanswered": bool(parsed.get("unanswered")),
            "overclaim": bool(parsed.get("overclaim")),
            "rewrite_required": bool(parsed.get("rewrite_required")) or not parsed_ok,
            "reason": _clean_text(parsed.get("reason"), limit=200) or None,
        }

    def _verify_answer_deterministic(self, answer: str) -> dict[str, Any]:
        return _verify_answer_deterministic(answer)

    async def _resolve_runtime_profile(
        self,
        *,
        function_name: str,
        prompt_key: str,
        fallback_max_tokens: int,
        fallback_temperature: float,
    ) -> dict[str, Any]:
        try:
            profile = await AIFunctionRuntimeProfileService(self.db).resolve_profile(
                function_name=function_name,
                scene_key=PMFLOW_AI_ANSWER_PRESENTATION_SCENE_KEY,
                prompt_key=prompt_key,
                stage="active",
            )
            return {
                "max_tokens": profile.max_tokens,
                "temperature": profile.temperature,
                "routing_meta": profile.routing_meta(),
            }
        except Exception:
            return {
                "max_tokens": fallback_max_tokens,
                "temperature": fallback_temperature,
                "routing_meta": {
                    "function_runtime_profile_source": "code_fallback",
                    "function_runtime_max_tokens": fallback_max_tokens,
                    "scene_key": PMFLOW_AI_ANSWER_PRESENTATION_SCENE_KEY,
                },
            }

    def _render_safe_fallback(self, packet: dict[str, Any]) -> str:
        answer = self._render_deterministic_answer(packet)
        if answer:
            return answer
        fallback = _clean_text(packet.get("fallback_summary"), limit=1200)
        if fallback and _verify_answer_deterministic(fallback).get("ok"):
            return fallback
        limitations = list(packet.get("limitations") or [])
        if limitations:
            return "我现在没有拿到足够明确的可展示证据来回答这个问题。你可以补充项目名、对象或范围后我再查。"
        return "我现在没有生成足够明确的可展示回答。你可以换一种问法，或补充项目/对象范围后我再查。"


def _standardize_canvas_direct_answer(value: Any) -> str:
    text = _strip_canvas_source_tail(_strip_internal_answer_text(value))
    if not text:
        return ""
    return _strip_generated_canvas_sections(text)[:2000]


def _strip_canvas_source_tail(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    text = re.sub(r"(?:\n|\s)*(?:来源摘要|来源|参考来源)[:：].*$", "", text, flags=re.IGNORECASE | re.DOTALL)
    text = re.sub(r"\s*URL\[[^\]]+\]\s*[;；]?", "", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


def _packet_to_dict(value: dict[str, Any] | CustomerEvidencePacket) -> dict[str, Any]:
    if isinstance(value, CustomerEvidencePacket):
        return value.to_customer_prompt_json()
    if isinstance(value, dict):
        try:
            return CustomerEvidencePacket.model_validate(value).to_customer_prompt_json()
        except Exception:
            return dict(value)
    return CustomerEvidencePacket().to_customer_prompt_json()


def _packet_for_model_prompt(packet: dict[str, Any]) -> dict[str, Any]:
    """Strip navigation anchors before customer prose composition/verification."""

    safe = _packet_to_dict(packet)
    actions = []
    for item in list(safe.get("suggested_actions") or []):
        if not isinstance(item, dict):
            continue
        actions.append(
            {
                "title": _display_action_title(item.get("title")),
                "project": _clean_text(item.get("project"), limit=120),
                "priority": _clean_text(item.get("priority"), limit=60),
            }
        )
    safe["suggested_actions"] = [item for item in actions if item.get("title")][:8]
    return safe


def _normalize_fact_model(value: Any) -> CustomerEvidenceFact:
    if isinstance(value, CustomerEvidenceFact):
        return value
    if isinstance(value, dict):
        return CustomerEvidenceFact.model_validate(value)
    return CustomerEvidenceFact(type="general", display_label="事实", display_summary=_clean_text(value, limit=400))


def _fact_type(value: dict[str, Any]) -> str:
    return str(value.get("type") or value.get("fact_type") or "").strip().lower()


def _fact_dict(
    *,
    fact_type: str,
    label: str,
    summary: str,
    values: dict[str, Any] | None = None,
    source_labels: list[str] | None = None,
    confidence: str = "unknown",
    action_hint: str | None = None,
) -> dict[str, Any]:
    return CustomerEvidenceFact(
        type=str(fact_type or "general").strip() or "general",
        display_label=_clean_text(label, limit=160),
        display_summary=_clean_text(summary, limit=1200),
        business_values=dict(values or {}),
        source_display_labels=list(source_labels or []),
        confidence=confidence if confidence in {"high", "medium", "low", "unknown"} else "unknown",
        action_hint=_clean_text(action_hint, limit=240) or None,
    ).model_dump(mode="json", exclude_none=True)


def _external_project_fit_fact(value: dict[str, Any]) -> dict[str, Any]:
    summary = _clean_text(
        value.get("display_summary")
        or value.get("summary")
        or value.get("fit_summary")
        or value.get("recommendation"),
        limit=800,
    )
    action = _clean_text(value.get("action") or value.get("next_action") or value.get("suggestion"), limit=240)
    return _fact_dict(
        fact_type="external_project_fit",
        label="外部信息与当前项目匹配判断",
        summary=summary or action or "外部信息需要结合当前项目继续判断。",
        values={"匹配判断": summary, "建议动作": action},
        source_labels=["外部公开来源", "当前项目背景"],
        confidence="medium" if summary else "low",
        action_hint=action or None,
    )


def _canvas_relevant_facts(*, answer_type: str, facts: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Keep answer-canvas evidence tied to the current answer instead of the whole page surface."""

    normalized = str(answer_type or "").strip()
    type_map: dict[str, set[str]] = {
        "strategic_judgement": {"runtime_evidence_finding", "external_summary", "retrieved_context"},
        "status_answer": {"runtime_evidence_finding", "project", "space_status", "retrieved_context"},
        "clarification": {"runtime_evidence_finding", "space_status"},
        "execution_guidance": {"runtime_evidence_finding", "project", "space_status", "retrieved_context"},
        "fact_list": {"runtime_evidence_finding", "project", "saved_idea", "space_status", "retrieved_context"},
        "composite_answer": {"runtime_evidence_finding", "project", "space_status", "retrieved_context"},
        "general_answer": {"runtime_evidence_finding", "external_summary", "retrieved_context", "project", "space_status"},
    }
    allowed_types = type_map.get(normalized)
    if not allowed_types:
        return list(facts)
    return [fact for fact in facts if _fact_type(fact) in allowed_types]


def _canvas_relevant_actions(*, answer_type: str, actions: list[dict[str, Any]]) -> list[dict[str, Any]]:
    _ = answer_type
    return actions


def _route_display(route_plan: dict[str, Any], research_contract: dict[str, Any]) -> str | None:
    strategy = str(route_plan.get("source_strategy") or research_contract.get("source_strategy") or "").strip().lower()
    lane = str(research_contract.get("effective_lane") or "").strip().lower()
    if strategy.startswith("hybrid") or lane == "hybrid":
        return "结合当前项目与外部公开信息"
    if "external" in strategy or lane == "external":
        return "外部公开信息"
    if "internal" in strategy or lane == "internal":
        return "当前空间和项目数据"
    if strategy == "clarify":
        return "需要补充范围"
    return None


def _visible_objects_from_packet(
    packet: dict[str, Any],
    *,
    research_workspace: dict[str, Any] | None = None,
) -> list[CustomerVisibleObject]:
    objects: list[CustomerVisibleObject] = []
    seen: set[str] = set()
    for fact in list(packet.get("facts") or []):
        if not isinstance(fact, dict) or _fact_type(fact) != "project":
            continue
        values = _normalize_dict(fact.get("business_values"))
        project_id = _safe_int(values.get("项目编号"))
        title = _clean_text(values.get("项目名称") or fact.get("display_label"), limit=160)
        if not title:
            continue
        object_ref = f"project:{project_id}" if project_id is not None else f"project:title:{title}"
        if object_ref in seen:
            continue
        seen.add(object_ref)
        objects.append(
            CustomerVisibleObject(
                object_type="project",
                object_ref=object_ref,
                object_id=project_id,
                title=title,
                status=_clean_text(values.get("状态"), limit=80) or None,
                href=_clean_text(values.get("入口"), limit=240) or None,
                display_summary=_clean_text(fact.get("display_summary"), limit=800),
            )
        )
    for fact in list(packet.get("facts") or []):
        if not isinstance(fact, dict) or _fact_type(fact) != "saved_idea":
            continue
        values = _normalize_dict(fact.get("business_values"))
        idea_id = _clean_text(values.get("想法编号"), limit=120)
        title = _clean_text(values.get("想法标题") or fact.get("display_label"), limit=160)
        if not title:
            continue
        object_ref = f"saved_idea:{idea_id}" if idea_id else f"saved_idea:title:{title}"
        if object_ref in seen:
            continue
        seen.add(object_ref)
        objects.append(
            CustomerVisibleObject(
                object_type="saved_idea",
                object_ref=object_ref,
                title=title,
                status=_clean_text(values.get("状态"), limit=80) or None,
                href=_clean_text(values.get("入口"), limit=240) or None,
                display_summary=_clean_text(fact.get("display_summary"), limit=800),
            )
        )
    workspace = normalize_research_workspace_payload(research_workspace)
    for item in list(workspace.get("candidate_items") or [])[:6]:
        if not isinstance(item, dict):
            continue
        candidate_ref = _clean_text(item.get("candidate_ref"), limit=120)
        title = _clean_text(item.get("display_name"), limit=160)
        if not candidate_ref or not title or candidate_ref in seen:
            continue
        seen.add(candidate_ref)
        objects.append(
            CustomerVisibleObject(
                object_type="research_candidate",
                object_ref=candidate_ref,
                title=title,
                status=_clean_text(item.get("status"), limit=80) or None,
                display_summary=_clean_text(item.get("summary"), limit=800),
            )
        )
    return objects[:12]


def _workspace_primary_focus(workspace: dict[str, Any]) -> CustomerVisibleObject | None:
    normalized = normalize_research_workspace_payload(workspace)
    primary_ref = _clean_text(normalized.get("primary_object_ref"), limit=120)
    primary_title = _clean_text(normalized.get("primary_object_title"), limit=160)
    if not primary_ref or not primary_title:
        return None
    return CustomerVisibleObject(
        object_type="project" if primary_ref.startswith("project:") else "workspace_object",
        object_ref=primary_ref,
        object_id=_safe_int(normalized.get("primary_object_id")),
        title=primary_title,
        display_summary=_clean_text(normalized.get("research_goal"), limit=400),
    )


def _active_workspace_summary(workspace: dict[str, Any]) -> CustomerActiveWorkspaceSummary | None:
    normalized = normalize_research_workspace_payload(workspace)
    if not normalized:
        return None
    return CustomerActiveWorkspaceSummary(
        workspace_kind=_clean_text(normalized.get("workspace_kind"), limit=64) or "external_research",
        topic=_clean_text(normalized.get("topic"), limit=120),
        primary_object_ref=_clean_text(normalized.get("primary_object_ref"), limit=120) or None,
        primary_object_title=_clean_text(normalized.get("primary_object_title"), limit=160) or None,
        candidate_set_binding=_clean_text(normalized.get("candidate_set_binding"), limit=120) or None,
        candidate_count=len(list(normalized.get("candidate_items") or [])),
        status=_clean_text(normalized.get("status"), limit=32) or None,
    )


def _explicit_process_steps_from_packet(packet: dict[str, Any]) -> list[CustomerAnswerProcessStep]:
    raw_steps = packet.get("process_steps") or packet.get("runtime_steps")
    if not isinstance(raw_steps, list):
        return []
    steps: list[CustomerAnswerProcessStep] = []
    for index, raw in enumerate(raw_steps[:8], start=1):
        if not isinstance(raw, dict):
            continue
        label = _clean_text(raw.get("label") or raw.get("display_summary"), limit=160)
        if not label:
            continue
        steps.append(
            CustomerAnswerProcessStep(
                key=_clean_text(raw.get("key"), limit=80) or f"runtime_step_{index}",
                label=label,
                status=_clean_text(raw.get("status"), limit=24) or "completed",
                display_summary=_clean_text(raw.get("display_summary"), limit=240) or None,
                duration_ms=_safe_int(raw.get("duration_ms")),
                blocking=bool(raw.get("blocking", False)),
            )
        )
    return steps


def _customer_anchor_sources_from_packet(packet: dict[str, Any]) -> list[dict[str, Any]]:
    anchors = []
    for obj in _visible_objects_from_packet(packet):
        if obj.object_type != "project" or obj.object_id is None:
            continue
        anchors.append(
            {
                "source": "pmflow",
                "label": "项目数据",
                "score": 1.0,
                "metadata": {
                    "project_id": int(obj.object_id),
                    "query_keyword": obj.title,
                    "subject_name": obj.title,
                    "object_ref": obj.object_ref,
                    "source_type": "customer_visible_object",
                },
            }
        )
    return anchors


def _answer_shape_from_runtime_payload(payload: dict[str, Any]) -> str:
    turn_state = _normalize_dict(payload.get("pmflow_ai_turn_state"))
    answer_state = _normalize_dict(turn_state.get("answer_state"))
    shape = _clean_text(answer_state.get("answer_shape"), limit=80)
    if shape in _RUNTIME_ANSWER_SHAPES:
        return shape
    display_contract = _normalize_dict(
        payload.get("display_contract")
        or _normalize_dict(payload.get("pmflow_ai_route_plan")).get("display_contract")
    )
    shape = _clean_text(display_contract.get("answer_shape"), limit=80)
    if shape in _RUNTIME_ANSWER_SHAPES:
        return shape
    return ""


def _canonical_runtime_answer_type(value: Any) -> str:
    text = _clean_text(value, limit=80)
    if text in _RUNTIME_ANSWER_SHAPES or text == _PROTECTED_ANSWER_TYPE:
        return text
    return "general_answer" if text else "general_answer"


def _render_state_shape_answer(packet: dict[str, Any]) -> str:
    fallback = _clean_text(packet.get("fallback_summary"), limit=1400)
    if fallback:
        return _strip_generated_canvas_sections(fallback)
    facts = [fact for fact in list(packet.get("facts") or []) if isinstance(fact, dict)]
    summaries: list[str] = []
    for fact in facts[:6]:
        summary = _clean_text(fact.get("display_summary") or fact.get("display_label"), limit=300)
        if summary and summary not in summaries:
            summaries.append(summary)
    if not summaries:
        return ""
    return _strip_generated_canvas_sections("；".join(summaries[:4]) + "。")


def _strip_generated_canvas_sections(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    rows = [row.rstrip() for row in text.splitlines()]
    cleaned: list[str] = []
    skip_next_blank = False
    for row in rows:
        stripped = row.strip()
        if re.fullmatch(r"\*\*[^*\n]{1,16}\*\*", stripped) or re.fullmatch(r"[^，。；.!?！？\n:：]{1,16}[:：]", stripped):
            skip_next_blank = True
            continue
        inline_heading = re.match(r"^([^，。；.!?！？\n:：]{1,16})[:：]\s*(.+)$", stripped)
        if inline_heading:
            row = inline_heading.group(2).strip()
            stripped = row
        if skip_next_blank and not stripped:
            continue
        skip_next_blank = False
        cleaned.append(row)
    result = "\n".join(cleaned).strip()
    result = re.sub(r"\n{3,}", "\n\n", result)
    return result


def _display_action_title(value: Any) -> str:
    text = _clean_text(value, limit=160)
    if not text:
        return ""
    text = re.sub(r"\bapproval:\d+\b", "待确认事项", text, flags=re.IGNORECASE)
    text = re.sub(r"\bchange_set:\d+\b", "审查退回事项", text, flags=re.IGNORECASE)
    text = re.sub(r"\btask:\d+\b", "任务事项", text, flags=re.IGNORECASE)
    text = re.sub(r"(确认变更|变更)\s*\d+\b", "确认一项变更", text)
    text = re.sub(r"修改后再审\s*\d+\b", "复审一项变更", text)
    text = re.sub(r"暂缓处理\s*\d+\b", "处理一项暂缓事项", text)
    return _strip_internal_anchors(text)


def _project_facts_from_surface(surface_context: Any) -> list[dict[str, Any]]:
    surface = _normalize_dict(surface_context)
    workbench = _normalize_dict(surface.get("workbench"))
    projects = []
    project_rows = list(
        workbench.get("recent_projects")
        or workbench.get("recentProjects")
        or workbench.get("projects")
        or []
    )
    for item in project_rows[:12]:
        if not isinstance(item, dict):
            continue
        name = _clean_text(item.get("title") or item.get("name"), limit=120)
        if not name:
            continue
        status = _display_status(item.get("status"))
        next_action = _clean_text(item.get("next_action") or item.get("nextAction"), limit=160)
        project_id = _safe_int(item.get("id") or item.get("project_id") or item.get("projectId"))
        projects.append(
            _fact_dict(
                fact_type="project",
                label=name,
                summary=f"{name}，状态：{status}" + (f"，下一步：{next_action}" if next_action else ""),
                values={
                    "项目名称": name,
                    "项目编号": project_id,
                    "状态": status,
                    "阶段": _clean_text(item.get("stage"), limit=80),
                    "进度": _safe_int(item.get("progress")),
                    "待确认": _safe_int(item.get("pendingCount") or item.get("pending_count")),
                    "进行中任务": _safe_int(item.get("activeTaskCount") or item.get("active_task_count")),
                    "风险": _clean_text(item.get("riskLabel") or item.get("risk_label"), limit=80),
                    "下一步": next_action,
                    "入口": _clean_text(item.get("href"), limit=200),
                },
                source_labels=["当前空间状态"],
                confidence="high",
                action_hint=next_action or None,
            )
        )
    return projects


def _should_quarantine_surface_context(
    *,
    request: RAGAskRequest,
    payload: dict[str, Any],
    route_plan: dict[str, Any],
) -> bool:
    if bool(payload.get("preserve_customer_evidence_packet")):
        return True
    turn_state = _normalize_dict(payload.get("pmflow_ai_turn_state"))
    answer_state = _normalize_dict(turn_state.get("answer_state"))
    object_state = _normalize_dict(turn_state.get("object_state"))
    primary_object = _normalize_dict(object_state.get("primary_object"))
    shape = str(answer_state.get("answer_shape") or "").strip().lower()
    object_type = str(primary_object.get("object_type") or primary_object.get("kind") or "").strip().lower()
    if shape == "strategic_judgement" and object_type in {"new_idea", "external_subject", "unknown"}:
        return True
    _ = request, route_plan
    return False


def _saved_idea_facts_from_surface(surface_context: Any) -> list[dict[str, Any]]:
    surface = _normalize_dict(surface_context)
    workbench = _normalize_dict(surface.get("workbench"))
    ideas: list[dict[str, Any]] = []
    for item in list(workbench.get("saved_ideas") or workbench.get("savedIdeas") or [])[:8]:
        if not isinstance(item, dict):
            continue
        title = _clean_text(item.get("title") or item.get("name"), limit=120)
        if not title:
            continue
        summary = _clean_text(item.get("summary") or item.get("ai_summary") or item.get("raw_text"), limit=500)
        status = _display_saved_idea_status(item.get("status"))
        idea_id = _clean_text(item.get("id") or item.get("idea_id") or item.get("ideaId"), limit=120)
        ideas.append(
            _fact_dict(
                fact_type="saved_idea",
                label=title,
                summary=f"{title}，状态：{status}" + (f"，摘要：{summary}" if summary else ""),
                values={
                    "想法编号": idea_id,
                    "想法标题": title,
                    "状态": status,
                    "摘要": summary,
                    "准备度": item.get("readiness_score") or item.get("readinessScore"),
                    "入口": _clean_text(item.get("href"), limit=200),
                },
                source_labels=["当前空间状态"],
                confidence="high",
                action_hint="继续完善或生成项目草案",
            )
        )
    return ideas


def _project_facts_from_answer(value: Any) -> list[dict[str, Any]]:
    text = str(value or "")
    facts: list[dict[str, Any]] = []
    for match in _PROJECT_LINE_RE.finditer(text):
        name = _clean_text(match.group("name"), limit=120).strip("-：: ")
        if not name:
            continue
        status = _display_status(match.group("status"))
        facts.append(
            _fact_dict(
                fact_type="project",
                label=name,
                summary=f"{name}，状态：{status}",
                values={"项目名称": name, "项目编号": _safe_int(match.group("id")), "状态": status},
                source_labels=["项目数据"],
                confidence="medium",
            )
        )
    return facts


def _space_status_facts_from_surface(surface_context: Any) -> list[dict[str, Any]]:
    surface = _normalize_dict(surface_context)
    workbench = _normalize_dict(surface.get("workbench"))
    counts = _normalize_dict(workbench.get("counts"))
    if not counts:
        return []
    label_map = {
        "nextActions": "下一步",
        "pendingConfirmations": "待确认",
        "aiProgress": "AI 正在推进",
        "projects": "项目",
        "savedIdeas": "暂存想法",
        "drafts": "草案",
        "releaseAttention": "发布关注",
        "feedbackAttention": "反馈关注",
    }
    display_counts = {
        label_map.get(str(key), str(key)): _safe_int(value)
        for key, value in counts.items()
        if label_map.get(str(key), str(key))
    }
    summary = "，".join(f"{key} {value} 个" for key, value in display_counts.items() if value is not None)
    return [
        _fact_dict(
            fact_type="space_status",
            label="当前空间状态",
            summary=summary or "当前空间已有状态数据",
            values=display_counts,
            source_labels=["当前空间状态"],
            confidence="high",
        )
    ]


def _ai_readiness_facts_from_surface(surface_context: Any) -> list[dict[str, Any]]:
    surface = _normalize_dict(surface_context)
    readiness = _normalize_dict(surface.get("ai_readiness"))
    operating_packet = _normalize_dict(surface.get("space_operating_packet"))
    if not readiness:
        readiness = _normalize_dict(operating_packet.get("readiness"))
    if not readiness:
        return []
    status = _clean_text(readiness.get("status") or readiness.get("title"), limit=120) or "状态待确认"
    blockers = []
    for item in list(readiness.get("blockers") or [])[:6]:
        if isinstance(item, dict):
            text = _clean_text(item.get("description") or item.get("label"), limit=160)
        else:
            text = _clean_text(item, limit=160)
        if text:
            blockers.append(text)
    return [
        _fact_dict(
            fact_type="ai_readiness",
            label="AI 能力状态",
            summary=status,
            values={
                "状态": status,
                "可生成项目草案": bool(readiness.get("can_generate_project_draft")),
                "可管理": bool(readiness.get("can_manage")),
                "影响说明": blockers,
            },
            source_labels=["当前空间状态"],
            confidence="high",
        )
    ]


def _release_blocker_facts_from_surface(surface_context: Any) -> list[dict[str, Any]]:
    surface = _normalize_dict(surface_context)
    workbench = _normalize_dict(surface.get("workbench"))
    rows = list(workbench.get("release_attention") or workbench.get("releaseAttention") or [])
    facts: list[dict[str, Any]] = []
    for item in rows[:8]:
        if not isinstance(item, dict):
            continue
        title = _clean_text(item.get("title") or item.get("next_action"), limit=160)
        project = _clean_text(item.get("project") or item.get("projectTitle"), limit=120)
        summary = _clean_text(item.get("summary") or item.get("reason"), limit=500)
        reasons = [
            _clean_text(reason, limit=140)
            for reason in list(item.get("reasons") or [])
            if _clean_text(reason, limit=140)
        ]
        severity = _clean_text(item.get("severity"), limit=40)
        gate_status = _clean_text(item.get("gate_status") or item.get("gateStatus"), limit=60)
        is_blocking = gate_status == "blocked" or severity in {"high", "critical"} or "阻塞" in title
        if not title or not is_blocking:
            continue
        facts.append(
            _fact_dict(
                fact_type="release_blocker",
                label=title,
                summary=summary or "发布门禁或验证状态需要处理。",
                values={
                    "项目名称": project,
                    "阻塞事项": title,
                    "原因": reasons,
                    "严重程度": severity,
                    "门禁状态": gate_status,
                    "入口": _clean_text(item.get("href"), limit=200),
                },
                source_labels=["当前空间状态", "发布验证记录"],
                confidence="high" if reasons or summary else "medium",
                action_hint=_clean_text(item.get("next_action") or "处理发布阻塞", limit=160),
            )
        )
    return facts


def _external_facts_from_payload(payload: dict[str, Any]) -> list[dict[str, Any]]:
    facts: list[dict[str, Any]] = []
    recommendation = _normalize_dict(payload.get("external_recommendation"))
    if recommendation:
        summary = _clean_text(recommendation.get("external_summary"), limit=600)
        impact = _clean_text(recommendation.get("external_impact"), limit=600)
        action = _clean_text(recommendation.get("external_action"), limit=240)
        facts.append(
            _fact_dict(
                fact_type="external_summary",
                label="外部公开信息摘要",
                summary=summary or impact or "外部公开信息暂未形成明确摘要",
                values={"外部摘要": summary, "对当前项目的影响": impact, "建议动作": action},
                source_labels=["外部公开来源"],
                confidence="medium" if summary or impact else "low",
                action_hint=action or None,
            )
        )
    return facts


def _evidence_findings_from_payload(payload: dict[str, Any]) -> list[dict[str, Any]]:
    rows = payload.get("evidence_findings")
    if not isinstance(rows, list):
        turn_state = _normalize_dict(payload.get("pmflow_ai_turn_state"))
        evidence_state = _normalize_dict(turn_state.get("evidence_state"))
        rows = evidence_state.get("known_facts") if isinstance(evidence_state.get("known_facts"), list) else []
    facts: list[dict[str, Any]] = []
    for item in list(rows or [])[:10]:
        if not isinstance(item, dict):
            continue
        fact = _clean_text(item.get("fact"), limit=700)
        implication = _clean_text(item.get("implication"), limit=700)
        if not fact:
            continue
        source = _clean_text(item.get("source") or "PMFlow 运行时证据", limit=120)
        relevance = _clean_text(item.get("relevance_to_goal"), limit=120)
        facts.append(
            _fact_dict(
                fact_type="runtime_evidence_finding",
                label=source,
                summary=implication or fact,
                values={
                    "事实": fact,
                    "含义": implication,
                    "目标相关性": relevance,
                    "来源": source,
                },
                source_labels=[source],
                confidence=_confidence_value(item.get("confidence")),
                action_hint=implication or None,
            )
        )
    return facts


def _retrieved_summary_facts_from_payload(payload: dict[str, Any]) -> list[dict[str, Any]]:
    rows = list(payload.get("customer_retrieved_summaries") or [])
    facts: list[dict[str, Any]] = []
    for item in rows[:8]:
        if not isinstance(item, dict):
            continue
        summary = _clean_text(item.get("display_summary") or item.get("content"), limit=700)
        if not summary:
            continue
        source_label = _clean_text(item.get("display_label") or "本轮证据", limit=80)
        facts.append(
            _fact_dict(
                fact_type=_clean_text(item.get("type") or "retrieved_context", limit=60),
                label=source_label,
                summary=summary,
                values={"内容摘要": summary},
                source_labels=[source_label],
                confidence="medium",
            )
        )
    return facts


def _dedupe_facts(facts: list[dict[str, Any]]) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    seen: set[str] = set()
    for fact in facts:
        values = _normalize_dict(fact.get("business_values"))
        key = f"{_fact_type(fact)}:{values.get('项目编号')}:{fact.get('display_label')}:{fact.get('display_summary')}"
        if key in seen:
            continue
        seen.add(key)
        out.append(fact)
    return out


def _merge_project_facts(primary: list[dict[str, Any]], secondary: list[dict[str, Any]]) -> list[dict[str, Any]]:
    if not primary:
        return secondary
    merged = [dict(item) for item in primary]
    for source in secondary:
        source_values = _normalize_dict(source.get("business_values"))
        source_name = _clean_text(source_values.get("项目名称") or source.get("display_label"), limit=120)
        source_id = _safe_int(source_values.get("项目编号"))
        matched = None
        for target in merged:
            target_values = _normalize_dict(target.get("business_values"))
            target_name = _clean_text(target_values.get("项目名称") or target.get("display_label"), limit=120)
            target_id = _safe_int(target_values.get("项目编号"))
            if source_id is not None and target_id == source_id:
                matched = target
                break
            if source_name and target_name and source_name == target_name:
                matched = target
                break
        if matched is None:
            merged.append(dict(source))
            continue
        matched_values = _normalize_dict(matched.get("business_values"))
        for key, value in source_values.items():
            if matched_values.get(key) in (None, "", [], {}) and value not in (None, "", [], {}):
                matched_values[key] = value
        matched["business_values"] = matched_values
        if not matched.get("action_hint") and source.get("action_hint"):
            matched["action_hint"] = source.get("action_hint")
    return merged


def _customer_source_summaries(payload: dict[str, Any]) -> list[str]:
    rows = []
    for key in ("sources_used", "citations", "sources", "evidence"):
        raw = payload.get(key)
        if isinstance(raw, list):
            rows.extend(item for item in raw if isinstance(item, dict))
    labels: list[str] = []
    for item in rows:
        label = _customer_source_label(item)
        if label and label not in labels:
            labels.append(label)
    return labels[:6]


def _customer_source_label(item: dict[str, Any]) -> str:
    raw = str(item.get("display_label") or item.get("label") or item.get("raw_source_label") or "").strip()
    source = str(item.get("source") or item.get("source_type") or "").strip().lower()
    lower = raw.lower()
    if source in {"url", "web", "external", "mcp_search"} or lower.startswith(("http", "url[")):
        return "外部公开来源"
    if "project database" in lower or source in {"db", "pmflow"}:
        return "项目数据"
    if "knowledge graph" in lower or source == "graph":
        return "项目记忆"
    if "feishu" in lower or source == "feishu":
        return "飞书记录"
    if "snapshot" in lower or source == "system":
        return "当前空间状态"
    if raw and not _INTERNAL_TERM_RE.search(raw):
        return raw
    return ""


def _build_limitations(payload: dict[str, Any]) -> list[str]:
    fields = []
    for item in list(payload.get("missing_fields") or [])[:5]:
        text = _clean_text(item, limit=80)
        if text:
            fields.append(text)
    grounding = str(payload.get("grounding_status") or "").strip().lower()
    if grounding in {"insufficient_evidence", "blocked", "needs_clarification"}:
        fields.append("当前证据不足")
    return fields[:6]


def _suggested_actions_from_surface(surface_context: Any) -> list[dict[str, Any]]:
    surface = _normalize_dict(surface_context)
    workbench = _normalize_dict(surface.get("workbench"))
    operating_packet = _normalize_dict(surface.get("space_operating_packet"))
    actions: list[dict[str, Any]] = []
    for item in list(
        workbench.get("attention_queue")
        or workbench.get("attentionQueue")
        or operating_packet.get("attention_queue")
        or []
    )[:5]:
        if not isinstance(item, dict):
            continue
        title = _clean_text(item.get("next_action") or item.get("title"), limit=120)
        if not title:
            continue
        action = {
            "title": title,
            "project": _clean_text(
                    (_normalize_dict(item.get("object_anchor") or item.get("objectAnchor")).get("project_title"))
                    or item.get("project")
                    or item.get("projectTitle")
                    or item.get("reason"),
                limit=120,
            ),
            "priority": _clean_text(item.get("urgency"), limit=40),
            "href": _clean_text(item.get("href"), limit=200),
            "object_ref": _clean_text(
                item.get("object_ref")
                or item.get("objectRef")
                or (_normalize_dict(item.get("object_anchor") or item.get("objectAnchor")).get("object_ref")),
                limit=120,
            ),
            "object_anchor": _clean_object_anchor(item.get("object_anchor") or item.get("objectAnchor")),
            "action_type": _clean_text(item.get("action_type") or item.get("actionType"), limit=80),
            "object_type": _clean_text(item.get("object_type") or item.get("objectType"), limit=80),
            "surface": _clean_text(item.get("surface"), limit=80),
        }
        actions.append(_normalize_surface_action(action))
    for item in list(workbench.get("top_next_actions") or workbench.get("topNextActions") or [])[:5]:
        if not isinstance(item, dict):
            continue
        title = _clean_text(item.get("title"), limit=120)
        if not title:
            continue
        action = {
            "title": title,
            "project": _clean_text(item.get("project"), limit=80),
            "priority": _clean_text(item.get("priority"), limit=40),
            "href": _clean_text(item.get("href"), limit=200),
            "object_ref": _clean_text(
                item.get("object_ref")
                or item.get("objectRef")
                or (_normalize_dict(item.get("object_anchor") or item.get("objectAnchor")).get("object_ref")),
                limit=120,
            ),
            "object_anchor": _clean_object_anchor(item.get("object_anchor") or item.get("objectAnchor")),
            "action_type": _clean_text(item.get("action_type") or item.get("actionType"), limit=80),
            "object_type": _clean_text(item.get("object_type") or item.get("objectType"), limit=80),
            "surface": _clean_text(item.get("surface"), limit=80),
        }
        actions.append(_normalize_surface_action(action))
    deduped: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for action in actions:
        key = (str(action.get("title") or ""), str(action.get("href") or ""))
        if key in seen:
            continue
        seen.add(key)
        deduped.append(action)
    return deduped[:5]


def _confidence_label(payload: dict[str, Any]) -> str:
    try:
        score = float(payload.get("confidence") or 0)
    except Exception:
        score = 0
    grounding = str(payload.get("grounding_status") or "").strip().lower()
    if grounding in {"insufficient_evidence", "blocked", "needs_clarification"}:
        return "low"
    if score >= 0.75:
        return "high"
    if score >= 0.4:
        return "medium"
    return "unknown"


def _confidence_value(value: Any) -> str:
    normalized = str(value or "").strip().lower()
    return normalized if normalized in {"high", "medium", "low", "unknown"} else "unknown"


def _normalize_surface_action(action: dict[str, Any]) -> dict[str, Any]:
    out = dict(action or {})
    anchor = _normalize_dict(out.get("object_anchor") or out.get("objectAnchor"))
    if not _clean_text(out.get("surface"), limit=80):
        out["surface"] = _clean_text(anchor.get("surface"), limit=80)
    if not _clean_text(out.get("object_type") or out.get("objectType"), limit=80):
        out["object_type"] = _clean_text(anchor.get("object_type") or anchor.get("objectType"), limit=80)
    if not _clean_text(out.get("action_type") or out.get("actionType"), limit=80):
        out["action_type"] = "navigate"
    return out


def _canvas_source_type(fact: dict[str, Any]) -> str:
    fact_type = _fact_type(fact)
    labels = " ".join(str(item or "") for item in list(fact.get("source_display_labels") or [])).lower()
    if fact_type.startswith("external") or "外部" in labels:
        return "external"
    if fact_type == "project" or "项目数据" in labels:
        return "project"
    if fact_type == "release_blocker":
        return "execution"
    if fact_type == "space_status" or "当前空间" in labels:
        return "space"
    if "项目记忆" in labels:
        return "memory"
    return "pmflow"


def _canvas_action_type_from_href(href: str | None) -> str:
    target = str(href or "").strip().lower()
    return "navigate" if target else "none"


def _answer_canvas_headline(*, answer_type: str, answer: str) -> str:
    answer_type = _canonical_runtime_answer_type(answer_type)
    mapping = {
        "strategic_judgement": "方向判断",
        "status_answer": "状态判断",
        "clarification": "需要补充范围",
        "execution_guidance": "执行建议",
        "fact_list": "事实依据",
        "composite_answer": "综合判断",
        "general_answer": "回答",
        "protected_scope": "用户侧不可回答范围",
    }
    headline = mapping.get(str(answer_type or "").strip(), "")
    if headline:
        return headline
    first_line = _clean_text(str(answer or "").splitlines()[0] if answer else "", limit=80)
    return first_line or "PMFlow AI 回答"


def _strip_internal_answer_text(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    lines: list[str] = []
    for raw in text.splitlines():
        line = raw.rstrip()
        compact = line.strip()
        if not compact:
            lines.append(line)
            continue
        compact = _strip_internal_anchors(compact)
        if _INTERNAL_TERM_RE.search(compact) or _INTERNAL_EVIDENCE_TONE_RE.search(compact):
            continue
        lines.append(_display_status_in_text(compact))
    cleaned = "\n".join(lines)
    cleaned = re.sub(r"\n{3,}", "\n\n", cleaned).strip()
    return cleaned


def _verify_answer_deterministic(answer: str) -> dict[str, Any]:
    text = str(answer or "").strip()
    leak_terms = sorted({match.group(0) for match in _INTERNAL_TERM_RE.finditer(text)})
    anchor_terms = sorted({match.group(0) for match in _INTERNAL_ANCHOR_RE.finditer(text)})
    tone_leak = bool(_INTERNAL_EVIDENCE_TONE_RE.search(text))
    incomplete = _is_incomplete_customer_answer(text)
    ok = bool(text) and not leak_terms and not anchor_terms and not tone_leak and not incomplete
    if leak_terms or anchor_terms or tone_leak:
        reason = "internal_term_or_anchor_or_evidence_tone"
    elif incomplete:
        reason = "incomplete_answer"
    elif text:
        reason = None
    else:
        reason = "empty_answer"
    return {
        "ok": ok,
        "origin": "deterministic_guard",
        "leak_found": bool(leak_terms or anchor_terms or tone_leak),
        "untranslated_terms": (leak_terms + anchor_terms)[:10],
        "unanswered": not bool(text),
        "overclaim": False,
        "rewrite_required": not ok,
        "reason": reason,
    }


_INCOMPLETE_ANSWER_TAIL_RE = re.compile(
    r"("
    r"[，、：；（(《“\"']$|"
    r"(?:我会|我将|会|将|把|并|和|与|以及|或者|然后|同时|如果|但|因为|所以|包括|例如|比如|作为|进入|生成|整理|提交|补充)$"
    r")"
)


def _is_incomplete_customer_answer(value: Any) -> bool:
    text = str(value or "").strip()
    if not text:
        return False
    if text.count("```") % 2 == 1:
        return True
    last_line = next((line.strip() for line in reversed(text.splitlines()) if line.strip()), "")
    if not last_line:
        return False
    if len(text) >= 20 and _INCOMPLETE_ANSWER_TAIL_RE.search(last_line):
        return True
    if len(text) < 80:
        return False
    if last_line.startswith(("- ", "* ", "1. ", "2. ", "3. ", "4. ")) and not re.search(r"[。！？.!?）)】]$", last_line):
        return True
    return False


def _verify_answer_relevance(*, packet: dict[str, Any], answer: str) -> dict[str, Any]:
    _ = answer
    packet_dict = packet if isinstance(packet, dict) else {}
    answer_type = str(packet_dict.get("answer_type") or "").strip().lower()
    object_binding = _normalize_dict(packet_dict.get("object_binding"))
    if answer_type == "strategic_judgement" and object_binding.get("bound_existing_project") is True:
        return {
            "ok": False,
            "origin": "deterministic_relevance_guard",
            "leak_found": False,
            "untranslated_terms": [],
            "unanswered": False,
            "overclaim": True,
            "rewrite_required": True,
            "reason": "strategic_judgement_bound_existing_project",
        }
    return {
        "ok": True,
        "origin": "deterministic_relevance_guard",
        "leak_found": False,
        "untranslated_terms": [],
        "unanswered": False,
        "overclaim": False,
        "rewrite_required": False,
        "reason": None,
    }


def _strip_internal_anchors(value: Any) -> str:
    text = str(value or "")
    if not text:
        return ""
    text = re.sub(r"\[([^\]]+)\]\(/[^)\s]+\)", r"\1", text)
    text = _INTERNAL_ANCHOR_RE.sub("", text)
    text = re.sub(r"\s+([，。；：,.!?])", r"\1", text)
    text = re.sub(r"\s{2,}", " ", text)
    return text.strip()


def _repair_customer_answer_text(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    text = re.sub(r"\[([^\]]+)\]\(/[^)\s]+\)", r"\1", text)
    text = re.sub(r"\bapproval:\d+\b", "待确认事项", text, flags=re.IGNORECASE)
    text = re.sub(r"\bchange_set:\d+\b", "审查退回事项", text, flags=re.IGNORECASE)
    text = re.sub(r"\btask:\d+\b", "任务事项", text, flags=re.IGNORECASE)
    text = re.sub(r"(确认变更|变更)\s*\d+\b", "确认一项变更", text)
    text = re.sub(r"修改后再审\s*\d+\b", "复审一项变更", text)
    text = re.sub(r"暂缓处理\s*\d+\b", "处理一项暂缓事项", text)
    text = _strip_internal_anchors(text)
    text = re.sub(r"(来源摘要|来源|参考来源)[:：]\s*(Knowledge Graph|Project database|System Snapshot|User Snapshot)", "", text, flags=re.IGNORECASE)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


def _has_hard_internal_anchor(value: Any) -> bool:
    return bool(_INTERNAL_ANCHOR_RE.search(str(value or "")))


def _needs_soft_display_repair(value: Any) -> bool:
    text = str(value or "")
    if not text:
        return False
    return bool(
        re.search(r"(确认变更|变更)\s*\d+\b", text)
        or re.search(r"修改后再审\s*\d+\b", text)
        or re.search(r"暂缓处理\s*\d+\b", text)
        or re.search(r"\bidea_[0-9a-f]{8,}\b", text, flags=re.IGNORECASE)
        or re.search(r"项目编号\s*[:：]?\s*\d+\b", text)
    )


def _display_status(value: Any) -> str:
    raw = str(value or "").strip().lower()
    mapping = {
        "active": "进行中",
        "in_progress": "进行中",
        "running": "进行中",
        "draft": "草稿",
        "archived": "已归档",
        "completed": "已完成",
        "done": "已完成",
        "blocked": "受阻",
    }
    return mapping.get(raw, _clean_text(value, limit=60) or "未知")


def _display_saved_idea_status(value: Any) -> str:
    raw = str(value or "").strip().lower()
    mapping = {
        "saved": "未整理",
        "enriched": "已整理",
        "needs_more_info": "信息不足",
        "draft_ready": "可生成草案",
        "promoted": "已转为草案",
        "archived": "已归档",
    }
    return mapping.get(raw, _clean_text(value, limit=60) or "未整理")


def _display_status_in_text(value: str) -> str:
    out = str(value or "")
    out = re.sub(r"状态[：:]\s*active", "状态：进行中", out, flags=re.IGNORECASE)
    out = re.sub(r"状态[：:]\s*draft", "状态：草稿", out, flags=re.IGNORECASE)
    out = re.sub(r"状态[：:]\s*archived", "状态：已归档", out, flags=re.IGNORECASE)
    out = re.sub(r"状态[：:]\s*completed", "状态：已完成", out, flags=re.IGNORECASE)
    return out


def _normalize_interaction_request_from_planner(value: Any, *, fallback: dict[str, Any]) -> dict[str, Any]:
    raw = dict(value) if isinstance(value, dict) else {}
    fallback = dict(fallback or {})
    fallback_questions = [
        item for item in list(fallback.get("questions") or []) if isinstance(item, dict)
    ]
    raw_questions = [
        item for item in list(raw.get("questions") or []) if isinstance(item, dict)
    ]
    questions: list[dict[str, Any]] = []
    for index, fallback_question in enumerate(fallback_questions):
        planned = raw_questions[index] if index < len(raw_questions) else {}
        merged = _normalize_interaction_question_from_planner(planned, fallback_question=fallback_question, index=index + 1)
        if merged:
            questions.append(merged)
    if not questions:
        for index, planned in enumerate(raw_questions[:6], start=1):
            merged = _normalize_interaction_question_from_planner(planned, fallback_question={}, index=index)
            if merged:
                questions.append(merged)
    if not questions:
        return {}
    supplement = raw.get("supplement") if isinstance(raw.get("supplement"), dict) else {}
    fallback_supplement = fallback.get("supplement") if isinstance(fallback.get("supplement"), dict) else {}
    return {
        "id": _clean_text(raw.get("id") or fallback.get("id"), limit=100) or "pmflow-ai-interaction-planned",
        "type": "confirm_and_supplement",
        "title": _clean_text(raw.get("title") or fallback.get("title"), limit=80) or "还要你确认的点",
        "mode": _clean_text(raw.get("mode") or fallback.get("mode"), limit=24) or "blocking",
        "presentation": "stepper",
        "questions": questions[:8],
        "supplement": {
            "enabled": True,
            "required": bool(supplement.get("required", fallback_supplement.get("required", False))),
            "placeholder": _clean_text(
                supplement.get("placeholder") or fallback_supplement.get("placeholder"),
                limit=180,
            )
            or "如果这些选项都不准确，直接补充你的真实对象、边界、限制或判断...",
        },
        "submit_label": _clean_text(raw.get("submit_label") or fallback.get("submit_label"), limit=40) or "确认并继续",
        "cancel_label": _clean_text(raw.get("cancel_label") or fallback.get("cancel_label"), limit=40) or "先不补充",
    }


def _normalize_interaction_question_from_planner(
    value: dict[str, Any],
    *,
    fallback_question: dict[str, Any],
    index: int,
) -> dict[str, Any]:
    raw = dict(value or {})
    fallback = dict(fallback_question or {})
    label = _clean_text(raw.get("label") or raw.get("question") or fallback.get("label") or fallback.get("question"), limit=180)
    if not label or _has_hard_internal_anchor(label):
        return {}
    options = _normalize_interaction_options_from_planner(raw.get("options"))
    if len(options) < 2:
        options = _normalize_interaction_options_from_planner(fallback.get("options"))
    if len(options) < 2:
        return {}
    option_ids = {item["id"] for item in options}
    if "custom_in_supplement" not in option_ids:
        options.append({"id": "custom_in_supplement", "label": "都不准确，我最后补充", "description": None})
    input_type = _clean_text(raw.get("input_type") or fallback.get("input_type"), limit=32) or "single_choice"
    if input_type not in {"single_choice", "multi_choice"}:
        input_type = "single_choice"
    return {
        "id": _clean_text(raw.get("id") or fallback.get("id"), limit=80) or f"question_{index}",
        "label": label,
        "input_type": input_type,
        "required": bool(raw.get("required", fallback.get("required", index == 1))),
        "placeholder": None,
        "options": options[:6],
    }


def _normalize_interaction_options_from_planner(value: Any) -> list[dict[str, Any]]:
    rows = list(value or []) if isinstance(value, list) else []
    out: list[dict[str, Any]] = []
    seen: set[str] = set()
    for index, item in enumerate(rows[:8], start=1):
        raw = dict(item) if isinstance(item, dict) else {"label": item}
        label = _clean_text(raw.get("label") or raw.get("title") or raw.get("name"), limit=120)
        if not label or _has_hard_internal_anchor(label):
            continue
        description = _clean_text(raw.get("description") or raw.get("summary") or raw.get("reason"), limit=180)
        if description and _has_hard_internal_anchor(description):
            description = ""
        option_id = _clean_text(raw.get("id") or raw.get("value") or raw.get("key"), limit=80)
        if not option_id:
            option_id = re.sub(r"[^a-zA-Z0-9_]+", "_", label).strip("_").lower()[:48] or f"option_{index}"
        if option_id in seen:
            continue
        seen.add(option_id)
        out.append({"id": option_id, "label": label, "description": description or None})
    return out


def _parse_json_object(value: Any) -> dict[str, Any]:
    text = str(value or "").strip()
    if not text:
        return {}
    try:
        parsed = json.loads(text)
        return parsed if isinstance(parsed, dict) else {}
    except Exception:
        pass
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if not match:
        return {}
    try:
        parsed = json.loads(match.group(0))
        return parsed if isinstance(parsed, dict) else {}
    except Exception:
        return {}


def _append_policy_version(value: Any, suffix: str) -> str:
    base = str(value or "customer-answer-policy-v1").strip() or "customer-answer-policy-v1"
    if suffix in base:
        return base
    return f"{base}|{suffix}"


def _normalize_dict(value: Any) -> dict[str, Any]:
    return dict(value) if isinstance(value, dict) else {}


def _safe_int(value: Any) -> int | None:
    try:
        if value is None or value == "":
            return None
        return int(value)
    except Exception:
        return None


def _safe_float(value: Any) -> float | None:
    try:
        if value is None or value == "":
            return None
        return float(value)
    except Exception:
        return None


def _clean_text(value: Any, *, limit: int) -> str:
    return " ".join(str(value or "").strip().split())[:limit]


def _clean_object_anchor(value: Any) -> dict[str, Any] | None:
    if not isinstance(value, dict):
        return None
    object_ref = _clean_text(value.get("object_ref") or value.get("objectRef"), limit=120)
    object_type = _clean_text(value.get("object_type") or value.get("objectType"), limit=60)
    object_id = _clean_text(value.get("object_id") or value.get("objectId"), limit=120)
    if not object_ref and object_type and object_id:
        object_ref = f"{object_type}:{object_id}"
    if not object_ref:
        return None
    return {
        "object_ref": object_ref,
        "object_type": object_type,
        "object_id": object_id,
        "surface": _clean_text(value.get("surface"), limit=80),
        "project_id": _clean_text(value.get("project_id") or value.get("projectId"), limit=80),
        "project_title": _clean_text(value.get("project_title") or value.get("projectTitle"), limit=160),
        "href": _clean_text(value.get("href"), limit=240),
        "parent_object_ref": _clean_text(value.get("parent_object_ref") or value.get("parentObjectRef"), limit=120) or None,
    }


__all__ = [
    "CustomerAnswerPresentationService",
    "CUSTOMER_ANSWER_PACKET_VERSION",
    "PROMPT_CUSTOMER_ANSWER_COMPOSER_KEY",
    "PROMPT_CUSTOMER_ANSWER_VERIFIER_KEY",
    "_customer_anchor_sources_from_packet",
]